Merge pull request #7501 from cloudflare/jasnell/ts-streams-pipe-dest-erroring-pin

Pin that a pipe waits for an erroring destination's in-flight write

Author: jasnell

src/tests/streams/piping/AGENTS.md

@@ -24,7 +24,7 @@ a deliberate defect pin, not a hole).
 | 4 | ws.close() queued BEFORE pipeTo | pipe locks both ends, waits, cancels source with 'This destination writable stream is closed.', then RESOLVES (see the TODO(conform) in the test) | pipe REJECTS IMMEDIATELY 'Destination closed before the pipe completed', cancels source with the same error (preventCancel suppresses), locks never observed held | `pipeToJsToJsCloseQueuedDestination`(+`PreventCancel`) |
 | 5 | pipeTo brand check on a broken `this` | THROWS synchronously (before the capture_async_api_throws wrapper; the WPT general.any seed); a real stream with a bad destination REJECTS | both reject (spec) | `brandChecks` |
 | 6 | destination hwm 0 (never desires) | pipe writes an available chunk anyway — ignores desiredSize (the WPT 'dest never desires chunks' seed family) | never writes (spec) | `sourceErroredAfterChunkHwmZero` |
-| 7 | dest controller error()s while the pipe waits on a read | HALF-PROPAGATES: cancels the source with the error but FULFILLS the pipe promise | rejects the pipe and cancels the source with the error (spec) | `destControllerErrorsMidPipe` |
+| 7 | dest controller error()s while the pipe waits on a read | HALF-PROPAGATES: cancels the source with the error but FULFILLS the pipe promise | rejects the pipe and cancels the source with the error (spec) | `destControllerErrorsMidPipe`, `destErroringWaitsForInFlightWrite` |
 | 8 | FixedLengthStream length violations via pipe | overflow: pipe NEVER SETTLES (bounded); underflow: never settles | overflow: rejects RangeError; underflow: never settles (parity of nonconformance) | `fixedLengthStreamPipeOverflow`/`Underflow` |
 | 9 | already-closed source → already-closed dest | rejects TypeError (a C++ deviation: the spec's ordered shutdown conditions give closing-forward priority) | FULFILLS (spec; WPT multiple-propagation 'closed readable to closed writable' pins the fulfillment) | `closedSourceToClosedDest` |
 | 10 | SharedArrayBuffer-backed views into CompressionStream | copies the shared bytes; round-trips | write path REJECTS TypeError 'The provided value is not of type (ArrayBuffer or ArrayBufferView)' (spec: BufferSource without [AllowShared]) — while its identity stream ACCEPTS the same views | `sabViewThroughCompressionRoundTrip` |
@@ -77,7 +77,7 @@ the source FIRST, then releasing the write (`pipeStopsPullingWhenDestStalls`).
 | `pipe-matrix.js` | migrated pipe-streams-test.js wholesale (35): pipeThrough + pipeTo across JS↔native in all directions, prevent* combos, pre-aborted and mid-read AbortSignals, tee'd pipes, queued-destination close (ledger #1-#4, #13) |
 | `api-surface.js` | brand checks (ledger #5), option getter order, throwing getters, invalid signal, locked pipeThrough endpoints |
 | `error-propagation.js` | forward matrix (starts-errored × prevent* × truthy), hwm-0 dest (ledger #6), custom-error preservation (migrated from streams-error-edge-cases-test.js) |
-| `close-propagation.js` | the WPT-disabled backward territory, bounded: external close/abort on piped dest, write-throw backward propagation, idle dest-controller error (ledger #7) |
+| `close-propagation.js` | the WPT-disabled backward territory, bounded: external close/abort on piped dest, write-throw backward propagation, idle dest-controller error and one with a write in flight (ledger #7) |
 | `flow-control.js` | backpressure chain (migrated from streams-backpressure-test.js), stalled-dest read-ahead bound |
 | `shutdown-backlog.js` | shutdown with chunks buffered in the source: abort, source error, invalid identity chunk; what is written and what stays readable (ledger #11, #12) |
 | `shutdown-pending-read.js` | abort while the pipe waits on a read: a chunk arriving during the shutdown's wait, in the abort's turn, or after an idle abort (ledger #14, #15) |

src/tests/streams/piping/close-propagation.js

@@ -164,3 +164,50 @@ export const destControllerErrorsMidPipe = {
     }
   },
 };
+
+// The destination's controller errors while a write is in flight: the
+// destination is erroring until the write settles, and only then errored.
+// Backward propagation waits for errored (spec), so the source is not
+// cancelled while the write is outstanding. DIVERGENCE (ledger #7): C++
+// then cancels the source with the error but FULFILLS the pipe promise.
+export const destErroringWaitsForInFlightWrite = {
+  async test() {
+    const derr = new Error('dest-err');
+    let cancelArg = 'not-called';
+    const rs = new ReadableStream({
+      start(c) {
+        c.enqueue('a');
+      },
+      cancel(r) {
+        cancelArg = r;
+      },
+    });
+    let wc;
+    let releaseWrite;
+    const writeStarted = Promise.withResolvers();
+    const ws = new WritableStream({
+      start(c) {
+        wc = c;
+      },
+      write() {
+        writeStarted.resolve();
+        return new Promise((r) => (releaseWrite = r));
+      },
+    });
+    const pipeP = rs.pipeTo(ws);
+    await writeStarted.promise;
+    wc.error(derr);
+    const early = await outcomeOf(pipeP, 50);
+    strictEqual(early.state, 'pending');
+    strictEqual(cancelArg, 'not-called');
+    releaseWrite();
+    const outcome = await outcomeOf(pipeP);
+    if (usingTsImpl) {
+      strictEqual(outcome.state, 'rejected');
+      strictEqual(outcome.reason, derr);
+    } else {
+      strictEqual(outcome.state, 'fulfilled');
+    }
+    strictEqual(cancelArg, derr);
+  },
+};

src/tests/streams/piping/main.js

@@ -70,6 +70,7 @@ export {
   destWriteThrowsMidPipe,
   destWriteThrowsMidPipePreventCancel,
   destControllerErrorsMidPipe,
+  destErroringWaitsForInFlightWrite,
 } from 'close-propagation';
 
 export {