quic: retain readers for readable streams

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

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;
@@ -2061,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

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);
+  }
+}

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();