quic: do not destroy incoming streams that have a consumer

doc/api/quic.md

@@ -305,7 +305,11 @@ unidirectional (data flows in only one direction). The `quic` module provides
 separate APIs for creating each kind:
 [`session.createBidirectionalStream()`][] and
 [`session.createUnidirectionalStream()`][]. Streams initiated by a remote
-peer are delivered via the [`session.onstream`][] callback.
+peer are delivered via the [`session.onstream`][] callback. When the
+negotiated application protocol supports the stream-level callbacks (e.g.
+HTTP/3) and an `onheaders` callback is configured, incoming streams can
+instead be consumed entirely through it and registering `onstream` is
+optional.
 
 There are two ways to write data to a stream:
 
@@ -409,7 +413,9 @@ A typical client session progresses through these stages:
 
 On the server side, call [`quic.listen()`][] with a callback. The callback
 fires for each incoming session after the TLS handshake begins. Incoming
-streams arrive via the [`session.onstream`][] callback.
+streams arrive via the [`session.onstream`][] callback, or, for HTTP/3
+sessions with an `onheaders` callback configured, directly through that
+callback (see the [minimal HTTP/3 server][] example).
 
 [`session.destroy()`][] is available for immediate teardown — all open streams
 are destroyed and the session is closed without waiting for them to finish.
@@ -1110,6 +1116,15 @@ added: v23.8.0
 
 The callback to invoke when a new stream is initiated by a remote peer. Read/write.
 
+If no `onstream` callback is set and the stream has no other consumer, an
+incoming stream is destroyed on arrival and a warning is emitted. An
+`onheaders` callback counts as a consumer when the negotiated application
+protocol supports it (e.g. HTTP/3), because it is invoked for every incoming
+request stream. Other stream-level callbacks (`ontrailers`, `oninfo`,
+`onwanttrailers`) do not, since they are conditional or outbound-only and
+would leave the stream unobservable. An HTTP/3 server that handles requests
+entirely through `onheaders` does not need to set `onstream`.
+
 ### `session.ondatagram`
 
 <!-- YAML
@@ -4008,7 +4023,9 @@ import { listen } from 'node:quic';
 const encoder = new TextEncoder();
 
 const endpoint = await listen((session) => {
-  // The session.onstream callback fires for each new client-initiated stream.
+  // The session.onstream callback fires for each new client-initiated
+  // stream. It is optional here: with `onheaders` configured below,
+  // request streams are consumed through that callback.
 }, {
   sni: { '*': { keys: [defaultKey], certs: [defaultCert] } },
   // ALPN defaults to 'h3'.
@@ -4642,5 +4659,6 @@ throughput issues caused by flow control.
 [`stream.writer`]: #streamwriter
 [`writer.fail()`]: #streamwriter
 [`writer.fail(reason)`]: #streamwriter
+[minimal HTTP/3 server]: #minimal-http3-server
 [qlog]: https://datatracker.ietf.org/doc/draft-ietf-quic-qlog-main-schema/
 [qvis]: https://qvis.quictools.info/

lib/internal/quic/quic.js

@@ -4129,6 +4129,24 @@ class QuicSession {
     this.#inner.verifyPeer = value;
   }
 
+  /**
+   * True if an incoming stream has a consumer registered on this session:
+   * either an onstream callback, or - when the negotiated application
+   * supports headers (e.g. HTTP/3) - session-level stream callbacks that
+   * the application layer will invoke (onheaders et al).
+   * @returns {boolean}
+   */
+  #hasStreamConsumer() {
+    if (typeof this.#inner.onstream === 'function') return true;
+    // Only onheaders is guaranteed to fire for every incoming stream when
+    // the negotiated application supports stream callbacks (e.g. HTTP/3).
+    // Other stream callbacks are conditional (ontrailers, oninfo) or
+    // outbound-only (onwanttrailers) and do not expose the stream, so they
+    // do not count as a consumer.
+    if (typeof this[kStreamCallbacks]?.onheaders !== 'function') return false;
+    return getQuicSessionState(this).streamCallbacksSupported === 1;
+  }
+
   /**
    * @param {object} handle
    * @param {number} direction
@@ -4141,10 +4159,13 @@ class QuicSession {
     // Set the default byte budget for received streams.
     stream.budget = kDefaultBudget;
 
-    // A new stream was received. If we don't have an onstream callback, then
-    // there's nothing we can do about it. Destroy the stream in this case.
-    if (typeof inner.onstream !== 'function') {
-      process.emitWarning('A new stream was received but no onstream callback was provided');
+    // A new stream was received. If the session has no consumer for it -
+    // neither an onstream callback nor, on a session whose application
+    // supports headers (e.g. HTTP/3), any session-level stream callbacks -
+    // there's nothing that could ever read it. Destroy the stream in this
+    // case rather than letting it hold flow control credit.
+    if (!this.#hasStreamConsumer()) {
+      process.emitWarning('A new stream was received but no stream consumer callback was provided');
       stream.destroy();
       return;
     }
@@ -4175,7 +4196,14 @@ class QuicSession {
       });
     }
 
-    safeCallbackInvoke(inner.onstream, this, stream);
+    // Deliver the stream to the onstream consumer if one is registered.
+    // Reaching this point without one means #hasStreamConsumer accepted
+    // the stream on behalf of the application layer: the session-level
+    // stream callbacks were applied above and the application (e.g.
+    // HTTP/3) drives the stream, so there is nothing to invoke here.
+    if (typeof inner.onstream === 'function') {
+      safeCallbackInvoke(inner.onstream, this, stream);
+    }
   }
 
   [kRemoveStream](stream) {

lib/internal/quic/state.js

@@ -72,6 +72,7 @@ const {
   IDX_STATE_SESSION_STREAM_OPEN_ALLOWED,
   IDX_STATE_SESSION_PRIORITY_SUPPORTED,
   IDX_STATE_SESSION_HEADERS_SUPPORTED,
+  IDX_STATE_SESSION_STREAM_CALLBACKS_SUPPORTED,
   IDX_STATE_SESSION_WRAPPED,
   IDX_STATE_SESSION_APPLICATION_TYPE,
   IDX_STATE_SESSION_NO_ERROR_CODE,
@@ -119,6 +120,7 @@ assert(IDX_STATE_SESSION_HANDSHAKE_CONFIRMED !== undefined);
 assert(IDX_STATE_SESSION_STREAM_OPEN_ALLOWED !== undefined);
 assert(IDX_STATE_SESSION_PRIORITY_SUPPORTED !== undefined);
 assert(IDX_STATE_SESSION_HEADERS_SUPPORTED !== undefined);
+assert(IDX_STATE_SESSION_STREAM_CALLBACKS_SUPPORTED !== undefined);
 assert(IDX_STATE_SESSION_WRAPPED !== undefined);
 assert(IDX_STATE_SESSION_APPLICATION_TYPE !== undefined);
 assert(IDX_STATE_SESSION_NO_ERROR_CODE !== undefined);
@@ -493,6 +495,19 @@ class QuicSessionState {
     return DataViewPrototypeGetUint8(handle, this.#offset + IDX_STATE_SESSION_HEADERS_SUPPORTED);
   }
 
+  /**
+   * Whether the negotiated application dispatches the session-level
+   * stream callbacks (onheaders et al) for incoming streams.
+   * Returns 0 (unknown), 1 (supported), or 2 (not supported).
+   * @type {number}
+   */
+  get streamCallbacksSupported() {
+    const handle = this.#handle;
+    if (handle === undefined) return undefined;
+    return DataViewPrototypeGetUint8(
+      handle, this.#offset + IDX_STATE_SESSION_STREAM_CALLBACKS_SUPPORTED);
+  }
+
   /** @type {boolean} */
   get isWrapped() {
     const handle = this.#handle;

src/quic/application.h

@@ -210,6 +210,11 @@ class Session::Application : public MemoryRetainer {
   // do not support headers should return false (the default).
   virtual bool SupportsHeaders() const { return false; }
 
+  // True if this application dispatches the session-level stream
+  // callbacks (onheaders et al) for incoming streams when they are
+  // registered on the session.
+  virtual bool SupportsStreamCallbacks() const { return false; }
+
   // Initiates application-level graceful shutdown signaling (e.g.,
   // HTTP/3 GOAWAY). Called when Session::Close(GRACEFUL) is invoked.
   virtual void BeginShutdown() {}

src/quic/defs.h

@@ -328,6 +328,12 @@ enum class HeadersSupportState : uint8_t {
   UNSUPPORTED,
 };
 
+enum class StreamCallbacksSupportState : uint8_t {
+  UNKNOWN,
+  SUPPORTED,
+  UNSUPPORTED,
+};
+
 enum class PathValidationResult : uint8_t {
   SUCCESS = NGTCP2_PATH_VALIDATION_RESULT_SUCCESS,
   FAILURE = NGTCP2_PATH_VALIDATION_RESULT_FAILURE,

src/quic/http3.cc

@@ -202,6 +202,8 @@ class Http3ApplicationImpl final : public Session::Application {
 
   bool SupportsHeaders() const override { return true; }
 
+  bool SupportsStreamCallbacks() const override { return true; }
+
   bool is_started() const override { return started_; }
 
   bool Start() override {

src/quic/session.cc

@@ -136,6 +136,7 @@ uint64_t MaxDatagramPayload(uint64_t max_frame_size) {
   V(STREAM_OPEN_ALLOWED, stream_open_allowed, uint8_t)                         \
   V(PRIORITY_SUPPORTED, priority_supported, uint8_t)                           \
   V(HEADERS_SUPPORTED, headers_supported, uint8_t)                             \
+  V(STREAM_CALLBACKS_SUPPORTED, stream_callbacks_supported, uint8_t)           \
   V(WRAPPED, wrapped, uint8_t)                                                 \
   V(APPLICATION_TYPE, application_type, uint8_t)                               \
   V(NO_ERROR_CODE, no_error_code, error_code)                                  \
@@ -2649,6 +2650,10 @@ void Session::SetApplication(std::unique_ptr<Application> app) {
   impl_->state()->headers_supported = static_cast<uint8_t>(
       app->SupportsHeaders() ? HeadersSupportState::SUPPORTED
                              : HeadersSupportState::UNSUPPORTED);
+  impl_->state()->stream_callbacks_supported =
+      static_cast<uint8_t>(app->SupportsStreamCallbacks()
+                               ? StreamCallbacksSupportState::SUPPORTED
+                               : StreamCallbacksSupportState::UNSUPPORTED);
   // Surface the application's "no error" and "internal error" codes via
   // session state so that JS-side code (e.g. the stream writer's fail()
   // path) can resolve the right wire code for the negotiated ALPN