node:quic: create the endpoint socket without IP_RECVERR and count a refused datagram as sent

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++) {

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,

src/runtime/node/quic/endpoint.rs

@@ -244,6 +244,8 @@ pub(crate) struct QuicEndpoint {
     block_list_allow: Cell<bool>,
     pub(crate) event_loop_timer: JsCell<EventLoopTimer>,
     pending_endpoint_close: Cell<bool>,
+    /// What the deferred `onEndpointClose` reports.
+    close_context: Cell<u8>,
     /// 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::<QuicEndpoint>() };
+    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;

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;

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,

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<extern "C" fn(*mut Socket, c_int, c_int)>,
         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<extern "C" fn(*mut Socket, c_int, c_int)>,
         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<extern "C" fn(*mut Socket, c_int, c_int)>,
         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<extern "C" fn(*mut Socket, c_int, c_int)>,
         fd: LIBUS_SOCKET_DESCRIPTOR,
         shared: c_int,
         err: *mut c_int,