Make QUIC stream truncation behaviour configurable

Author: pimterry

doc/api/quic.md

@@ -3305,6 +3305,31 @@ 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 `'allow'`.
+* **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.
+
+* `'allow'` - Truncated reads are allowed: only a stream or connection error
+  is reported, and any clean abort/cancellation or similar simply ends the
+  stream. A non-zero peer reset still throws `ERR_QUIC_STREAM_RESET` and a
+  connection error still throws its real error, but a truncation that carried
+  no error (an idle timeout, a graceful close, a local `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

@@ -614,13 +614,7 @@ function createBlobReaderStream(reader) {
 // unbounded memory growth when the DataQueue has a large burst of data.
 const kMaxBatchChunks = 16;
 
-async function* createBlobReaderIterable(reader, options = kEmptyObject) {
-  // getEndError(status) lets the caller map a read ending to an error, or null.
-  // It is consulted both on a read failure (status < 0, always an error - we
-  // fall back to a generic one if the caller returns null) and on a clean EOS
-  // (status 0, an error only if the caller returns one, e.g. a stream that
-  // ended before its FIN).
-  const { getEndError } = options;
+async function* createBlobReaderIterable(reader) {
   let wakeup = PromiseWithResolvers();
   let immediate;
   let fin = false;
@@ -654,9 +648,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
           break;
         }
         if (pullResult.status < 0) {
-          error = (typeof getEndError === 'function' &&
-                   getEndError(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) {
@@ -671,11 +663,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
         yield batch;
       }
 
-      if (eos) {
-        const eosError = typeof getEndError === 'function' ? getEndError(0) : null;
-        if (eosError) throw eosError;
-        return;
-      }
+      if (eos) return;
       if (error) throw error;
 
       if (blocked) {

lib/internal/quic/quic.js

@@ -13,7 +13,6 @@ const {
   ErrorCaptureStackTrace,
   FunctionPrototypeBind,
   FunctionPrototypeCall,
-  Number,
   ObjectDefineProperties,
   ObjectKeys,
   PromisePrototypeThen,
@@ -114,8 +113,6 @@ const {
     ERR_QUIC_CONNECTION_FAILED,
     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');
@@ -398,6 +395,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'|'allow'} [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)
@@ -1580,6 +1578,8 @@ class QuicStream {
     state: undefined,
     stats: undefined,
     pendingClose: undefined,
+    destroyError: undefined,
+    truncatedReads: undefined,
     reader: undefined,
     destroying: false,
     iteratorLocked: false,
@@ -1627,9 +1627,10 @@ class QuicStream {
    * @param {object} handle
    * @param {QuicSession} session
    * @param {number} direction
-   * @param {boolean} [isLocal]
+   * @param {boolean} isLocal
+   * @param {'error'|'allow'} truncatedReads
    */
-  constructor(privateSymbol, handle, session, direction, isLocal) {
+  constructor(privateSymbol, handle, session, direction, isLocal, truncatedReads) {
     assertPrivateSymbol(privateSymbol);
 
     this.#handle = handle;
@@ -1638,6 +1639,7 @@ class QuicStream {
     inner.session = session;
     inner.direction = direction;
     inner.isLocal = isLocal;
+    inner.truncatedReads = truncatedReads;
     inner.state = new QuicStreamState(
       kPrivateConstructor, handle.state, handle.stateByteOffset);
 
@@ -1665,39 +1667,41 @@ class QuicStream {
     inner.iteratorLocked = true;
 
     inner.reader ??= this.#handle?.getReader();
-    // Non-readable stream (outbound-only unidirectional, or closed)
-    if (!inner.reader) return;
-
-    // Maps the read side's end state to the error to surface, or null if it
-    // ended cleanly. state.finReceived is set only when the peer explicitly
-    // sent a FIN, confirming we got the whole stream - the one and only clean
-    // ending. Returns null in that case. Without a FIN the read side is
-    // truncated, for one of these reasons:
-    //   * 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.stopSending(). This runs EndReadable
-    //     in C++, setting state.readEnded but not state.finReceived. There
-    //     is no peer code to surface.
-    //   * The session was torn down before a FIN arrived (a local or peer
-    //     connection close, an idle timeout, or any error), again leaving
-    //     state.finReceived false.
-    const readTruncationError = () => {
-      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');
+    // No reader means either a 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);
+    }
+
+    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.
+
+      // 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 });
       }
-      return null;
-    };
 
-    yield* createBlobReaderIterable(inner.reader, {
-      getEndError: readTruncationError,
-    });
+      // 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 });
+      }
+    }
   }
 
   /**
@@ -2068,9 +2072,10 @@ 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();
   }
 
   /**
@@ -2566,6 +2571,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 {
@@ -2788,6 +2796,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,
@@ -2821,8 +2830,9 @@ class QuicSession {
    * @param {symbol} privateSymbol
    * @param {object} handle
    * @param {QuicEndpoint} endpoint
+   * @param {{ truncatedReads?: 'error'|'allow' }} [options]
    */
-  constructor(privateSymbol, handle, endpoint) {
+  constructor(privateSymbol, handle, endpoint, options = kEmptyObject) {
     // Instances of QuicSession can only be created internally.
     assertPrivateSymbol(privateSymbol);
 
@@ -2831,6 +2841,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;
@@ -3376,7 +3388,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);
@@ -4161,7 +4174,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;
@@ -4278,6 +4291,7 @@ class QuicEndpoint {
     sessions: new SafeSet(),
     stat: undefined,
     stats: undefined,
+    truncatedReads: undefined,
     onsession: undefined,
     sessionCallbacks: undefined,
   };
@@ -4410,8 +4424,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;
@@ -4585,9 +4599,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,
@@ -4629,6 +4647,7 @@ class QuicEndpoint {
     validateObject(options, 'options');
     const {
       sessionTicket,
+      truncatedReads,
       ...rest
     } = options;
 
@@ -4637,7 +4656,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);
@@ -4871,7 +4890,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.
@@ -5357,6 +5377,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,
@@ -5408,6 +5429,9 @@ function processSessionOptions(options, config = kEmptyObject) {
   validateOneOf(verifyPeer, 'options.verifyPeer',
                 ['strict', 'auto', 'manual']);
 
+  validateOneOf(truncatedReads, 'options.truncatedReads',
+                ['error', 'allow']);
+
   validateInteger(drainingPeriodMultiplier, 'options.drainingPeriodMultiplier',
                   3, 255);
 
@@ -5470,6 +5494,7 @@ function processSessionOptions(options, config = kEmptyObject) {
       verifyHostname: verifyPeer !== 'manual',
     },
     verifyPeer,
+    truncatedReads,
     qlog,
     maxPayloadSize,
     unacknowledgedPacketThreshold,