diff --git a/packages/bun-usockets/src/bsd.c b/packages/bun-usockets/src/bsd.c index a1190b3a6515..fd62e49708c9 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) { @@ -105,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; @@ -152,6 +156,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..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, @@ -244,6 +246,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 +399,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 +434,30 @@ fn registry_find_by_addr(addr: &StoredAddr, not: *const QuicEndpoint) -> Option< }) } +/// Whether an unsent datagram stops the batch. Sets errno for lsquic. Twin: `us_quic_packets_out`. +#[cold] +#[inline(never)] +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 } { + // 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. + unsafe { *e = libc::EAGAIN }; + true +} + /// 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 +749,8 @@ lsquic_callback! { n: c_uint, ) -> c_int = 0; { let Some(socket) = this.socket.get() else { - return 0; + // 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(); // SAFETY: pure constant query. @@ -789,17 +826,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(this, rv) { break; } this.add_stat(IDX_STATS_PACKETS_SENT, 1); @@ -1147,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()), @@ -1162,6 +1190,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 +1402,7 @@ impl QuicEndpoint { on_data, on_drain, on_close, - on_recv_error, + None, cfg.host.as_ptr().cast(), cfg.port, 0, @@ -1463,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) }; } @@ -1530,9 +1564,10 @@ impl QuicEndpoint { self.sweep_provisional(); self.rearm_timer(); self.update_keepalive(); + // 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.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,28 @@ impl QuicEndpoint { true } + /// 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) { + 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..c5d5751674f3 100644 --- a/src/runtime/node/quic/session.rs +++ b/src/runtime/node/quic/session.rs @@ -1967,6 +1967,14 @@ impl QuicSession { }) } + /// 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(), + 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..a989987971e0 100644 --- a/src/uws_sys/udp.rs +++ b/src/uws_sys/udp.rs @@ -14,12 +14,13 @@ bun_opaque::opaque_ffi! { } impl Socket { + /// `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), 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 +55,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 +202,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 +236,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 868f9feae7c4..7c3a1bc0bcec 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 { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, normalizeBunSnapshot, tempDir } from "harness"; +import { socketFaultInjection as fault } from "bun:internal-for-testing"; +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"; +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, type EndpointOptions, type SessionOptions } from "node:quic"; @@ -310,3 +313,537 @@ 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(`127.0.0.1:${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(`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); + 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", () => { + // 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(); + 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], + // 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); + 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[] = []; + 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(`127.0.0.1:${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[] = []; + 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, + }); + }); + }); +});