stream: update ShareImpl to retain factory abort listener

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

Author: jasnell

lib/internal/streams/iter/share.js

@@ -432,7 +432,9 @@ class ShareImpl {
         this.#sourceError = error;
         this.#sourceExhausted = true;
       } finally {
-        if (this.#sourceExhausted) this.#cleanupFactorySignal();
+        if (this.#sourceExhausted && this.#consumers.size === 0) {
+          this.#cleanupFactorySignal();
+        }
         this.#pulling = false;
         for (let i = 0; i < this.#pullWaiters.length; i++) {
           this.#pullWaiters[i]();
@@ -496,7 +498,11 @@ class ShareImpl {
 
   #deleteConsumer(consumer) {
     if (this.#consumers.delete(consumer)) {
-      return this.#deleteConsumerFromMin(consumer);
+      const shouldTrimBuffer = this.#deleteConsumerFromMin(consumer);
+      if (this.#sourceExhausted && this.#consumers.size === 0) {
+        this.#cleanupFactorySignal();
+      }
+      return shouldTrimBuffer;
     }
     return false;
   }

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

@@ -62,6 +62,22 @@ async function testShareSignalLifetime() {
   assert.strictEqual(abortListenerCount(controller.signal), 0);
 }
 
+async function testShareSignalLifetimeWithSlowConsumer() {
+  const controller = new AbortController();
+  const reason = new Error('aborted');
+  const shared = share('x', { signal: controller.signal });
+  const fast = shared.pull()[Symbol.asyncIterator]();
+  const slow = shared.pull()[Symbol.asyncIterator]();
+
+  assert.strictEqual((await fast.next()).done, false);
+  assert.strictEqual((await fast.next()).done, true);
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+
+  controller.abort(reason);
+  await assert.rejects(slow.next(), (error) => error === reason);
+  assert.strictEqual(abortListenerCount(controller.signal), 0);
+}
+
 async function testDuplexSignalLifetime() {
   const controller = new AbortController();
   const [channelA, channelB] = duplex({ signal: controller.signal });
@@ -80,5 +96,6 @@ Promise.all([
   testPushSignalLifetime(),
   testBroadcastSignalLifetime(),
   testShareSignalLifetime(),
+  testShareSignalLifetimeWithSlowConsumer(),
   testDuplexSignalLifetime(),
 ]).then(common.mustCall());