quic: do not destroy incoming streams that have a consumer

An incoming stream was destroyed unless the session had an onstream
callback, even when session-level stream callbacks (onheaders et al)
were registered and the negotiated application (HTTP/3) would drive
the stream through them. Users had to register stub onstream handlers
just to keep their streams alive.

Destroy an incoming stream only when the session has no consumer for
it at all: no onstream callback, and no session-level stream callbacks
runnable on the negotiated application (checked via the existing
headersSupported session state, computed when the application is
selected from ALPN). Sessions with no consumers keep the current
destroy-and-warn behavior so unconsumed streams cannot accumulate and
hold flow control credit.

On HTTP/3 sessions only bidirectional request streams reach this path;
control and QPACK streams are consumed internally by nghttp3 and are
never exposed to JavaScript.

Fixes: #64192

Signed-off-by: Naman Trivedi <trivenay@amazon.com>

Author: trivenay

lib/internal/quic/quic.js

@@ -4129,6 +4129,19 @@ 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;
+    if (this[kStreamCallbacks] == null) return false;
+    return getQuicSessionState(this).headersSupported === 1;
+  }
+
   /**
    * @param {object} handle
    * @param {number} direction
@@ -4141,10 +4154,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 +4191,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) {

test/parallel/test-quic-h3-stream-without-onstream.mjs

@@ -0,0 +1,131 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: incoming stream consumer checks.
+// An incoming stream must not be destroyed just because `onstream` is
+// not set: on a session whose application supports headers (HTTP/3),
+// session-level stream callbacks (`onheaders` et al) are a consumer
+// and the stream must be kept and driven by the application layer.
+// Refs: https://github.com/nodejs/node/issues/64192
+//
+// A session with no stream consumers at all still destroys incoming
+// streams (and emits a warning), so unconsumed streams cannot
+// accumulate and hold flow control credit.
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import * as fixtures from '../common/fixtures.mjs';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+const { text } = await import('stream/iter');
+
+const key = createPrivateKey(fixtures.readKey('agent1-key.pem'));
+const cert = fixtures.readKey('agent1-cert.pem');
+
+// The consumer warning must never fire in the first block (onheaders is
+// a consumer) and must fire in the second (no runnable consumer).
+// common.expectWarning is not usable here: importing node:quic emits
+// ExperimentalWarning, which it would reject as unexpected.
+const kWarning =
+  'A new stream was received but no stream consumer callback was provided';
+function failOnConsumerWarning(warning) {
+  assert.notStrictEqual(warning.message, kWarning);
+}
+
+// --- An h3 request completes with only session-level stream callbacks ---
+{
+  process.on('warning', failOnConsumerWarning);
+  const serverDone = Promise.withResolvers();
+
+  // Note: no `onstream` callback anywhere on this session.
+  const serverEndpoint = await listen(mustCall((serverSession) => {
+    serverSession.onerror = () => {};
+  }), {
+    sni: { '*': { keys: [key], certs: [cert] } },
+    onheaders: mustCall(function(headers) {
+      assert.strictEqual(headers[':path'], '/test');
+      this.sendHeaders({
+        ':status': '200',
+        'content-type': 'text/plain',
+      });
+      const w = this.writer;
+      w.writeSync('kept without onstream');
+      w.endSync();
+      serverDone.resolve();
+    }),
+  });
+
+  const clientSession = await connect(serverEndpoint.address, {
+    servername: 'localhost',
+    verifyPeer: 'manual',
+  });
+  await clientSession.opened;
+
+  const headersReceived = Promise.withResolvers();
+  const stream = await clientSession.createBidirectionalStream({
+    headers: {
+      ':method': 'GET',
+      ':path': '/test',
+      ':scheme': 'https',
+      ':authority': 'localhost',
+    },
+    onheaders: mustCall((headers) => {
+      assert.strictEqual(headers[':status'], 200);
+      headersReceived.resolve();
+    }),
+  });
+
+  await headersReceived.promise;
+  const body = await text(stream);
+  assert.strictEqual(body, 'kept without onstream');
+
+  await serverDone.promise;
+  await clientSession.close();
+  await serverEndpoint.close();
+  process.off('warning', failOnConsumerWarning);
+}
+
+// --- Stream callbacks that cannot run are not a consumer ---
+// On a session whose negotiated application does not support headers,
+// registered session-level stream callbacks can never fire, so an
+// incoming stream with no onstream callback is destroyed with the
+// warning. The h3 block above must not trigger that warning.
+{
+  // Awaiting warned.promise is the assertion: the test times out if the
+  // warning never fires.
+  const warned = Promise.withResolvers();
+  process.on('warning', function onWarning(warning) {
+    if (warning.message === kWarning) {
+      process.off('warning', onWarning);
+      warned.resolve();
+    }
+  });
+
+  // The onheaders callback is registered but the ALPN is not h3,
+  // so it can never run.
+  const serverEndpoint = await listen(mustCall((serverSession) => {
+    serverSession.onerror = () => {};
+  }), {
+    sni: { '*': { keys: [key], certs: [cert] } },
+    alpn: ['test-proto'],
+    onheaders: () => {},
+  });
+
+  const clientSession = await connect(serverEndpoint.address, {
+    servername: 'localhost',
+    alpn: 'test-proto',
+    verifyPeer: 'manual',
+  });
+  await clientSession.opened;
+
+  const stream = await clientSession.createUnidirectionalStream();
+  stream.writer.writeSync('x');
+
+  await warned.promise;
+  await clientSession.close();
+  await serverEndpoint.close();
+}