diff --git a/doc/api/quic.md b/doc/api/quic.md index dad86a144682..fa09611a9f4e 100644 --- a/doc/api/quic.md +++ b/doc/api/quic.md @@ -1076,7 +1076,9 @@ session will be closed. If `error` is provided and [`session.onerror`][] is set, the `onerror` callback is invoked before destruction. The `session.closed` promise will reject with the error. If `options` is provided, the `CONNECTION_CLOSE` frame sent to the peer will include the -specified error code, type, and reason. +specified error code, type, and reason. The `CONNECTION_CLOSE` ends the +streams that are still open along with the session, so no `RESET_STREAM` or +`STOP_SENDING` frame is sent for them. ### `session.destroyed` diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index 683eba6925ea..a3046385bc8a 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -1574,6 +1574,7 @@ function isSyncIterable(obj) { // Functions used specifically for internal or assertion purposes only. let getQuicStreamState; let getQuicSessionState; +let isSessionEndingStreams; let getQuicEndpointState; let assertIsQuicEndpoint; let assertIsQuicStream; @@ -2114,13 +2115,17 @@ class QuicStream { error.errorCode : getQuicSessionState(inner.session).internalErrorCode; } + // A CONNECTION_CLOSE that ends the session ends this stream as well, + // so neither frame below is sent then. + const sendsFrames = !isSessionEndingStreams(inner.session); // When destroying with an error, ensure the peer stops sending // data we are about to discard by emitting STOP_SENDING. The // condition gates the emission to error-path destroys with a // still-open readable side. The C++ state.readEnded flag is // authoritative -- it is set for locally-initiated uni streams // (which have no readable side) and when reading completes. - if (abortCode !== undefined && + if (sendsFrames && + abortCode !== undefined && !inner.state.readEnded) { this.#handle.stopSending(abortCode); } @@ -2131,7 +2136,8 @@ class QuicStream { // (e.g. used setBody or never wrote at all) need an explicit // RESET_STREAM here so the write side does not dangle on the // wire. The C++ state.writeEnded flag is authoritative. - if (abortCode !== undefined && + if (sendsFrames && + abortCode !== undefined && inner.writer === undefined && !inner.state.writeEnded) { this.#handle.resetStream(abortCode); @@ -2482,10 +2488,16 @@ class QuicStream { // `H3_INTERNAL_ERROR` (0x102); for raw QUIC applications // it falls back to the QUIC transport-layer // `INTERNAL_ERROR` (0x1). - const code = QuicError.isQuicError(error) ? - error.errorCode : - getQuicSessionState(stream.#inner.session).internalErrorCode; - handle.resetStream(code); + // + // No RESET_STREAM is sent when a CONNECTION_CLOSE that ends the + // session ends the stream as well. + const session = stream.#inner.session; + if (!isSessionEndingStreams(session)) { + const code = QuicError.isQuicError(error) ? + error.errorCode : + getQuicSessionState(session).internalErrorCode; + handle.resetStream(code); + } if (drainWakeup != null) { markPromiseAsHandled(drainWakeup.promise); drainWakeup.reject(error); @@ -2901,6 +2913,7 @@ class QuicSession { isPendingClose: false, selfInitiatedClose: false, destroying: false, + destroyingWithConnectionClose: false, handshakeCompleted: false, pendingClose: PromiseWithResolvers(), pendingOpen: PromiseWithResolvers(), @@ -2957,6 +2970,22 @@ class QuicSession { assertIsQuicSession(session); return session.#inner.state; }; + + // Whether the session is being closed immediately with a CONNECTION_CLOSE. + // The connection then enters the closing state and its streams end with it + // (RFC 9000, Section 10.2), so they are torn down without sending + // RESET_STREAM or STOP_SENDING ahead of the CONNECTION_CLOSE. That is the + // case when destroy() is given a close code, and when the native side + // closes the session other than gracefully or silently. + isSessionEndingStreams = function(session) { + if (session === undefined) return false; + const inner = session.#inner; + if (inner.destroyingWithConnectionClose) return true; + const state = inner.state; + return state?.isClosing === true && + !state.isGracefulClose && + !state.isSilentClose; + }; } /** @@ -3790,6 +3819,10 @@ class QuicSession { } } + // A close code makes the native side send a CONNECTION_CLOSE, which ends + // the streams destroyed below without frames of their own. + inner.destroyingWithConnectionClose = options?.code !== undefined; + // First, forcefully and immediately destroy all open streams, if any. for (const stream of inner.streams) { stream.destroy(error); diff --git a/test/parallel/test-quic-session-destroy-no-stream-reset.mjs b/test/parallel/test-quic-session-destroy-no-stream-reset.mjs new file mode 100644 index 000000000000..9842180f540d --- /dev/null +++ b/test/parallel/test-quic-session-destroy-no-stream-reset.mjs @@ -0,0 +1,72 @@ +// Flags: --experimental-quic --no-warnings + +// session.destroy() with a close code sends a CONNECTION_CLOSE, which puts +// the connection into the closing state and ends all of its streams with it +// (RFC 9000, Section 10.2). A stream that is still open must therefore not +// be reset with a RESET_STREAM of its own ahead of the CONNECTION_CLOSE: the +// peer would see the stream fail, with an unrelated error code, before it +// learns how the session closed. +// Refs: https://github.com/nodejs/node/issues/67278 + +import { hasQuic, skip, mustCall, mustNotCall } from '../common/index.mjs'; +import assert from 'node:assert'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { listen, connect } = await import('../common/quic.mjs'); + +async function test(error, closeOptions, checkServerClosed) { + const serverReceived = Promise.withResolvers(); + const serverClosed = Promise.withResolvers(); + + const serverEndpoint = await listen(mustCall((serverSession) => { + serverClosed.resolve(checkServerClosed(serverSession.closed)); + serverSession.onstream = mustCall(async (stream) => { + stream.onreset = mustNotCall('the stream must not be reset'); + stream.closed.catch(() => {}); + try { + for await (const chunk of stream) { + assert.ok(chunk.length > 0); + serverReceived.resolve(); + } + } catch { + // The stream ends with the session. + } + }); + })); + + const resetFrames = []; + const clientSession = await connect(serverEndpoint.address, { + reuseEndpoint: false, + qlog: true, + onqlog(data) { + for (const [frame] of String(data).matchAll(/"frame_type":"reset_stream"/g)) { + resetFrames.push(frame); + } + }, + }); + await clientSession.opened; + + // A long-lived stream that is still open when the session is destroyed. + const stream = await clientSession.createUnidirectionalStream(); + stream.closed.catch(() => {}); + stream.writer.write(new Uint8Array([1, 2, 3])); + await serverReceived.promise; + + clientSession.destroy(error, closeOptions); + await serverClosed.promise; + + assert.deepStrictEqual(resetFrames, []); + await serverEndpoint.close(); +} + +// No error: the peer sees the session close cleanly with code 0. +await test(undefined, { code: 0, type: 'application' }, (closed) => closed); + +// With an error, the peer sees the close code, not a stream reset. +await test(new Error('boom'), { code: 5n, type: 'application' }, + (closed) => assert.rejects(closed, { + code: 'ERR_QUIC_APPLICATION_ERROR', + }));