quic: WT implement close session callback

martenrichter · martenrichter · commit 6d77818b7589 · 2026-06-20T17:57:18.000+02:00
diff --git a/lib/internal/quic/diagnostics.js b/lib/internal/quic/diagnostics.js
@@ -33,7 +33,8 @@ const onSessionGoawayChannel = dc.channel('quic.session.goaway');
 const onSessionEarlyRejectedChannel = dc.channel('quic.session.early.rejected');
 const onStreamClosedChannel = dc.channel('quic.stream.closed');
 const onStreamHeadersChannel = dc.channel('quic.stream.headers');
-const onStreamSessionIdChannel = dc.channel('quic.stream.sessioid');
+const onStreamSessionIdChannel = dc.channel('quic.stream.sessionid');
+const onStreamWTSessionCloseChannel = dc.channel('quic.stream.wtsessionclose')
 const onStreamTrailersChannel = dc.channel('quic.stream.trailers');
 const onStreamInfoChannel = dc.channel('quic.stream.info');
 const onStreamResetChannel = dc.channel('quic.stream.reset');
@@ -70,6 +71,7 @@ module.exports = {
   onStreamClosedChannel,
   onStreamHeadersChannel,
   onStreamSessionIdChannel,
+  onStreamWTSessionCloseChannel,
   onStreamTrailersChannel,
   onStreamInfoChannel,
   onStreamResetChannel,
diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js
@@ -191,6 +191,7 @@ const {
   kVerifyPeer,
   kHeaders,
   kSessionId,
+  kWTSessionClose,
   kOwner,
   kRemoveSession,
   kKeylog,
@@ -266,6 +267,7 @@ const {
   onStreamClosedChannel,
   onStreamHeadersChannel,
   onStreamSessionIdChannel,
+  onStreamWTSessionCloseChannel,
   onStreamTrailersChannel,
   onStreamInfoChannel,
   onStreamResetChannel,
@@ -306,6 +308,7 @@ const endpointRegistry = new SafeSet();
  *   webtransport support. If no headers are provided, it has no effect.
  * @property {OnHeadersCallback} [onheaders] Callback for incoming initial headers
  * @property {OnSessionIdCallback} [onsessionid] Callback for incoming sessionid
+ * @property {OnWTSessionCloseCallback} [onwtsessionclose] Callback for incoming close capsules
  * @property {OnTrailersCallback} [ontrailers] Callback for incoming trailing headers
  * @property {OnInfoCallback} [oninfo] Callback for informational (1xx) headers
  * @property {OnWantTrailersCallback} [onwanttrailers] Callback fired when the
@@ -469,6 +472,7 @@ const endpointRegistry = new SafeSet();
  * @property {OnApplicationCallback} [onapplication] application options callback.
  * @property {OnHeadersCallback} [onheaders] Default per-stream initial-headers callback.
  * @property {OnSessionIdCallback} [onsessionid] Default perstream initial callback for incoming sessionid.
+ * @property {OnWTSessionCloseCallback} [onwtsessionclose] Default perstream  initial callback for incoming close capsules
  * @property {OnTrailersCallback} [ontrailers] Default per-stream trailing-headers callback.
  * @property {OnInfoCallback} [oninfo] Default per-stream informational-headers callback.
  * @property {OnWantTrailersCallback} [onwanttrailers] Default per-stream
@@ -725,6 +729,15 @@ const endpointRegistry = new SafeSet();
  * @returns {void}
  */
 
+/**
+ * Called when session id (e.g. for Webtransport streams) is determined
+ * @callback OnWTSessionCloseCallback
+ * @this {QuicStream}
+ * @param {number} error code from the webtransport session
+ * @param {string|undefined} error message from the webtransport session 
+ * @returns {void}
+ */
+
 
 
 /**
@@ -1021,6 +1034,12 @@ setCallbacks({
     this[kOwner][kSessionId](sessionId);
   },
 
+  onStreamWTSessionClose(errorcode, errormessage) {
+    // Called when the stream C++ handle has received a close capsule
+    debug(`stream ${this[kOwner].id} wtsessionclose callback`, errorcode, errormessage);
+    this[kOwner][kWTSessionClose](errorcode, errormessage);
+  },
+
   onStreamTrailers() {
     // Called when the stream C++ handle is ready to receive trailing headers.
     debug('stream want trailers callback', this[kOwner]);
@@ -1343,6 +1362,7 @@ function applyCallbacks(session, cbs) {
     };
   }
   if (cbs.onsessionid) session.onsessionid = cbs.onsessionid;
+  if (cbs.onwtsessionclose) session.onwtsessionclose = cbs.onwtsessionclose;
 }
 
 /**
@@ -1603,6 +1623,7 @@ class QuicStream {
     onreset: undefined,
     onheaders: undefined,
     onsessionid: undefined,
+    onwtsessionclose: undefined,
     ontrailers: undefined,
     oninfo: undefined,
     onwanttrailers: undefined,
@@ -1846,6 +1867,26 @@ class QuicStream {
     }
   }
 
+  /** @type {OnWTSessionCloseCallback} */
+  get onwtsessionclose() {
+    assertIsQuicStream(this);
+    return this.#inner.onwtsessionclose;
+  }
+
+  set onwtsessionclose(fn) {
+    assertIsQuicStream(this);
+    const inner = this.#inner;
+    if (fn === undefined) {
+      inner.onwtsessionclose = undefined;
+      inner.state.wantsWTSessionClose = false;
+    } else {
+      console.log('Set wantsWTSessionClose');
+      validateFunction(fn, 'onwtsessionclose');
+      inner.onwtsessionclose = FunctionPrototypeBind(fn, this);
+      inner.state.wantsWTSessionClose = true;
+    }
+  }
+
   /** @type {Function|undefined} */
   get oninfo() {
     assertIsQuicStream(this);
@@ -2610,6 +2651,7 @@ class QuicStream {
     inner.onreset = undefined;
     inner.onheaders = undefined;
     inner.onsessionid = undefined;
+    inner.onwtsessionclose = undefined;
     inner.onerror = undefined;
     inner.ontrailers = undefined;
     inner.oninfo = undefined;
@@ -2724,6 +2766,23 @@ class QuicStream {
     }
   }
 
+  [kWTSessionClose](errorcode, errormessage) {
+    if (this.destroyed) return;
+    const inner = this.#inner;
+    if (onStreamWTSessionCloseChannel.hasSubscribers) {
+      onStreamWTSessionCloseChannel.publish({
+        __proto__: null,
+        stream: this,
+        session: inner.session,
+        errorcode,
+        errormessage
+      });
+    }
+    if (typeof inner.onwtsessionclose === 'function') {
+      safeCallbackInvoke(inner.onwtsessionclose, this, errorcode, errormessage);
+    }
+  }
+
   [kTrailers]() {
     if (this.destroyed) return;
     const inner = this.#inner;
@@ -3331,6 +3390,7 @@ class QuicSession {
       headers,
       onheaders,
       onsessionid,
+      onwtsessionclose,
       ontrailers,
       oninfo,
       onwanttrailers,
@@ -3388,6 +3448,7 @@ class QuicSession {
     // Set stream callbacks before sending headers to avoid missing events.
     if (onheaders) stream.onheaders = onheaders;
     if (onsessionid) stream.onsessionid = onsessionid;
+    if (onwtsessionclose) stream.onwtsessionclose = onwtsessionclose;
     if (ontrailers) stream.ontrailers = ontrailers;
     if (oninfo) stream.oninfo = oninfo;
     if (onwanttrailers) stream.onwanttrailers = onwanttrailers;
@@ -4172,6 +4233,7 @@ class QuicSession {
     if (scbs) {
       if (scbs.onheaders) stream.onheaders = scbs.onheaders;
       if (scbs.onsessionid) stream.onsessionid = scbs.onsessionid;
+      if (scbs.onwtsessionclose) stream.onwtsessionclose = scbs.onwtsessionclose;
       if (scbs.ontrailers) stream.ontrailers = scbs.ontrailers;
       if (scbs.oninfo) stream.oninfo = scbs.oninfo;
       if (scbs.onwanttrailers) stream.onwanttrailers = scbs.onwanttrailers;
@@ -4550,6 +4612,7 @@ class QuicEndpoint {
       // Stream-level callbacks applied to each incoming stream.
       onheaders,
       onsessionid,
+      onwtsessionclose,
       ontrailers,
       oninfo,
       onwanttrailers,
@@ -4576,6 +4639,7 @@ class QuicEndpoint {
       onapplication,
       onheaders,
       onsessionid,
+      onwtsessionclose,
       ontrailers,
       oninfo,
       onwanttrailers,
@@ -5312,6 +5376,7 @@ function processSessionOptions(options, config = kEmptyObject) {
     // Stream-level callbacks.
     onheaders,
     onsessionid,
+    onwtsessionclose,
     ontrailers,
     oninfo,
     onwanttrailers,
@@ -5434,6 +5499,7 @@ function processSessionOptions(options, config = kEmptyObject) {
     onapplication,
     onheaders,
     onsessionid,
+    onwtsessionclose,
     ontrailers,
     oninfo,
     onwanttrailers,
diff --git a/lib/internal/quic/state.js b/lib/internal/quic/state.js
@@ -101,6 +101,7 @@ const {
   IDX_STATE_STREAM_WANTS_BLOCK,
   IDX_STATE_STREAM_WANTS_HEADERS,
   IDX_STATE_STREAM_WANTS_SESSIONID,
+  IDX_STATE_STREAM_WANTS_WTSESSIONCLOSE,
   IDX_STATE_STREAM_WANTS_RESET,
   IDX_STATE_STREAM_WANTS_TRAILERS,
   IDX_STATE_STREAM_RECEIVED_EARLY_DATA,
@@ -828,6 +829,20 @@ class QuicStreamState {
     DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_SESSIONID, val ? 1 : 0);
   }
 
+  /** @type {boolean} */
+  get wantsWTSessionClose() {
+    const handle = this.#handle;
+    if (handle === undefined) return undefined;
+    return DataViewPrototypeGetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_WTSESSIONCLOSE) !== 0;
+  }
+
+  /** @type {boolean} */
+  set wantsWTSessionClose(val) {
+    const handle = this.#handle;
+    if (handle === undefined) return;
+    DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_WTSESSIONCLOSE, val ? 1 : 0);
+  }
+
 
   /** @type {boolean} */
   get wantsReset() {
@@ -922,6 +937,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsSessionId,
+      wantsWTSessionClose,
       wantsTrailers,
       early,
       resetCode,
@@ -943,6 +959,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsSessionId,
+      wantsWTSessionClose,
       wantsTrailers,
       early,
       resetCode,
@@ -980,6 +997,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsSessionId,
+      wantsWTSessionClose,
       wantsTrailers,
       early,
       resetCode,
@@ -1001,6 +1019,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsSessionId,
+      wantsWTSessionClose,
       wantsTrailers,
       early,
       resetCode,
diff --git a/lib/internal/quic/symbols.js b/lib/internal/quic/symbols.js
@@ -41,6 +41,7 @@ const kHandshakeCompleted = Symbol('kHandshakeCompleted');
 const kVerifyPeer = Symbol('kVerifyPeer');
 const kHeaders = Symbol('kHeaders');
 const kSessionId = Symbol('kSessionId');
+const kWTSessionClose = Symbol('kWTSessionClose'); 
 const kKeylog = Symbol('kKeylog');
 const kListen = Symbol('kListen');
 const kQlog = Symbol('kQlog');
diff --git a/src/quic/bindingdata.h b/src/quic/bindingdata.h
@@ -61,6 +61,7 @@ class SessionManager;
   V(stream_drain, StreamDrain)                                                 \
   V(stream_headers, StreamHeaders)                                             \
   V(stream_sessionid, StreamSessionId)                                         \
+  V(stream_wtsessionclose, StreamWTSessionClose)                               \
   V(stream_reset, StreamReset)                                                 \
   V(stream_trailers, StreamTrailers)
 
diff --git a/src/quic/http3.cc b/src/quic/http3.cc
@@ -1297,6 +1297,22 @@ class Http3ApplicationImpl final : public Session::Application {
     return NGHTTP3_ERR_CALLBACK_FAILURE;
   }
 
+  static int on_recv_wt_close_session(nghttp3_conn *conn,
+                                      int64_t session_id,
+                                      uint32_t wt_error_code,
+                                      const uint8_t *msg,
+                                      size_t msglen,
+                                      void *conn_user_data,
+                                      void *stream_user_data) {
+    NGHTTP3_CALLBACK_SCOPE(app);
+    auto& session = app.session();
+    if (auto stream = FindOrCreateStream(conn, &session, session_id)) [[likely]] {
+      stream->NotifyWTSessionClose(wt_error_code, msg, msglen);
+      return NGTCP2_SUCCESS;
+    }
+    return NGHTTP3_ERR_CALLBACK_FAILURE;
+  }
+
   static int on_deferred_consume(nghttp3_conn* conn,
                                  stream_id id,
                                  size_t consumed,
@@ -1496,7 +1512,8 @@ class Http3ApplicationImpl final : public Session::Application {
       on_rand,
       on_receive_settings,
       on_receive_wt_data,
-      on_wt_data_stream_open};
+      on_wt_data_stream_open,
+      on_recv_wt_close_session};
 };
 
 std::optional<PendingTicketAppData> ParseHttp3TicketData(const uv_buf_t& data) {
diff --git a/src/quic/streams.cc b/src/quic/streams.cc
@@ -62,6 +62,8 @@ namespace quic {
   V(WANTS_HEADERS, wants_headers, uint8_t)                                     \
   /* Set when the stream has a sessionid event handler */                      \
   V(WANTS_SESSIONID, wants_sessionid, uint8_t)                                 \
+  /* Set when the stream has a event handler for closing a WT session */       \
+  V(WANTS_WTSESSIONCLOSE, wants_wtsessionclose, uint8_t)                       \
   /* Set when the stream has a reset event handler */                          \
   V(WANTS_RESET, wants_reset, uint8_t)                                         \
   /* Set when the stream has a trailers event handler */                       \
@@ -1628,12 +1630,18 @@ bool Stream::AddHeader(std::unique_ptr<Header> header) {
   return true;
 }
 
- void Stream::NotifyWTSession(stream_id session_id) {
+void Stream::NotifyWTSession(stream_id session_id) {
   if (state()->session_id != session_id) {
     state()->session_id = session_id;
     EmitSessionid(session_id);
   }
- }
+}
+
+void Stream::NotifyWTSessionClose(uint32_t wt_error_code,
+                                   const uint8_t *msg,
+                                   size_t msglen) {
+  EmitWTSessionClose(wt_error_code, msg, msglen);
+}
 
 void Stream::Acknowledge(size_t datalen) {
   if (outbound_ == nullptr) return;
@@ -1984,6 +1992,22 @@ void Stream::EmitSessionid(stream_id session_id) {
   MakeCallback(BindingData::Get(env()).stream_sessionid_callback(), 1, &sid);
 }
 
+  
+void Stream::EmitWTSessionClose(uint32_t wt_error_code,
+                                const uint8_t *msg,
+                                size_t msglen) {       
+  if (!env()->can_call_into_js()  || !state()->wants_wtsessionclose) return;
+  CallbackScope<Stream> cb_scope(this);
+  Local<Value> argv[] = {
+      Integer::NewFromUnsigned(env()->isolate(),
+                               wt_error_code),
+      String::NewFromUtf8(env()->isolate(), reinterpret_cast<const char *>(msg), 
+          v8::NewStringType::kNormal, msglen).ToLocalChecked()
+  };
+  MakeCallback(BindingData::Get(env()).stream_wtsessionclose_callback(), 
+    arraysize(argv), argv);
+}
+
 void Stream::EmitReset(const QuicError& error) {
   // state()->wants_reset will be set from the javascript side if the
   // stream object has a handler for the reset event.
diff --git a/src/quic/streams.h b/src/quic/streams.h
@@ -359,6 +359,11 @@ class Stream final : public AsyncWrap,
   // Currently only http/3 can have a session stream in WebTransport
   void NotifyWTSession(stream_id session_id);
 
+  // Currently only http/3 can have a session stream that receives a close capsule
+  void NotifyWTSessionClose(uint32_t wt_error_code,
+                            const uint8_t *msg,
+                            size_t msglen);
+
   // TODO(@jasnell): Implement MemoryInfo to track outbound_, inbound_,
   // reader_, headers_, and pending_headers_queue_.
   SET_NO_MEMORY_INFO()
@@ -434,6 +439,11 @@ class Stream final : public AsyncWrap,
   // Delivers the session_id aka the stream that holds e.g. the WT session.
   void EmitSessionid(stream_id session_id);
 
+  // delivers the content of the close capsule
+  void EmitWTSessionClose(uint32_t wt_error_code,
+                          const uint8_t *msg,
+                          size_t msglen);
+
   void NotifyReadableEnded(error_code code);
   void NotifyWritableEnded(error_code code);
 
diff --git a/test/parallel/test-quic-internal-setcallbacks.mjs b/test/parallel/test-quic-internal-setcallbacks.mjs
