quic: fix readable stream truncation on stop-sending, abort & timeout

Signed-off-by: Tim Perry <pimterry@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/63967
Reviewed-By: James M Snell <jasnell@gmail.com>

doc/api/quic.md

@@ -3438,6 +3438,30 @@ value, PING frames will be sent automatically to keep the connection alive
 before the idle timeout fires. The value should be less than the effective
 idle timeout (`maxIdleTimeout` transport parameter) to be useful.
 
+#### `sessionOptions.truncatedReads`
+
+* Type: {string} One of `'error'` or `'ignore'`.
+* **Default:** `'error'`
+
+Controls how reading a stream reports a truncated read. A stream's read side
+can end without receiving a QUIC FIN, meaning the peer never signalled that
+the whole stream had been sent and the data received may be incomplete. This
+selects how the stream's async iterator reports this:
+
+* `'error'` - The default. Peers are expected to always send a FIN to end
+  their data explicitly, and so any truncation is an error. The iterator yields
+  the data that did arrive and then throws, so an incomplete stream can never
+  be mistaken for a complete one. Incomplete streams will either throw a
+  `ERR_QUIC_STREAM_RESET` carrying the peer's error code, a connection error,
+  or `ERR_QUIC_STREAM_ABORTED` for other cases.
+
+* `'ignore'` - The truncation itself is ignored: only a stream or connection
+  error is reported, and any clean abort/cancellation or similar simply ends
+  the stream. A non-zero peer reset, non-zero local stop-sending or connection
+  error still fails, but a truncation with no error at all (an idle timeout,
+  a graceful close, or a plain `stopSending()`) ends the read cleanly with the
+  data received. This matches `stream.closed`, which rejects only on an error.
+
 #### `sessionOptions.verifyPeer` (client only)
 
 * Type: {string} One of `'strict'`, `'auto'`, or `'manual'`.

lib/internal/blob.js

@@ -631,7 +631,7 @@ const kMaxBatchChunks = 16;
 const kDefaultMaxBatchBytes = 65536;
 
 async function* createBlobReaderIterable(reader, options = kEmptyObject) {
-  const { getReadError, maxBatchBytes = kDefaultMaxBatchBytes } = options;
+  const { maxBatchBytes = kDefaultMaxBatchBytes } = options;
   let wakeup = PromiseWithResolvers();
   let immediate;
   reader.setWakeup(() => {
@@ -664,9 +664,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
           break;
         }
         if (pullResult.status < 0) {
-          error = typeof getReadError === 'function' ?
-            getReadError(pullResult.status) :
-            new ERR_INVALID_STATE('The reader is not readable');
+          error = new ERR_INVALID_STATE('The reader is not readable');
           break;
         }
         if (pullResult.status === 2) {

lib/internal/errors.js

@@ -1718,8 +1718,6 @@ E('ERR_QUIC_CONNECTION_FAILED', 'QUIC connection failed', Error);
 E('ERR_QUIC_ENDPOINT_CLOSED', 'QUIC endpoint closed: %s (%d)', Error);
 E('ERR_QUIC_OPEN_STREAM_FAILED', 'Failed to open QUIC stream', Error);
 E('ERR_QUIC_STREAM_ABORTED', '%s', Error);
-E('ERR_QUIC_STREAM_RESET',
-  'The QUIC stream was reset by the peer with error code %d', Error);
 E('ERR_QUIC_VERSION_NEGOTIATION_ERROR', 'The QUIC session requires version negotiation', Error);
 E('ERR_REQUIRE_ASYNC_MODULE', function(filename, parent, locations) {
   let message = 'require() cannot be used on an ESM graph with top-level await. Use import() instead.';

lib/internal/quic/quic.js

@@ -13,7 +13,6 @@ const {
   ErrorCaptureStackTrace,
   FunctionPrototypeBind,
   FunctionPrototypeCall,
-  Number,
   ObjectDefineProperties,
   ObjectKeys,
   PromisePrototypeThen,
@@ -117,7 +116,6 @@ const {
     ERR_QUIC_ENDPOINT_CLOSED,
     ERR_QUIC_OPEN_STREAM_FAILED,
     ERR_QUIC_STREAM_ABORTED,
-    ERR_QUIC_STREAM_RESET,
     ERR_QUIC_VERSION_NEGOTIATION_ERROR,
   },
 } = require('internal/errors');
@@ -401,6 +399,7 @@ const endpointRegistry = new SafeSet();
  * @property {number} [minVersion] The minimum acceptable QUIC version
  * @property {'use'|'ignore'|'default'} [preferredAddressPolicy] The preferred address policy
  * @property {'strict'|'auto'|'manual'} [verifyPeer='auto'] Peer certificate verification policy (client only)
+ * @property {'error'|'ignore'} [truncatedReads] Truncated read policy
  * @property {ApplicationOptions} [application] The application options
  * @property {TransportParams} [transportParams] The transport parameters
  * @property {string} [servername] The server name identifier (client only)
@@ -1600,6 +1599,9 @@ class QuicStream {
     state: undefined,
     stats: undefined,
     pendingClose: undefined,
+    destroyError: undefined,
+    stopSendingCode: undefined,
+    truncatedReads: undefined,
     reader: undefined,
     destroying: false,
     iteratorLocked: false,
@@ -1648,9 +1650,10 @@ class QuicStream {
    * @param {object} handle
    * @param {QuicSession} session
    * @param {number} direction
-   * @param {boolean} [isLocal]
+   * @param {boolean} isLocal
+   * @param {'error'|'ignore'} truncatedReads
    */
-  constructor(privateSymbol, handle, session, direction, isLocal) {
+  constructor(privateSymbol, handle, session, direction, isLocal, truncatedReads) {
     assertPrivateSymbol(privateSymbol);
 
     this.#handle = handle;
@@ -1659,6 +1662,7 @@ class QuicStream {
     inner.session = session;
     inner.direction = direction;
     inner.isLocal = isLocal;
+    inner.truncatedReads = truncatedReads;
     inner.state = new QuicStreamState(
       kPrivateConstructor, handle.state, handle.stateByteOffset);
 
@@ -1690,34 +1694,49 @@ class QuicStream {
     inner.iteratorLocked = true;
 
     inner.reader ??= this.#handle?.getReader();
-    // Non-readable stream (outbound-only unidirectional, or closed)
-    if (!inner.reader) return;
+    // No reader means either an outbound-only unidirectional stream, or a
+    // stream already destroyed (data gone, but truncation must still be
+    // checked below).
+    if (inner.reader) {
+      yield* createBlobReaderIterable(inner.reader);
+    }
 
-    yield* createBlobReaderIterable(inner.reader, {
-      getReadError: () => {
-        // The read side ends for one of three reasons:
-        //   * Clean FIN received from the peer (state.finReceived
-        //     === true). The iterator stops without calling this;
-        //     fall through to the generic state error if it does.
-        //   * Peer sent us a RESET_STREAM. The C++ side records the
-        //     code in state.resetCode regardless of whether the JS
-        //     onreset handler was attached. state.finReceived stays
-        //     false because no FIN was seen.
-        //   * We aborted locally via stream.resetStream() or
-        //     stream.stopSending(). Both paths run EndReadable in
-        //     C++, setting state.readEnded without setting
-        //     state.finReceived. There is no peer code to surface.
-        if (inner.state.readEnded && !inner.state.finReceived) {
-          const peerResetCode = inner.state.resetCode;
-          if (peerResetCode !== undefined && peerResetCode > 0n) {
-            return new ERR_QUIC_STREAM_RESET(Number(peerResetCode));
-          }
-          return new ERR_QUIC_STREAM_ABORTED(
-            'Stream aborted before FIN was received');
-        }
-        return new ERR_INVALID_STATE('The stream is not readable');
-      },
-    });
+    if (inner.state.readEnded && !inner.state.finReceived) {
+      // The readable has been truncated - ended with no clean FIN. We expose
+      // this in different ways depending on the truncatedReads option.
+
+      // If we cancelled ourselves, check our own stop-sending code (not the
+      // result mirrored by the remote peer)
+      if (inner.stopSendingCode > 0n) {
+        throw new QuicError('Stream aborted before FIN was received',
+                            { __proto__: null,
+                              errorCode: inner.stopSendingCode });
+      }
+
+      // Non-zero reset is always an error:
+      const peerResetCode = inner.state.resetCode;
+      if (peerResetCode > 0n) {
+        throw new QuicError(
+          `The QUIC stream was reset by the peer with error code ${peerResetCode}`,
+          { __proto__: null,
+            code: 'ERR_QUIC_STREAM_RESET',
+            errorCode: peerResetCode });
+      }
+
+      // If stream teardown has started (stats.destroyedAt is set) then a
+      // close event confirming a final error/clean close will settle
+      // imminently (might be settled already). We await here to rethrow
+      // any errors if the connection has failed.
+      if (this.destroyed || this.stats.destroyedAt !== 0n) {
+        await this.closed;
+      }
+
+      // Clean abort is truncation, but not necessarily an error:
+      if (inner.truncatedReads === 'error') {
+        throw new QuicError('Stream aborted before FIN was received',
+                            { __proto__: null, errorCode: peerResetCode ?? 0n });
+      }
+    }
   }
 
   /**
@@ -2099,9 +2118,11 @@ class QuicStream {
     if (error !== undefined && typeof inner.onerror === 'function') {
       invokeOnerror(inner.onerror, error);
     }
-    const handle = this.#handle;
-    this[kFinishClose](error);
-    handle.destroy();
+    // handle.destroy() kicks off all the cleanup internals, eventually
+    // including [kFinishClose] which needs this original destroy error:
+    inner.destroyError = error;
+    this.#handle.destroy();
+    this[kFinishClose](error); // A no-op, unless destroy() failed
   }
 
   /**
@@ -2519,7 +2540,9 @@ class QuicStream {
   stopSending(code = 0n) {
     assertIsQuicStream(this);
     if (this.destroyed) return;
-    this.#handle.stopSending(BigInt(code));
+    const abortCode = BigInt(code);
+    this.#inner.stopSendingCode = abortCode;
+    this.#handle.stopSending(abortCode);
   }
 
   /**
@@ -2612,6 +2635,9 @@ class QuicStream {
     if (this.destroyed) {
       return inner.pendingClose.promise;
     }
+    // Prefer an error staged by destroy() (the original object the caller
+    // passed) over the error delivered by the native close callback.
+    error = inner.destroyError ?? error;
     if (error !== undefined) {
       inner.pendingClose.reject(error);
     } else {
@@ -2649,6 +2675,7 @@ class QuicStream {
     inner.session = undefined;
     inner.pendingClose.reject = undefined;
     inner.pendingClose.resolve = undefined;
+    inner.destroyError = undefined;
     inner.onblocked = undefined;
     inner.onreset = undefined;
     inner.onstopsending = undefined;
@@ -2859,6 +2886,7 @@ class QuicSession {
     // because server-side cert validation is handled by rejectUnauthorized
     // at the C++ level.
     verifyPeer: 'manual',
+    truncatedReads: 'error',
     handshakeInfo: undefined,
     /** @type {QuicSessionPath|undefined} */
     path: undefined,
@@ -2892,8 +2920,9 @@ class QuicSession {
    * @param {symbol} privateSymbol
    * @param {object} handle
    * @param {QuicEndpoint} endpoint
+   * @param {{ truncatedReads?: 'error'|'ignore' }} [options]
    */
-  constructor(privateSymbol, handle, endpoint) {
+  constructor(privateSymbol, handle, endpoint, options = kEmptyObject) {
     // Instances of QuicSession can only be created internally.
     assertPrivateSymbol(privateSymbol);
 
@@ -2902,6 +2931,8 @@ class QuicSession {
 
     const inner = this.#inner;
     inner.endpoint = endpoint;
+    const { truncatedReads } = options;
+    if (truncatedReads !== undefined) inner.truncatedReads = truncatedReads;
     // Move any qlog entries that arrived before the wrapper existed.
     if (handle._pendingQlog !== undefined) {
       inner.pendingQlog = handle._pendingQlog;
@@ -3447,7 +3478,8 @@ class QuicSession {
     }
 
     const stream = new QuicStream(
-      kPrivateConstructor, handle, this, direction, true /* isLocal */);
+      kPrivateConstructor, handle, this, direction, true /* isLocal */,
+      inner.truncatedReads);
     inner.streams.add(stream);
     if (typeof this.#inner.onerror === 'function') {
       markPromiseAsHandled(stream.closed);
@@ -4232,7 +4264,7 @@ class QuicSession {
   [kNewStream](handle, direction) {
     const inner = this.#inner;
     const stream = new QuicStream(kPrivateConstructor, handle, this, direction,
-                                  false /* isLocal */);
+                                  false /* isLocal */, inner.truncatedReads);
 
     // Set the default byte budget for received streams.
     stream.budget = kDefaultBudget;
@@ -4349,6 +4381,7 @@ class QuicEndpoint {
     sessions: new SafeSet(),
     stat: undefined,
     stats: undefined,
+    truncatedReads: undefined,
     onsession: undefined,
     sessionCallbacks: undefined,
   };
@@ -4481,8 +4514,8 @@ class QuicEndpoint {
     };
   }
 
-  #newSession(handle) {
-    const session = new QuicSession(kPrivateConstructor, handle, this);
+  #newSession(handle, options) {
+    const session = new QuicSession(kPrivateConstructor, handle, this, options);
     this.#inner.sessions.add(session);
     // Set default pending datagram queue size.
     session.maxPendingDatagrams = kDefaultMaxPendingDatagrams;
@@ -4656,9 +4689,13 @@ class QuicEndpoint {
       ontrailers,
       oninfo,
       onwanttrailers,
+      // Stored on the endpoint and applied to each incoming session.
+      truncatedReads,
       ...rest
     } = options;
 
+    inner.truncatedReads = truncatedReads;
+
     // Store session and stream callbacks to apply to each new incoming session.
     inner.sessionCallbacks = {
       __proto__: null,
@@ -4700,6 +4737,7 @@ class QuicEndpoint {
     validateObject(options, 'options');
     const {
       sessionTicket,
+      truncatedReads,
       ...rest
     } = options;
 
@@ -4708,7 +4746,7 @@ class QuicEndpoint {
     if (handle === undefined) {
       throw new ERR_QUIC_CONNECTION_FAILED();
     }
-    const session = this.#newSession(handle);
+    const session = this.#newSession(handle, { __proto__: null, truncatedReads });
     // Set callbacks before any async work to avoid missing events
     // that fire during or immediately after the handshake.
     applyCallbacks(session, options);
@@ -4942,7 +4980,8 @@ class QuicEndpoint {
     const inner = this.#inner;
     assert(typeof inner.onsession === 'function',
            'onsession callback not specified');
-    const session = this.#newSession(handle);
+    const session = this.#newSession(handle,
+                                     { __proto__: null, truncatedReads: inner.truncatedReads });
     // Apply session callbacks stored at listen time before notifying
     // the onsession callback, to avoid missing events that fire
     // during or immediately after the handshake.
@@ -5434,6 +5473,7 @@ function processSessionOptions(options, config = kEmptyObject) {
     maxDatagramSendAttempts = 5,
     streamIdleTimeout,
     verifyPeer = 'auto',
+    truncatedReads = 'error',
     // HTTP/3 application-specific options. Nested under `application`
     // to separate protocol-specific settings from transport-level ones.
     application = kEmptyObject,
@@ -5485,6 +5525,9 @@ function processSessionOptions(options, config = kEmptyObject) {
   validateOneOf(verifyPeer, 'options.verifyPeer',
                 ['strict', 'auto', 'manual']);
 
+  validateOneOf(truncatedReads, 'options.truncatedReads',
+                ['error', 'ignore']);
+
   validateInteger(drainingPeriodMultiplier, 'options.drainingPeriodMultiplier',
                   3, 255);
 
@@ -5547,6 +5590,7 @@ function processSessionOptions(options, config = kEmptyObject) {
       verifyHostname: verifyPeer !== 'manual',
     },
     verifyPeer,
+    truncatedReads,
     qlog,
     maxPayloadSize,
     unacknowledgedPacketThreshold,

lib/internal/quic/state.js

@@ -8,7 +8,9 @@ const {
   DataView,
   DataViewPrototypeGetBigInt64,
   DataViewPrototypeGetBigUint64,
+  DataViewPrototypeGetBuffer,
   DataViewPrototypeGetByteLength,
+  DataViewPrototypeGetByteOffset,
   DataViewPrototypeGetUint16,
   DataViewPrototypeGetUint32,
   DataViewPrototypeGetUint8,
@@ -17,6 +19,9 @@ const {
   DataViewPrototypeSetUint8,
   JSONStringify,
   Number,
+  TypedArrayPrototypeGetBuffer,
+  TypedArrayPrototypeSlice,
+  Uint8Array,
 } = primordials;
 
 const {
@@ -107,6 +112,7 @@ const {
   IDX_STATE_STREAM_WRITE_DESIRED_SIZE,
   IDX_STATE_STREAM_BUDGET,
   IDX_STATE_STREAM_RESET_CODE,
+  IDX_STATE_STREAM_SIZE,
 } = internalBinding('quic');
 
 assert(IDX_STATE_SESSION_LISTENER_FLAGS !== undefined);
@@ -723,8 +729,8 @@ class QuicStreamState {
   #handle;
   /** @type {number} */
   #offset = 0;
-  /** @type {bigint|undefined} */
-  #id = undefined;
+  /** @type {boolean} */
+  #disconnected = false;
 
   /**
    * @param {symbol} privateSymbol
@@ -746,7 +752,7 @@ class QuicStreamState {
   /** @type {bigint} */
   get id() {
     const handle = this.#handle;
-    if (handle === undefined) return this.#id;
+    if (handle === undefined) return undefined;
     return DataViewPrototypeGetBigInt64(handle, this.#offset + IDX_STATE_STREAM_ID, kIsLittleEndian);
   }
 
@@ -943,7 +949,7 @@ class QuicStreamState {
   }
 
   [kInspect](depth, options) {
-    if (this.#handle === undefined ||
+    if (this.#disconnected || this.#handle === undefined ||
       DataViewPrototypeGetByteLength(this.#handle) === 0) {
       return 'QuicStreamState { <Closed> }';
     }
@@ -998,9 +1004,24 @@ class QuicStreamState {
   }
 
   [kFinishClose]() {
-    // Cache the stream ID since the buffer will be zeroed out and the ID will be lost.
-    this.#id = this.id;
-    this.#handle = undefined;
+    // Copy the final values out of the shared buffer (zeroed when the
+    // underlying stream goes away) so getters keep reporting the final
+    // state, the same way QuicStreamStats does.
+    const handle = this.#handle;
+    if (handle === undefined ||
+        DataViewPrototypeGetByteLength(handle) <
+          this.#offset + IDX_STATE_STREAM_SIZE) {
+      this.#handle = undefined;
+      this.#disconnected = true;
+      return;
+    }
+    const copy = TypedArrayPrototypeSlice(new Uint8Array(
+      DataViewPrototypeGetBuffer(handle),
+      DataViewPrototypeGetByteOffset(handle) + this.#offset,
+      IDX_STATE_STREAM_SIZE));
+    this.#handle = new DataView(TypedArrayPrototypeGetBuffer(copy));
+    this.#offset = 0;
+    this.#disconnected = true;
   }
 }
 

src/quic/streams.cc

@@ -1008,6 +1008,9 @@ void Stream::InitPerContext(Realm* realm, Local<Object> target) {
 #undef V
 
   NODE_DEFINE_CONSTANT(target, IDX_STATS_STREAM_COUNT);
+
+  constexpr auto IDX_STATE_STREAM_SIZE = sizeof(Stream::State);
+  NODE_DEFINE_CONSTANT(target, IDX_STATE_STREAM_SIZE);
 }
 
 Stream* Stream::From(void* stream_user_data) {
@@ -1321,13 +1324,6 @@ BaseObjectPtr<Blob::Reader> Stream::get_reader() {
   return reader;
 }
 
-void Stream::set_final_size(uint64_t final_size) {
-  DCHECK_IMPLIES(state()->fin_received == 1,
-                 final_size <= STAT_GET(Stats, final_size));
-  state()->fin_received = 1;
-  STAT_SET(Stats, final_size, final_size);
-}
-
 void Stream::set_outbound(std::shared_ptr<DataQueue> source) {
   if (!source || !is_writable()) return;
   Debug(this, "Setting the outbound data source");
@@ -1471,7 +1467,7 @@ void Stream::FlushAccumulation() {
     return;
   }
   // Should be unreachable: append() only fails once the queue has been capped,
-  // EndReadable() flushes before capping, and ReceiveData() accumulates
+  // CapReadable() flushes before capping, and ReceiveData() accumulates
   // nothing once read_ended is set. Reaching here means received stream data
   // is being dropped on the floor, so say so and at least do not also leak
   // the flow control credit for it.
@@ -1536,13 +1532,26 @@ void Stream::EndWritable() {
   state()->write_ended = 1;
 }
 
-void Stream::EndReadable(std::optional<uint64_t> maybe_final_size) {
+void Stream::FinishReadable() {
   if (!is_readable()) return;
+  state()->fin_received = 1;
+  CapReadable(std::nullopt);
+}
+
+void Stream::TruncateReadable(std::optional<uint64_t> maybe_final_size) {
+  if (!is_readable()) return;
+  CapReadable(maybe_final_size);
+}
+
+void Stream::CapReadable(std::optional<uint64_t> maybe_final_size) {
+  DCHECK(is_readable());
   state()->read_ended = 1;
   // Flush any accumulated data before capping so the reader can see it.
   FlushAccumulation();
-  set_final_size(maybe_final_size.value_or(STAT_GET(Stats, bytes_received)));
-  inbound_->cap(STAT_GET(Stats, final_size));
+  const uint64_t final_size =
+      maybe_final_size.value_or(STAT_GET(Stats, bytes_received));
+  STAT_SET(Stats, final_size, final_size);
+  inbound_->cap(final_size);
   // Notify the JS reader so it can see EOS. The subsequent pull observes
   // the now-capped DataQueue and returns EOS.
   if (reader_) reader_->NotifyPull();
@@ -1570,14 +1579,15 @@ void Stream::Destroy(QuicError error) {
   // End the writable before marking as destroyed.
   EndWritable();
 
-  // Also end the readable side if it isn't already.
-  EndReadable();
+  // Also end the readable side if it isn't already. If not already ended,
+  // this will eventually surface as an error, since the data is truncated.
+  TruncateReadable();
 
   // We are going to release our reference to the outbound_ queue here.
   outbound_.reset();
   application_state_.reset();
 
-  // EndReadable() above already flushed accumulated data. Just release
+  // TruncateReadable() above already flushed accumulated data. Just release
   // the ring buffer memory.
   recv_accumulator_.reset();
 
@@ -1632,7 +1642,7 @@ void Stream::ReceiveData(const uint8_t* data,
   // end the readable side if this is the last bit of data we've received.
   Debug(this, "Receiving %zu bytes of data", len);
   if (state()->read_ended == 1 || len == 0) {
-    if (flags.fin) EndReadable();
+    if (flags.fin) FinishReadable();
     // Nothing will ever consume these bytes, so return the connection-level
     // credit ngtcp2 charged for them. The stream window is deliberately left
     // alone: there is no point inviting more data onto a stream we have
@@ -1715,7 +1725,7 @@ void Stream::ReceiveData(const uint8_t* data,
 
   if (flags.fin) {
     FlushAccumulation();
-    EndReadable();
+    FinishReadable();
   } else if (reader_ && was_empty) {
     // Notify the reader once when the accumulator transitions from empty
     // to non-empty. This wakes the reader exactly once per accumulation
@@ -1746,7 +1756,7 @@ void Stream::ReceiveStreamReset(uint64_t final_size, QuicError error) {
         final_size,
         error);
   state()->reset_code = error.code();
-  EndReadable(final_size);
+  TruncateReadable(final_size);
   EmitReset(error);
 }
 
@@ -1769,7 +1779,7 @@ void Stream::DoStreamReset(error_code code) {
 }
 
 void Stream::SendStopSending(error_code code) {
-  EndReadable();
+  TruncateReadable();
 
   if (!is_pending()) {
     // If the stream is a local unidirectional there's nothing to do here.

src/quic/streams.h

@@ -325,7 +325,12 @@ class Stream final : public AsyncWrap,
   void UpdateWriteDesiredSize();
 
   void EndWritable();
-  void EndReadable(std::optional<uint64_t> maybe_final_size = std::nullopt);
+  // The read side ended cleanly with a peer FIN: the content is complete.
+  void FinishReadable();
+  // The read side ended without a FIN (a reset, a local abort, or the session
+  // being torn down) so the content is truncated.
+  void TruncateReadable(
+      std::optional<uint64_t> maybe_final_size = std::nullopt);
   void EntryRead(size_t amount) override;
   void BeforePull() override;
 
@@ -416,10 +421,13 @@ class Stream final : public AsyncWrap,
   // consumer or dropped before reaching one.
   void CreditConsumedBytes(uint64_t amount);
 
+  // Common tail of FinishReadable()/TruncateReadable(): marks the read side
+  // ended, caps it at the final size, and notifies the reader.
+  void CapReadable(std::optional<uint64_t> maybe_final_size);
+
   // Gets a reader for the data received for this stream from the peer,
   BaseObjectPtr<Blob::Reader> get_reader();
 
-  void set_final_size(uint64_t amount);
   void set_outbound(std::shared_ptr<DataQueue> source);
 
   // Streaming outbound support

test/common/quic.mjs

@@ -7,6 +7,7 @@
 // listen/connect that apply default options suitable for most tests.
 
 import * as fixtures from '../common/fixtures.mjs';
+import { setTimeout } from 'node:timers/promises';
 
 const { createPrivateKey } = await import('node:crypto');
 const quic = await import('node:quic');
@@ -98,6 +99,81 @@ function hashBytes(buf) {
   return h >>> 0;
 }
 
+/**
+ * Write `size` bytes to the stream and wait until the peer has
+ * acknowledged them.
+ */
+async function writeAndAwaitAck(stream, size) {
+  stream.writer.write(new Uint8Array(size).fill(7));
+  while (stream.stats.maxOffsetAcknowledged < BigInt(size)) await setTimeout(5);
+}
+
+/**
+ * Send `size` bytes and then stall forever without a FIN
+ * @yields {Uint8Array}
+ */
+async function* stallingBody(size) {
+  yield new Uint8Array(size).fill(7);
+  await new Promise(() => {});
+}
+
+/**
+ * Do a full single stream-read scenario, with the given client options and
+ * server body, and two hooks available to tweak behaviour, returning the
+ * details of the result after completion.
+ * @param {Function} serverBody
+ * @param {object} [options]
+ * @param {object} [options.clientOptions]  Options forwarded to connect().
+ * @param {Function} [options.beforeIterate]
+ * @param {Function} [options.onFirstChunk]
+ * @returns {Promise<{received: number, threw: any, closedError: any}>}
+ *   Bytes received, the iteration error if any, and the client stream's
+ *   closed rejection if any.
+ */
+async function readStream(serverBody, options = {}) {
+  const { clientOptions, beforeIterate, onFirstChunk } = options;
+
+  const serverEndpoint = await listen((session) => {
+    session.closed.catch(() => {});
+    session.onstream = (stream) => {
+      stream.closed.catch(() => {});
+      serverBody(stream, session);
+    };
+  });
+
+  const session = await connect(serverEndpoint.address, clientOptions);
+  await session.opened;
+  session.closed.catch(() => {});
+
+  const stream = await session.createBidirectionalStream();
+  await stream.writer.write(new Uint8Array([1]));
+
+  let closedError;
+  const closedSettled = stream.closed.catch((err) => { closedError = err; });
+
+  await beforeIterate?.({ stream, session });
+
+  let received = 0;
+  let threw;
+  let firstChunk = true;
+  try {
+    for await (const chunk of stream) {
+      for (const c of chunk) received += c.byteLength;
+      if (firstChunk) {
+        firstChunk = false;
+        await onFirstChunk?.({ stream, session });
+      }
+    }
+  } catch (err) {
+    threw = err;
+  }
+
+  session.close();
+  await closedSettled;
+  await serverEndpoint.close();
+  return { received, threw, closedError };
+}
+
 export {
   key,
   cert,
@@ -105,4 +181,7 @@ export {
   connect,
   makePayload,
   hashBytes,
+  readStream,
+  stallingBody,
+  writeAndAwaitAck,
 };

test/parallel/test-quic-stream-iteration-destroyed.mjs

@@ -1,37 +0,0 @@
-// Flags: --experimental-quic --no-warnings
-
-// Test: destroyed stream returns finished iterator.
-
-import { hasQuic, skip, mustCall } from '../common/index.mjs';
-import * as assert from 'node:assert';
-
-if (!hasQuic) {
-  skip('QUIC is not enabled');
-}
-
-const { listen, connect } = await import('../common/quic.mjs');
-
-const encoder = new TextEncoder();
-
-const serverEndpoint = await listen(mustCall(async (serverSession) => {
-  await serverSession.closed;
-}));
-
-const clientSession = await connect(serverEndpoint.address);
-await clientSession.opened;
-
-const stream = await clientSession.createBidirectionalStream({
-  body: encoder.encode('destroy test'),
-});
-
-// Destroy the stream immediately.
-stream.destroy();
-
-// Iterating a destroyed stream should immediately finish.
-const iter = stream[Symbol.asyncIterator]();
-const { done } = await iter.next();
-assert.strictEqual(done, true);
-
-await stream.closed;
-await clientSession.close();
-await serverEndpoint.destroy();

test/parallel/test-quic-stream-iteration-reset.mjs

@@ -1,64 +0,0 @@
-// Flags: --experimental-quic --experimental-stream-iter --no-warnings
-
-// Test: peer RESET_STREAM causes iterator to error.
-// When the server resets the stream, the client's async iterator
-// should throw or return early.
-
-import { hasQuic, skip, mustCall } from '../common/index.mjs';
-import * as assert from 'node:assert';
-
-if (!hasQuic) {
-  skip('QUIC is not enabled');
-}
-
-const { listen, connect } = await import('../common/quic.mjs');
-
-const encoder = new TextEncoder();
-
-const serverReady = Promise.withResolvers();
-
-const serverEndpoint = await listen(mustCall((serverSession) => {
-  serverSession.onstream = mustCall(async (stream) => {
-    // Reset the stream from the server side.
-    stream.resetStream(42n);
-    await assert.rejects(stream.closed, mustCall((err) => {
-      assert.ok(err);
-      return true;
-    }));
-    serverReady.resolve();
-    await serverSession.closed;
-  });
-}), { transportParams: { maxIdleTimeout: 1 } });
-
-const clientSession = await connect(serverEndpoint.address, {
-  transportParams: { maxIdleTimeout: 1 },
-});
-await clientSession.opened;
-
-const stream = await clientSession.createBidirectionalStream({
-  body: encoder.encode('will be reset by server'),
-});
-
-// Set up the closed handler before the reset to avoid unhandled rejection.
-const closedPromise = assert.rejects(stream.closed, mustCall((err) => {
-  assert.ok(err);
-  return true;
-}));
-
-await serverReady.promise;
-
-// The async iterator should either throw or return early when the
-// peer resets the readable side.
-try {
-  for await (const batch of stream) {
-    // May receive some data before the reset arrives.
-    assert.ok(Array.isArray(batch));
-  }
-} catch {
-  // The iterator may throw when the reset arrives mid-iteration.
-}
-
-// Either way, the stream should close.
-await closedPromise;
-await clientSession.closed;
-await serverEndpoint.close();

test/parallel/test-quic-stream-setbody-errors.mjs

@@ -53,7 +53,13 @@ await clientSession.opened;
     message: /writer already accessed/,
   });
 
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  // The server handles only the first stream and then closes its session, so
+  // this stream is never answered and never receives a FIN. Reading it
+  // therefore surfaces the truncation rather than ending cleanly.
+  await assert.rejects((async () => {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of stream) { /* drain */ }
+  })(), { code: 'ERR_QUIC_STREAM_ABORTED' });
   await stream.closed;
 }
 

test/parallel/test-quic-stream-truncated-reads-destroy.mjs

@@ -0,0 +1,72 @@
+// Flags: --experimental-quic --no-warnings
+
+// Test: truncation is still reported when the stream is torn down locally
+// before the iterator finishes draining. The stream's final read state and
+// close error are persisted at close, so a consumer slower than the teardown
+// (or one that only starts reading after it) still observes the truncation
+// rather than a clean end that would make an incomplete stream look complete.
+
+import { hasQuic, skip } from '../common/index.mjs';
+import { setTimeout } from 'node:timers/promises';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { readStream, stallingBody } = await import('../common/quic.mjs');
+
+// The server sends 1000 bytes then stalls without a FIN, so the client's
+// read side can only end by truncation.
+const serve = (stream) => { stream.setBody(stallingBody(1000)); };
+
+// The session is destroyed with an error while the consumer is mid-iteration:
+// the iterator delivers the data that arrived, then throws that same error
+// object.
+{
+  const boom = new Error('boom');
+  const { received, threw } = await readStream(serve, {
+    onFirstChunk: ({ session }) => session.destroy(boom),
+  });
+  assert.ok(received > 0);
+  assert.strictEqual(threw, boom);
+}
+
+// The session is destroyed without an error mid-iteration: an errorless
+// truncation, reported as aborted under the default policy.
+{
+  const { received, threw } = await readStream(serve, {
+    onFirstChunk: ({ session }) => session.destroy(),
+  });
+  assert.ok(received > 0);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+  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.
+{
+  const { received, threw } = await readStream(serve, {
+    beforeIterate: async ({ stream, session }) => {
+      while (stream.stats.bytesReceived === 0n) await setTimeout(5);
+      session.destroy();
+    },
+  });
+  assert.strictEqual(received, 0);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+}
+
+// Destroying the stream itself (rather than the session) before iterating
+// reports the truncation the same way, with closed resolving cleanly.
+{
+  const { received, threw, closedError } = await readStream(serve, {
+    beforeIterate: async ({ stream }) => {
+      while (stream.stats.bytesReceived === 0n) await setTimeout(5);
+      stream.destroy();
+    },
+  });
+  assert.strictEqual(received, 0);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+  assert.strictEqual(closedError, undefined);
+}

test/parallel/test-quic-stream-truncated-reads-timeout.mjs

@@ -0,0 +1,35 @@
+// Flags: --experimental-quic --no-warnings
+
+// Test: a readable truncated by the connection idle timeout delivers the
+// data it received, then ends per the truncatedReads policy: an error under
+// the default (so an incomplete stream can never look complete), and a clean
+// end under 'ignore' (an idle timeout carries no error).
+
+import { hasQuic, skip } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { readStream, stallingBody } = await import('../common/quic.mjs');
+
+// The server sends 1000 bytes then stalls without a FIN, so the short (1s)
+// connection idle timeout is the only thing that ends the read side.
+const serve = (stream) => { stream.setBody(stallingBody(1000)); };
+const transportParams = { maxIdleTimeout: 1 };
+
+{
+  const { received, threw } =
+    await readStream(serve, { clientOptions: { transportParams } });
+  assert.strictEqual(received, 1000);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+  assert.strictEqual(threw.errorCode, 0n);
+}
+{
+  const { received, threw } = await readStream(serve, {
+    clientOptions: { transportParams, truncatedReads: 'ignore' },
+  });
+  assert.strictEqual(received, 1000);
+  assert.strictEqual(threw, undefined);
+}

test/parallel/test-quic-stream-truncated-reads.mjs

@@ -0,0 +1,184 @@
+// Flags: --experimental-quic --no-warnings
+
+// Test: a stream read that ends without a FIN is a truncation. What the read
+// reports is decided in this order:
+//
+//   1. the peer reset the stream with a non-zero code: ERR_QUIC_STREAM_RESET
+//      carrying that code, under either policy;
+//   2. the stream was already being torn down and its closed promise rejected:
+//      that same error, under either policy;
+//   3. otherwise the truncation carries no error of its own and the
+//      truncatedReads policy decides - 'error' (the default) throws
+//      ERR_QUIC_STREAM_ABORTED so an incomplete stream can never look
+//      complete, 'ignore' ends the read cleanly.
+//
+// This file covers rules 1 and 3 on a live connection: peer resets and local
+// aborts. Rule 2 needs the stream to already be tearing down when the reader
+// resumes, which is covered by the sibling
+// test-quic-stream-truncated-reads-destroy test, and truncation by idle
+// timeout by test-quic-stream-truncated-reads-timeout.
+
+import { hasQuic, skip } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { connect, listen, readStream, stallingBody, writeAndAwaitAck } =
+  await import('../common/quic.mjs');
+
+// Sends 1000 bytes, waits for them to land, then resets with the given code.
+const resetWith = (code) => async (stream) => {
+  await writeAndAwaitAck(stream, 1000);
+  stream.resetStream(code);
+};
+
+// Sends 1000 bytes and then stalls without a FIN, so only the client's own
+// abort can end the read.
+const stall = (stream) => { stream.setBody(stallingBody(1000)); };
+
+// A peer reset with code 0 is a clean abort: a truncation, but not an error.
+// Rule 3 - the default reports it, 'ignore' treats it as a clean end.
+{
+  const { received, threw } = await readStream(resetWith(0n));
+  assert.strictEqual(received, 1000);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+  assert.strictEqual(threw.errorCode, 0n);
+}
+{
+  const { received, threw } =
+    await readStream(resetWith(0n), { clientOptions: { truncatedReads: 'ignore' } });
+  assert.strictEqual(received, 1000);
+  assert.strictEqual(threw, undefined);
+}
+
+// A peer reset with a nonzero code is rule 1: an error under either policy,
+// carrying the peer's code. It also rejects the closed promise on both sides.
+{
+  const serverClosed = Promise.withResolvers();
+  const { received, threw, closedError } = await readStream(async (stream) => {
+    await writeAndAwaitAck(stream, 1000);
+    stream.resetStream(42n);
+    // Our own reset closes the server-side stream with an error too.
+    serverClosed.resolve(stream.closed.then(() => undefined, (err) => err));
+  });
+  assert.strictEqual(received, 1000);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_RESET');
+  assert.strictEqual(threw.errorCode, 42n);
+  assert.strictEqual(threw.message,
+                     'The QUIC stream was reset by the peer with error code 42');
+  assert.strictEqual(closedError?.code, 'ERR_QUIC_APPLICATION_ERROR');
+  assert.strictEqual(closedError.errorCode, 42n);
+  const serverClosedError = await serverClosed.promise;
+  assert.strictEqual(serverClosedError?.code, 'ERR_QUIC_APPLICATION_ERROR');
+  assert.strictEqual(serverClosedError.errorCode, 42n);
+}
+{
+  const { received, threw } =
+    await readStream(resetWith(42n), { clientOptions: { truncatedReads: 'ignore' } });
+  assert.strictEqual(received, 1000);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_RESET');
+  assert.strictEqual(threw.errorCode, 42n);
+}
+
+// Aborting our own read with stopSending() is rule 3, not rule 1: we asked for
+// the truncation, so it is not an error the peer inflicted on us, and the code
+// we send does not come back as one. The read still stops short, so the
+// default policy reports it and 'ignore' does not.
+{
+  const { received, threw, closedError } = await readStream(stall, {
+    onFirstChunk: ({ stream }) => stream.stopSending(0n),
+  });
+  assert.ok(received > 0);
+  assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+  assert.strictEqual(threw.errorCode, 0n);
+  assert.strictEqual(closedError, undefined);
+}
+{
+  const { received, threw } = await readStream(stall, {
+    clientOptions: { truncatedReads: 'ignore' },
+    onFirstChunk: ({ stream }) => stream.stopSending(0n),
+  });
+  assert.ok(received > 0);
+  assert.strictEqual(threw, undefined);
+}
+
+// A nonzero stopSending code is an error this end raised, so it is reported
+// under either policy and carries that code. The peer answers our
+// STOP_SENDING with a RESET_STREAM echoing it, which is what rejects closed -
+// but the read reports our own abort rather than attributing it to the peer,
+// and does so whether or not that answer has arrived yet.
+for (const truncatedReads of ['error', 'ignore']) {
+  for (const awaitEcho of [false, true]) {
+    const peerReset = Promise.withResolvers();
+    const { received, threw, closedError } = await readStream(stall, {
+      clientOptions: { truncatedReads },
+      beforeIterate: ({ stream }) => { stream.onreset = () => peerReset.resolve(); },
+      onFirstChunk: async ({ stream }) => {
+        stream.stopSending(7n);
+        // For the 2nd pass, wait until we receive the corresponding reset:
+        if (awaitEcho) await peerReset.promise;
+      },
+    });
+    assert.ok(received > 0);
+    assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED');
+    assert.strictEqual(threw.errorCode, 7n);
+    assert.strictEqual(closedError?.code, 'ERR_QUIC_APPLICATION_ERROR');
+    assert.strictEqual(closedError.errorCode, 7n);
+  }
+}
+
+// A peer abruptly destroying its session truncates the read too. That
+// currently reaches us as an implicit reset, so which rule applies depends on
+// the code the peer's teardown puts on the wire - not part of this contract,
+// and deliberately not pinned here. What must hold either way is the default
+// policy's promise: the incomplete stream never looks complete.
+{
+  const { received, threw } = await readStream(async (stream, session) => {
+    await writeAndAwaitAck(stream, 1000);
+    session.destroy(new Error('connection boom'));
+  });
+  assert.strictEqual(received, 1000);
+  assert.ok(threw);
+}
+
+// Check the option in the server case as well:
+{
+  const serverRead = Promise.withResolvers();
+  const serverEndpoint = await listen((session) => {
+    session.closed.catch(() => {});
+    session.onstream = async (stream) => {
+      stream.closed.catch(() => {});
+      let received = 0;
+      try {
+        for await (const chunk of stream) {
+          for (const c of chunk) received += c.byteLength;
+        }
+        serverRead.resolve({ received, threw: undefined });
+      } catch (err) {
+        serverRead.resolve({ received, threw: err });
+      }
+    };
+  }, { truncatedReads: 'ignore' });
+
+  const session = await connect(serverEndpoint.address);
+  await session.opened;
+  session.closed.catch(() => {});
+
+  const stream = await session.createBidirectionalStream();
+  stream.closed.catch(() => {});
+  await writeAndAwaitAck(stream, 100);
+  stream.resetStream(0n);
+
+  const { threw } = await serverRead.promise;
+  assert.strictEqual(threw, undefined);
+
+  session.close();
+  await serverEndpoint.close();
+}
+
+// The option is validated.
+await assert.rejects(connect('127.0.0.1:1234', { truncatedReads: 'nope' }), {
+  code: 'ERR_INVALID_ARG_VALUE',
+});