Ensure local stop-sending is handled correctly

Author: pimterry

doc/api/quic.md

@@ -3444,11 +3444,10 @@ selects how the stream's async iterator reports this:
 
 * `'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.
+  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)
 

lib/internal/quic/quic.js

@@ -1579,6 +1579,7 @@ class QuicStream {
     stats: undefined,
     pendingClose: undefined,
     destroyError: undefined,
+    stopSendingCode: undefined,
     truncatedReads: undefined,
     reader: undefined,
     destroying: false,
@@ -1678,6 +1679,14 @@ class QuicStream {
       // 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) {
@@ -2479,7 +2488,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);
   }
 
   /**

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

@@ -104,20 +104,29 @@ const stall = (stream) => { stream.setBody(stallingBody(1000)); };
   assert.strictEqual(threw, undefined);
 }
 
-// Same with a nonzero stopSending code. The peer answers our STOP_SENDING with
-// a RESET_STREAM carrying it, which rejects closed - but the read is still our
-// own abort, so it is reported by policy rather than as that reset. This is
-// the one case where the read and closed facets deliberately disagree.
+// 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', 'allow']) {
-  const { received, threw, closedError } = await readStream(stall, {
-    clientOptions: { truncatedReads },
-    onFirstChunk: ({ stream }) => stream.stopSending(7n),
-  });
-  assert.ok(received > 0);
-  assert.strictEqual(threw?.code,
-                     truncatedReads === 'error' ? 'ERR_QUIC_STREAM_ABORTED' : undefined);
-  assert.strictEqual(closedError?.code, 'ERR_QUIC_APPLICATION_ERROR');
-  assert.strictEqual(closedError.errorCode, 7n);
+  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