From f057aabf68e7fe89a416819909d274f1c3c601df Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Wed, 30 Sep 2026 16:58:30 +0900 Subject: [PATCH 1/2] quic: retain readers for readable streams Create readers when readable QuicStream instances are constructed so data received before the first async iterator pull survives native stream cleanup. Add a regression test for a peer-initiated unidirectional stream that sends data and FIN before the reader is pulled. Assisted-by: Codex Signed-off-by: GiHoon1123 --- lib/internal/quic/quic.js | 6 +++ .../test-quic-stream-uni-delayed-reader.mjs | 47 +++++++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 test/parallel/test-quic-stream-uni-delayed-reader.mjs diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index 6603bd452899..b83b69a884d3 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -1678,6 +1678,12 @@ class QuicStream { this.#handle = handle; handle[kOwner] = this; const inner = this.#inner; + // Keep a reader alive for readable streams before the native handle can + // be destroyed. A peer may send data and FIN before the consumer asks + // for its async iterator. + if (!isLocal || direction === kStreamDirectionBidirectional) { + inner.reader = handle.getReader(); + } inner.session = session; inner.direction = direction; inner.isLocal = isLocal; diff --git a/test/parallel/test-quic-stream-uni-delayed-reader.mjs b/test/parallel/test-quic-stream-uni-delayed-reader.mjs new file mode 100644 index 000000000000..45a233617161 --- /dev/null +++ b/test/parallel/test-quic-stream-uni-delayed-reader.mjs @@ -0,0 +1,47 @@ +// Flags: --experimental-quic --experimental-stream-iter --no-warnings + +// Test: data received before the async iterator is pulled remains readable +// after a peer-initiated unidirectional stream closes. + +import { hasQuic, skip, mustCall } from '../common/index.mjs'; +import assert from 'node:assert'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { listen, connect } = await import('../common/quic.mjs'); +const { bytes } = await import('stream/iter'); + +const encoder = new TextEncoder(); +const expected = encoder.encode('data before the first pull'); +const done = Promise.withResolvers(); + +const serverEndpoint = await listen(mustCall(async (serverSession) => { + await serverSession.opened; + const stream = await serverSession.createUnidirectionalStream({ + body: expected, + }); + await stream.closed; + serverSession.close(); +})); + +const clientSession = await connect(serverEndpoint.address); +await clientSession.opened; + +clientSession.onstream = mustCall(async (stream) => { + const iterator = stream[Symbol.asyncIterator](); + + // Delay the first pull until after the peer has sent FIN and the native + // stream handle has been closed. + await stream.closed; + + const received = await bytes(iterator); + assert.deepStrictEqual(received, expected); + clientSession.close(); + done.resolve(); +}); + +await done.promise; +await clientSession.closed; +await serverEndpoint.close(); From c3cf3300a115fe4f699da1d5ff10c8790a13018f Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Sat, 24 Oct 2026 17:31:06 +0900 Subject: [PATCH 2/2] test: cover QUIC reads after local teardown Wait for the full payload before destroying a session or stream and check that buffered data remains readable. Cover explicit errors after multiple reader batches under both truncated-read policies, and clarify the destroy documentation. Assisted-by: Codex Signed-off-by: GiHoon1123 --- doc/api/quic.md | 6 +++ lib/internal/quic/quic.js | 4 +- ...st-quic-stream-truncated-reads-destroy.mjs | 42 +++++++++++++++---- 3 files changed, 43 insertions(+), 9 deletions(-) diff --git a/doc/api/quic.md b/doc/api/quic.md index 6df3f62da170..dad86a144682 100644 --- a/doc/api/quic.md +++ b/doc/api/quic.md @@ -1993,6 +1993,11 @@ Immediately and abruptly destroys the stream. If `error` is provided and [`stream.onerror`][] is set, the `onerror` callback is invoked before destruction. The `stream.closed` promise rejects with the error. +Buffered incoming data remains readable after destruction, even if iteration +has not started. If the readable side ended without a FIN, the iterator yields +that data before reporting the truncation or error according to +[`sessionOptions.truncatedReads`][]. + When the stream is destroyed with an `error` (or with an explicit `options.code`), the QUIC stack signals the abort to the peer: @@ -4888,6 +4893,7 @@ throughput issues caused by flow control. [`sessionOptions.sessionTicket`]: #sessionoptionssessionticket [`sessionOptions.sni`]: #sessionoptionssni-server-only [`sessionOptions.token`]: #sessionoptionstoken-client-only +[`sessionOptions.truncatedReads`]: #sessionoptionstruncatedreads [`stream.destroy()`]: #streamdestroyerror-options [`stream.headers`]: #streamheaders [`stream.onerror`]: #streamonerror diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index b83b69a884d3..d34196ff938b 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -2067,8 +2067,8 @@ class QuicStream { } /** - * Immediately destroys the stream. Any queued data is discarded. If - * an error is given, the closed promise will be rejected with that + * Immediately destroys the stream. Buffered incoming data remains readable. + * If an error is given, the closed promise will be rejected with that * error. If no error is given, the closed promise will be resolved. * When destroying with an error, RESET_STREAM and/or STOP_SENDING * are emitted to the peer for any still-open writable / readable diff --git a/test/parallel/test-quic-stream-truncated-reads-destroy.mjs b/test/parallel/test-quic-stream-truncated-reads-destroy.mjs index 01eddd3b0f0f..f7b7ae981ba1 100644 --- a/test/parallel/test-quic-stream-truncated-reads-destroy.mjs +++ b/test/parallel/test-quic-stream-truncated-reads-destroy.mjs @@ -43,17 +43,16 @@ const serve = (stream) => { stream.setBody(stallingBody(1000)); }; assert.strictEqual(threw.errorCode, 0n); } -// The session is destroyed before iteration even starts: the buffered data -// is gone, but the truncation is still reported rather than a clean empty -// end. +// Destroying the session before iteration preserves the buffered data, +// but still reports truncation after it has been read. { const { received, threw } = await readStream(serve, { beforeIterate: async ({ stream, session }) => { - while (stream.stats.bytesReceived === 0n) await setTimeout(5); + while (stream.stats.bytesReceived < 1000n) await setTimeout(5); session.destroy(); }, }); - assert.strictEqual(received, 0); + assert.strictEqual(received, 1000); assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); } @@ -62,11 +61,40 @@ const serve = (stream) => { stream.setBody(stallingBody(1000)); }; { const { received, threw, closedError } = await readStream(serve, { beforeIterate: async ({ stream }) => { - while (stream.stats.bytesReceived === 0n) await setTimeout(5); + while (stream.stats.bytesReceived < 1000n) await setTimeout(5); stream.destroy(); }, }); - assert.strictEqual(received, 0); + assert.strictEqual(received, 1000); assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); assert.strictEqual(closedError, undefined); } + +// An explicit error must survive draining the buffered data, even when +// truncation itself is ignored. Use more than one reader batch of data. +for (const target of ['session', 'stream']) { + for (const truncatedReads of ['error', 'ignore']) { + const size = 128 * 1024; + const boom = new Error(`${target}: ${truncatedReads}`); + const { received, threw, closedError } = await readStream((stream) => { + stream.setBody(stallingBody(size)); + }, { + clientOptions: { + truncatedReads, + transportParams: { + initialMaxStreamDataBidiLocal: size * 2, + initialMaxData: size * 2, + }, + }, + beforeIterate: async (context) => { + while (context.stream.stats.bytesReceived < BigInt(size)) { + await setTimeout(5); + } + context[target].destroy(boom); + }, + }); + assert.strictEqual(received, size); + assert.strictEqual(threw, boom); + assert.strictEqual(closedError, boom); + } +}