From fa8e30a92d642bf9da30aa7973c7d0aae1f22887 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:59:19 +0000 Subject: [PATCH 1/3] node:quic: create the endpoint socket without IP_RECVERR and count a refused datagram as sent A UDP send error that belongs to one destination stopped the sends of every session on the endpoint. `packets_out` reported each errno except EMSGSIZE to lsquic as EAGAIN. lsquic reads that as a blocked socket: it pauses the engine, and at the next tick it drops each waiting packet that holds only DATAGRAM frames. - The endpoint passes no receive-error callback, so its socket never gets IP_RECVERR. An ICMP error about one peer no longer fails a send to another peer. `uws::udp::Socket::create` takes the callback as an `Option`. - `packets_out` counts a datagram that the kernel refuses as sent, which lsquic sees as a packet lost on the wire. Only a socket that would block pauses the engine. EMSGSIZE still reaches lsquic. - With no receive-error callback, uSockets closes the socket when a receive fails. `on_close` now ends each session with no packet and closes the endpoint with RECEIVE_FAILURE. `packets_out` with no socket reports the batch as sent, and the close does not wait for the connections that an engine still holds. - `QuicEndpoint.closing` and the endpoint reuse of `connect()` also read the native closing state. A `connect()` after a receive error uses another endpoint. - `bsd_sendmmsg` and `bsd_recvmmsg` get the fault-injection hooks that `bsd_sendmsg` and `bsd_recvmsg` have, so a test can make a UDP send or receive fail. Release builds compile them out. --- packages/bun-usockets/src/bsd.c | 6 + src/js/internal/quic/quic.ts | 16 +- src/runtime/node/quic/endpoint.rs | 94 ++++- src/runtime/node/quic/session.rs | 9 + src/runtime/socket/udp_socket.rs | 4 +- src/uws_sys/udp.rs | 12 +- test/js/node/quic/quic-endpoint.test.ts | 536 +++++++++++++++++++++++- 7 files changed, 648 insertions(+), 29 deletions(-) diff --git a/packages/bun-usockets/src/bsd.c b/packages/bun-usockets/src/bsd.c index a1190b3a6515..1d6d4aa69c75 100644 --- a/packages/bun-usockets/src/bsd.c +++ b/packages/bun-usockets/src/bsd.c @@ -83,6 +83,9 @@ extern void uv__winsock_ensure(void); /* We need to emulate sendmmsg, recvmmsg on platform who don't have it */ int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int flags) { + ssize_t injected = 0; int unused = 0; + if (US_FAULT_CHECK(US_FAULT_SENDMSG, fd, injected, unused)) return (int) injected; + (void)unused; #if defined(_WIN32)// || defined(__APPLE__) for (int i = 0; i < sendbuf->num; i++) { while (1) { @@ -152,6 +155,9 @@ int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int fl } int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags, int max_packets) { + ssize_t injected = 0; int unused = 0; + if (US_FAULT_CHECK(US_FAULT_RECVMSG, fd, injected, unused)) return (int) injected; + (void)unused; if (max_packets > LIBUS_UDP_RECV_COUNT) max_packets = LIBUS_UDP_RECV_COUNT; #if defined(_WIN32) for (int i = 0; i < LIBUS_UDP_RECV_COUNT; i++) { diff --git a/src/js/internal/quic/quic.ts b/src/js/internal/quic/quic.ts index dc474e25e4f3..4da05c95eda6 100644 --- a/src/js/internal/quic/quic.ts +++ b/src/js/internal/quic/quic.ts @@ -4470,8 +4470,14 @@ class QuicEndpoint { return this.#inner.stats; } + // The native side also starts a close: a receive error ends the endpoint. + get #isClosing() { + const inner = this.#inner; + return inner.isPendingClose || inner.state.isClosing === true; + } + get #isClosedOrClosing() { - return this.destroyed || this.#inner.isPendingClose; + return this.destroyed || this.#isClosing; } /** @@ -4687,7 +4693,7 @@ class QuicEndpoint { */ get closing() { assertIsQuicEndpoint(this); - return this.#inner.isPendingClose; + return this.#isClosing; } /** @type {boolean} */ @@ -4846,7 +4852,7 @@ class QuicEndpoint { [kRemoveSession](session) { const inner = this.#inner; inner.sessions.delete(session); - if (inner.sessions.size !== 0 || inner.listening || this.destroyed || inner.isPendingClose) { + if (inner.sessions.size !== 0 || inner.listening || this.destroyed || this.#isClosing) { return; } // Node parity: idle endpoints are unref'd, not destroyed — see test-quic-endpoint-idle-timeout. @@ -4873,14 +4879,14 @@ class QuicEndpoint { depth: options.depth == null ? null : options.depth - 1, }; - const { address, busy, isPendingClose: closing, listening, sessions, stats, state } = this.#inner; + const { address, busy, listening, sessions, stats, state } = this.#inner; return `QuicEndpoint ${inspect( { address, busy, closed: this.closed, - closing, + closing: this.#isClosing, destroyed: this.destroyed, listening, sessions, diff --git a/src/runtime/node/quic/endpoint.rs b/src/runtime/node/quic/endpoint.rs index 9a9577fb313a..9b0071c9c775 100644 --- a/src/runtime/node/quic/endpoint.rs +++ b/src/runtime/node/quic/endpoint.rs @@ -244,6 +244,8 @@ pub(crate) struct QuicEndpoint { block_list_allow: Cell, pub(crate) event_loop_timer: JsCell, pending_endpoint_close: Cell, + /// What the deferred `onEndpointClose` reports. + close_context: Cell, /// Intrusive node on the loop's node:quic driver list; linked only while /// this endpoint holds a socket. `us_nq_loop_flush_if_pending` runs the /// process pass once per loop turn when `pending` is set. @@ -395,8 +397,16 @@ extern "C" fn on_drain(socket: *mut uws::udp::Socket) { } this.schedule_process(); } -extern "C" fn on_close(_socket: *mut uws::udp::Socket) {} -extern "C" fn on_recv_error(_socket: *mut uws::udp::Socket, _errno: c_int, _is_errqueue: c_int) {} +extern "C" fn on_close(socket: *mut uws::udp::Socket) { + let user = uws::udp::Socket::opaque_mut(socket).user(); + if user.is_null() { + return; + } + // SAFETY: `user` is the QuicEndpoint that owns this socket. It closes the + // socket in `release_native` before it is freed, so it is alive here. + let this = unsafe { &*user.cast::() }; + this.on_socket_closed(socket); +} thread_local! { /// A server may advertise another local endpoint's address as its @@ -422,6 +432,36 @@ fn registry_find_by_addr(addr: &StoredAddr, not: *const QuicEndpoint) -> Option< }) } +/// `packets_out` has a datagram that the socket did not send. Returns whether +/// the batch stops there, with `errno` set to what lsquic reads from a short +/// return. `us_quic_packets_out` (quic.c) is the same verdict for HTTP/3: +/// change both together. +/// +/// - `rv == 0`: the socket would block, and uSockets armed WRITABLE for it. +/// EAGAIN pauses the engine until `on_drain`. +/// - EMSGSIZE: lsquic takes the packet back (ci_packet_too_large). An MTU +/// probe feeds DPLPMTUD, and any other packet ends its connection. +/// - Any other errno: the kernel refused this one datagram (EINVAL, EACCES, +/// EPERM, ENETUNREACH, ...). lsquic has no outcome for that: a short return +/// pauses every session of the engine or closes one. So the datagram counts +/// as sent, which lsquic sees as a packet lost on the wire. Node drops it +/// too: Endpoint::PacketDone ignores the status of a send +/// (node/src/quic/endpoint.cc). ENOBUFS is in this group while uSockets +/// reports it as an error: WRITABLE is armed only for `rv == 0`, so nothing +/// would end a pause (quic.c arms the poll itself). +#[cold] +#[inline(never)] +fn unsent_stops_batch(rv: c_int) -> bool { + let e = bun_core::ffi::errno_ptr(); + if rv == 0 { + // SAFETY: `errno_ptr()` is this thread's errno slot. + unsafe { *e = libc::EAGAIN }; + return true; + } + // SAFETY: as above. + unsafe { *e == libc::EMSGSIZE } +} + /// Header byte 0 bit 7 (RFC 8999 sec 5.1): 1 = long header, 0 = short. const HEADER_FORM_LONG: u8 = 0x80; /// Minimum stateless reset size (RFC 9000 sec 10.3: header byte + 4+ random @@ -713,7 +753,9 @@ lsquic_callback! { n: c_uint, ) -> c_int = 0; { let Some(socket) = this.socket.get() else { - return 0; + // uSockets closed the socket (`on_socket_closed`). Nothing can + // leave, and a short return would make lsquic read a stale errno. + return n as c_int; }; let my_addr = this.local_addr.get(); // SAFETY: pure constant query. @@ -789,17 +831,7 @@ lsquic_callback! { &[total], &[dest.as_ptr().cast()], ); - if rv < 1 { - // SAFETY: `errno_ptr()` is this thread's errno slot. - unsafe { - let e = bun_core::ffi::errno_ptr(); - // EMSGSIZE has to survive: it is how lsquic learns to drop - // an oversized packet and feed DPLPMTUD (ci_packet_too_large). - // Anything else it cannot act on becomes backpressure. - if *e != libc::EAGAIN && *e != libc::EWOULDBLOCK && *e != libc::EMSGSIZE { - *e = libc::EAGAIN; - } - } + if rv < 1 && unsent_stops_batch(rv) { break; } this.add_stat(IDX_STATS_PACKETS_SENT, 1); @@ -1162,6 +1194,7 @@ impl QuicEndpoint { EventLoopTimerTag::QuicEndpoint, )), pending_endpoint_close: Cell::new(false), + close_context: Cell::new(CLOSECONTEXT_CLOSE), nq_driver: JsCell::new(UsNqDriver::default()), nq_registered: Cell::new(false), defer_closes: Cell::new(false), @@ -1373,7 +1406,7 @@ impl QuicEndpoint { on_data, on_drain, on_close, - on_recv_error, + None, cfg.host.as_ptr().cast(), cfg.port, 0, @@ -1530,9 +1563,11 @@ impl QuicEndpoint { self.sweep_provisional(); self.rearm_timer(); self.update_keepalive(); + // With no socket the engines can send nothing, so the conns they still + // hold (handshakes with no session yet) are not waited for. if self.closing.get() && self.sessions.get().is_empty() - && self.engine_conn_count() == 0 + && (self.engine_conn_count() == 0 || self.socket.get().is_none()) && !self.closed.get() { self.finish_close(); @@ -1647,7 +1682,7 @@ impl QuicEndpoint { // SAFETY: as in `on_data`. let global = unsafe { &*global_ptr }; if this_ref.pending_endpoint_close.replace(false) { - this_ref.deliver_endpoint_close(global, CLOSECONTEXT_CLOSE, 0); + this_ref.deliver_endpoint_close(global, this_ref.close_context.get(), 0); return; } this_ref.process(global); @@ -2118,6 +2153,31 @@ impl QuicEndpoint { true } + /// uSockets closed the socket: a receive or a poll failed, and there is no + /// receive-error callback. Node's `Endpoint::Destroy(RECEIVE_FAILURE)` + /// (node/src/quic/endpoint.cc): each session ends with no packet sent, + /// then the endpoint closes. + fn on_socket_closed(&self, socket: *mut uws::udp::Socket) { + // `release_native` takes the handle before it closes the socket. + if self.socket.get() != Some(socket) { + return; + } + self.socket.set(None); + self.close_context.set(CLOSECONTEXT_RECEIVE_FAILURE); + self.closing.set(true); + self.with_state(|s| { + s.closing = 1; + s.listening = 0; + }); + for session in self.sessions.get().clone() { + if let Some(session) = self.live_session(session) { + session.close_silently(); + } + } + // `process` delivers each session's close to JS, then `finish_close`. + self.schedule_process(); + } + fn finish_close(&self) { if !self.teardown() { return; diff --git a/src/runtime/node/quic/session.rs b/src/runtime/node/quic/session.rs index 0ef2b1a94882..ae2662253b85 100644 --- a/src/runtime/node/quic/session.rs +++ b/src/runtime/node/quic/session.rs @@ -1967,6 +1967,15 @@ impl QuicSession { }) } + /// Node's `Session::Close(CloseMethod::SILENT)`: no CONNECTION_CLOSE + /// leaves, and JS sees a close with no error. + pub(super) fn close_silently(&self) { + match self.conn() { + Some(conn) => conn.abort_silent(), + None => self.push_event(SessionEvent::Closed), + } + } + pub(super) fn maybe_finish_deferred_close(&self) { if self.deferred_close.get().is_none() || self.destroyed.get() { return; diff --git a/src/runtime/socket/udp_socket.rs b/src/runtime/socket/udp_socket.rs index d919f2af5184..088a220e9cb7 100644 --- a/src/runtime/socket/udp_socket.rs +++ b/src/runtime/socket/udp_socket.rs @@ -707,7 +707,7 @@ impl UDPSocket { on_data, on_drain, on_close, - on_recv_error, + Some(on_recv_error), fd, config.shared_fd, Some(&mut err), @@ -719,7 +719,7 @@ impl UDPSocket { on_data, on_drain, on_close, - on_recv_error, + Some(on_recv_error), hostname_z.as_ptr(), config.port, config.flags, diff --git a/src/uws_sys/udp.rs b/src/uws_sys/udp.rs index dfe2b06247c5..a465c39c7446 100644 --- a/src/uws_sys/udp.rs +++ b/src/uws_sys/udp.rs @@ -14,12 +14,16 @@ bun_opaque::opaque_ffi! { } impl Socket { + /// Without `recv_error_cb` the socket never gets `IP_RECVERR` (Linux), so + /// an ICMP error about one peer cannot fail a send to another. A failing + /// receive then has no handler: uSockets closes the socket and calls + /// `close_cb`, which must drop the handle. pub fn create( loop_: *mut Loop, data_cb: extern "C" fn(*mut Socket, *mut PacketBuffer, c_int), drain_cb: extern "C" fn(*mut Socket), close_cb: extern "C" fn(*mut Socket), - recv_error_cb: extern "C" fn(*mut Socket, c_int, c_int), + recv_error_cb: Option, host: *const c_char, port: c_ushort, options: c_int, @@ -54,7 +58,7 @@ impl Socket { data_cb: extern "C" fn(*mut Socket, *mut PacketBuffer, c_int), drain_cb: extern "C" fn(*mut Socket), close_cb: extern "C" fn(*mut Socket), - recv_error_cb: extern "C" fn(*mut Socket, c_int, c_int), + recv_error_cb: Option, fd: c_int, shared: bool, err: Option<&mut c_int>, @@ -201,7 +205,7 @@ unsafe extern "C" { data_cb: extern "C" fn(*mut Socket, *mut PacketBuffer, c_int), drain_cb: extern "C" fn(*mut Socket), close_cb: extern "C" fn(*mut Socket), - recv_error_cb: extern "C" fn(*mut Socket, c_int, c_int), + recv_error_cb: Option, host: *const c_char, port: c_ushort, options: c_int, @@ -235,7 +239,7 @@ unsafe extern "C" { data_cb: extern "C" fn(*mut Socket, *mut PacketBuffer, c_int), drain_cb: extern "C" fn(*mut Socket), close_cb: extern "C" fn(*mut Socket), - recv_error_cb: extern "C" fn(*mut Socket, c_int, c_int), + recv_error_cb: Option, fd: LIBUS_SOCKET_DESCRIPTOR, shared: c_int, err: *mut c_int, diff --git a/test/js/node/quic/quic-endpoint.test.ts b/test/js/node/quic/quic-endpoint.test.ts index edaced224456..49d15883b866 100644 --- a/test/js/node/quic/quic-endpoint.test.ts +++ b/test/js/node/quic/quic-endpoint.test.ts @@ -1,12 +1,15 @@ // lsquic fixes HTTP/3-vs-raw framing per client *engine*, set by the first // connect() through an endpoint; a later connect in the other mode must fail // loudly instead of silently reusing an engine that cannot frame it. +import { socketFaultInjection as fault } from "bun:internal-for-testing"; import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, normalizeBunSnapshot, tempDir } from "harness"; +import { bunEnv, bunExe, isLinux, isWindows, normalizeBunSnapshot, tempDir } from "harness"; import { createPrivateKey } from "node:crypto"; import { createSocket } from "node:dgram"; +import { once } from "node:events"; import { readFileSync } from "node:fs"; import { BlockList } from "node:net"; +import { constants } from "node:os"; import { join } from "node:path"; import { connect, listen, QuicEndpoint } from "node:quic"; @@ -310,3 +313,534 @@ describe("endpoint.close() while a session is live", () => { expect({ announced, resolved }).toEqual({ announced: 1, resolved: true }); }); }); + +// lsquic reads a short return of the endpoint's send callback as "the socket +// is blocked". It then stops every session of the engine, and at its next +// tick it drops each waiting packet that holds only DATAGRAM frames. So an +// error that belongs to one destination must not reach lsquic that way, and +// the socket must not report the error of one peer on a send to another. +describe("an error on the endpoint's socket", () => { + const tp = { maxIdleTimeout: 30 }; + const sni = { "*": { keys: [key], certs: [cert] } }; + const clientOptions = { alpn: "quic-test", servername: "localhost", verifyPeer: "manual", transportParams: tp }; + const loopTurn = () => new Promise(resolve => setImmediate(resolve)); + const ignoreSettlement = (session: any) => { + session.opened.catch(() => {}); + session.closed.catch(() => {}); + return session; + }; + /** Ends the sessions at once. A graceful close waits for a peer that cannot answer. */ + const destroyAtExit = (...sessions: any[][]) => ({ + [Symbol.dispose]() { + for (const session of sessions.flat()) session.destroy(); + }, + }); + /** Writes how `closed` settles into `order`. */ + const logClose = (order: string[], name: string, closed: Promise) => + closed.then( + () => void order.push(`${name} closed`), + (error: any) => void order.push(`${name} closed with ${error.code}`), + ); + + /** The server answers each stream with one byte. */ + const onServerSession = (sessions: any[]) => (session: any) => { + session.onerror = () => {}; + session.closed.catch(() => {}); + session.onstream = (stream: any) => { + stream.closed.catch(() => {}); + stream.writer.writeSync(new Uint8Array(1)); + stream.writer.endSync(); + }; + sessions.push(session); + }; + + /** + * A stream round trip. When it ends, each side has every packet that the + * other sent before it. Returns the number of bytes in the answer. + */ + async function roundTrip(session: any) { + const stream = await session.createBidirectionalStream({ body: new Uint8Array(1) }); + stream.closed.catch(() => {}); + let answered = 0; + for await (const batch of stream) for (const chunk of [batch].flat()) answered += chunk.byteLength; + return answered; + } + + /** Resolves when `session` gets the answer to a stream. It stays pending when the session ends first. */ + const answered = (session: any) => + new Promise( + resolve => + void roundTrip(session).then( + bytes => bytes === 1 && resolve("answered"), + () => {}, + ), + ); + + /** A UDP port of 127.0.0.1 that nothing is bound to. */ + async function closedPort() { + const socket = createSocket("udp4"); + socket.bind(0, "127.0.0.1"); + await once(socket, "listening"); + const { port } = socket.address(); + await new Promise(resolve => socket.close(() => resolve())); + return port; + } + + /** A UDP relay in front of the server. `cut()` closes the socket that the server sends to. */ + async function relayTo(serverPort: number) { + const front = createSocket("udp4"); + const upstream = createSocket("udp4"); + let clientPort = 0; + let cut = false; + upstream.on("message", packet => front.send(packet, clientPort, "127.0.0.1")); + front.on("message", (packet, from) => { + clientPort = from.port; + if (!cut) 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")]); + return { + port: front.address().port, + /** The kernel answers each later packet of the server with ICMP port unreachable. */ + cut() { + cut = true; + upstream.close(); + }, + [Symbol.dispose]() { + front.close(); + if (!cut) upstream.close(); + }, + }; + } + + // The endpoint is bound to 127.0.0.1. An IPv6 destination cannot leave its + // IPv4 socket on any platform. Linux refuses the other two with EINVAL and + // EACCES. + const refused = ["[::1]:443", ...(isLinux ? ["192.0.2.1:443", "255.255.255.255:443"] : [])]; + + test.each(refused)("a datagram that the kernel refuses for %s does not stop another session", async address => { + const received: number[] = []; + const serverSessions: any[] = []; + const clientSessions: any[] = []; + await using server = await listen(onServerSession(serverSessions), { + sni, + alpn: ["quic-test"], + transportParams: tp, + ondatagram(datagram: Uint8Array) { + received.push(datagram[0]); + }, + }); + await using endpoint = new QuicEndpoint(); + using _sessions = destroyAtExit(clientSessions, serverSessions); + const live = await connect(server.address, { ...clientOptions, endpoint }); + clientSessions.push(live); + await live.opened; + const refusedSession = ignoreSettlement(await connect(address, { ...clientOptions, endpoint })); + clientSessions.push(refusedSession); + + // A loop turn after each datagram, so each one is a send pass of its own. + const sent = Array.from({ length: 10 }, (_, id) => id); + for (const id of sent) { + await live.sendDatagram(new Uint8Array([id])); + await loopTurn(); + } + // The session that cannot send goes first. Without it the round trip + // ends on an endpoint that held the live session back too. + refusedSession.destroy(); + await roundTrip(live); + + expect(received).toEqual(sent); + }); + + test("a session opens and exchanges a stream while an earlier session of the endpoint cannot send", async () => { + const serverSessions: any[] = []; + const clientSessions: any[] = []; + await using server = await listen(onServerSession(serverSessions), { + sni, + alpn: ["quic-test"], + transportParams: tp, + }); + await using endpoint = new QuicEndpoint(); + using _sessions = destroyAtExit(clientSessions, serverSessions); + const refusedSession = ignoreSettlement(await connect("[::1]:443", { ...clientOptions, endpoint })); + clientSessions.push(refusedSession); + + const live = await connect(server.address, { ...clientOptions, endpoint }); + clientSessions.push(live); + await live.opened; + + expect({ answered: await roundTrip(live), refusedIsPending: !refusedSession.destroyed }).toEqual({ + answered: 1, + refusedIsPending: true, + }); + }); + + // Linux only: a socket with IP_RECVERR gets the ICMP error about one peer + // on its next send, whatever the destination of that send is. + describe.skipIf(!isLinux)("an ICMP error about a dead peer", () => { + test.each([2, 3])("does not fail the sends to a live peer (%d dead peers)", async dead => { + const received: number[] = []; + const serverSessions: any[] = []; + const clientSessions: any[] = []; + let onRound = () => {}; + await using server = await listen(onServerSession(serverSessions), { + sni, + alpn: ["quic-test"], + transportParams: tp, + // Only the live client sends. The answers for all sessions leave in one pass. + ondatagram(datagram: Uint8Array) { + for (const session of serverSessions) session.sendDatagram(datagram); + onRound(); + }, + }); + await using endpoint = new QuicEndpoint(); + using _sessions = destroyAtExit(clientSessions, serverSessions); + using relays = new DisposableStack(); + + const live = await connect(server.address, { + ...clientOptions, + endpoint, + ondatagram(datagram: Uint8Array) { + received.push(datagram[0]); + }, + }); + clientSessions.push(live); + await live.opened; + const cuts: (() => void)[] = []; + for (let i = 0; i < dead; i++) { + const relay = relays.use(await relayTo(server.address.port)); + const session = ignoreSettlement( + await connect({ address: "127.0.0.1", port: relay.port }, { ...clientOptions, endpoint }), + ); + clientSessions.push(session); + await session.opened; + cuts.push(relay.cut); + } + // The server announces a session when its first packet arrives. + const [, ...sessionsOfDeadPeers] = serverSessions; + expect(sessionsOfDeadPeers).toHaveLength(dead); + for (const cut of cuts) cut(); + + const rounds = Array.from({ length: 10 }, (_, round) => round); + for (const round of rounds) { + const handled = new Promise(resolve => (onRound = resolve)); + await live.sendDatagram(new Uint8Array([round])); + await handled; + await loopTurn(); + } + // The sessions with a dead peer go first. Without them no new error + // arrives, so the round trip ends on an endpoint that lost datagrams too. + for (const session of sessionsOfDeadPeers) session.destroy(); + await roundTrip(live); + + expect(received).toEqual(rounds); + }); + + test("does not eat the version negotiation probe of another session", async () => { + const serverSessions: any[] = []; + const clientSessions: any[] = []; + await using server = await listen(onServerSession(serverSessions), { + sni, + alpn: ["quic-test"], + transportParams: tp, + }); + const deadPort = await closedPort(); + await using endpoint = new QuicEndpoint(); + using _sessions = destroyAtExit(clientSessions, serverSessions); + + // Both leave in one turn, so the error about the first is still pending + // on the socket when the probe of the second is sent. + const toDeadPeer = connect({ address: "127.0.0.1", port: deadPort }, { ...clientOptions, endpoint }); + const probing = connect(server.address, { ...clientOptions, endpoint, version: 0x1a1a1a1a, onerror() {} }); + clientSessions.push(ignoreSettlement(await toDeadPeer)); + const session = await probing; + clientSessions.push(session); + session.opened.catch(() => {}); + + expect( + await session.closed.then( + () => "closed", + (error: any) => error.code, + ), + ).toBe("ERR_QUIC_VERSION_NEGOTIATION_ERROR"); + }); + }); + + // These need the fault hooks of a debug or ASAN build. A rule has no fd + // here, so it hits the next send or receive of a UDP socket in the process. + // On Windows an injected errno does not pass through the Winsock mapping. + describe.skipIf(!fault.available() || isWindows)("injected in uSockets", () => { + /** One server, and `count` open sessions to it on one client endpoint. */ + async function connected(count: number, serverOptions: object = {}, sessionOptions: object = {}) { + const stack = new AsyncDisposableStack(); + stack.defer(() => fault.clear()); + const serverSessions: any[] = []; + const clientSessions: any[] = []; + const server = await listen(onServerSession(serverSessions), { + sni, + alpn: ["quic-test"], + transportParams: tp, + ...serverOptions, + }); + // After a receive error the endpoint is closed, and `close()` rejects. + stack.defer(() => server.close().catch(() => {})); + const endpoint = new QuicEndpoint(); + stack.defer(() => endpoint.close().catch(() => {})); + stack.use(destroyAtExit(clientSessions, serverSessions)); + for (let i = 0; i < count; i++) { + const session = ignoreSettlement( + await connect(server.address, { ...clientOptions, endpoint, ...sessionOptions }), + ); + clientSessions.push(session); + await session.opened; + } + return { server, endpoint, serverSessions, clientSessions, [Symbol.asyncDispose]: () => stack.disposeAsync() }; + } + const receiveError = { syscall: "recvmsg", action: "errno", errno: "ENOMEM" } as const; + + // Both endpoints are in this process, and the loopback loses nothing. So + // the packets that one endpoint counts and the other never gets are the + // failed sends, plus the few that are still on their way. + test.each([ + ["the socket would block", { action: "errno", errno: "EAGAIN", repeat: 5 }, false], + ["a send returns 0", { action: "zero", repeat: 5 }, false], + ["the kernel refuses 3 sends", { action: "errno", errno: "ENETUNREACH", repeat: 3 }, true], + ] as const)("a stream round trip completes when %s", async (_case, rule, countedAsSent) => { + await using net = await connected(1); + const [session] = net.clientSessions; + /** The packets that `from` counts as sent and `to` did not receive. */ + const missing = (from: any, to: any) => Number(from.stats.packetsSent - to.stats.packetsReceived); + + fault.set({ syscall: "sendmsg", ...rule }); + // Two datagrams take the first failures. lsquic does not send a datagram + // again, so the stream data waits for one retransmission, not for three. + for (const id of [0, 1]) { + await session.sendDatagram(new Uint8Array([id])); + await loopTurn(); + } + const answered = await roundTrip(session); + const neverArrived = missing(net.endpoint, net.server) + missing(net.server, net.endpoint); + + expect({ answered, failedSendsCountedAsSent: neverArrived >= rule.repeat }).toEqual({ + answered: 1, + failedSendsCountedAsSent: countedAsSent, + }); + }); + + /** + * Two sessions on a `listen()` endpoint. The server answers a datagram of + * the first client with one datagram to each session, in one pass, and the + * first send of the first pass fails with `errno`. + */ + async function twoSessionsOneFailedSend(errno: "ENETUNREACH" | number) { + const received: number[][] = [[], []]; + const gotSecond = [Promise.withResolvers(), Promise.withResolvers()]; + const firstHandled = Promise.withResolvers(); + const net = await connected( + 2, + { + ondatagram(datagram: Uint8Array) { + if (datagram[0] === 0) fault.set({ syscall: "sendmsg", action: "errno", errno }); + for (const session of net.serverSessions) if (!session.destroyed) session.sendDatagram(datagram); + firstHandled.resolve(); + }, + }, + { + ondatagram(this: any, datagram: Uint8Array) { + const index = net.clientSessions.indexOf(this); + received[index].push(datagram[0]); + if (datagram[0] === 1) gotSecond[index].resolve(); + }, + }, + ); + await net.clientSessions[0].sendDatagram(new Uint8Array([0])); + await firstHandled.promise; + await loopTurn(); + return { net, received, gotSecond: gotSecond.map(second => second.promise) }; + } + + test("on a listen() endpoint, a refused datagram costs only that datagram", async () => { + const { net, received, gotSecond } = await twoSessionsOneFailedSend("ENETUNREACH"); + await using _net = net; + + await net.clientSessions[0].sendDatagram(new Uint8Array([1])); + await Promise.all(gotSecond); + + expect(received.toSorted((a, b) => a.length - b.length)).toEqual([[1], [0, 1]]); + }); + + // lsquic takes EMSGSIZE for a packet that is not an MTU probe as fatal for + // the connection that owns it. The rest of the batch still leaves. + test("on a listen() endpoint, EMSGSIZE ends only the session that owns the packet", async () => { + const { net, received } = await twoSessionsOneFailedSend(constants.errno.EMSGSIZE); + await using _net = net; + + const answers = await Promise.allSettled(net.clientSessions.map(roundTrip)); + const ended = answers.findIndex(answer => answer.status !== "fulfilled" || answer.value !== 1); + const other = ended === 0 ? 1 : 0; + + expect({ + endedOnTheServer: await net.serverSessions[ended]?.closed.then( + () => "closed", + (error: any) => error.code, + ), + endedGot: received[ended], + otherGot: received[other], + otherAnswered: answers[other], + }).toEqual({ + endedOnTheServer: "ERR_QUIC_TRANSPORT_ERROR", + endedGot: [], + otherGot: [0], + otherAnswered: { status: "fulfilled", value: 1 }, + }); + }); + + // The socket has no receive-error callback, so uSockets closes it when a + // receive fails. Node destroys the endpoint then: its sessions end with no + // packet, and `closed` rejects. + test("a receive error ends each session, then closes the endpoint", async () => { + const order: string[] = []; + const armed = Promise.withResolvers(); + await using net = await connected(1, { + // This runs inside the receive loop of the server's socket, so the + // next receive on that socket is the one that fails. + ondatagram() { + fault.set(receiveError); + armed.resolve(); + }, + }); + logClose(order, "session", net.serverSessions[0].closed); + const endpointClosed = logClose(order, "endpoint", net.server.closed); + + await net.clientSessions[0].sendDatagram(new Uint8Array(1)); + await armed.promise; + // A server that is still open answers this stream, and that ends the wait. + await Promise.race([endpointClosed, answered(net.clientSessions[0])]); + + expect({ order, destroyed: net.server.destroyed }).toEqual({ + order: ["session closed", "endpoint closed with ERR_QUIC_ENDPOINT_CLOSED"], + destroyed: true, + }); + }); + + // The engine still holds the connection of the handshake, and the session + // has no connection yet. The endpoint waits for neither. + test("a receive error during a handshake ends the announced session, then closes the endpoint", async () => { + const order: string[] = []; + const serverSessions: any[] = []; + using _fault = { [Symbol.dispose]: () => fault.clear() }; + const server = await listen( + (session: any) => { + // Inside the receive loop of the server's socket, as above. + fault.set(receiveError); + serverSessions.push(session); + session.onerror = () => {}; + session.opened.then( + () => void order.push("session opened"), + (error: any) => void order.push(`session opened rejected with ${error.code}`), + ); + logClose(order, "session", session.closed); + }, + { sni, alpn: ["quic-test"], transportParams: tp }, + ); + await using _server = { [Symbol.asyncDispose]: () => server.close().catch(() => {}) }; + const endpointClosed = logClose(order, "endpoint", server.closed); + await using endpoint = new QuicEndpoint(); + const client = ignoreSettlement(await connect(server.address, { ...clientOptions, endpoint })); + using _sessions = destroyAtExit([client], serverSessions); + + // A server that is still open completes the handshake, and that ends the wait. + await Promise.race([endpointClosed, client.opened.catch(() => {})]); + + expect({ order, destroyed: server.destroyed }).toEqual({ + order: [ + "session opened rejected with ERR_INVALID_STATE", + "session closed", + "endpoint closed with ERR_QUIC_ENDPOINT_CLOSED", + ], + destroyed: true, + }); + }); + + test("a receive error on a client endpoint also ends a session that waits for version negotiation", async () => { + const order: string[] = []; + const armed = Promise.withResolvers(); + await using net = await connected( + 1, + {}, + { + // Inside the receive loop of the client's socket. + ondatagram() { + fault.set(receiveError); + armed.resolve(); + }, + }, + ); + const [live] = net.clientSessions; + // Nothing answers this probe, so the session has no connection. + const probe = ignoreSettlement( + await connect( + { address: "127.0.0.1", port: await closedPort() }, + { ...clientOptions, endpoint: net.endpoint, version: 0x1a1a1a1a, onerror() {} }, + ), + ); + net.clientSessions.push(probe); + const sessionsClosed = Promise.all([ + logClose(order, "live", live.closed), + logClose(order, "probe", probe.closed), + ]); + const endpointClosed = logClose(order, "endpoint", net.endpoint.closed); + + await net.serverSessions[0].sendDatagram(new Uint8Array(1)); + await armed.promise; + // A client that is still open gets the answer to this stream. + await Promise.race([Promise.all([sessionsClosed, endpointClosed]), answered(live)]); + + expect({ sessions: order.slice(0, 2).toSorted(), then: order.slice(2) }).toEqual({ + sessions: ["live closed", "probe closed"], + then: ["endpoint closed with ERR_QUIC_ENDPOINT_CLOSED"], + }); + }); + + // `connect()` with no `endpoint` option reuses an endpoint. It must not + // pick one that a receive error is closing. + test("a connect() after a receive error uses another endpoint", async () => { + const armed = Promise.withResolvers(); + const serverSessions: any[] = []; + const clientSessions: any[] = []; + using _fault = { [Symbol.dispose]: () => fault.clear() }; + await using server = await listen(onServerSession(serverSessions), { + sni, + alpn: ["quic-test"], + transportParams: tp, + }); + using _sessions = destroyAtExit(clientSessions, serverSessions); + const options = { + ...clientOptions, + ondatagram() { + fault.set(receiveError); + armed.resolve(); + }, + }; + const first = ignoreSettlement(await connect(server.address, options)); + clientSessions.push(first); + await first.opened; + const firstEndpoint = first.endpoint; + firstEndpoint.closed.catch(() => {}); + + await serverSessions[0].sendDatagram(new Uint8Array(1)); + await armed.promise; + // A client that is still open gets the answer to this stream. + const firstEnded = await Promise.race([first.closed.then(() => "closed"), answered(first)]); + const second = ignoreSettlement(await connect(server.address, clientOptions)); + clientSessions.push(second); + await second.opened; + + expect({ firstEnded, sameEndpoint: second.endpoint === firstEndpoint }).toEqual({ + firstEnded: "closed", + sameEndpoint: false, + }); + }); + }); +}); From 7108350f5c719ad8d9dc5d5f07c4a41dab145464 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:23:50 +0000 Subject: [PATCH 2/3] node:quic: take ENOBUFS for backpressure and wait for the socket A send that fails with ENOBUFS counted as sent, so lsquic saw a lost packet. ENOBUFS means that the output queue of the interface is full. It is common on macOS and it clears by itself. - `packets_out` reports ENOBUFS to lsquic as a blocked socket, as `us_quic_packets_out` does for HTTP/3. uSockets arms the writable poll only when a send would block, so the endpoint arms it with the new `us_udp_socket_poll_writable`. `on_drain` then sends the packets. - `bsd_sendmmsg` on Windows reports WSAENOBUFS as ENOBUFS and not as EIO. `Bun.udpSocket` reads the Winsock code, so its errors do not change. - The tests in quic-endpoint.test.ts have no type errors, and one `afterEach` clears the fault rules. --- packages/bun-usockets/src/bsd.c | 1 + packages/bun-usockets/src/libusockets.h | 3 ++ packages/bun-usockets/src/udp.c | 7 ++++ src/runtime/node/quic/endpoint.rs | 22 +++++++----- src/uws_sys/udp.rs | 7 ++++ test/js/node/quic/quic-endpoint.test.ts | 45 +++++++++++++------------ 6 files changed, 55 insertions(+), 30 deletions(-) diff --git a/packages/bun-usockets/src/bsd.c b/packages/bun-usockets/src/bsd.c index 1d6d4aa69c75..fd62e49708c9 100644 --- a/packages/bun-usockets/src/bsd.c +++ b/packages/bun-usockets/src/bsd.c @@ -108,6 +108,7 @@ int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int fl if (err == WSAEINTR) continue; switch (err) { case WSAEWOULDBLOCK: errno = EAGAIN; break; + case WSAENOBUFS: errno = ENOBUFS; break; case WSAEMSGSIZE: errno = EMSGSIZE; break; case WSAECONNREFUSED: errno = ECONNREFUSED; break; case WSAENETUNREACH: errno = ENETUNREACH; break; diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index bb78632bb6a0..4d02605608d8 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -209,6 +209,9 @@ void us_udp_buffer_set_packet_payload(struct us_udp_packet_buffer_t *send_buf, i int us_udp_socket_send(struct us_udp_socket_t *s, void** payloads, size_t* lengths, void** addresses, int num); +/* Calls drain_cb once the socket is writable. us_udp_socket_send does this itself when the send would block. */ +void us_udp_socket_poll_writable(struct us_udp_socket_t *s); + /* Allocates a packet buffer that is reuable per thread. Mutated by us_udp_socket_receive. */ struct us_udp_packet_buffer_t *us_create_udp_packet_buffer(); diff --git a/packages/bun-usockets/src/udp.c b/packages/bun-usockets/src/udp.c index d9a9bfc0fe14..992340990dc7 100644 --- a/packages/bun-usockets/src/udp.c +++ b/packages/bun-usockets/src/udp.c @@ -72,6 +72,13 @@ int us_udp_socket_send(struct us_udp_socket_t *s, void** payloads, size_t* lengt return total_sent; } +/* on_drain runs once the socket is writable. us_udp_socket_send asks for this + * itself when the send would block. A caller that takes another send error for + * backpressure (ENOBUFS) asks for it here. */ +void us_udp_socket_poll_writable(struct us_udp_socket_t *s) { + us_poll_change((struct us_poll_t *) s, s->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); +} + int us_udp_socket_bound_port(struct us_udp_socket_t *s) { /* An adopted-but-unbound descriptor cached port 0 at creation; the kernel * auto-binds on the first sendto(), so re-query then. */ diff --git a/src/runtime/node/quic/endpoint.rs b/src/runtime/node/quic/endpoint.rs index 9b0071c9c775..decae69b1c94 100644 --- a/src/runtime/node/quic/endpoint.rs +++ b/src/runtime/node/quic/endpoint.rs @@ -439,6 +439,9 @@ fn registry_find_by_addr(addr: &StoredAddr, not: *const QuicEndpoint) -> Option< /// /// - `rv == 0`: the socket would block, and uSockets armed WRITABLE for it. /// EAGAIN pauses the engine until `on_drain`. +/// - ENOBUFS: the output queue of the interface is full, which is common on +/// macOS. It clears by itself, so it is the same pause. uSockets reports it +/// as an error and arms nothing, so the endpoint arms `socket`, its own. /// - EMSGSIZE: lsquic takes the packet back (ci_packet_too_large). An MTU /// probe feeds DPLPMTUD, and any other packet ends its connection. /// - Any other errno: the kernel refused this one datagram (EINVAL, EACCES, @@ -446,20 +449,21 @@ fn registry_find_by_addr(addr: &StoredAddr, not: *const QuicEndpoint) -> Option< /// pauses every session of the engine or closes one. So the datagram counts /// as sent, which lsquic sees as a packet lost on the wire. Node drops it /// too: Endpoint::PacketDone ignores the status of a send -/// (node/src/quic/endpoint.cc). ENOBUFS is in this group while uSockets -/// reports it as an error: WRITABLE is armed only for `rv == 0`, so nothing -/// would end a pause (quic.c arms the poll itself). +/// (node/src/quic/endpoint.cc). #[cold] #[inline(never)] -fn unsent_stops_batch(rv: c_int) -> bool { +fn unsent_stops_batch(socket: *mut uws::udp::Socket, rv: c_int) -> bool { let e = bun_core::ffi::errno_ptr(); - if rv == 0 { + if rv < 0 { // SAFETY: `errno_ptr()` is this thread's errno slot. - unsafe { *e = libc::EAGAIN }; - return true; + match unsafe { *e } { + libc::ENOBUFS => uws::udp::Socket::opaque_mut(socket).poll_writable(), + errno => return errno == libc::EMSGSIZE, + } } // SAFETY: as above. - unsafe { *e == libc::EMSGSIZE } + unsafe { *e = libc::EAGAIN }; + true } /// Header byte 0 bit 7 (RFC 8999 sec 5.1): 1 = long header, 0 = short. @@ -831,7 +835,7 @@ lsquic_callback! { &[total], &[dest.as_ptr().cast()], ); - if rv < 1 && unsent_stops_batch(rv) { + if rv < 1 && unsent_stops_batch(socket, rv) { break; } this.add_stat(IDX_STATS_PACKETS_SENT, 1); diff --git a/src/uws_sys/udp.rs b/src/uws_sys/udp.rs index a465c39c7446..e3fdea6aa402 100644 --- a/src/uws_sys/udp.rs +++ b/src/uws_sys/udp.rs @@ -105,6 +105,12 @@ impl Socket { } } + /// `drain_cb` runs once the socket is writable. `send` asks for this itself + /// when the send would block. + pub fn poll_writable(&mut self) { + us_udp_socket_poll_writable(self) + } + pub fn user(&mut self) -> *mut c_void { us_udp_socket_user(self) } @@ -221,6 +227,7 @@ unsafe extern "C" { addresses: *const *const c_void, num: c_int, ) -> c_int; + safe fn us_udp_socket_poll_writable(socket: &mut Socket); safe fn us_udp_socket_user(socket: &mut Socket) -> *mut c_void; safe fn us_udp_socket_bound_port(socket: &mut Socket) -> c_int; fn us_udp_socket_bound_ip(socket: *mut Socket, buf: *mut u8, length: *mut i32); diff --git a/test/js/node/quic/quic-endpoint.test.ts b/test/js/node/quic/quic-endpoint.test.ts index 73c7a8fc4b95..7c3a1bc0bcec 100644 --- a/test/js/node/quic/quic-endpoint.test.ts +++ b/test/js/node/quic/quic-endpoint.test.ts @@ -2,7 +2,7 @@ // connect() through an endpoint; a later connect in the other mode must fail // loudly instead of silently reusing an engine that cannot frame it. import { socketFaultInjection as fault } from "bun:internal-for-testing"; -import { describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, test } from "bun:test"; import { bunEnv, bunExe, isLinux, isWindows, normalizeBunSnapshot, tempDir } from "harness"; import { createPrivateKey } from "node:crypto"; import { createSocket } from "node:dgram"; @@ -433,7 +433,7 @@ describe("an error on the endpoint's socket", () => { }); await using endpoint = new QuicEndpoint(); using _sessions = destroyAtExit(clientSessions, serverSessions); - const live = await connect(server.address, { ...clientOptions, endpoint }); + const live = await connect(server.address!, { ...clientOptions, endpoint }); clientSessions.push(live); await live.opened; const refusedSession = ignoreSettlement(await connect(address, { ...clientOptions, endpoint })); @@ -466,7 +466,7 @@ describe("an error on the endpoint's socket", () => { const refusedSession = ignoreSettlement(await connect("[::1]:443", { ...clientOptions, endpoint })); clientSessions.push(refusedSession); - const live = await connect(server.address, { ...clientOptions, endpoint }); + const live = await connect(server.address!, { ...clientOptions, endpoint }); clientSessions.push(live); await live.opened; @@ -498,7 +498,7 @@ describe("an error on the endpoint's socket", () => { using _sessions = destroyAtExit(clientSessions, serverSessions); using relays = new DisposableStack(); - const live = await connect(server.address, { + const live = await connect(server.address!, { ...clientOptions, endpoint, ondatagram(datagram: Uint8Array) { @@ -509,10 +509,8 @@ describe("an error on the endpoint's socket", () => { await live.opened; const cuts: (() => void)[] = []; for (let i = 0; i < dead; i++) { - const relay = relays.use(await relayTo(server.address.port)); - const session = ignoreSettlement( - await connect({ address: "127.0.0.1", port: relay.port }, { ...clientOptions, endpoint }), - ); + const relay = relays.use(await relayTo(server.address!.port)); + const session = ignoreSettlement(await connect(`127.0.0.1:${relay.port}`, { ...clientOptions, endpoint })); clientSessions.push(session); await session.opened; cuts.push(relay.cut); @@ -551,8 +549,8 @@ describe("an error on the endpoint's socket", () => { // Both leave in one turn, so the error about the first is still pending // on the socket when the probe of the second is sent. - const toDeadPeer = connect({ address: "127.0.0.1", port: deadPort }, { ...clientOptions, endpoint }); - const probing = connect(server.address, { ...clientOptions, endpoint, version: 0x1a1a1a1a, onerror() {} }); + const toDeadPeer = connect(`127.0.0.1:${deadPort}`, { ...clientOptions, endpoint }); + const probing = connect(server.address!, { ...clientOptions, endpoint, version: 0x1a1a1a1a, onerror() {} }); clientSessions.push(ignoreSettlement(await toDeadPeer)); const session = await probing; clientSessions.push(session); @@ -571,10 +569,12 @@ describe("an error on the endpoint's socket", () => { // here, so it hits the next send or receive of a UDP socket in the process. // On Windows an injected errno does not pass through the Winsock mapping. describe.skipIf(!fault.available() || isWindows)("injected in uSockets", () => { + // Also after a timeout, so a rule that is left cannot fail the next test. + afterEach(() => fault.clear()); + /** One server, and `count` open sessions to it on one client endpoint. */ async function connected(count: number, serverOptions: object = {}, sessionOptions: object = {}) { const stack = new AsyncDisposableStack(); - stack.defer(() => fault.clear()); const serverSessions: any[] = []; const clientSessions: any[] = []; const server = await listen(onServerSession(serverSessions), { @@ -590,7 +590,7 @@ describe("an error on the endpoint's socket", () => { stack.use(destroyAtExit(clientSessions, serverSessions)); for (let i = 0; i < count; i++) { const session = ignoreSettlement( - await connect(server.address, { ...clientOptions, endpoint, ...sessionOptions }), + await connect(server.address!, { ...clientOptions, endpoint, ...sessionOptions }), ); clientSessions.push(session); await session.opened; @@ -605,6 +605,9 @@ describe("an error on the endpoint's socket", () => { test.each([ ["the socket would block", { action: "errno", errno: "EAGAIN", repeat: 5 }, false], ["a send returns 0", { action: "zero", repeat: 5 }, false], + // uSockets does not wait for the socket after ENOBUFS, so the endpoint + // does. With no wait, lsquic tries again only 1 s after each failure. + ["the kernel has no buffer space", { action: "errno", errno: "ENOBUFS", repeat: 30 }, false], ["the kernel refuses 3 sends", { action: "errno", errno: "ENETUNREACH", repeat: 3 }, true], ] as const)("a stream round trip completes when %s", async (_case, rule, countedAsSent) => { await using net = await connected(1); @@ -729,7 +732,6 @@ describe("an error on the endpoint's socket", () => { test("a receive error during a handshake ends the announced session, then closes the endpoint", async () => { const order: string[] = []; const serverSessions: any[] = []; - using _fault = { [Symbol.dispose]: () => fault.clear() }; const server = await listen( (session: any) => { // Inside the receive loop of the server's socket, as above. @@ -747,7 +749,7 @@ describe("an error on the endpoint's socket", () => { await using _server = { [Symbol.asyncDispose]: () => server.close().catch(() => {}) }; const endpointClosed = logClose(order, "endpoint", server.closed); await using endpoint = new QuicEndpoint(); - const client = ignoreSettlement(await connect(server.address, { ...clientOptions, endpoint })); + const client = ignoreSettlement(await connect(server.address!, { ...clientOptions, endpoint })); using _sessions = destroyAtExit([client], serverSessions); // A server that is still open completes the handshake, and that ends the wait. @@ -780,10 +782,12 @@ describe("an error on the endpoint's socket", () => { const [live] = net.clientSessions; // Nothing answers this probe, so the session has no connection. const probe = ignoreSettlement( - await connect( - { address: "127.0.0.1", port: await closedPort() }, - { ...clientOptions, endpoint: net.endpoint, version: 0x1a1a1a1a, onerror() {} }, - ), + await connect(`127.0.0.1:${await closedPort()}`, { + ...clientOptions, + endpoint: net.endpoint, + version: 0x1a1a1a1a, + onerror() {}, + }), ); net.clientSessions.push(probe); const sessionsClosed = Promise.all([ @@ -809,7 +813,6 @@ describe("an error on the endpoint's socket", () => { const armed = Promise.withResolvers(); const serverSessions: any[] = []; const clientSessions: any[] = []; - using _fault = { [Symbol.dispose]: () => fault.clear() }; await using server = await listen(onServerSession(serverSessions), { sni, alpn: ["quic-test"], @@ -823,7 +826,7 @@ describe("an error on the endpoint's socket", () => { armed.resolve(); }, }; - const first = ignoreSettlement(await connect(server.address, options)); + const first = ignoreSettlement(await connect(server.address!, options)); clientSessions.push(first); await first.opened; const firstEndpoint = first.endpoint; @@ -833,7 +836,7 @@ describe("an error on the endpoint's socket", () => { await armed.promise; // A client that is still open gets the answer to this stream. const firstEnded = await Promise.race([first.closed.then(() => "closed"), answered(first)]); - const second = ignoreSettlement(await connect(server.address, clientOptions)); + const second = ignoreSettlement(await connect(server.address!, clientOptions)); clientSessions.push(second); await second.opened; From 803bd64077859da0fedcec50c1c521619afb3eb7 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:02:58 +0000 Subject: [PATCH 3/3] node:quic: after ENOBUFS, send again in the next pass The writable poll is the wrong signal for ENOBUFS. The socket stays writable while the output queue of the interface is full, so the poll made the loop send again at once, without a pause. - `packets_out` still reports ENOBUFS to lsquic as a blocked socket. It now sets `retry_unsent`, and the next `process` pass sends the waiting packets again. The timer runs that pass 1 ms later. - `us_udp_socket_poll_writable` is removed again. - Each comment that this branch adds in src/ is one line. --- packages/bun-usockets/src/libusockets.h | 3 -- packages/bun-usockets/src/udp.c | 7 ---- src/runtime/node/quic/endpoint.rs | 53 +++++++++++-------------- src/runtime/node/quic/session.rs | 3 +- src/uws_sys/udp.rs | 12 +----- 5 files changed, 25 insertions(+), 53 deletions(-) diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index 4d02605608d8..bb78632bb6a0 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -209,9 +209,6 @@ void us_udp_buffer_set_packet_payload(struct us_udp_packet_buffer_t *send_buf, i int us_udp_socket_send(struct us_udp_socket_t *s, void** payloads, size_t* lengths, void** addresses, int num); -/* Calls drain_cb once the socket is writable. us_udp_socket_send does this itself when the send would block. */ -void us_udp_socket_poll_writable(struct us_udp_socket_t *s); - /* Allocates a packet buffer that is reuable per thread. Mutated by us_udp_socket_receive. */ struct us_udp_packet_buffer_t *us_create_udp_packet_buffer(); diff --git a/packages/bun-usockets/src/udp.c b/packages/bun-usockets/src/udp.c index 992340990dc7..d9a9bfc0fe14 100644 --- a/packages/bun-usockets/src/udp.c +++ b/packages/bun-usockets/src/udp.c @@ -72,13 +72,6 @@ int us_udp_socket_send(struct us_udp_socket_t *s, void** payloads, size_t* lengt return total_sent; } -/* on_drain runs once the socket is writable. us_udp_socket_send asks for this - * itself when the send would block. A caller that takes another send error for - * backpressure (ENOBUFS) asks for it here. */ -void us_udp_socket_poll_writable(struct us_udp_socket_t *s) { - us_poll_change((struct us_poll_t *) s, s->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); -} - int us_udp_socket_bound_port(struct us_udp_socket_t *s) { /* An adopted-but-unbound descriptor cached port 0 at creation; the kernel * auto-binds on the first sendto(), so re-query then. */ diff --git a/src/runtime/node/quic/endpoint.rs b/src/runtime/node/quic/endpoint.rs index decae69b1c94..efff13357c8c 100644 --- a/src/runtime/node/quic/endpoint.rs +++ b/src/runtime/node/quic/endpoint.rs @@ -230,6 +230,8 @@ pub(crate) struct QuicEndpoint { processing: Cell, followup_due: Cell, + /// A send failed with ENOBUFS. The next `process` pass sends the waiting packets again. + retry_unsent: Cell, sessions: JsCell>, pub(super) server_local_tp: JsCell, pub(super) client_local_tp: JsCell, @@ -432,33 +434,23 @@ fn registry_find_by_addr(addr: &StoredAddr, not: *const QuicEndpoint) -> Option< }) } -/// `packets_out` has a datagram that the socket did not send. Returns whether -/// the batch stops there, with `errno` set to what lsquic reads from a short -/// return. `us_quic_packets_out` (quic.c) is the same verdict for HTTP/3: -/// change both together. -/// -/// - `rv == 0`: the socket would block, and uSockets armed WRITABLE for it. -/// EAGAIN pauses the engine until `on_drain`. -/// - ENOBUFS: the output queue of the interface is full, which is common on -/// macOS. It clears by itself, so it is the same pause. uSockets reports it -/// as an error and arms nothing, so the endpoint arms `socket`, its own. -/// - EMSGSIZE: lsquic takes the packet back (ci_packet_too_large). An MTU -/// probe feeds DPLPMTUD, and any other packet ends its connection. -/// - Any other errno: the kernel refused this one datagram (EINVAL, EACCES, -/// EPERM, ENETUNREACH, ...). lsquic has no outcome for that: a short return -/// pauses every session of the engine or closes one. So the datagram counts -/// as sent, which lsquic sees as a packet lost on the wire. Node drops it -/// too: Endpoint::PacketDone ignores the status of a send -/// (node/src/quic/endpoint.cc). +/// Whether an unsent datagram stops the batch. Sets errno for lsquic. Twin: `us_quic_packets_out`. #[cold] #[inline(never)] -fn unsent_stops_batch(socket: *mut uws::udp::Socket, rv: c_int) -> bool { +fn unsent_stops_batch(this: &QuicEndpoint, rv: c_int) -> bool { let e = bun_core::ffi::errno_ptr(); if rv < 0 { // SAFETY: `errno_ptr()` is this thread's errno slot. match unsafe { *e } { - libc::ENOBUFS => uws::udp::Socket::opaque_mut(socket).poll_writable(), - errno => return errno == libc::EMSGSIZE, + // The interface queue is full, and no poll reports when it has room: try again in 1 ms. + libc::ENOBUFS => { + this.retry_unsent.set(true); + this.followup_due.set(true); + } + // lsquic retires the packet (ci_packet_too_large). + libc::EMSGSIZE => return true, + // The kernel refused this datagram. It counts as sent, so lsquic sees a lost packet. + _ => return false, } } // SAFETY: as above. @@ -757,8 +749,7 @@ lsquic_callback! { n: c_uint, ) -> c_int = 0; { let Some(socket) = this.socket.get() else { - // uSockets closed the socket (`on_socket_closed`). Nothing can - // leave, and a short return would make lsquic read a stale errno. + // uSockets closed the socket (`on_socket_closed`), so the batch counts as sent. return n as c_int; }; let my_addr = this.local_addr.get(); @@ -835,7 +826,7 @@ lsquic_callback! { &[total], &[dest.as_ptr().cast()], ); - if rv < 1 && unsent_stops_batch(socket, rv) { + if rv < 1 && unsent_stops_batch(this, rv) { break; } this.add_stat(IDX_STATS_PACKETS_SENT, 1); @@ -1183,6 +1174,7 @@ impl QuicEndpoint { client_is_http: Cell::new(false), processing: Cell::new(false), followup_due: Cell::new(false), + retry_unsent: Cell::new(false), sessions: JsCell::new(Vec::new()), server_local_tp: JsCell::new(lsquic::NqTransportParams::default()), client_local_tp: JsCell::new(lsquic::NqTransportParams::default()), @@ -1500,8 +1492,13 @@ impl QuicEndpoint { } } bun_boringssl_sys::ERR_clear_error(); + let retry_unsent = self.retry_unsent.replace(false); for engine in [self.server_engine.get(), self.client_engine.get()] { if !engine.is_null() { + if retry_unsent { + // SAFETY: engine is live while the endpoint is. + unsafe { lsquic::lsquic_engine_send_unsent_packets(engine) }; + } // SAFETY: engine is live while the endpoint is. unsafe { lsquic::lsquic_engine_process_conns(engine) }; } @@ -1567,8 +1564,7 @@ impl QuicEndpoint { self.sweep_provisional(); self.rearm_timer(); self.update_keepalive(); - // With no socket the engines can send nothing, so the conns they still - // hold (handshakes with no session yet) are not waited for. + // With no socket, the conns that an engine still holds can send nothing. if self.closing.get() && self.sessions.get().is_empty() && (self.engine_conn_count() == 0 || self.socket.get().is_none()) @@ -2157,10 +2153,7 @@ impl QuicEndpoint { true } - /// uSockets closed the socket: a receive or a poll failed, and there is no - /// receive-error callback. Node's `Endpoint::Destroy(RECEIVE_FAILURE)` - /// (node/src/quic/endpoint.cc): each session ends with no packet sent, - /// then the endpoint closes. + /// uSockets closed the socket on a receive error: Node's `Endpoint::Destroy(RECEIVE_FAILURE)`. fn on_socket_closed(&self, socket: *mut uws::udp::Socket) { // `release_native` takes the handle before it closes the socket. if self.socket.get() != Some(socket) { diff --git a/src/runtime/node/quic/session.rs b/src/runtime/node/quic/session.rs index ae2662253b85..c5d5751674f3 100644 --- a/src/runtime/node/quic/session.rs +++ b/src/runtime/node/quic/session.rs @@ -1967,8 +1967,7 @@ impl QuicSession { }) } - /// Node's `Session::Close(CloseMethod::SILENT)`: no CONNECTION_CLOSE - /// leaves, and JS sees a close with no error. + /// Node's `Session::Close(CloseMethod::SILENT)`: JS sees a close with no error. pub(super) fn close_silently(&self) { match self.conn() { Some(conn) => conn.abort_silent(), diff --git a/src/uws_sys/udp.rs b/src/uws_sys/udp.rs index e3fdea6aa402..a989987971e0 100644 --- a/src/uws_sys/udp.rs +++ b/src/uws_sys/udp.rs @@ -14,10 +14,7 @@ bun_opaque::opaque_ffi! { } impl Socket { - /// Without `recv_error_cb` the socket never gets `IP_RECVERR` (Linux), so - /// an ICMP error about one peer cannot fail a send to another. A failing - /// receive then has no handler: uSockets closes the socket and calls - /// `close_cb`, which must drop the handle. + /// `recv_error_cb: None`: no `IP_RECVERR`, and a receive error closes the socket (`close_cb`). pub fn create( loop_: *mut Loop, data_cb: extern "C" fn(*mut Socket, *mut PacketBuffer, c_int), @@ -105,12 +102,6 @@ impl Socket { } } - /// `drain_cb` runs once the socket is writable. `send` asks for this itself - /// when the send would block. - pub fn poll_writable(&mut self) { - us_udp_socket_poll_writable(self) - } - pub fn user(&mut self) -> *mut c_void { us_udp_socket_user(self) } @@ -227,7 +218,6 @@ unsafe extern "C" { addresses: *const *const c_void, num: c_int, ) -> c_int; - safe fn us_udp_socket_poll_writable(socket: &mut Socket); safe fn us_udp_socket_user(socket: &mut Socket) -> *mut c_void; safe fn us_udp_socket_bound_port(socket: &mut Socket) -> c_int; fn us_udp_socket_bound_ip(socket: *mut Socket, buf: *mut u8, length: *mut i32);