stream: destroy source when map iterator closes early

Fixes: #64261
Signed-off-by: y1d7ng <y1d7ng@yeah.net>
PR-URL: #64264
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: Y1D7NG

lib/internal/streams/operators.js

@@ -29,6 +29,7 @@ const {
   validateFunction,
 } = require('internal/validators');
 const { kWeakHandler, kResistStopPropagation } = require('internal/event_target');
+const destroyImpl = require('internal/streams/destroy');
 const { finished } = require('internal/streams/end-of-stream');
 
 const kEmpty = Symbol('kEmpty');
@@ -177,6 +178,7 @@ function map(fn, options) {
         resume();
         resume = null;
       }
+      destroyImpl.destroyer(stream, null);
     }
   }.call(this);
 }

test/parallel/test-stream-some-find-every.mjs

@@ -1,6 +1,6 @@
 import * as common from '../common/index.mjs';
 import { setTimeout } from 'timers/promises';
-import { Readable } from 'stream';
+import { PassThrough, Readable } from 'stream';
 import assert from 'assert';
 
 
@@ -77,6 +77,36 @@ function oneTo5Async() {
     await findStream.find(common.mustCall((x) => x > 1, 2));
     await checkDestroyed(findStream);
 
+    const openSomeStream = new PassThrough({ objectMode: true });
+    openSomeStream.write(1);
+    openSomeStream.write(2);
+    openSomeStream.write(3);
+    assert.strictEqual(
+      await openSomeStream.some(common.mustCall((x) => x > 2, 3)),
+      true,
+    );
+    await checkDestroyed(openSomeStream);
+
+    const openEveryStream = new PassThrough({ objectMode: true });
+    openEveryStream.write(1);
+    openEveryStream.write(2);
+    openEveryStream.write(3);
+    assert.strictEqual(
+      await openEveryStream.every(common.mustCall((x) => x < 3, 3)),
+      false,
+    );
+    await checkDestroyed(openEveryStream);
+
+    const openFindStream = new PassThrough({ objectMode: true });
+    openFindStream.write(1);
+    openFindStream.write(2);
+    openFindStream.write(3);
+    assert.strictEqual(
+      await openFindStream.find(common.mustCall((x) => x > 1, 2)),
+      2,
+    );
+    await checkDestroyed(openFindStream);
+
     // When short circuit isn't possible the whole stream is iterated
     await oneTo5().some(common.mustCall(() => false, 5));
     await oneTo5().every(common.mustCall(() => true, 5));