node:quic: send destroy()'s RESET_STREAM/STOP_SENDING in the call, or when the session's dispatch returns

patches/lsquic/stream-by-id.patch

@@ -0,0 +1,64 @@
+--- a/include/lsquic.h
++++ b/include/lsquic.h
+@@ -1700,6 +1700,14 @@
+ unsigned
+ lsquic_conn_n_avail_streams (const lsquic_conn_t *);
+
++/* bun: the connection's live stream with this ID, or NULL.  A stream ID is
++ * never reused, so it names a stream that lsquic can free on any tick more
++ * safely than the pointer does.  conn_iface has carried this lookup since
++ * gQUIC; it was never filled in for IETF connections.
++ */
++lsquic_stream_t *
++lsquic_conn_get_stream_by_id (lsquic_conn_t *, lsquic_stream_id_t);
++
+ /**
+  * Create a new request stream.  This causes @ref on_new_stream() callback
+  * to be called.  If creating more requests is not permitted at the moment
+--- a/src/liblsquic/lsquic_conn.c
++++ b/src/liblsquic/lsquic_conn.c
+@@ -246,6 +246,18 @@
+ }
+
+
++/* bun: see lsquic_conn_get_stream_by_id in lsquic.h. */
++lsquic_stream_t *
++lsquic_conn_get_stream_by_id (struct lsquic_conn *lconn,
++                                            lsquic_stream_id_t stream_id)
++{
++    if (lconn->cn_if->ci_get_stream_by_id)
++        return lconn->cn_if->ci_get_stream_by_id(lconn, stream_id);
++    else
++        return NULL;
++}
++
++
+ unsigned
+ lsquic_conn_cancel_pending_streams (struct lsquic_conn *lconn, unsigned count)
+ {
+--- a/src/liblsquic/lsquic_full_conn_ietf.c
++++ b/src/liblsquic/lsquic_full_conn_ietf.c
+@@ -5334,6 +5334,15 @@
+ }
+
+
++/* bun: upstream fills the conn_iface lookup for gQUIC connections only. */
++static lsquic_stream_t *
++ietf_full_conn_ci_get_stream_by_id (struct lsquic_conn *lconn,
++                                            lsquic_stream_id_t stream_id)
++{
++    return find_stream_by_id((struct ietf_full_conn *) lconn, stream_id);
++}
++
++
+ static void
+ maybe_schedule_ss_for_stream (struct ietf_full_conn *conn,
+                 lsquic_stream_id_t stream_id, enum http_error_code error_code)
+@@ -9346,6 +9355,7 @@
+     .ci_get_engine           =  ietf_full_conn_ci_get_engine, \
+     .ci_get_min_datagram_size=  ietf_full_conn_ci_get_min_datagram_size, \
+     .ci_get_path             =  ietf_full_conn_ci_get_path, \
++    .ci_get_stream_by_id     =  ietf_full_conn_ci_get_stream_by_id, \
+     .ci_going_away           =  ietf_full_conn_ci_going_away, \
+     .ci_hsk_done             =  ietf_full_conn_ci_hsk_done, \
+     .ci_internal_error       =  ietf_full_conn_ci_internal_error, \

scripts/build/deps/lsquic.ts

@@ -137,6 +137,11 @@ export const lsquic: Dependency = {
     // lsquic_prq_destroy() only frees the free list. A stateless reset whose
     // send failed is on the returned list, so destroying the engine leaks it.
     "patches/lsquic/prq-free-returned-conns.patch",
+    // conn_iface carries a lookup of a stream by its ID that only gQUIC
+    // connections fill in. node:quic holds the frames of a destroyed stream
+    // until its session can send them, and lsquic frees a stream on the tick
+    // that finishes it, so those frames name the stream by ID.
+    "patches/lsquic/stream-by-id.patch",
   ],
 
   fetchDeps: ["zlib", "lshpack", "lsqpack", "boringssl"],

src/js/internal/quic/quic.ts

@@ -2065,25 +2065,22 @@ class QuicStream {
           ? error.errorCode
           : getQuicSessionState(inner.session).internalErrorCode;
     } else {
-      // No error: Node still shuts the stream down on the wire with
-      // RESET_STREAM(0)/STOP_SENDING(0) (ngtcp2_conn_shutdown_stream with
-      // app code 0) — a clean FIN would make the peer surface a normal
-      // empty stream instead of never seeing it.
-      abortCode = 0n;
+      // Node v26.3.0 sends nothing here and loses the stream's place
+      // (https://github.com/nodejs/node/blob/v26.3.0/doc/api/quic.md#L1844-L1846).
+      abortCode = 0n; // the one reset code after which the peer's `closed` resolves
     }
     // When the whole session is being torn down, the cascaded per-stream
     // destroys must not emit wire frames: the session goes away immediately
     // (Node destroys the native session before its deferred send runs), and
     // the surviving peer is expected to clean up via its own close/idle
     // timeout rather than via per-stream RESET/STOP_SENDING.
     const cascadingFromSessionDestroy = inner.session !== undefined && isQuicSessionDestroying(inner.session);
-    // Node parity for streams created and abandoned in one turn.
+    // Records what this destroy owes the peer. handle.destroy() below sends it
+    // or parks it, together with the reset of a live writer.
     if (abortCode !== undefined && !cascadingFromSessionDestroy) {
       const wantStop = !inner.state.readEnded;
       const wantReset = inner.writer === undefined && !inner.state.writeEnded;
-      if (wantStop || wantReset) {
-        this.#handle.abortForDestroy(wantStop ? abortCode : undefined, wantReset ? abortCode : undefined);
-      }
+      this.#handle.abortForDestroy(wantStop ? abortCode : undefined, wantReset ? abortCode : undefined);
     }
     const { onerror } = inner;
     if (error !== undefined && typeof onerror === "function") {

src/lsquic_sys/lib.rs

@@ -417,6 +417,14 @@
             v => Some(v != 0),
         }
     }
+    /// The live stream with this id, which lsquic never reuses.
+    pub fn stream_by_id(&self, id: u64) -> Option<Stream> {
+        unsafe extern "C" {
+            fn lsquic_conn_get_stream_by_id(c: *mut lsquic_conn, id: u64) -> *mut lsquic_stream;
+        }
+        // SAFETY: as above; lsquic returns one of this conn's live streams.
+        unsafe { Stream::from_raw(lsquic_conn_get_stream_by_id(self.0, id)) }
+    }
     pub fn make_stream(&self) {
         // SAFETY: as above.
         unsafe { lsquic_conn_make_stream(self.0) }
@@ -495,7 +503,7 @@
     pub unsafe fn from_raw(raw: *mut lsquic_stream) -> Option<Self> {
         (!raw.is_null()).then_some(Self(raw))
     }
     pub fn raw(&self) -> *mut lsquic_stream {
         self.0
     }
     /// # Safety

src/runtime/node/quic/endpoint.rs

@@ -1452,16 +1452,6 @@ impl QuicEndpoint {
         }
         self.send_scope_depth.set(self.send_scope_depth.get() + 1);
         self.followup_due.set(false);
-        // Not from the mid-turn drain pass: a deferred abort is dropped when
-        // a later stream writes first, and node decides that over the whole
-        // turn -- flushing mid-chain puts a RESET on the wire node never sends.
-        if !self.defer_closes.get() {
-            for session in self.sessions.get().clone() {
-                if let Some(session) = self.live_session(session) {
-                    session.flush_deferred_aborts();
-                }
-            }
-        }
         bun_boringssl_sys::ERR_clear_error();
         for engine in [self.server_engine.get(), self.client_engine.get()] {
             if !engine.is_null() {
@@ -1526,6 +1516,7 @@ impl QuicEndpoint {
             };
             session.process_events(global);
             session.maybe_finish_deferred_close();
+            session.flush_deferred_aborts();
         }
         self.sweep_provisional();
         self.rearm_timer();

src/runtime/node/quic/session.rs

@@ -275,13 +275,16 @@ pub(super) struct HskSnapshot {
 }
 
 pub(super) struct DeferredAbort {
-    /// The raw lsquic stream — alive until the conn dies (entries are
-    /// cleared first) and only ever passed back to lsquic, never
-    /// dereferenced here.
-    ls: *mut lsquic::lsquic_stream,
+    /// The stream, by name: lsquic frees one on the tick that finishes it,
+    /// and the peer can finish a held stream at any time.
+    id: u64,
     reset: Option<u64>,
     stop: Option<u64>,
-    marker: u64,
+    /// The app stopped the stream itself, or holds its writer: these frames
+    /// are always sent (`QuicStream::abort_always_sent`).
+    always_sent: bool,
+    /// The peer never hears of the stream (`silence_opening_flight_aborts`).
+    silent: bool,
 }
 
 pub(super) enum SessionEvent {
@@ -380,7 +383,8 @@ pub(crate) struct QuicSession {
     events: JsCell<Vec<SessionEvent>>,
     origin_buf: JsCell<Vec<u8>>,
     deferred_aborts: JsCell<Vec<DeferredAbort>>,
-    write_marker: Cell<u64>,
+    /// `process_events` of this session is on the stack.
+    dispatching: Cell<bool>,
     /// A graceful close requested while a dispatch was on the stack: applied
     /// on the next depth-0 pass so its GOAWAY/CONNECTION_CLOSE does not share
     /// a flight with data written by the same dispatch (node's close lands an
@@ -434,7 +438,7 @@ impl QuicSession {
             events: JsCell::new(Vec::new()),
             origin_buf: JsCell::new(Vec::new()),
             deferred_aborts: JsCell::new(Vec::new()),
-            write_marker: Cell::new(0),
+            dispatching: Cell::new(false),
             pending_graceful: JsCell::new(None),
             verneg: Cell::new(None),
             peer_close: JsCell::new(None),
@@ -668,64 +672,85 @@ impl QuicSession {
             ep.schedule_process();
         }
     }
-    pub(super) fn note_stream_write(&self) {
-        self.write_marker
-            .set(self.write_marker.get().wrapping_add(1));
+    /// How many stream aborts are parked. A stream records it at each write.
+    pub(super) fn parked_aborts(&self) -> u32 {
+        self.deferred_aborts.get().len() as u32
+    }
+    /// A client before its handshake is reported. Until LSCONN_HANDSHAKE_DONE
+    /// lsquic sends a packet that holds only RESET_STREAM / STOP_SENDING with a
+    /// Handshake header, which no peer can read.
+    fn awaits_handshake(&self) -> bool {
+        !self.is_server.get() && !self.handshake_reported.get()
+    }
+    /// Whether `destroy()` leaves the frames of a stream that never sent to
+    /// `flush_deferred_aborts`: a client before its handshake is reported, and
+    /// JS of this session's dispatch unless the frames are `always_sent`.
+    pub(super) fn parks_stream_abort(&self, always_sent: bool) -> bool {
+        self.awaits_handshake() || (self.dispatching.get() && !always_sent)
     }
     pub(super) fn defer_stream_abort(
         &self,
-        ls: *mut lsquic::lsquic_stream,
-        reset: Option<u64>,
+        id: u64,
         stop: Option<u64>,
+        reset: Option<u64>,
+        always_sent: bool,
     ) {
-        let marker = self.write_marker.get();
         self.deferred_aborts.with_mut(|v| {
-            if let Some(e) = v.iter_mut().find(|e| e.ls == ls) {
-                e.reset = e.reset.or(reset);
-                e.stop = e.stop.or(stop);
-            } else {
-                v.push(DeferredAbort {
-                    ls,
-                    reset,
-                    stop,
-                    marker,
-                });
-            }
+            v.push(DeferredAbort {
+                id,
+                reset,
+                stop,
+                always_sent,
+                silent: false,
+            });
         });
     }
-    pub(super) fn has_deferred_abort(&self, ls: *mut lsquic::lsquic_stream) -> bool {
-        self.deferred_aborts.get().iter().any(|e| e.ls == ls)
+    /// After a client's handshake callback, in which its `opened` continuation
+    /// ran. A stream destroyed there stays hidden from the peer when an open
+    /// stream wrote after the destroy: Node's peer does not see it either, and
+    /// test-quic-callback-error-stream-onerror.mjs depends on that.
+    fn silence_opening_flight_aborts(&self, parked_before: usize) {
+        let streams = self.streams.get();
+        self.deferred_aborts.with_mut(|v| {
+            for (index, e) in v.iter_mut().enumerate().skip(parked_before) {
+                // SAFETY: as in `has_opened_stream`.
+                e.silent = !e.always_sent
+                    && streams
+                        .iter()
+                        .any(|&(_, s)| unsafe { (*s).wrote_after_park(index) });
+            }
+        });
     }
-    pub(super) fn flush_deferred_aborts(&self) {
-        if self.deferred_aborts.get().is_empty() {
-            return;
-        }
+    /// Sends the parked aborts. One that is not `always_sent` is retired with
+    /// no frame when it is `silent`, and when the connection closes.
+    fn settle_deferred_aborts(&self, closing: bool) {
         let entries = self.deferred_aborts.with_mut(core::mem::take);
-        if self.conn.get().is_null() {
-            return;
-        }
-        let marker = self.write_marker.get();
+        let Some(conn) = self.conn() else { return };
         for e in entries {
-            // SAFETY: entries are cleared before the conn (and thus its
-            // streams) is freed, so `ls` is a live lsquic stream.
-            let Some(s) = (unsafe { lsquic::Stream::from_raw(e.ls) }) else {
+            // The peer can end a held stream itself (a RESET_STREAM and a
+            // STOP_SENDING finish it), and lsquic frees it on that tick.
+            let Some(s) = conn.stream_by_id(e.id) else {
                 continue;
             };
-            if e.marker != marker {
-                // Node parity — the queued frames are dropped.
+            if e.silent || (closing && !e.always_sent) {
                 s.shutdown_internal();
-                continue;
-            }
-            if let Some(code) = e.stop {
-                s.stop_sending(code);
-            }
-            if let Some(code) = e.reset {
-                s.reset(code);
             } else {
-                s.close();
+                super::stream::send_abort(s, e.stop, e.reset);
             }
         }
     }
+    /// Runs after this session's dispatch. The aborts wait for a client's
+    /// handshake report, and for a stashed `close()` (`flush_pending_graceful`).
+    pub(super) fn flush_deferred_aborts(&self) {
+        if self.deferred_aborts.get().is_empty()
+            || self.awaits_handshake()
+            || self.pending_graceful.get().is_some()
+        {
+            return;
+        }
+        self.settle_deferred_aborts(false);
+        self.schedule_process();
+    }
     pub(super) fn take_pending_local_stream(
         &self,
         uni: bool,
@@ -918,6 +943,7 @@ impl QuicSession {
         // (dropping the wrapper Strong) and can trigger GC, so hold a Strong
         // for the duration -- `deliver_pending_tickets` touches `self` after.
         let _keep_alive = Strong::create(self.handle(), global);
+        let was_dispatching = self.dispatching.replace(true);
         self.deliver_pending_tickets(global);
         // node leaves a microtask window between a session's lifecycle events
         // and its close; the loop driver can batch a whole exchange into one
@@ -957,6 +983,7 @@ impl QuicSession {
                 }
             }
         }
+        self.dispatching.set(was_dispatching);
     }
 
     /// One queued event's delivery. A value that cannot be built for its
@@ -1615,6 +1642,7 @@ impl QuicSession {
             ),
             None => (JSValue::UNDEFINED, JSValue::UNDEFINED),
         };
+        let parked_before = self.deferred_aborts.get().len();
         if let Some(callback) = callbacks::get(global, "onSessionHandshake") {
             let vm = global.bun_vm().as_mut();
             vm.event_loop_ref().run_callback(
@@ -1634,6 +1662,9 @@ impl QuicSession {
                 ],
             );
         }
+        if !self.is_server.get() {
+            self.silence_opening_flight_aborts(parked_before);
+        }
 
         if self.qlog_enabled.get() {
             let t = now_ns() / 1_000_000;
@@ -1939,13 +1970,16 @@ impl QuicSession {
         }
         if let Some((app, code, reason)) = self.pending_graceful.with_mut(Option::take) {
             self.apply_graceful_close(app, code, reason);
+            // The close waits for streams: the aborts it held go out first.
+            self.flush_deferred_aborts();
             return true;
         }
         false
     }
 
     fn apply_close(&self, app: bool, code: u64, reason: &[u8]) {
         let Some(c) = self.conn() else { return };
+        self.settle_deferred_aborts(true);
         if app || code != 0 || reason.len() > 1 {
             let creason = core::ffi::CStr::from_bytes_until_nul(reason).unwrap_or(c"close");
             c.abort_error(app, code.min(u32::MAX as u64) as core::ffi::c_uint, creason);