From 8e451fb55b1bfb68e30be576af2b72a2c8e7f26f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 6 Oct 2026 11:57:26 +0000 Subject: [PATCH 1/5] node:quic: report a datagram's status under the id kept in its lsquic frame `ondatagramstatus(id, status)` gave a status to the wrong id as soon as a packet was declared lost ahead of acknowledged ones. lsquic reported `(count, acked)` for a packet, and the session popped that many ids from a send-order queue. `on_dg_write` now names the frame with `lsquic_conn_set_datagram_id`. `write_datagram` stores the id in the DATAGRAM frame record. The send controller reports each frame as `(id, status)` when its packet is acknowledged, when its packet is declared lost, and when lsquic drops the frame before the send. The session reports a frame that was dropped before the send as 'lost'. The session keeps no queue of in-flight ids and no `sent` event for each datagram. --- packages/bun-usockets/src/node_quic_shim.c | 9 +- patches/lsquic/node-quic-accessors.patch | 261 ++++++---- src/lsquic_sys/lib.rs | 16 +- src/runtime/node/quic/session.rs | 171 +++--- test/js/node/quic/quic-datagram.test.ts | 577 +++++++++++++++++++++ 5 files changed, 845 insertions(+), 189 deletions(-) create mode 100644 test/js/node/quic/quic-datagram.test.ts diff --git a/packages/bun-usockets/src/node_quic_shim.c b/packages/bun-usockets/src/node_quic_shim.c index 95b40de6d9ce..a5076d0d1dea 100644 --- a/packages/bun-usockets/src/node_quic_shim.c +++ b/packages/bun-usockets/src/node_quic_shim.c @@ -25,7 +25,7 @@ struct us_nq_vtable { void (*on_stream_reset)(void *stream_ctx, int how, uint64_t error_code); ssize_t (*on_dg_write)(void *conn_ctx, void *buf, size_t buf_sz); void (*on_datagram)(void *conn_ctx, const void *buf, size_t sz); - void (*on_datagram_status)(void *conn_ctx, unsigned count, int acked); + void (*on_datagram_status)(void *conn_ctx, uint64_t id, int status); void (*on_early_data_failed)(void *conn_ctx); void (*on_path_switch)(void *conn_ctx, int validated, int is_preferred, const struct sockaddr *new_local, @@ -149,11 +149,14 @@ static void nq_on_datagram(lsquic_conn_t *c, const void *buf, size_t sz) { vt->on_datagram(ctx, buf, sz); } } -static void nq_on_datagram_status(lsquic_conn_t *c, unsigned count, int acked) { +/* lsquic_sys mirrors these values as LSQ_DG_*. */ +_Static_assert(LSQ_DG_ACKED == 0 && LSQ_DG_LOST == 1 && LSQ_DG_UNSENT == 2, + "lsquic_sys::LSQ_DG_* must match enum lsquic_dg_status"); +static void nq_on_datagram_status(lsquic_conn_t *c, uint64_t id, int status) { void *ctx = (void *) lsquic_conn_get_ctx(c); if (ctx) { struct us_nq_vtable *vt = *(struct us_nq_vtable **) ctx; - vt->on_datagram_status(ctx, count, acked); + vt->on_datagram_status(ctx, id, status); } } static void nq_on_early_data_failed(lsquic_conn_t *c) { diff --git a/patches/lsquic/node-quic-accessors.patch b/patches/lsquic/node-quic-accessors.patch index 8317ea477a95..80a7cda0ce0c 100644 --- a/patches/lsquic/node-quic-accessors.patch +++ b/patches/lsquic/node-quic-accessors.patch @@ -1,6 +1,6 @@ --- a/include/lsquic.h +++ b/include/lsquic.h -@@ -243,6 +243,51 @@ struct lsquic_stream_if { +@@ -243,6 +243,53 @@ struct lsquic_stream_if { * The user may pick it off immediately using lsquic_stream_get_hset(). */ void (*on_hset_in) (lsquic_stream_t *s, lsquic_stream_ctx_t *h); @@ -18,12 +18,14 @@ + const struct sockaddr *peer_sa, + uint64_t error_code); + /** -+ * Optional callback fired when a packet carrying DATAGRAM frames is -+ * acknowledged by the peer (`acked` is 1) or declared lost (`acked` is -+ * 0). `count` is the number of DATAGRAM frames the packet carried. -+ * bun addition. ++ * Optional callback fired once for each DATAGRAM frame when its packet ++ * is acknowledged, when its packet is declared lost, or when the frame ++ * is dropped before its packet is sent. `dgram_id` is the value that ++ * on_dg_write gave lsquic_conn_set_datagram_id() for that frame (0 if it ++ * gave none). `status` is one of enum lsquic_dg_status. bun addition. + */ -+ void (*on_datagram_status)(lsquic_conn_t *c, unsigned count, int acked); ++ void (*on_datagram_status)(lsquic_conn_t *c, uint64_t dgram_id, ++ int status); + /** + * Optional callback fired on the client the moment the server rejects + * its early data (before any of the 0-RTT streams are torn down). @@ -52,7 +54,7 @@ }; struct ssl_ctx_st; -@@ -880,6 +925,12 @@ struct lsquic_engine_settings { +@@ -880,6 +927,12 @@ struct lsquic_engine_settings { */ unsigned es_ping_period; @@ -65,7 +67,7 @@ /** * Source Connection ID length. Only applicable to the IETF QUIC * versions. Valid values are 0 through 20, inclusive. -@@ -1209,6 +1260,57 @@ struct lsquic_engine_settings { +@@ -1209,6 +1262,57 @@ struct lsquic_engine_settings { * and @ref LSQUIC_DF_MAX_HEADER_SETS_CLIENT in client mode. */ unsigned es_max_header_sets; @@ -123,7 +125,7 @@ }; /* Initialize `settings' to default values */ -@@ -1630,6 +1732,13 @@ lsquic_conn_cancel_pending_streams (lsquic_conn_t *, unsigned n); +@@ -1630,6 +1734,13 @@ lsquic_conn_cancel_pending_streams (lsquic_conn_t *, unsigned n); void lsquic_conn_going_away (lsquic_conn_t *); @@ -137,7 +139,7 @@ /** * This forces connection close. on_conn_closed and on_close callbacks * will be called. Closing an established IETF QUIC connection sends a -@@ -1864,6 +1973,20 @@ lsquic_stream_is_pushed (const lsquic_stream_t *s); +@@ -1864,6 +1975,20 @@ lsquic_stream_is_pushed (const lsquic_stream_t *s); int lsquic_stream_is_rejected (const lsquic_stream_t *s); @@ -158,7 +160,7 @@ /** * Retained for API compatibility with removed server push support. Current * LSQUIC versions do not create pushed streams. -@@ -2061,6 +2184,112 @@ lsquic_logger_lopt (const char *optarg); +@@ -2061,6 +2186,131 @@ lsquic_logger_lopt (const char *optarg); */ unsigned lsquic_engine_quic_versions (const lsquic_engine_t *); @@ -226,6 +228,25 @@ +lsquic_conn_datagram_early (const lsquic_conn_t *); + +/** ++ * Name the DATAGRAM frame that the on_dg_write callback is writing. Valid ++ * only during that callback. The id stays with the frame and comes back in ++ * on_datagram_status. ++ */ ++void ++lsquic_conn_set_datagram_id (lsquic_conn_t *, uint64_t dgram_id); ++ ++/** What became of a DATAGRAM frame (see on_datagram_status). */ ++enum lsquic_dg_status ++{ ++ /** The peer acknowledged the packet that carried the frame. */ ++ LSQ_DG_ACKED, ++ /** The packet that carried the frame was sent and declared lost. */ ++ LSQ_DG_LOST, ++ /** The frame was dropped before its packet was sent. */ ++ LSQ_DG_UNSENT, ++}; ++ ++/** + * Abort the connection without sending CONNECTION_CLOSE. The peer + * discovers the death via stateless reset or idle timeout. + */ @@ -822,7 +843,7 @@ } ifc_peer_hq_settings; struct dcid_elem *ifc_dces[MAX_IETF_CONN_DCIDS]; TAILQ_HEAD(, dcid_elem) ifc_to_retire; -@@ -513,6 +526,19 @@ struct ietf_full_conn +@@ -513,6 +526,24 @@ struct ietf_full_conn } ifc_u; lsquic_time_t ifc_idle_to; lsquic_time_t ifc_ping_period; @@ -831,18 +852,23 @@ + * (calloc'd conns default to not set, falling back to engine settings). + */ + lsquic_time_t ifc_ping_us_override; -+ int ifc_ping_us_override_set; ++ /* bun: the id that on_dg_write gave lsquic_conn_set_datagram_id for the ++ * DATAGRAM frame that write_datagram is generating. The three flags ++ * below are one byte each so that this field adds nothing to the size ++ * of the struct. */ ++ uint64_t ifc_dg_id; ++ unsigned char ifc_ping_us_override_set; + /* bun: the DATAGRAM frame currently being delivered arrived in a 0-RTT + * packet (valid only during the on_datagram callback). */ -+ int ifc_datagram_early; ++ unsigned char ifc_datagram_early; + /* bun: lsquic_conn_abort_silent — die without a CONNECTION_CLOSE. */ -+ int ifc_abort_silent; ++ unsigned char ifc_abort_silent; + /* bun: PING frames received, for session stats. */ + uint64_t ifc_pings_in; lsquic_time_t ifc_last_tick; struct lsquic_hash *ifc_bpus; uint64_t ifc_last_max_data_off_sent; -@@ -562,6 +588,11 @@ ietf_full_conn_ci_packet_in (struct lsquic_conn *, struct lsquic_packet_in *); +@@ -562,6 +593,11 @@ ietf_full_conn_ci_packet_in (struct lsquic_conn *, struct lsquic_packet_in *); static int handshake_ok (struct lsquic_conn *); @@ -854,7 +880,7 @@ static void ignore_init (struct ietf_full_conn *); -@@ -1277,6 +1308,7 @@ ietf_full_conn_init (struct ietf_full_conn *conn, +@@ -1277,6 +1313,7 @@ ietf_full_conn_init (struct ietf_full_conn *conn, conn->ifc_peer_hq_settings.header_table_size = HQ_DF_QPACK_MAX_TABLE_CAPACITY; conn->ifc_peer_hq_settings.qpack_blocked_streams = HQ_DF_QPACK_BLOCKED_STREAMS; @@ -862,7 +888,7 @@ conn->ifc_flags = flags | IFC_FIRST_TICK; conn->ifc_max_ack_packno[PNS_INIT] = IQUIC_INVALID_PACKNO; -@@ -1377,9 +1409,12 @@ lsquic_ietf_full_conn_client_new (struct lsquic_engine_public *enpub, +@@ -1377,9 +1414,12 @@ lsquic_ietf_full_conn_client_new (struct lsquic_engine_public *enpub, if (conn->ifc_settings->es_handshake_to) lsquic_alarmset_set(&conn->ifc_alset, AL_HANDSHAKE, lsquic_time_now() + conn->ifc_settings->es_handshake_to); @@ -877,7 +903,7 @@ conn->ifc_conn.cn_pf = select_pf_by_ver(ver); conn->ifc_conn.cn_esf_c = &lsquic_enc_session_common_ietf_v1; conn->ifc_conn.cn_esf.i = esfi; -@@ -1646,12 +1681,14 @@ lsquic_ietf_full_conn_server_new (struct lsquic_engine_public *enpub, +@@ -1646,12 +1686,14 @@ lsquic_ietf_full_conn_server_new (struct lsquic_engine_public *enpub, LSQ_DEBUG("Calling on_new_conn callback"); conn->ifc_conn.cn_conn_ctx = conn->ifc_enpub->enp_stream_if->on_new_conn( conn->ifc_enpub->enp_stream_if_ctx, &conn->ifc_conn); @@ -894,7 +920,7 @@ while ((packet_in = TAILQ_FIRST(&imc->imc_app_packets))) { TAILQ_REMOVE(&imc->imc_app_packets, packet_in, pi_next); -@@ -2505,7 +2542,9 @@ static int +@@ -2505,7 +2547,9 @@ static int generate_stop_sending_frame (struct ietf_full_conn *conn, struct lsquic_stream *stream) { @@ -905,7 +931,7 @@ { lsquic_stream_ss_frame_sent(stream); return 1; -@@ -2736,6 +2775,29 @@ maybe_create_delayed_streams (struct ietf_full_conn *conn) +@@ -2736,6 +2780,29 @@ maybe_create_delayed_streams (struct ietf_full_conn *conn) } @@ -935,7 +961,7 @@ static int have_bidi_streams (const struct ietf_full_conn *conn) { -@@ -2836,6 +2898,7 @@ service_streams (struct ietf_full_conn *conn) +@@ -2836,6 +2903,7 @@ service_streams (struct ietf_full_conn *conn) } else maybe_create_delayed_streams(conn); @@ -943,7 +969,7 @@ } -@@ -2981,6 +3044,46 @@ ietf_full_conn_ci_abort (struct lsquic_conn *lconn) +@@ -2981,6 +3049,46 @@ ietf_full_conn_ci_abort (struct lsquic_conn *lconn) } @@ -990,7 +1016,7 @@ static void retire_dcid (struct ietf_full_conn *conn, struct dcid_elem **dce) { -@@ -3374,6 +3477,14 @@ try_to_begin_migration (struct ietf_full_conn *conn, +@@ -3374,6 +3482,14 @@ try_to_begin_migration (struct ietf_full_conn *conn, return BM_NOT_MIGRATING; } @@ -1005,7 +1031,7 @@ if (conn->ifc_conn.cn_version <= LSQVER_ID27 /* Starting with ID-29, disable_active_migration TP applies only to the time period during the handshake. Our client does not migrate during the handshake: -@@ -3565,7 +3676,12 @@ apply_trans_params (struct ietf_full_conn *conn, +@@ -3565,7 +3681,12 @@ apply_trans_params (struct ietf_full_conn *conn, = params->tp_init_max_stream_data_bidi_remote; conn->ifc_cfg.ack_exp = params->tp_ack_delay_exponent; @@ -1019,7 +1045,7 @@ | !!params->tp_max_idle_timeout) { case (0 << 1) | 0: -@@ -3578,26 +3694,36 @@ apply_trans_params (struct ietf_full_conn *conn, +@@ -3578,26 +3699,36 @@ apply_trans_params (struct ietf_full_conn *conn, break; case (1 << 1) | 0: LSQ_DEBUG("peer did not specify max idle timeout, while ours is " @@ -1066,7 +1092,7 @@ LSQ_DEBUG("PING period is set to %"PRIu64" usec", conn->ifc_ping_period); if (conn->ifc_settings->es_delayed_acks -@@ -3726,7 +3852,9 @@ init_http (struct ietf_full_conn *conn) +@@ -3726,7 +3857,9 @@ init_http (struct ietf_full_conn *conn) &max_risked_streams); if (0 != lsquic_hcso_write_settings(&conn->ifc_hcso, conn->ifc_settings->es_max_header_list_size, dyn_table_size, @@ -1077,7 +1103,7 @@ #if LSQUIC_WEBTRANSPORT_SERVER_SUPPORT , conn->ifc_settings->es_webtransport_server, -@@ -3737,6 +3865,17 @@ init_http (struct ietf_full_conn *conn) +@@ -3737,6 +3870,17 @@ init_http (struct ietf_full_conn *conn) ABORT_WARN("cannot write SETTINGS"); return -1; } @@ -1095,7 +1121,7 @@ /* QDH initialization must precede outgoing decoder stream creation: * create_qdec_stream_out() synchronously invokes qdh_out_on_new(). -@@ -3867,6 +4006,7 @@ handshake_ok (struct lsquic_conn *lconn) +@@ -3867,6 +4011,7 @@ handshake_ok (struct lsquic_conn *lconn) conn->ifc_send_flags |= SF_SEND_NEW_CID; maybe_install_noprogress_timer(conn); maybe_create_delayed_streams(conn); @@ -1103,7 +1129,7 @@ if (!(conn->ifc_flags & IFC_SERVER)) lsquic_send_ctl_0rtt_to_1rtt(&conn->ifc_send_ctl); -@@ -4059,8 +4199,9 @@ immediate_close (struct ietf_full_conn *conn) +@@ -4059,8 +4204,9 @@ immediate_close (struct ietf_full_conn *conn) */ lsquic_send_ctl_drop_scheduled(&conn->ifc_send_ctl); @@ -1114,7 +1140,7 @@ return TICK_CLOSE; if ((conn->ifc_flags & (IFC_SERVER|IFC_IGNORE_HSK)) || -@@ -4080,7 +4221,7 @@ immediate_close (struct ietf_full_conn *conn) +@@ -4080,7 +4226,7 @@ immediate_close (struct ietf_full_conn *conn) return TICK_CLOSE; } @@ -1123,7 +1149,7 @@ if (pns == PNS_HSK && conn->ifc_error.app_error) { /* [RFC 9000] Section 10.2.3: clear app code and reason: */ -@@ -4117,7 +4258,11 @@ immediate_close (struct ietf_full_conn *conn) +@@ -4117,7 +4263,11 @@ immediate_close (struct ietf_full_conn *conn) sz = conn->ifc_conn.cn_pf->pf_gen_connect_close_frame( packet_out->po_data + packet_out->po_data_sz, lsquic_packet_out_avail(packet_out), conn_err.app_error, @@ -1136,7 +1162,7 @@ error_reason ? strlen(error_reason) : 0); if (sz < 0) { LSQ_WARN("%s failed", __func__); -@@ -4757,6 +4902,9 @@ handshake_confirmed (struct ietf_full_conn *conn) +@@ -4757,6 +4907,9 @@ handshake_confirmed (struct ietf_full_conn *conn) */ conn->ifc_conn.cn_esf.i->esfi_handshake_confirmed( conn->ifc_conn.cn_enc_session); @@ -1146,7 +1172,7 @@ if (!(conn->ifc_flags & (IFC_SERVER|IFC_MIGRA))) { conn->ifc_flags |= IFC_MIGRA; /* Perform migration just once */ -@@ -5049,6 +5197,18 @@ switch_path_to (struct ietf_full_conn *conn, unsigned char path_id) +@@ -5049,6 +5202,18 @@ switch_path_to (struct ietf_full_conn *conn, unsigned char path_id) &conn->ifc_paths[path_id].cop_path); assert(conn->ifc_cur_path_id != path_id); @@ -1165,7 +1191,7 @@ CUR_CPATH(conn)->cop_flags |= COP_RETIRED; EV_LOG_CONN_EVENT(LSQUIC_LOG_CONN_ID, "switched paths"); -@@ -5733,6 +5893,9 @@ process_stream_frame (struct ietf_full_conn *conn, +@@ -5733,6 +5898,9 @@ process_stream_frame (struct ietf_full_conn *conn, conn->ifc_max_req_id = stream_frame->stream_id; } @@ -1175,7 +1201,7 @@ stream_frame->packet_in = lsquic_packet_in_get(packet_in); if (0 != lsquic_stream_frame_in(stream, stream_frame)) { -@@ -5852,6 +6015,8 @@ process_ping_frame (struct ietf_full_conn *conn, +@@ -5852,6 +6020,8 @@ process_ping_frame (struct ietf_full_conn *conn, * return the length of this frame. */ EV_LOG_PING_FRAME_IN(LSQUIC_LOG_CONN_ID); @@ -1184,7 +1210,7 @@ if (conn->ifc_flags & IFC_SERVER) log_conn_flow_control(conn); -@@ -5859,6 +6024,67 @@ process_ping_frame (struct ietf_full_conn *conn, +@@ -5859,6 +6029,67 @@ process_ping_frame (struct ietf_full_conn *conn, } @@ -1252,7 +1278,7 @@ static int is_benign_transport_error_code (uint64_t error_code) { -@@ -6080,6 +6306,11 @@ process_max_streams_frame (struct ietf_full_conn *conn, +@@ -6080,6 +6311,11 @@ process_max_streams_frame (struct ietf_full_conn *conn, sd == SD_BIDI ? "bidi" : "uni", conn->ifc_max_allowed_stream_id[sit], max_stream_id); conn->ifc_max_allowed_stream_id[sit] = max_stream_id; @@ -1264,7 +1290,7 @@ } else LSQ_DEBUG("ignore old max %s streams value of %"PRIu64, -@@ -6184,6 +6415,16 @@ insert_new_dcid (struct ietf_full_conn *conn, uint64_t seqno, +@@ -6184,6 +6420,16 @@ insert_new_dcid (struct ietf_full_conn *conn, uint64_t seqno, (*dce)->de_cid = *cid; memcpy((*dce)->de_srst, token, sizeof((*dce)->de_srst)); (*dce)->de_flags |= DE_SRST; @@ -1281,7 +1307,7 @@ if (update_cur_dcid) { *CUR_DCID(conn) = *cid; -@@ -6414,9 +6655,12 @@ process_new_token_frame (struct ietf_full_conn *conn, +@@ -6414,9 +6660,12 @@ process_new_token_frame (struct ietf_full_conn *conn, free(token_str); } } @@ -1295,7 +1321,7 @@ return parsed_len; } -@@ -6659,12 +6903,30 @@ process_datagram_frame (struct ietf_full_conn *conn, +@@ -6659,12 +6908,46 @@ process_datagram_frame (struct ietf_full_conn *conn, EV_LOG_CONN_EVENT(LSQUIC_LOG_CONN_ID, "%zd-byte DATAGRAM", data_sz); LSQ_DEBUG("%zd-byte DATAGRAM", data_sz); @@ -1322,11 +1348,27 @@ + return conn->ifc_datagram_early; +} + ++ ++/* bun: on_dg_write names the DATAGRAM frame it is writing. write_datagram ++ * keeps the id in the frame record, so it stays with the frame whatever ++ * happens to its packet, and the send controller reports the frame under ++ * it (on_datagram_status). */ ++void ++lsquic_conn_set_datagram_id (struct lsquic_conn *lconn, uint64_t dgram_id) ++{ ++ struct ietf_full_conn *conn; ++ ++ if (!(lconn->cn_flags & LSCONN_IETF) || (lconn->cn_flags & LSCONN_MINI)) ++ return; ++ conn = (struct ietf_full_conn *) lconn; ++ conn->ifc_dg_id = dgram_id; ++} ++ + typedef unsigned (*process_frame_f)( struct ietf_full_conn *, struct lsquic_packet_in *, const unsigned char *p, size_t); -@@ -7702,6 +7964,45 @@ process_incoming_packet_fast (struct ietf_full_conn *conn, +@@ -7702,6 +7985,45 @@ process_incoming_packet_fast (struct ietf_full_conn *conn, } @@ -1372,7 +1414,7 @@ static void ietf_full_conn_ci_packet_in (struct lsquic_conn *lconn, struct lsquic_packet_in *packet_in) -@@ -7711,7 +8012,7 @@ ietf_full_conn_ci_packet_in (struct lsquic_conn *lconn, +@@ -7711,7 +8033,7 @@ ietf_full_conn_ci_packet_in (struct lsquic_conn *lconn, CONN_STATS(in.bytes, packet_in->pi_data_sz); if (conn->ifc_idle_to) lsquic_alarmset_set(&conn->ifc_alset, AL_IDLE, @@ -1381,7 +1423,7 @@ if (0 == (conn->ifc_flags & IFC_IMMEDIATE_CLOSE_FLAGS)) if (0 != conn->ifc_process_incoming_packet(conn, packet_in)) conn->ifc_flags |= IFC_ERROR; -@@ -8094,6 +8395,9 @@ ietf_full_conn_ci_early_data_failed (struct lsquic_conn *lconn) +@@ -8094,6 +8416,9 @@ ietf_full_conn_ci_early_data_failed (struct lsquic_conn *lconn) struct ietf_full_conn *conn = (struct ietf_full_conn *) lconn; LSQ_DEBUG("early data failed"); @@ -1391,7 +1433,7 @@ lsquic_send_ctl_stash_0rtt_packets(&conn->ifc_send_ctl); } -@@ -8153,6 +8457,11 @@ write_datagram (struct ietf_full_conn *conn) +@@ -8153,7 +8478,17 @@ write_datagram (struct ietf_full_conn *conn) packet_out = get_writeable_packet(conn, need); if (!packet_out) return 0; @@ -1401,9 +1443,26 @@ + if (!(conn->ifc_conn.cn_flags & LSCONN_HANDSHAKE_DONE)) + packet_out->po_header_type = HETY_0RTT; ++ /* bun: on_dg_write names the frame (lsquic_conn_set_datagram_id); the ++ * frame record carries that id to on_datagram_status. */ ++ _Static_assert(sizeof(uintptr_t) >= sizeof(conn->ifc_dg_id), ++ "a frame record holds a whole datagram id"); ++ conn->ifc_dg_id = 0; w = conn->ifc_conn.cn_pf->pf_gen_datagram_frame( packet_out->po_data + packet_out->po_data_sz, -@@ -8415,7 +8724,12 @@ ietf_full_conn_ci_tick (struct lsquic_conn *lconn, lsquic_time_t now) + lsquic_packet_out_avail(packet_out), conn->ifc_min_dg_sz, +@@ -8164,8 +8499,8 @@ write_datagram (struct ietf_full_conn *conn) + LSQ_DEBUG("could not generate DATAGRAM frame"); + return 0; + } +- if (0 != lsquic_packet_out_add_frame(packet_out, conn->ifc_pub.mm, 0, +- QUIC_FRAME_DATAGRAM, packet_out->po_data_sz, w)) ++ if (0 != lsquic_packet_out_add_frame(packet_out, conn->ifc_pub.mm, ++ conn->ifc_dg_id, QUIC_FRAME_DATAGRAM, packet_out->po_data_sz, w)) + { + ABORT_ERROR("adding DATAGRAME frame to packet failed: %d", errno); + return 0; +@@ -8415,7 +8750,12 @@ ietf_full_conn_ci_tick (struct lsquic_conn *lconn, lsquic_time_t now) if (!write_is_possible(conn)) goto end_write; @@ -1417,7 +1476,7 @@ if (!write_is_possible(conn)) goto end_write; -@@ -8474,6 +8788,7 @@ ietf_full_conn_ci_tick (struct lsquic_conn *lconn, lsquic_time_t now) +@@ -8474,6 +8814,7 @@ ietf_full_conn_ci_tick (struct lsquic_conn *lconn, lsquic_time_t now) } } else if (conn->ifc_ping_period @@ -1425,7 +1484,7 @@ && (conn->ifc_conn.cn_flags & LSCONN_HANDSHAKE_DONE)) { lsquic_alarmset_unset(&conn->ifc_alset, AL_PING); -@@ -8492,9 +8807,12 @@ ietf_full_conn_ci_tick (struct lsquic_conn *lconn, lsquic_time_t now) +@@ -8492,9 +8833,12 @@ ietf_full_conn_ci_tick (struct lsquic_conn *lconn, lsquic_time_t now) * frames without coordination can produce an excessive number of * packets and poor performance. */ @@ -1440,7 +1499,7 @@ lsquic_alarmset_set(&conn->ifc_alset, AL_PING, now + conn->ifc_ping_period); -@@ -8636,6 +8954,39 @@ ietf_full_conn_ci_make_stream (struct lsquic_conn *lconn) +@@ -8636,6 +8980,39 @@ ietf_full_conn_ci_make_stream (struct lsquic_conn *lconn) } @@ -1480,7 +1539,7 @@ static void ietf_full_conn_ci_internal_error (struct lsquic_conn *lconn, const char *format, ...) -@@ -9137,6 +9488,7 @@ on_settings_frame (void *ctx) +@@ -9137,6 +9514,7 @@ on_settings_frame (void *ctx) LSQ_DEBUG("cannot create QPACK encoder stream due to unidir limit"); } maybe_create_delayed_streams(conn); @@ -1488,7 +1547,7 @@ } -@@ -9159,6 +9511,10 @@ on_setting (void *ctx, uint64_t setting_id, uint64_t value) +@@ -9159,6 +9537,10 @@ on_setting (void *ctx, uint64_t setting_id, uint64_t value) LSQ_DEBUG("Peer's SETTINGS_MAX_HEADER_LIST_SIZE=%"PRIu64"; " "we ignore it", value); break; @@ -1499,7 +1558,7 @@ default: LSQ_DEBUG("received unknown SETTING 0x%"PRIX64"=0x%"PRIX64 "; ignore it", setting_id, value); -@@ -9455,6 +9811,18 @@ on_frame_error (void *ctx, unsigned code, uint64_t frame_type) +@@ -9455,6 +9837,18 @@ on_frame_error (void *ctx, unsigned code, uint64_t frame_type) } @@ -1518,7 +1577,7 @@ static const struct hcsi_callbacks hcsi_callbacks_server_27 = { .on_cancel_push = on_cancel_push_server, -@@ -9472,6 +9840,7 @@ static const struct hcsi_callbacks hcsi_callbacks_client_27 = +@@ -9472,6 +9866,7 @@ static const struct hcsi_callbacks hcsi_callbacks_client_27 = .on_max_push_id = on_max_push_id_client, .on_settings_frame = on_settings_frame, .on_setting = on_setting, @@ -1526,7 +1585,7 @@ .on_goaway = on_goaway_client_27, .on_frame_error = on_frame_error, .on_priority_update = on_priority_update_client, -@@ -9495,6 +9864,7 @@ static const struct hcsi_callbacks hcsi_callbacks_client_29 = +@@ -9495,6 +9890,7 @@ static const struct hcsi_callbacks hcsi_callbacks_client_29 = .on_max_push_id = on_max_push_id_client, .on_settings_frame = on_settings_frame, .on_setting = on_setting, @@ -1534,7 +1593,7 @@ .on_goaway = on_goaway_client, .on_frame_error = on_frame_error, .on_priority_update = on_priority_update_client, -@@ -9966,3 +10336,23 @@ lsquic_ietf_full_conn_test_conn_close (unsigned results[13]) +@@ -9966,3 +10362,23 @@ lsquic_ietf_full_conn_test_conn_close (unsigned results[13]) typedef char dcid_elem_fits_in_128_bytes[sizeof(struct dcid_elem) <= 128 ? 1 : - 1]; @@ -1770,73 +1829,81 @@ else { snprintf(reason_buf, sizeof(reason_buf), "TLS alert %"PRIu8, ---- a/src/liblsquic/lsquic_packet_out.c -+++ b/src/liblsquic/lsquic_packet_out.c -@@ -401,6 +401,20 @@ lsquic_packet_out_ack_streams (lsquic_packet_out_t *packet_out) +--- a/src/liblsquic/lsquic_send_ctl.c ++++ b/src/liblsquic/lsquic_send_ctl.c +@@ -1164,6 +1164,31 @@ send_ctl_record_loss (struct lsquic_send_ctl *ctl, } -+unsigned -+lsquic_packet_out_count_datagram_frames (lsquic_packet_out_t *packet_out) ++/* bun: tell the application what became of each DATAGRAM frame in the ++ * packet, under the id that write_datagram stored in the frame record. ++ * DATAGRAM frames are never retransmitted (RFC 9221 Section 5), so a caller ++ * is the last place that can name them: the packet was acknowledged, it was ++ * declared lost, or it is dropped or stripped before it is sent. ++ */ ++static void ++send_ctl_report_datagrams (struct lsquic_send_ctl *ctl, ++ struct lsquic_packet_out *packet_out, enum lsquic_dg_status status) +{ + struct packet_out_frec_iter pofi; + struct frame_rec *frec; -+ unsigned count = 0; ++ ++ if (!(packet_out->po_frame_types & QUIC_FTBIT_DATAGRAM) ++ || !ctl->sc_enpub->enp_stream_if->on_datagram_status) ++ return; ++ + for (frec = lsquic_pofi_first(&pofi, packet_out); frec; + frec = lsquic_pofi_next(&pofi)) + if (QUIC_FRAME_DATAGRAM == frec->fe_frame_type) -+ ++count; -+ return count; ++ ctl->sc_enpub->enp_stream_if->on_datagram_status( ++ ctl->sc_conn_pub->lconn, frec->fe_u.data, status); +} + + - void - lsquic_packet_out_zero_pad (lsquic_packet_out_t *packet_out) - { ---- a/src/liblsquic/lsquic_packet_out.h -+++ b/src/liblsquic/lsquic_packet_out.h -@@ -376,6 +376,9 @@ lsquic_pofi_first (struct packet_out_frec_iter *pofi, lsquic_packet_out_t *); - struct frame_rec * - lsquic_pofi_next (struct packet_out_frec_iter *pofi); - -+unsigned -+lsquic_packet_out_count_datagram_frames (lsquic_packet_out_t *); -+ - lsquic_packet_out_t * - lsquic_packet_out_new (struct lsquic_mm *, struct malo *, int use_cid, - const struct lsquic_conn *, enum packno_bits, ---- a/src/liblsquic/lsquic_send_ctl.c -+++ b/src/liblsquic/lsquic_send_ctl.c -@@ -1174,6 +1174,16 @@ send_ctl_handle_regular_lost_packet (struct lsquic_send_ctl *ctl, + static struct lsquic_packet_out * + send_ctl_handle_regular_lost_packet (struct lsquic_send_ctl *ctl, + lsquic_packet_out_t *packet_out, struct lsquic_packet_out **next) +@@ -1174,6 +1199,10 @@ send_ctl_handle_regular_lost_packet (struct lsquic_send_ctl *ctl, assert(ctl->sc_n_in_flight_all); packet_sz = packet_out_sent_sz(packet_out); -+ /* DATAGRAM frames are never retransmitted (RFC 9221 Section 5): even if -+ * the packet's other frames are rescheduled, the datagrams it carried -+ * are gone. Report them as lost. -+ */ -+ if ((packet_out->po_frame_types & QUIC_FTBIT_DATAGRAM) -+ && ctl->sc_enpub->enp_stream_if->on_datagram_status) -+ ctl->sc_enpub->enp_stream_if->on_datagram_status( -+ ctl->sc_conn_pub->lconn, -+ lsquic_packet_out_count_datagram_frames(packet_out), 0); ++ /* bun: even if the packet's other frames are rescheduled, the datagrams ++ * it carried are gone. */ ++ send_ctl_report_datagrams(ctl, packet_out, LSQ_DG_LOST); + ++ctl->sc_loss_count; #if LSQUIC_CONN_STATS ++ctl->sc_conn_pub->conn_stats->out.lost_packets; -@@ -1507,6 +1517,12 @@ lsquic_send_ctl_got_ack (lsquic_send_ctl_t *ctl, +@@ -1507,6 +1536,7 @@ lsquic_send_ctl_got_ack (lsquic_send_ctl_t *ctl, packet_sz = packet_out_sent_sz(packet_out); send_ctl_unacked_remove(ctl, packet_out, packet_sz); lsquic_packet_out_ack_streams(packet_out); -+ if ((packet_out->po_frame_types & QUIC_FTBIT_DATAGRAM) -+ && ctl->sc_enpub->enp_stream_if->on_datagram_status) -+ ctl->sc_enpub->enp_stream_if->on_datagram_status( -+ ctl->sc_conn_pub->lconn, -+ lsquic_packet_out_count_datagram_frames(packet_out), -+ 1); ++ send_ctl_report_datagrams(ctl, packet_out, LSQ_DG_ACKED); LSQ_DEBUG("acking via regular record #%"PRIu64, packet_out->po_packno); } +@@ -2288,6 +2318,10 @@ lsquic_send_ctl_next_packet_to_send (struct lsquic_send_ctl *ctl, + + if (packet_out->po_flags & PO_REPACKNO) + { ++ /* bun: a scheduled packet that gets a new number loses its DATAGRAM ++ * frames: it is dropped below, or update_for_resending chops them ++ * out with the other regenerated frames. */ ++ send_ctl_report_datagrams(ctl, packet_out, LSQ_DG_UNSENT); + if (packet_out->po_regen_sz < packet_out->po_data_sz + && packet_out->po_frame_types != QUIC_FTBIT_PADDING) + { +@@ -2871,6 +2905,10 @@ lsquic_send_ctl_squeeze_sched (lsquic_send_ctl_t *ctl) + send_ctl_sched_remove(ctl, packet_out); + LSQ_DEBUG("Dropping packet #%"PRIu64" from scheduled queue", + packet_out->po_packno); ++ /* bun: a delayed packet that holds nothing but regenerated ++ * frames is not sent later, and a DATAGRAM frame counts as ++ * one (write_datagram adds it to po_regen_sz). */ ++ send_ctl_report_datagrams(ctl, packet_out, LSQ_DG_UNSENT); + send_ctl_destroy_chain(ctl, packet_out, NULL); + send_ctl_destroy_packet(ctl, packet_out); + ++dropped; --- a/src/liblsquic/lsquic_stream.c +++ b/src/liblsquic/lsquic_stream.c @@ -694,6 +694,8 @@ lsquic_stream_destroy (lsquic_stream_t *stream) diff --git a/src/lsquic_sys/lib.rs b/src/lsquic_sys/lib.rs index 5997a5a2bcb8..ee343ddf8a64 100644 --- a/src/lsquic_sys/lib.rs +++ b/src/lsquic_sys/lib.rs @@ -44,6 +44,11 @@ pub const LSCONN_ST_RESET: c_int = 5; pub const LSCONN_ST_ERROR: c_int = 7; pub const LSCONN_ST_VERNEG_FAILURE: c_int = 10; +/// `enum lsquic_dg_status`, the `status` of `NqVtable::on_datagram_status`. +pub const LSQ_DG_ACKED: c_int = 0; +pub const LSQ_DG_LOST: c_int = 1; +pub const LSQ_DG_UNSENT: c_int = 2; + pub const N_LSQVER: c_int = 8; pub const LSQUIC_GLOBAL_CLIENT: c_int = 1; @@ -81,8 +86,7 @@ pub struct NqVtable { pub on_dg_write: unsafe extern "C" fn(conn_ctx: *mut c_void, buf: *mut c_void, buf_sz: usize) -> isize, pub on_datagram: unsafe extern "C" fn(conn_ctx: *mut c_void, buf: *const c_void, sz: usize), - pub on_datagram_status: - unsafe extern "C" fn(conn_ctx: *mut c_void, count: c_uint, acked: c_int), + pub on_datagram_status: unsafe extern "C" fn(conn_ctx: *mut c_void, id: u64, status: c_int), pub on_early_data_failed: unsafe extern "C" fn(conn_ctx: *mut c_void), pub on_path_switch: unsafe extern "C" fn( conn_ctx: *mut c_void, @@ -461,6 +465,14 @@ impl Conn { // SAFETY: `self.0` is a live conn (constructor contract). unsafe { lsquic_conn_use_preferred_address(self.0, on as c_int) } } + /// Only valid inside `on_dg_write`: the id `on_datagram_status` reports the frame under. + pub fn set_datagram_id(&self, id: u64) { + unsafe extern "C" { + fn lsquic_conn_set_datagram_id(c: *mut lsquic_conn, id: u64); + } + // SAFETY: `self.0` is a live conn (constructor contract). + unsafe { lsquic_conn_set_datagram_id(self.0, id) } + } pub fn datagram_early(&self) -> bool { unsafe extern "C" { fn lsquic_conn_datagram_early(c: *const lsquic_conn) -> c_int; diff --git a/src/runtime/node/quic/session.rs b/src/runtime/node/quic/session.rs index 0ef2b1a94882..5a5af486ae65 100644 --- a/src/runtime/node/quic/session.rs +++ b/src/runtime/node/quic/session.rs @@ -266,6 +266,14 @@ pub(super) struct DeferredAbort { marker: u64, } +/// The second argument of `ondatagramstatus` (node's `quic::DatagramStatus`). +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum DatagramStatus { + Acknowledged, + Lost, + Abandoned, +} + pub(super) enum SessionEvent { HandshakeDone { ok: bool, @@ -317,13 +325,11 @@ pub(super) enum SessionEvent { payload: Vec, early: bool, }, + /// `count` datagrams, with consecutive ids from `first_id`, got `status`. DatagramStatus { - id: u64, - sent: bool, - }, - DatagramAckStatus { - count: u32, - acked: bool, + first_id: u64, + count: u64, + status: DatagramStatus, }, EarlyDataFailed, /// HTTP/3 ORIGIN frame payload (RFC 9412). @@ -377,7 +383,6 @@ pub(crate) struct QuicSession { datagram_queue: JsCell)>>, /// Monotonic id assigned by `sendDatagram` (Node returns it as a BigInt). next_datagram_id: Cell, - inflight_datagrams: JsCell>, ticket_delivered: Cell, pending_tickets: JsCell)>>, close_after_streams: Cell, @@ -425,7 +430,6 @@ impl QuicSession { qlog_fin_sent: Cell::new(false), datagram_queue: JsCell::new(VecDeque::new()), next_datagram_id: Cell::new(1), - inflight_datagrams: JsCell::new(VecDeque::new()), ticket_delivered: Cell::new(false), pending_tickets: JsCell::new(VecDeque::new()), close_after_streams: Cell::new(false), @@ -1246,27 +1250,31 @@ impl QuicSession { ); } } - SessionEvent::DatagramStatus { id, sent } => { - if sent { - self.add_stat(IDX_STATS_SESSION_DATAGRAMS_SENT, 1); - self.inflight_datagrams.with_mut(|q| q.push_back(id)); - return Ok(()); - } - if !self.has_listener(LISTENER_FLAG_DATAGRAM_STATUS) { - return Ok(()); - } - let id_js = JSValue::from_uint64_no_truncate(global, id)?; - let status_js = global.common_strings().quic_datagram_abandoned(); - if let Some(cb) = callbacks::get(global, "onSessionDatagramStatus") { - let vm = global.bun_vm().as_mut(); - vm.event_loop_ref().run_callback( - bun_event_loop::ContextId::NONE, - cb, - global, - self.handle(), - &[id_js, status_js], - ); + SessionEvent::DatagramStatus { + first_id, + count, + status, + } => { + let stat = match status { + DatagramStatus::Acknowledged => Some(IDX_STATS_SESSION_DATAGRAMS_ACKNOWLEDGED), + DatagramStatus::Lost => Some(IDX_STATS_SESSION_DATAGRAMS_LOST), + DatagramStatus::Abandoned => None, + }; + // Each datagram is counted even when its status cannot be delivered. + let mut undelivered = Ok(()); + for id in first_id..first_id + count { + if let Some(stat) = stat { + self.add_stat(stat, 1); + } + if undelivered.is_ok() { + undelivered = self.emit_datagram_status(global, id, status); + } + // `ondatagramstatus` can destroy this session, which ends the run. + if self.destroyed.get() { + break; + } } + undelivered?; } SessionEvent::EarlyDataFailed => { // Node parity: their `closed` promises reject with an @@ -1285,47 +1293,6 @@ impl QuicSession { } } } - SessionEvent::DatagramAckStatus { count, acked } => { - let status_js = if acked { - global.common_strings().quic_datagram_acknowledged() - } else { - global.common_strings().quic_datagram_lost() - }; - // Every acknowledged/lost datagram is popped and counted even - // when its status cannot be delivered. - let mut undelivered = Ok(()); - for _ in 0..count { - let Some(id) = self.inflight_datagrams.with_mut(VecDeque::pop_front) else { - break; - }; - if acked { - self.add_stat(IDX_STATS_SESSION_DATAGRAMS_ACKNOWLEDGED, 1); - } else { - self.add_stat(IDX_STATS_SESSION_DATAGRAMS_LOST, 1); - } - if !self.has_listener(LISTENER_FLAG_DATAGRAM_STATUS) || undelivered.is_err() { - continue; - } - let id_js = match JSValue::from_uint64_no_truncate(global, id) { - Ok(id_js) => id_js, - Err(err) => { - undelivered = Err(err); - continue; - } - }; - if let Some(cb) = callbacks::get(global, "onSessionDatagramStatus") { - let vm = global.bun_vm().as_mut(); - vm.event_loop_ref().run_callback( - bun_event_loop::ContextId::NONE, - cb, - global, - self.handle(), - &[id_js, status_js], - ); - } - } - undelivered?; - } SessionEvent::VersionNegotiation { server_versions } => { let Some((requested, min)) = self.verneg.get() else { return Ok(()); @@ -2145,7 +2112,7 @@ impl QuicSession { // Node reports the abandonment synchronously from within // sendDatagram. if self.datagram_drop_newest.get() { - self.report_datagram_abandoned(global, id)?; + self.emit_datagram_status(global, id, DatagramStatus::Abandoned)?; // SAFETY: as above. unsafe { (&raw mut (*self.state_mut()).last_datagram_id).write_unaligned(id) }; return JSValue::from_uint64_no_truncate(global, id); @@ -2153,7 +2120,7 @@ impl QuicSession { if let Some((dropped_id, _)) = self.datagram_queue.with_mut(VecDeque::pop_front) { // Runs the user's `ondatagramstatus`, which can destroy this // session or close the conn before we get back here. - self.report_datagram_abandoned(global, dropped_id)?; + self.emit_datagram_status(global, dropped_id, DatagramStatus::Abandoned)?; if self.destroyed.get() || self.conn.get().is_null() { return JSValue::from_uint64_no_truncate(global, 0); } @@ -2169,14 +2136,23 @@ impl QuicSession { self.schedule_process(); JSValue::from_uint64_no_truncate(global, id) } - /// Runs inside `sendDatagram`: what building the status left pending is - /// thrown from there. - fn report_datagram_abandoned(&self, global: &JSGlobalObject, id: u64) -> JsResult<()> { + /// One `ondatagramstatus(id, status)` call. Inside `sendDatagram`, an `Err` is thrown from there. + fn emit_datagram_status( + &self, + global: &JSGlobalObject, + id: u64, + status: DatagramStatus, + ) -> JsResult<()> { if !self.has_listener(LISTENER_FLAG_DATAGRAM_STATUS) { return Ok(()); } let id_js = JSValue::from_uint64_no_truncate(global, id)?; - let status_js = global.common_strings().quic_datagram_abandoned(); + let strings = global.common_strings(); + let status_js = match status { + DatagramStatus::Acknowledged => strings.quic_datagram_acknowledged(), + DatagramStatus::Lost => strings.quic_datagram_lost(), + DatagramStatus::Abandoned => strings.quic_datagram_abandoned(), + }; if let Some(cb) = callbacks::get(global, "onSessionDatagramStatus") { let vm = global.bun_vm().as_mut(); vm.event_loop_ref().run_callback( @@ -2565,7 +2541,11 @@ lsquic_callback! { let max = session.with_state(|s| s.max_datagram_size) as usize; if max == 0 || len > max { session.datagram_queue.with_mut(VecDeque::pop_front); - session.push_event(SessionEvent::DatagramStatus { id, sent: false }); + session.push_event(SessionEvent::DatagramStatus { + first_id: id, + count: 1, + status: DatagramStatus::Abandoned, + }); // SAFETY: the conn is live for this callback. if let Some(c) = unsafe { lsquic::Conn::from_raw(session.conn.get()) } { c.want_datagram_write(!session.datagram_queue.get().is_empty()); @@ -2580,23 +2560,40 @@ lsquic_callback! { unsafe { core::ptr::copy_nonoverlapping(payload.as_ptr(), buf.cast::(), payload.len()) }; - session.push_event(SessionEvent::DatagramStatus { id, sent: true }); - if !session.datagram_queue.get().is_empty() { - // SAFETY: the conn is live for this callback. - if let Some(c) = unsafe { lsquic::Conn::from_raw(session.conn.get()) } { + session.add_stat(IDX_STATS_SESSION_DATAGRAMS_SENT, 1); + // SAFETY: the conn is live for this callback. + if let Some(c) = unsafe { lsquic::Conn::from_raw(session.conn.get()) } { + c.set_datagram_id(id); + if !session.datagram_queue.get().is_empty() { c.want_datagram_write(true); } } payload.len() as isize } - pub(super) fn on_datagram_status(session: &QuicSession, count: c_uint, acked: c_int) { - if count == 0 { - return; - } - session.push_event(SessionEvent::DatagramAckStatus { - count, - acked: acked != 0, + pub(super) fn on_datagram_status(session: &QuicSession, id: u64, status: c_int) { + let status = match status { + lsquic::LSQ_DG_ACKED => DatagramStatus::Acknowledged, + lsquic::LSQ_DG_LOST => DatagramStatus::Lost, + // Node keeps a datagram it cannot send queued and retries it: https://github.com/nodejs/node/blob/v26.3.0/src/quic/application.cc#L280-L349 + lsquic::LSQ_DG_UNSENT => DatagramStatus::Lost, + _ => { + debug_assert!(false, "unknown lsquic_dg_status"); + DatagramStatus::Lost + } + }; + // A run of consecutive ids with one status is one event. + session.events.with_mut(|events| match events.last_mut() { + Some(SessionEvent::DatagramStatus { + first_id, + count, + status: last, + }) if *last == status && *first_id + *count == id => *count += 1, + _ => events.push(SessionEvent::DatagramStatus { + first_id: id, + count: 1, + status, + }), }); } diff --git a/test/js/node/quic/quic-datagram.test.ts b/test/js/node/quic/quic-datagram.test.ts new file mode 100644 index 000000000000..f707f66784ec --- /dev/null +++ b/test/js/node/quic/quic-datagram.test.ts @@ -0,0 +1,577 @@ +// `ondatagramstatus(id, status)` reports each datagram under the id that +// `sendDatagram()` returned for it, whatever order the acknowledgements and +// the losses are declared in, and whatever lsquic does with the packet. +import { describe, expect, test } from "bun:test"; +import { createPrivateKey, randomBytes } from "node:crypto"; +import { createSocket } from "node:dgram"; +import { once } from "node:events"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { connect, listen } from "node:quic"; + +const keysDir = join(import.meta.dir, "..", "test", "fixtures", "keys"); +const key = createPrivateKey(readFileSync(join(keysDir, "agent1-key.pem"))); +const cert = readFileSync(join(keysDir, "agent1-cert.pem")); +const sni = { "*": { keys: [key], certs: [cert] } }; + +// The relay sees ciphertext, so it tells 1-RTT packets apart by length. The +// sizes below put each kind of packet in a range of its own, whatever small +// frames lsquic adds to it. Two datagrams of LARGE or of MEDIUM bytes do not +// fit in one packet, so each one travels in a packet of its own. +const LARGE = 1100; +const MEDIUM = 750; +const SMALL = 200; +const CHUNK = 300; +const lengthIn = (min: number, max: number) => (packet: Buffer) => + !(packet[0] & 0x80) && packet.length >= min && packet.length < max; +/** A packet with a LARGE datagram. A path MTU probe is longer than 1400 bytes. */ +const carriesLarge = lengthIn(LARGE, 1400); +const carriesMedium = lengthIn(MEDIUM, LARGE); +const carriesThreeSmall = lengthIn(3 * SMALL, MEDIUM); +/** A packet with CHUNK bytes of stream data. */ +const carriesChunk = lengthIn(CHUNK, 3 * SMALL); +// The tests with a retransmission timeout start with this many datagrams. +const WARM_UP = 24; +// A datagram that arrived is acknowledged, unless lsquic gave up on its +// packet before the ACK came. +const arrived = expect.stringMatching(/^(acknowledged|lost)$/); + +/** A UDP relay between a client and the server at `serverPort`. */ +async function relayTo(serverPort: number) { + const front = createSocket("udp4"); + const upstream = createSocket("udp4"); + let clientPort = 0; + const forServer: Buffer[] = []; + const forClient: Buffer[] = []; + const relay = { + port: 0, + /** Returns true for a client packet that the relay must lose. */ + lose: (_packet: Buffer): boolean => false, + /** While true, the relay keeps the packets of both sides. */ + outage: false, + /** How many of the packets kept for the client carry a MEDIUM datagram. */ + keptMedium: () => forClient.filter(carriesMedium).length, + /** Forwards the packets kept for the client, through the one with the `count`th MEDIUM datagram. */ + deliverMedium(count: number) { + let end = 0; + for (let seen = 0; end < forClient.length && seen < count; end++) seen += +carriesMedium(forClient[end]); + for (const packet of forClient.splice(0, end)) front.send(packet, clientPort, "127.0.0.1"); + }, + /** Ends the outage. What the relay kept goes out, in order. */ + release() { + relay.outage = false; + for (const packet of forServer.splice(0)) upstream.send(packet, serverPort, "127.0.0.1"); + for (const packet of forClient.splice(0)) front.send(packet, clientPort, "127.0.0.1"); + }, + /** Called for each packet that arrives at the relay from that side. */ + onClientPacket: (_packet: Buffer) => {}, + onServerPacket: () => {}, + [Symbol.dispose]() { + front.close(); + upstream.close(); + }, + }; + upstream.on("message", packet => { + if (relay.outage) forClient.push(packet); + else front.send(packet, clientPort, "127.0.0.1"); + relay.onServerPacket(); + }); + front.on("message", (packet, from) => { + clientPort = from.port; + relay.onClientPacket(packet); + if (relay.lose(packet)) return; + if (relay.outage) forServer.push(packet); + else upstream.send(packet, serverPort, "127.0.0.1"); + }); + front.bind(0, "127.0.0.1"); + upstream.bind(0, "127.0.0.1"); + await Promise.all([once(front, "listening"), once(upstream, "listening")]); + relay.port = front.address().port; + return relay; +} + +/** + * A server, a relay in front of it, and a client session through the relay. + * The first byte of each datagram the client sends is its id, so the server + * side knows which ids arrived. + */ +async function sessionThroughRelay() { + const received: number[] = []; + let onReceived = () => {}; + const serverSession = Promise.withResolvers(); + const serverSessionClosed = Promise.withResolvers(); + const server = await listen( + session => { + serverSession.resolve(session); + session.onerror = () => {}; + session.closed.then(serverSessionClosed.resolve, serverSessionClosed.resolve); + // The server echoes what the client writes to a stream. + session.onstream = async (stream: any) => { + stream.closed.catch(() => {}); + try { + for await (const batch of stream) for (const chunk of [batch].flat()) stream.writer.writeSync(chunk); + } catch {} + }; + }, + { + sni, + alpn: ["quic-test"], + transportParams: { maxIdleTimeout: 10 }, + ondatagram(datagram: Uint8Array) { + received.push(datagram[0]); + onReceived(); + }, + }, + ); + const relay = await relayTo(server.address.port); + + const statuses = new Map(); + let onStatus = () => {}; + let fromServer = 0; + let onServerDatagram = () => {}; + let onServerStream = () => {}; + const client = await connect( + { address: "127.0.0.1", port: relay.port }, + { + alpn: "quic-test", + servername: "localhost", + verifyPeer: "manual", + transportParams: { maxIdleTimeout: 10 }, + onerror() {}, + ondatagram() { + fromServer++; + onServerDatagram(); + }, + ondatagramstatus(id: bigint, status: string) { + statuses.set(Number(id), [...(statuses.get(Number(id)) ?? []), status]); + onStatus(); + }, + async onstream(stream: any) { + stream.closed.catch(() => {}); + try { + for await (const _ of stream); + } catch {} + onServerStream(); + }, + }, + ); + await client.opened; + + const ids: Promise[] = []; + /** Resolves once `done()` holds. `listen` installs the check at its source. */ + const until = (done: () => boolean, listen: (check: () => void) => void) => + new Promise(resolve => { + const check = () => done() && resolve(); + listen(check); + check(); + }); + /** Sends a datagram of `size` bytes. Returns its id: ids count from 1. */ + const send = (size = LARGE) => { + ids.push(client.sendDatagram(Buffer.alloc(size, ids.length + 1))); + return ids.length; + }; + /** The first status reported for `id`. */ + const status = (id: number) => + until( + () => statuses.has(id), + check => (onStatus = check), + ).then(() => statuses.get(id)![0]); + return { + client, + relay, + send, + status, + /** The ids of the datagrams that the server received, in ascending order. */ + received: () => received.toSorted((a, b) => a - b), + /** Every status reported for each id: one per id when all is right. */ + statuses: () => Object.fromEntries(statuses), + /** + * Round trips of two datagrams, because the server acknowledges every + * second packet at once. After them the retransmission timers of the + * client are short. These are datagrams 1 to WARM_UP. + */ + async warmUp() { + for (let i = 0; i < WARM_UP; i += 2) { + send(); + expect(await status(send())).toBe("acknowledged"); + } + }, + /** + * Resolves once the client has reported what the server's ACKs say about + * every datagram sent so far. The last one must pass the relay. When the + * server has it, the server opens a stream, and the packet with the + * stream data carries the ACKs that the server still owes. The client + * queues the statuses from that packet behind the stream, for the same + * pass: they are out when the pass is over. + */ + async settled() { + const last = ids.length; + await until( + () => received.includes(last), + check => (onReceived = check), + ); + const read = new Promise(resolve => (onServerStream = resolve)); + const stream = await (await serverSession.promise).createUnidirectionalStream({ body: new Uint8Array(1) }); + stream.closed.catch(() => {}); + await read; + await new Promise(resolve => setImmediate(resolve)); + expect((await Promise.all(ids)).map(Number)).toEqual(ids.map((_, i) => i + 1)); + }, + /** A stream to the server. `echoed(n)` resolves once `n` bytes have come back. */ + async echoStream() { + const stream = await client.createBidirectionalStream(); + stream.closed.catch(() => {}); + let count = 0; + let onEcho = () => {}; + (async () => { + try { + for await (const batch of stream) { + for (const chunk of [batch].flat()) count += chunk.byteLength; + onEcho(); + } + } catch {} + })(); + return { + write: (size: number) => void stream.writer.writeSync(new Uint8Array(size)), + echoed: (size: number) => + until( + () => count >= size, + check => (onEcho = check), + ), + }; + }, + /** + * The server sends `count` datagrams during an outage, each in a packet + * of its own. Resolves once the relay has them. + */ + async serverDatagrams(count: number) { + const session = await serverSession.promise; + for (let i = 0; i < count; i++) session.sendDatagram(new Uint8Array(MEDIUM)); + await until( + () => relay.keptMedium() >= count, + check => (relay.onServerPacket = check), + ); + }, + /** + * The relay hands the client the next two datagrams of the server. The + * client acknowledges every second packet at once, so it owes no ACK + * when it dispatches the second one. Resolves inside that dispatch, + * after `inDispatch` ran there. + */ + async deliverTwo(inDispatch = () => {}) { + const last = fromServer + 2; + const dispatched = until( + () => fromServer >= last, + check => + (onServerDatagram = () => { + if (fromServer === last) inDispatch(); + check(); + }), + ); + relay.deliverMedium(2); + await dispatched; + }, + /** + * An outage in which the retransmission timeout of the client fires. A + * datagram leaves in a packet of its own, then stream data keeps the + * retransmission timer running. Nothing comes back, so only that timeout + * can declare the datagram lost. Resolves outside the session's dispatch + * of that status: from there each `send()` is one lsquic tick. + */ + async timeout(echo: { write(size: number): void }) { + relay.lose = carriesLarge; + relay.outage = true; + const inOutage = send(); + await new Promise(resolve => (relay.onClientPacket = () => resolve())); + echo.write(64); + expect(await status(inOutage)).toBe("lost"); + await new Promise(resolve => setImmediate(resolve)); + }, + async [Symbol.asyncDispose]() { + // Both sides close through the relay, which passes everything again. + relay.lose = () => false; + relay.release(); + client.close(); + await Promise.all([client.closed.catch(() => {}), serverSessionClosed.promise]); + relay[Symbol.dispose](); + await server[Symbol.asyncDispose](); + }, + }; +} + +/** The ids from `from` to `to`. */ +const range = (from: number, to: number) => Array.from({ length: to - from + 1 }, (_, i) => from + i); +/** `{ [id]: [status] }` for each id from `from` to `to`. */ +const all = (status: unknown, from: number, to: number) => + Object.fromEntries(range(from, to).map(id => [id, [status]])); + +describe("ondatagramstatus", () => { + test.each([ + { lost: 1, of: 6 }, + { lost: 3, of: 8 }, + ])("reports the loss of datagram $lost of $of under its id", async ({ lost, of }) => { + await using net = await sessionThroughRelay(); + let carriers = 0; + net.relay.lose = packet => carriesLarge(packet) && ++carriers === lost; + + for (let i = 0; i < of; i++) net.send(); + await net.settled(); + + expect(net.received()).toEqual(range(1, of).filter(id => id !== lost)); + expect(net.statuses()).toEqual({ ...all("acknowledged", 1, of), [lost]: ["lost"] }); + }); + + test("reports every datagram as acknowledged when no packet is lost", async () => { + await using net = await sessionThroughRelay(); + + for (let i = 0; i < 6; i++) net.send(); + await net.settled(); + + expect(net.received()).toEqual(range(1, 6)); + expect(net.statuses()).toEqual(all("acknowledged", 1, 6)); + }); + + test("reports each datagram of a lost packet that carried several", async () => { + await using net = await sessionThroughRelay(); + net.relay.lose = carriesThreeSmall; + + // This `await` resumes inside the session's dispatch of the status, and + // writes made there leave together: datagrams 2, 3 and 4 share the one + // packet that the relay loses. + expect(await net.status(net.send())).toBe("acknowledged"); + for (let i = 0; i < 3; i++) net.send(SMALL); + for (let i = 0; i < 6; i++) net.send(); + await net.settled(); + + expect(net.received()).toEqual([1, ...range(5, 10)]); + expect(net.statuses()).toEqual({ ...all("acknowledged", 1, 10), ...all("lost", 2, 4) }); + }); + + test("keeps an acknowledged datagram and the lost one after it apart", async () => { + await using net = await sessionThroughRelay(); + const { relay } = net; + const echo = await net.echoStream(); + // Outside a dispatch of the session, each write below is one packet. + await new Promise(resolve => setImmediate(resolve)); + + // The relay keeps datagram 1, loses datagram 2, and keeps six packets of + // stream data. Then the server gets them all at once. Its ACK reaches + // past datagram 2 with stream data alone, so the client acknowledges + // datagram 1 and then declares datagram 2 lost, in one pass. + relay.lose = carriesLarge; + relay.outage = true; + let chunks = 0; + const kept = new Promise(resolve => { + relay.onClientPacket = packet => carriesChunk(packet) && ++chunks === 6 && resolve(); + }); + net.send(MEDIUM); + net.send(); + for (let i = 0; i < 6; i++) echo.write(CHUNK); + await kept; + relay.release(); + + await echo.echoed(6 * CHUNK); + net.send(MEDIUM); + await net.settled(); + + expect(net.received()).toEqual([1, 3]); + expect(net.statuses()).toEqual({ 1: [arrived], 2: ["lost"], 3: [arrived] }); + }); + + // After a retransmission timeout lsquic sends at most two packets until an + // ACK arrives, and the stream data it resends is one of them. It still + // frames each datagram into a packet, and drops the packets it may not + // send. Node keeps such a datagram queued. Both tests below pin the status + // that Bun gives it: 'lost', under its id. + test.each([{ listener: true }, { listener: false }])( + "counts a datagram that is dropped before the send as lost (ondatagramstatus: $listener)", + async ({ listener }) => { + await using net = await sessionThroughRelay(); + const { client, relay } = net; + const echo = await net.echoStream(); + + await net.warmUp(); + await net.timeout(echo); + if (!listener) client.ondatagramstatus = undefined; + + // The outage ends for all but the packets with a LARGE datagram: the + // relay loses the one of these three that may leave. + relay.release(); + for (let i = 0; i < 3; i++) net.send(); + + // The echo arrives when the connection has recovered. The datagrams + // after it pass the relay, and their acknowledgements let the client + // declare a datagram that did leave lost. + await echo.echoed(64); + for (let i = 0; i < 6; i++) net.send(MEDIUM); + await net.settled(); + + // Datagrams 1 to 24 are the warm-up, and 25 is the one of the timeout. + expect(net.received()).toEqual([...range(1, 24), ...range(29, 34)]); + expect(net.statuses()).toEqual( + listener + ? { ...all("acknowledged", 1, 24), ...all("lost", 25, 28), ...all(arrived, 29, 34) } + : { ...all("acknowledged", 1, 24), 25: ["lost"] }, + ); + const { datagramsSent: sent, datagramsAcknowledged: acknowledged, datagramsLost: lost } = client.stats; + expect({ sent, counted: acknowledged + lost }).toEqual({ sent: 34n, counted: 34n }); + expect(lost).toBeGreaterThanOrEqual(4n); + }, + ); + + test("reports a datagram that is cut out of a packet before the send as lost, under its id", async () => { + await using net = await sessionThroughRelay(); + const { relay } = net; + const echo = await net.echoStream(); + + await net.warmUp(); + // The relay keeps four datagrams of the server for the steps below. + relay.outage = true; + await net.serverDatagrams(4); + await net.timeout(echo); + // Datagram 26 uses up what lsquic may still send before an ACK. + net.send(); + + // Datagram 27 and stream data, written in one dispatch of the session, + // share a packet. lsquic may not send that packet yet, and keeps it + // because of the stream data. + await net.deliverTwo(() => { + echo.write(8); + net.send(); + }); + // The client sends an ACK for the next two. lsquic gives new numbers to + // the packets that wait behind an ACK, and it cuts the DATAGRAM frame out + // of a packet that it renumbers. + await net.deliverTwo(); + + relay.release(); + await echo.echoed(72); + for (let i = 0; i < 6; i++) net.send(MEDIUM); + await net.settled(); + + expect(net.received()).toEqual([...range(1, 24), ...range(28, 33)]); + expect(net.statuses()).toEqual({ + ...all("acknowledged", 1, 24), + ...all("lost", 25, 27), + ...all(arrived, 28, 33), + }); + }); + + test("stops at a callback that destroys the session", async () => { + const serverSession = Promise.withResolvers(); + await using server = await listen( + session => { + session.onerror = () => {}; + session.closed.catch(() => {}); + serverSession.resolve(session); + }, + { sni, alpn: ["quic-test"], ondatagram() {} }, + ); + const calls: string[] = []; + const client = await connect(server.address, { + alpn: "quic-test", + servername: "localhost", + verifyPeer: "manual", + onerror() {}, + ondatagramstatus(id: bigint, status: string) { + calls.push(`${id}:${status}`); + if (id > 1n) throw new Error("from ondatagramstatus"); + // Writes made inside a session callback leave together: one packet, + // and so one acknowledgement, for datagrams 2, 3 and 4. + for (let i = 0; i < 3; i++) client.sendDatagram(new Uint8Array(SMALL)); + }, + }); + await client.opened; + await client.sendDatagram(new Uint8Array(SMALL)); + + // The throw destroys the session. As in Node, datagrams 3 and 4 are then + // neither reported nor counted: an uncaught ERR_INTERNAL_ASSERTION for + // each would fail this test. + const closed = await client.closed.then( + () => "resolved", + (error: Error) => error.message, + ); + (await serverSession.promise).destroy(); + expect({ closed, calls, acknowledged: client.stats.datagramsAcknowledged }).toEqual({ + closed: "from ondatagramstatus", + calls: ["1:acknowledged", "2:acknowledged"], + acknowledged: 2n, + }); + }); + + test("skips a datagram of a rejected 0-RTT flight without moving the ids after it", async () => { + const tokenSecret = randomBytes(16); + const ticket = Promise.withResolvers(); + const token = Promise.withResolvers(); + const onSession = (session: any) => { + session.onerror = () => {}; + session.closed.catch(() => {}); + }; + + // A first server hands out a session ticket and a token. + const first = await listen(onSession, { sni, alpn: ["quic-test"], endpoint: { tokenSecret } }); + const firstClient = await connect(first.address, { + alpn: "quic-test", + verifyPeer: "manual", + onsessionticket: ticket.resolve, + onnewtoken: token.resolve, + }); + await Promise.all([firstClient.opened, ticket.promise, token.promise]); + await firstClient.close(); + await first.close(); + + // A second one takes the token and rejects the early data. + const received: number[] = []; + let onProgress = () => {}; + const serverSession = Promise.withResolvers(); + await using second = await listen( + (session: any) => { + onSession(session); + serverSession.resolve(session); + }, + { + sni, + alpn: ["quic-test"], + endpoint: { tokenSecret }, + enableEarlyData: false, + ondatagram(datagram: Uint8Array) { + received.push(datagram[0]); + onProgress(); + }, + }, + ); + const statuses = new Map(); + const client = await connect(second.address, { + alpn: "quic-test", + verifyPeer: "manual", + sessionTicket: await ticket.promise, + token: await token.promise, + onerror() {}, + onearlyrejected() {}, + ondatagramstatus(id: bigint, status: string) { + statuses.set(Number(id), [...(statuses.get(Number(id)) ?? []), status]); + onProgress(); + }, + }); + client.closed.catch(() => {}); + + // Datagram 1 leaves in the 0-RTT flight, datagrams 2 to 4 after the + // handshake. No packet is lost. As in Node, datagram 1 gets no status. + const ids = [await client.sendDatagram(new Uint8Array([1]))]; + const { earlyDataAttempted, earlyDataAccepted } = await client.opened; + for (let i = 2; i <= 4; i++) ids.push(await client.sendDatagram(new Uint8Array([i]))); + await new Promise(resolve => { + onProgress = () => statuses.size >= 3 && received.length >= 3 && resolve(); + }); + client.destroy(); + (await serverSession.promise).destroy(); + + expect({ earlyDataAttempted, earlyDataAccepted, ids }).toEqual({ + earlyDataAttempted: true, + earlyDataAccepted: false, + ids: [1n, 2n, 3n, 4n], + }); + expect(received.toSorted((a, b) => a - b)).toEqual([2, 3, 4]); + expect(Object.fromEntries(statuses)).toEqual(all("acknowledged", 2, 4)); + }); +}); From 7426614a3ef323ac907894d2fd13d6972cf43e1a Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:28:05 +0000 Subject: [PATCH 2/5] test(node:quic): fail the datagram tests with the cause when a session fails Each wait of the relay harness now races the `closed` promise and the `onerror` event of both sessions. A session that fails rejects the wait, and the test does not run into its timeout. The teardown destroys the server session, so a client that failed does not leave it to the idle timeout. --- test/js/node/quic/quic-datagram.test.ts | 82 +++++++++++++++++++------ 1 file changed, 62 insertions(+), 20 deletions(-) diff --git a/test/js/node/quic/quic-datagram.test.ts b/test/js/node/quic/quic-datagram.test.ts index f707f66784ec..62debe33c342 100644 --- a/test/js/node/quic/quic-datagram.test.ts +++ b/test/js/node/quic/quic-datagram.test.ts @@ -36,6 +36,13 @@ const WARM_UP = 24; // packet before the ACK came. const arrived = expect.stringMatching(/^(acknowledged|lost)$/); +/** Rejects `waits` when `session` fails or closes before they settle. */ +function failWith(session: any, ...waits: PromiseWithResolvers[]) { + const fail = (error: unknown) => waits.forEach(wait => wait.reject(error)); + for (const wait of waits) wait.promise.catch(() => {}); + session.closed.then(() => fail(new Error("the session closed")), fail); +} + /** A UDP relay between a client and the server at `serverPort`. */ async function relayTo(serverPort: number) { const front = createSocket("udp4"); @@ -98,12 +105,16 @@ async function relayTo(serverPort: number) { async function sessionThroughRelay() { const received: number[] = []; let onReceived = () => {}; + // Each wait below races this, so a test that loses a session fails with + // the cause. + const failed = Promise.withResolvers(); const serverSession = Promise.withResolvers(); const serverSessionClosed = Promise.withResolvers(); const server = await listen( session => { serverSession.resolve(session); - session.onerror = () => {}; + session.onerror = failed.reject; + failWith(session, failed); session.closed.then(serverSessionClosed.resolve, serverSessionClosed.resolve); // The server echoes what the client writes to a stream. session.onstream = async (stream: any) => { @@ -137,7 +148,7 @@ async function sessionThroughRelay() { servername: "localhost", verifyPeer: "manual", transportParams: { maxIdleTimeout: 10 }, - onerror() {}, + onerror: failed.reject, ondatagram() { fromServer++; onServerDatagram(); @@ -155,16 +166,20 @@ async function sessionThroughRelay() { }, }, ); + failWith(client, failed); await client.opened; const ids: Promise[] = []; - /** Resolves once `done()` holds. `listen` installs the check at its source. */ + /** Resolves once `done()` holds. `listen` installs the check at its source. Rejects when a session fails. */ const until = (done: () => boolean, listen: (check: () => void) => void) => - new Promise(resolve => { - const check = () => done() && resolve(); - listen(check); - check(); - }); + Promise.race([ + failed.promise, + new Promise(resolve => { + const check = () => done() && resolve(); + listen(check); + check(); + }), + ]); /** Sends a datagram of `size` bytes. Returns its id: ids count from 1. */ const send = (size = LARGE) => { ids.push(client.sendDatagram(Buffer.alloc(size, ids.length + 1))); @@ -181,6 +196,7 @@ async function sessionThroughRelay() { relay, send, status, + until, /** The ids of the datagrams that the server received, in ascending order. */ received: () => received.toSorted((a, b) => a - b), /** Every status reported for each id: one per id when all is right. */ @@ -210,10 +226,18 @@ async function sessionThroughRelay() { () => received.includes(last), check => (onReceived = check), ); - const read = new Promise(resolve => (onServerStream = resolve)); + let read = false; + let onRead = () => {}; + onServerStream = () => { + read = true; + onRead(); + }; const stream = await (await serverSession.promise).createUnidirectionalStream({ body: new Uint8Array(1) }); stream.closed.catch(() => {}); - await read; + await until( + () => read, + check => (onRead = check), + ); await new Promise(resolve => setImmediate(resolve)); expect((await Promise.all(ids)).map(Number)).toEqual(ids.map((_, i) => i + 1)); }, @@ -282,17 +306,28 @@ async function sessionThroughRelay() { relay.lose = carriesLarge; relay.outage = true; const inOutage = send(); - await new Promise(resolve => (relay.onClientPacket = () => resolve())); + let atRelay = false; + await until( + () => atRelay, + check => + (relay.onClientPacket = () => { + atRelay = true; + check(); + }), + ); echo.write(64); expect(await status(inOutage)).toBe("lost"); await new Promise(resolve => setImmediate(resolve)); }, async [Symbol.asyncDispose]() { - // Both sides close through the relay, which passes everything again. + // The client closes through the relay, which passes everything again. relay.lose = () => false; relay.release(); client.close(); - await Promise.all([client.closed.catch(() => {}), serverSessionClosed.promise]); + await client.closed.catch(() => {}); + // A client that failed did not tell the server. + (await serverSession.promise).destroy(); + await serverSessionClosed.promise; relay[Symbol.dispose](); await server[Symbol.asyncDispose](); }, @@ -361,9 +396,14 @@ describe("ondatagramstatus", () => { relay.lose = carriesLarge; relay.outage = true; let chunks = 0; - const kept = new Promise(resolve => { - relay.onClientPacket = packet => carriesChunk(packet) && ++chunks === 6 && resolve(); - }); + const kept = net.until( + () => chunks >= 6, + check => + (relay.onClientPacket = packet => { + chunks += +carriesChunk(packet); + check(); + }), + ); net.send(MEDIUM); net.send(); for (let i = 0; i < 6; i++) echo.write(CHUNK); @@ -516,6 +556,7 @@ describe("ondatagramstatus", () => { onsessionticket: ticket.resolve, onnewtoken: token.resolve, }); + failWith(firstClient, ticket, token); await Promise.all([firstClient.opened, ticket.promise, token.promise]); await firstClient.close(); await first.close(); @@ -541,6 +582,7 @@ describe("ondatagramstatus", () => { }, ); const statuses = new Map(); + const reported = Promise.withResolvers(); const client = await connect(second.address, { alpn: "quic-test", verifyPeer: "manual", @@ -553,16 +595,16 @@ describe("ondatagramstatus", () => { onProgress(); }, }); - client.closed.catch(() => {}); + failWith(client, reported); // Datagram 1 leaves in the 0-RTT flight, datagrams 2 to 4 after the // handshake. No packet is lost. As in Node, datagram 1 gets no status. const ids = [await client.sendDatagram(new Uint8Array([1]))]; const { earlyDataAttempted, earlyDataAccepted } = await client.opened; for (let i = 2; i <= 4; i++) ids.push(await client.sendDatagram(new Uint8Array([i]))); - await new Promise(resolve => { - onProgress = () => statuses.size >= 3 && received.length >= 3 && resolve(); - }); + onProgress = () => statuses.size >= 3 && received.length >= 3 && reported.resolve(); + onProgress(); + await reported.promise; client.destroy(); (await serverSession.promise).destroy(); From 44ae42c08658d737587cf457eaf5ec4b57f674d8 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:47:44 +0000 Subject: [PATCH 3/5] test(node:quic): release the endpoints and sessions of the datagram tests when a step fails The relay harness and the 0-RTT test now keep what they open in an `AsyncDisposableStack`. A handshake that fails releases the listening endpoint, the relay sockets and the sessions, and the test fails with the cause. --- test/js/node/quic/quic-datagram.test.ts | 94 ++++++++++++++----------- 1 file changed, 54 insertions(+), 40 deletions(-) diff --git a/test/js/node/quic/quic-datagram.test.ts b/test/js/node/quic/quic-datagram.test.ts index 62debe33c342..350fe5020f8a 100644 --- a/test/js/node/quic/quic-datagram.test.ts +++ b/test/js/node/quic/quic-datagram.test.ts @@ -103,6 +103,9 @@ async function relayTo(serverPort: number) { * side knows which ids arrived. */ async function sessionThroughRelay() { + // Releases what is open when a step of the setup fails. The caller owns it + // after that. + await using setup = new AsyncDisposableStack(); const received: number[] = []; let onReceived = () => {}; // Each wait below races this, so a test that loses a session fails with @@ -110,31 +113,34 @@ async function sessionThroughRelay() { const failed = Promise.withResolvers(); const serverSession = Promise.withResolvers(); const serverSessionClosed = Promise.withResolvers(); - const server = await listen( - session => { - serverSession.resolve(session); - session.onerror = failed.reject; - failWith(session, failed); - session.closed.then(serverSessionClosed.resolve, serverSessionClosed.resolve); - // The server echoes what the client writes to a stream. - session.onstream = async (stream: any) => { - stream.closed.catch(() => {}); - try { - for await (const batch of stream) for (const chunk of [batch].flat()) stream.writer.writeSync(chunk); - } catch {} - }; - }, - { - sni, - alpn: ["quic-test"], - transportParams: { maxIdleTimeout: 10 }, - ondatagram(datagram: Uint8Array) { - received.push(datagram[0]); - onReceived(); + const server = setup.use( + await listen( + session => { + serverSession.resolve(session); + session.onerror = failed.reject; + failWith(session, failed); + session.closed.then(serverSessionClosed.resolve, serverSessionClosed.resolve); + // The server echoes what the client writes to a stream. + session.onstream = async (stream: any) => { + stream.closed.catch(() => {}); + try { + for await (const batch of stream) for (const chunk of [batch].flat()) stream.writer.writeSync(chunk); + } catch {} + }; }, - }, + { + sni, + alpn: ["quic-test"], + transportParams: { maxIdleTimeout: 10 }, + ondatagram(datagram: Uint8Array) { + received.push(datagram[0]); + onReceived(); + }, + }, + ), ); - const relay = await relayTo(server.address.port); + setup.defer(() => void serverSession.promise.then(session => session.destroy())); + const relay = setup.use(await relayTo(server.address.port)); const statuses = new Map(); let onStatus = () => {}; @@ -166,8 +172,10 @@ async function sessionThroughRelay() { }, }, ); + setup.defer(() => client.destroy()); failWith(client, failed); await client.opened; + const resources = setup.move(); const ids: Promise[] = []; /** Resolves once `done()` holds. `listen` installs the check at its source. Rejects when a session fails. */ @@ -328,8 +336,7 @@ async function sessionThroughRelay() { // A client that failed did not tell the server. (await serverSession.promise).destroy(); await serverSessionClosed.promise; - relay[Symbol.dispose](); - await server[Symbol.asyncDispose](); + await resources.disposeAsync(); }, }; } @@ -540,22 +547,26 @@ describe("ondatagramstatus", () => { }); test("skips a datagram of a rejected 0-RTT flight without moving the ids after it", async () => { + // Releases the endpoints and the sessions, also when a step below fails. + await using cleanup = new AsyncDisposableStack(); const tokenSecret = randomBytes(16); const ticket = Promise.withResolvers(); const token = Promise.withResolvers(); const onSession = (session: any) => { session.onerror = () => {}; session.closed.catch(() => {}); + cleanup.defer(() => session.destroy()); }; // A first server hands out a session ticket and a token. - const first = await listen(onSession, { sni, alpn: ["quic-test"], endpoint: { tokenSecret } }); + const first = cleanup.use(await listen(onSession, { sni, alpn: ["quic-test"], endpoint: { tokenSecret } })); const firstClient = await connect(first.address, { alpn: "quic-test", verifyPeer: "manual", onsessionticket: ticket.resolve, onnewtoken: token.resolve, }); + cleanup.defer(() => firstClient.destroy()); failWith(firstClient, ticket, token); await Promise.all([firstClient.opened, ticket.promise, token.promise]); await firstClient.close(); @@ -565,21 +576,23 @@ describe("ondatagramstatus", () => { const received: number[] = []; let onProgress = () => {}; const serverSession = Promise.withResolvers(); - await using second = await listen( - (session: any) => { - onSession(session); - serverSession.resolve(session); - }, - { - sni, - alpn: ["quic-test"], - endpoint: { tokenSecret }, - enableEarlyData: false, - ondatagram(datagram: Uint8Array) { - received.push(datagram[0]); - onProgress(); + const second = cleanup.use( + await listen( + (session: any) => { + onSession(session); + serverSession.resolve(session); }, - }, + { + sni, + alpn: ["quic-test"], + endpoint: { tokenSecret }, + enableEarlyData: false, + ondatagram(datagram: Uint8Array) { + received.push(datagram[0]); + onProgress(); + }, + }, + ), ); const statuses = new Map(); const reported = Promise.withResolvers(); @@ -595,6 +608,7 @@ describe("ondatagramstatus", () => { onProgress(); }, }); + cleanup.defer(() => client.destroy()); failWith(client, reported); // Datagram 1 leaves in the 0-RTT flight, datagrams 2 to 4 after the From 274e848b05abfc6c4f95822acbd330dcfe8c4034 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 18 Oct 2026 02:24:58 +0000 Subject: [PATCH 4/5] test(node:quic): make the datagram tests pass the type check of test/ `cd test && bun check` reports no type error on main. The new file passes it too: `server.address!`, a `SocketAddress` for the relay, and `as SessionOptions` for `verifyPeer`, as in the other files of this directory. --- test/js/node/quic/quic-datagram.test.ts | 67 ++++++++++++------------- 1 file changed, 32 insertions(+), 35 deletions(-) diff --git a/test/js/node/quic/quic-datagram.test.ts b/test/js/node/quic/quic-datagram.test.ts index 350fe5020f8a..ae6755760b11 100644 --- a/test/js/node/quic/quic-datagram.test.ts +++ b/test/js/node/quic/quic-datagram.test.ts @@ -6,8 +6,9 @@ import { createPrivateKey, randomBytes } from "node:crypto"; import { createSocket } from "node:dgram"; import { once } from "node:events"; import { readFileSync } from "node:fs"; +import { SocketAddress } from "node:net"; import { join } from "node:path"; -import { connect, listen } from "node:quic"; +import { connect, listen, type SessionOptions } from "node:quic"; const keysDir = join(import.meta.dir, "..", "test", "fixtures", "keys"); const key = createPrivateKey(readFileSync(join(keysDir, "agent1-key.pem"))); @@ -140,38 +141,35 @@ async function sessionThroughRelay() { ), ); setup.defer(() => void serverSession.promise.then(session => session.destroy())); - const relay = setup.use(await relayTo(server.address.port)); + const relay = setup.use(await relayTo(server.address!.port)); const statuses = new Map(); let onStatus = () => {}; let fromServer = 0; let onServerDatagram = () => {}; let onServerStream = () => {}; - const client = await connect( - { address: "127.0.0.1", port: relay.port }, - { - alpn: "quic-test", - servername: "localhost", - verifyPeer: "manual", - transportParams: { maxIdleTimeout: 10 }, - onerror: failed.reject, - ondatagram() { - fromServer++; - onServerDatagram(); - }, - ondatagramstatus(id: bigint, status: string) { - statuses.set(Number(id), [...(statuses.get(Number(id)) ?? []), status]); - onStatus(); - }, - async onstream(stream: any) { - stream.closed.catch(() => {}); - try { - for await (const _ of stream); - } catch {} - onServerStream(); - }, + const client = await connect(new SocketAddress({ address: "127.0.0.1", port: relay.port }), { + alpn: "quic-test", + servername: "localhost", + verifyPeer: "manual", + transportParams: { maxIdleTimeout: 10 }, + onerror: failed.reject, + ondatagram() { + fromServer++; + onServerDatagram(); }, - ); + ondatagramstatus(id: bigint, status: string) { + statuses.set(Number(id), [...(statuses.get(Number(id)) ?? []), status]); + onStatus(); + }, + async onstream(stream: any) { + stream.closed.catch(() => {}); + try { + for await (const _ of stream); + } catch {} + onServerStream(); + }, + } as SessionOptions); setup.defer(() => client.destroy()); failWith(client, failed); await client.opened; @@ -208,7 +206,7 @@ async function sessionThroughRelay() { /** The ids of the datagrams that the server received, in ascending order. */ received: () => received.toSorted((a, b) => a - b), /** Every status reported for each id: one per id when all is right. */ - statuses: () => Object.fromEntries(statuses), + statuses: (): Record => Object.fromEntries(statuses), /** * Round trips of two datagrams, because the server acknowledges every * second packet at once. After them the retransmission timers of the @@ -344,8 +342,7 @@ async function sessionThroughRelay() { /** The ids from `from` to `to`. */ const range = (from: number, to: number) => Array.from({ length: to - from + 1 }, (_, i) => from + i); /** `{ [id]: [status] }` for each id from `from` to `to`. */ -const all = (status: unknown, from: number, to: number) => - Object.fromEntries(range(from, to).map(id => [id, [status]])); +const all = (status: T, from: number, to: number) => Object.fromEntries(range(from, to).map(id => [id, [status]])); describe("ondatagramstatus", () => { test.each([ @@ -515,7 +512,7 @@ describe("ondatagramstatus", () => { { sni, alpn: ["quic-test"], ondatagram() {} }, ); const calls: string[] = []; - const client = await connect(server.address, { + const client = await connect(server.address!, { alpn: "quic-test", servername: "localhost", verifyPeer: "manual", @@ -527,7 +524,7 @@ describe("ondatagramstatus", () => { // and so one acknowledgement, for datagrams 2, 3 and 4. for (let i = 0; i < 3; i++) client.sendDatagram(new Uint8Array(SMALL)); }, - }); + } as SessionOptions); await client.opened; await client.sendDatagram(new Uint8Array(SMALL)); @@ -560,12 +557,12 @@ describe("ondatagramstatus", () => { // A first server hands out a session ticket and a token. const first = cleanup.use(await listen(onSession, { sni, alpn: ["quic-test"], endpoint: { tokenSecret } })); - const firstClient = await connect(first.address, { + const firstClient = await connect(first.address!, { alpn: "quic-test", verifyPeer: "manual", onsessionticket: ticket.resolve, onnewtoken: token.resolve, - }); + } as SessionOptions); cleanup.defer(() => firstClient.destroy()); failWith(firstClient, ticket, token); await Promise.all([firstClient.opened, ticket.promise, token.promise]); @@ -596,7 +593,7 @@ describe("ondatagramstatus", () => { ); const statuses = new Map(); const reported = Promise.withResolvers(); - const client = await connect(second.address, { + const client = await connect(second.address!, { alpn: "quic-test", verifyPeer: "manual", sessionTicket: await ticket.promise, @@ -607,7 +604,7 @@ describe("ondatagramstatus", () => { statuses.set(Number(id), [...(statuses.get(Number(id)) ?? []), status]); onProgress(); }, - }); + } as SessionOptions); cleanup.defer(() => client.destroy()); failWith(client, reported); From 0840e1672385a3e78851e2d59bb3fee6a8cce999 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 18 Oct 2026 02:50:23 +0000 Subject: [PATCH 5/5] ci: retrigger