fixup

Author: ronag

lib/internal/streams/destroy.js

@@ -326,12 +326,12 @@ function emitErrorCloseLegacy(stream, err) {
 }
 
 // Normalize destroy for legacy.
-function destroyer(stream, err) {
+function destroyer(stream, err, options) {
   if (!stream || isDestroyed(stream)) {
     return;
   }
 
-  if (!err && !isFinished(stream)) {
+  if (!err && !isFinished(stream) && options?.autoAbort !== false) {
     err = new AbortError();
   }
 

lib/internal/streams/readable.js

@@ -1150,7 +1150,9 @@ async function* createAsyncIterator(stream, options) {
       (error || options?.destroyOnReturn !== false) &&
       (error === undefined || stream._readableState.autoDestroy)
     ) {
-      destroyImpl.destroyer(stream, null);
+      destroyImpl.destroyer(stream, null, {
+        autoAbort: false
+      });
     }
   }
 }

test/parallel/test-stream-pipeline.js

@@ -764,21 +764,21 @@ const tsp = require('timers/promises');
   }));
 }
 
-// {
-//   const s = new PassThrough();
-//   pipeline(async function*() {
-//     await Promise.resolve();
-//     yield 'hello';
-//     yield 'world';
-//   }, s, async function(source) {
-//     for await (const chunk of source) {
-//       throw new Error('kaboom');
-//     }
-//   }, common.mustCall((err, val) => {
-//     assert.strictEqual(err.message, 'kaboom');
-//     assert.strictEqual(s.destroyed, true);
-//   }));
-// }
+{
+  const s = new PassThrough();
+  pipeline(async function*() {
+    await Promise.resolve();
+    yield 'hello';
+    yield 'world';
+  }, s, async function(source) {
+    for await (const chunk of source) {
+      throw new Error('kaboom');
+    }
+  }, common.mustCall((err, val) => {
+    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(s.destroyed, true);
+  }));
+}
 
 {
   const s = new PassThrough();