diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index f645998e628d..fbb15a4fb8f0 100644 --- a/lib/internal/quic/quic.js +++ b/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).streamCallbacksSupported === 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) { diff --git a/lib/internal/quic/state.js b/lib/internal/quic/state.js index 7d2ff3a96c6b..a41790c52024 100644 --- a/lib/internal/quic/state.js +++ b/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; diff --git a/src/quic/application.h b/src/quic/application.h index 619b41dd8d0b..dec5cffb4243 100644 --- a/src/quic/application.h +++ b/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() {} diff --git a/src/quic/defs.h b/src/quic/defs.h index 75ae915335be..5184288475b1 100644 --- a/src/quic/defs.h +++ b/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, diff --git a/src/quic/http3.cc b/src/quic/http3.cc index b6d876af60f6..2cf4e2d6fdd1 100644 --- a/src/quic/http3.cc +++ b/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 { diff --git a/src/quic/session.cc b/src/quic/session.cc index 1bea15fbadb4..60b27fbfa358 100644 --- a/src/quic/session.cc +++ b/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 app) { impl_->state()->headers_supported = static_cast( app->SupportsHeaders() ? HeadersSupportState::SUPPORTED : HeadersSupportState::UNSUPPORTED); + impl_->state()->stream_callbacks_supported = + static_cast(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 diff --git a/test/parallel/test-quic-h3-stream-without-onstream.mjs b/test/parallel/test-quic-h3-stream-without-onstream.mjs new file mode 100644 index 000000000000..67cd4d4f04b2 --- /dev/null +++ b/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(); +}