stream: pre-aborted pipeTo now applies dest failure handling

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65658
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

doc/api/stream_iter.md

@@ -605,7 +605,8 @@ added: v24.20.0
 * `...transforms` {Function|Object} Zero or more transforms to apply.
 * `writer` {Object} Destination with `write(chunk)` method.
 * `options` {Object}
-  * `signal` {AbortSignal} Abort the pipeline.
+  * `signal` {AbortSignal} Abort the pipeline. Aborting fails the destination
+    writer unless `preventFail` is `true`.
   * `preventClose` {boolean} If `true`, do not call `writer.end()` when
     the source ends. **Default:** `false`.
   * `preventFail` {boolean} If `true`, do not call `writer.fail()` on

lib/internal/streams/iter/pull.js

@@ -1041,8 +1041,18 @@ async function pipeTo(source, ...args) {
 
   const signal = options?.signal;
 
-  // Check for abort
-  signal?.throwIfAborted();
+  function failWriter(error) {
+    if (!options?.preventFail) {
+      writer.fail?.(wrapError(error));
+    }
+  }
+
+  try {
+    signal?.throwIfAborted();
+  } catch (error) {
+    failWriter(error);
+    throw error;
+  }
 
   const hasWriteSync = typeof writer.writeSync === 'function';
   const useSyncIterableFastPath =
@@ -1169,9 +1179,7 @@ async function pipeTo(source, ...args) {
       }
     }
   } catch (error) {
-    if (!options?.preventFail) {
-      writer.fail?.(wrapError(error));
-    }
+    failWriter(error);
     throw error;
   }
 

test/parallel/test-stream-iter-pipeto-signal.js

@@ -9,6 +9,51 @@ const assert = require('assert');
 const { setTimeout } = require('timers/promises');
 const { pipeTo, from } = require('stream/iter');
 
+async function testPipeToPreAbortedSignalFailsWriter() {
+  const reason = new Error('already aborted');
+  let sourceTouched = false;
+  const source = {
+    [Symbol.asyncIterator]() {
+      sourceTouched = true;
+      return {};
+    },
+  };
+  const writer = {
+    write: common.mustNotCall(),
+    fail: common.mustCall((error) => assert.strictEqual(error, reason)),
+  };
+
+  await assert.rejects(
+    pipeTo(source, writer, { signal: AbortSignal.abort(reason) }),
+    (error) => error === reason,
+  );
+  assert.strictEqual(sourceTouched, false);
+}
+
+async function testPipeToPreAbortedSignalPreventFail() {
+  const reason = new Error('already aborted');
+  let sourceTouched = false;
+  const source = {
+    [Symbol.asyncIterator]() {
+      sourceTouched = true;
+      return {};
+    },
+  };
+  const writer = {
+    write: common.mustNotCall(),
+    fail: common.mustNotCall(),
+  };
+
+  await assert.rejects(
+    pipeTo(source, writer, {
+      signal: AbortSignal.abort(reason),
+      preventFail: true,
+    }),
+    (error) => error === reason,
+  );
+  assert.strictEqual(sourceTouched, false);
+}
+
 // pipeTo with live signal, no transforms — abort mid-stream
 async function testPipeToLiveSignalNoTransforms() {
   const ac = new AbortController();
@@ -116,6 +161,8 @@ async function testPipeToLiveSignalWithTransformsCompletes() {
 }
 
 Promise.all([
+  testPipeToPreAbortedSignalFailsWriter(),
+  testPipeToPreAbortedSignalPreventFail(),
   testPipeToLiveSignalNoTransforms(),
   testPipeToLiveSignalNoTransformsPendingNext(),
   testPipeToLiveSignalWithTransforms(),