quic: fix stop sending behaviour & callback

Signed-off-by: Tim Perry <pimterry@gmail.com>

Author: pimterry

doc/api/quic.md

@@ -2033,8 +2033,7 @@ added: v23.8.0
 
 The callback to invoke when the peer aborts a direction of the stream by
 sending a `RESET_STREAM` frame (the peer abandons their writable side, so
-no further data will arrive on our readable side) or a `STOP_SENDING`
-frame (the peer asks us to stop writing on our writable side).
+no further data will arrive on our readable side).
 
 The callback receives a Node.js error whose `errorCode` (`bigint`)
 property carries the application error code from the wire frame.
@@ -2045,6 +2044,21 @@ continue using the still-active direction on a bidirectional stream),
 abort the other direction with [`writer.fail()`][], or tear down the
 whole stream with [`stream.destroy()`][]. Read/write.
 
+### `stream.onstopsending`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {quic.OnStreamErrorCallback}
+
+The callback to invoke when the peer aborts a direction of the stream by
+sending a `STOP_SENDING` frame (the peer asks us to stop writing on our
+writable side).
+
+The callback receives a Node.js error whose `errorCode` (`bigint`)
+property carries the application error code from the wire frame. Read/write.
+
 ### `stream.headers`
 
 <!-- YAML
@@ -3600,8 +3614,8 @@ functions. If a callback throws synchronously or returns a promise that
 rejects, the error is caught and the owning session or stream is destroyed
 with that error:
 
-* Stream callbacks (`onblocked`, `onreset`, `onheaders`, `ontrailers`,
-  `oninfo`, `onwanttrailers`): the stream is destroyed.
+* Stream callbacks (`onblocked`, `onreset`, `onstopsending`, `onheaders`,
+  `ontrailers`, `oninfo`, `onwanttrailers`): the stream is destroyed.
 * Session callbacks (`onapplication`, `onstream`, `ondatagram`,
   `ondatagramstatus`, `onpathvalidation`, `onsessionticket`,
   `onnewtoken`, `onversionnegotiation`, `onorigin`, `ongoaway`,
@@ -4470,10 +4484,9 @@ added: v26.2.0
 * `session` {quic.QuicSession}
 * `error` {any} The QUIC error associated with the reset.
 
-Published when a stream receives a STOP\_SENDING or RESET\_STREAM frame
-from the peer, indicating the peer has aborted the stream. This is a
-key signal for diagnosing application-level issues such as cancelled
-requests.
+Published when a stream receives a RESET\_STREAM frame from the peer,
+indicating the peer has aborted its sending direction. This is a key signal
+for diagnosing application-level issues such as cancelled requests.
 
 ### Channel: `quic.stream.blocked`
 

lib/internal/quic/quic.js

@@ -12,6 +12,7 @@ const {
   DataViewPrototypeGetByteLength,
   ErrorCaptureStackTrace,
   FunctionPrototypeBind,
+  FunctionPrototypeCall,
   Number,
   ObjectDefineProperties,
   ObjectKeys,
@@ -210,6 +211,7 @@ const {
   kSendHeaders,
   kSessionApplication,
   kSessionTicket,
+  kStopSending,
   kTrailers,
   kVersionNegotiation,
   kInspect,
@@ -991,6 +993,14 @@ setCallbacks({
     this[kOwner][kReset](error);
   },
 
+  onStreamStopSending(error) {
+    if (error !== undefined) {
+      error = convertQuicError(error);
+    }
+    debug('stream stop sending callback', this[kOwner], error);
+    this[kOwner][kStopSending](error);
+  },
+
   onStreamHeaders(headers, kind) {
     // Called when the stream C++ handle has received a full block of headers.
     debug(`stream ${this[kOwner].id} headers callback`, headers, kind);
@@ -1576,6 +1586,7 @@ class QuicStream {
     onerror: undefined,
     onblocked: undefined,
     onreset: undefined,
+    onstopsending: undefined,
     onheaders: undefined,
     ontrailers: undefined,
     oninfo: undefined,
@@ -1781,6 +1792,25 @@ class QuicStream {
     }
   }
 
+  /** @type {OnStreamErrorCallback} */
+  get onstopsending() {
+    assertIsQuicStream(this);
+    return this.#inner.onstopsending;
+  }
+
+  set onstopsending(fn) {
+    assertIsQuicStream(this);
+    const inner = this.#inner;
+    if (fn === undefined) {
+      inner.onstopsending = undefined;
+      inner.state.wantsStopSending = false;
+    } else {
+      validateFunction(fn, 'onstopsending');
+      inner.onstopsending = FunctionPrototypeBind(fn, this);
+      inner.state.wantsStopSending = true;
+    }
+  }
+
   /** @type {OnHeadersCallback} */
   get onheaders() {
     assertIsQuicStream(this);
@@ -2143,6 +2173,19 @@ class QuicStream {
       }
     };
 
+    const onStopSending = stream[kStopSending];
+    stream[kStopSending] = (reason) => {
+      if (!closed && !errored) {
+        errored = true;
+        error = reason;
+        if (drainWakeup != null) {
+          drainWakeup.reject(error);
+          drainWakeup = null;
+        }
+      }
+      FunctionPrototypeCall(onStopSending, stream, reason);
+    };
+
     // A note on backpressure handling: per the stream/iter spec, the default
     // backpressure policy for writers is strict, meaning that if the stream
     // signals backpressure additional writes are rejected until the buffer has
@@ -2543,6 +2586,7 @@ class QuicStream {
     inner.pendingClose.resolve = undefined;
     inner.onblocked = undefined;
     inner.onreset = undefined;
+    inner.onstopsending = undefined;
     inner.onheaders = undefined;
     inner.onerror = undefined;
     inner.ontrailers = undefined;
@@ -2596,6 +2640,12 @@ class QuicStream {
     safeCallbackInvoke(inner.onreset, this, error);
   }
 
+  [kStopSending](error) {
+    const inner = this.#inner;
+    assert(inner.onstopsending, 'Unexpected stop sending event');
+    safeCallbackInvoke(inner.onstopsending, this, error);
+  }
+
   [kHeaders](headers, kind) {
     const block = parseHeaderPairs(headers);
     const kindName = kHeadersKindName[kind] ?? kind;

lib/internal/quic/state.js

@@ -101,6 +101,7 @@ const {
   IDX_STATE_STREAM_WANTS_BLOCK,
   IDX_STATE_STREAM_WANTS_HEADERS,
   IDX_STATE_STREAM_WANTS_RESET,
+  IDX_STATE_STREAM_WANTS_STOP_SENDING,
   IDX_STATE_STREAM_WANTS_TRAILERS,
   IDX_STATE_STREAM_RECEIVED_EARLY_DATA,
   IDX_STATE_STREAM_WRITE_DESIRED_SIZE,
@@ -142,6 +143,7 @@ assert(IDX_STATE_STREAM_HAS_READER !== undefined);
 assert(IDX_STATE_STREAM_WANTS_BLOCK !== undefined);
 assert(IDX_STATE_STREAM_WANTS_HEADERS !== undefined);
 assert(IDX_STATE_STREAM_WANTS_RESET !== undefined);
+assert(IDX_STATE_STREAM_WANTS_STOP_SENDING !== undefined);
 assert(IDX_STATE_STREAM_WANTS_TRAILERS !== undefined);
 assert(IDX_STATE_STREAM_WRITE_DESIRED_SIZE !== undefined);
 assert(IDX_STATE_STREAM_RESET_CODE !== undefined);
@@ -826,6 +828,24 @@ class QuicStreamState {
     DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_RESET, val ? 1 : 0);
   }
 
+  /** @type {boolean} */
+  get wantsStopSending() {
+    const handle = this.#handle;
+    if (handle === undefined) return undefined;
+    return DataViewPrototypeGetUint8(
+      handle, this.#offset + IDX_STATE_STREAM_WANTS_STOP_SENDING) !== 0;
+  }
+
+  /** @type {boolean} */
+  set wantsStopSending(val) {
+    const handle = this.#handle;
+    if (handle === undefined) return;
+    DataViewPrototypeSetUint8(
+      handle,
+      this.#offset + IDX_STATE_STREAM_WANTS_STOP_SENDING,
+      val ? 1 : 0);
+  }
+
   /** @type {boolean} */
   get wantsTrailers() {
     const handle = this.#handle;
@@ -903,6 +923,7 @@ class QuicStreamState {
       hasReader,
       wantsBlock,
       wantsReset,
+      wantsStopSending,
       wantsHeaders,
       wantsTrailers,
       early,
@@ -923,6 +944,7 @@ class QuicStreamState {
       hasReader,
       wantsBlock,
       wantsReset,
+      wantsStopSending,
       wantsHeaders,
       wantsTrailers,
       early,
@@ -960,6 +982,7 @@ class QuicStreamState {
       hasReader,
       wantsBlock,
       wantsReset,
+      wantsStopSending,
       wantsHeaders,
       wantsTrailers,
       early,
@@ -980,6 +1003,7 @@ class QuicStreamState {
       hasReader,
       wantsBlock,
       wantsReset,
+      wantsStopSending,
       wantsHeaders,
       wantsTrailers,
       early,

lib/internal/quic/symbols.js

@@ -57,6 +57,7 @@ const kReset = Symbol('kReset');
 const kSendHeaders = Symbol('kSendHeaders');
 const kSessionApplication = Symbol('kSessionApplication');
 const kSessionTicket = Symbol('kSessionTicket');
+const kStopSending = Symbol('kStopSending');
 const kTrailers = Symbol('kTrailers');
 const kVersionNegotiation = Symbol('kVersionNegotiation');
 
@@ -93,6 +94,7 @@ module.exports = {
   kSendHeaders,
   kSessionApplication,
   kSessionTicket,
+  kStopSending,
   kTrailers,
   kVersionNegotiation,
 };

src/quic/bindingdata.h

@@ -61,6 +61,7 @@ class SessionManager;
   V(stream_drain, StreamDrain)                                                 \
   V(stream_headers, StreamHeaders)                                             \
   V(stream_reset, StreamReset)                                                 \
+  V(stream_stop_sending, StreamStopSending)                                    \
   V(stream_trailers, StreamTrailers)
 
 // The various JS strings the implementation uses.

src/quic/session.cc

@@ -1611,11 +1611,11 @@ struct Session::Impl final : public MemoryRetainer {
     return NGTCP2_SUCCESS;
   }
 
-  static int on_stream_stop_sending(ngtcp2_conn* conn,
-                                    stream_id stream_id,
-                                    error_code app_error_code,
-                                    void* user_data,
-                                    void* stream_user_data) {
+  static int on_receive_stream_stop_sending(ngtcp2_conn* conn,
+                                            stream_id stream_id,
+                                            error_code app_error_code,
+                                            void* user_data,
+                                            void* stream_user_data) {
     NGTCP2_CALLBACK_SCOPE(session)
     auto* stream = Stream::From(stream_user_data);
     if (stream == nullptr) return NGTCP2_SUCCESS;
@@ -1652,7 +1652,7 @@ struct Session::Impl final : public MemoryRetainer {
 
   static constexpr ngtcp2_callbacks CLIENT = {
       ngtcp2_crypto_client_initial_cb,
-      nullptr,
+      nullptr,  // stream_stop_sending
       ngtcp2_crypto_recv_crypto_data_cb,
       on_handshake_completed,
       on_receive_version_negotiation,
@@ -1686,7 +1686,7 @@ struct Session::Impl final : public MemoryRetainer {
       on_acknowledge_datagram,
       on_lost_datagram,
       nullptr,  // get_path_challenge_data (deprecated, use v2 below)
-      on_stream_stop_sending,
+      nullptr,  // stream_stop_sending
       ngtcp2_crypto_version_negotiation_cb,
       on_receive_rx_key,
       on_receive_tx_key,
@@ -1697,12 +1697,12 @@ struct Session::Impl final : public MemoryRetainer {
       on_cid_status,
       ngtcp2_crypto_get_path_challenge_data2_cb,
 #ifdef NGTCP2_CALLBACKS_V4
-      nullptr,
+      on_receive_stream_stop_sending,
 #endif
   };
 
   static constexpr ngtcp2_callbacks SERVER = {
-      nullptr,
+      nullptr,  // stream_stop_sending
       ngtcp2_crypto_recv_client_initial_cb,
       ngtcp2_crypto_recv_crypto_data_cb,
       on_handshake_completed,
@@ -1737,7 +1737,7 @@ struct Session::Impl final : public MemoryRetainer {
       on_acknowledge_datagram,
       on_lost_datagram,
       nullptr,  // get_path_challenge_data (deprecated, use v2 below)
-      on_stream_stop_sending,
+      nullptr,  // stream_stop_sending
       ngtcp2_crypto_version_negotiation_cb,
       nullptr,
       on_receive_tx_key,
@@ -1748,7 +1748,7 @@ struct Session::Impl final : public MemoryRetainer {
       on_cid_status,
       ngtcp2_crypto_get_path_challenge_data2_cb,
 #ifdef NGTCP2_CALLBACKS_V4
-      nullptr,
+      on_receive_stream_stop_sending,
 #endif
   };
 };

src/quic/streams.cc

@@ -61,6 +61,8 @@ namespace quic {
   V(WANTS_HEADERS, wants_headers, uint8_t)                                     \
   /* Set when the stream has a reset event handler */                          \
   V(WANTS_RESET, wants_reset, uint8_t)                                         \
+  /* Set when the stream has a stop sending event handler */                   \
+  V(WANTS_STOP_SENDING, wants_stop_sending, uint8_t)                           \
   /* Set when the stream has a trailers event handler */                       \
   V(WANTS_TRAILERS, wants_trailers, uint8_t)                                   \
   /* True when 0-RTT early data was received */                                \
@@ -1774,19 +1776,11 @@ void Stream::ReceiveData(const uint8_t* data,
 }
 
 void Stream::ReceiveStopSending(QuicError error) {
-  // STOP_SENDING from the peer asks us to stop sending. Per RFC 9000
-  // §3.5 the receiver SHOULD respond with RESET_STREAM, which is what
-  // ngtcp2_conn_shutdown_stream_write below schedules. If our
-  // writable side has already been shut down (e.g. we already sent
-  // RESET_STREAM ourselves or finished sending with FIN) there is
-  // nothing more to do here. The previous guard checked
-  // `state()->read_ended` which is unrelated to the writable side and
-  // suppressed STOP_SENDING handling whenever a sibling RESET_STREAM
-  // frame had been processed first within the same packet.
-  if (state()->write_ended) return;
+  // STOP_SENDING from the peer asks us to stop sending. The required
+  // RESET_STREAM response is scheduled automatically.
   Debug(this, "Received stop sending with error %s", error);
-  ngtcp2_conn_shutdown_stream_write(session(), 0, id(), error.code());
   EndWritable();
+  EmitStopSending(error);
 }
 
 void Stream::ReceiveStreamReset(uint64_t final_size, QuicError error) {
@@ -1958,6 +1952,17 @@ void Stream::EmitReset(const QuicError& error) {
   MakeCallback(BindingData::Get(env()).stream_reset_callback(), 1, &err);
 }
 
+void Stream::EmitStopSending(const QuicError& error) {
+  if (!env()->can_call_into_js() || !state()->wants_stop_sending) {
+    return;
+  }
+  CallbackScope<Stream> cb_scope(this);
+  Local<Value> err;
+  if (!error.ToV8Value(env()).ToLocal(&err)) return;
+
+  MakeCallback(BindingData::Get(env()).stream_stop_sending_callback(), 1, &err);
+}
+
 void Stream::EmitWantTrailers() {
   // state()->wants_trailers will be set from the javascript side if the
   // stream object has a handler for the trailers event.

src/quic/streams.h

@@ -417,6 +417,9 @@ class Stream final : public AsyncWrap,
   // Notifies the JavaScript side that the stream has been reset.
   void EmitReset(const QuicError& error);
 
+  // Notifies the JavaScript side that the peer asked it to stop sending.
+  void EmitStopSending(const QuicError& error);
+
   // Notifies the JavaScript side that the application is ready to receive
   // trailing headers. Any trailing headers must be sent immediately, and
   // synchronously when this callback is triggered.

test/parallel/test-quic-internal-endpoint-stats-state.mjs

@@ -158,6 +158,7 @@ strictEqual(streamState.reset, false);
 strictEqual(streamState.hasReader, false);
 strictEqual(streamState.wantsBlock, false);
 strictEqual(streamState.wantsReset, false);
+strictEqual(streamState.wantsStopSending, false);
 
 strictEqual(sessionState.hasPathValidationListener, false);
 strictEqual(sessionState.hasDatagramListener, false);