stream: make share budget failures detach before throwing

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: https://github.com/nodejs/node/pull/66028
Reviewed-By: Filip Skokan <panva.ip@gmail.com>
Reviewed-By: Tim Perry <pimterry@gmail.com>

lib/internal/streams/iter/share.js

@@ -200,7 +200,16 @@ class ShareImpl {
             }
 
             // Need to pull from source - check buffer limit
-            const shouldBuffer = await self.#waitForBufferSpace();
+            let shouldBuffer;
+            try {
+              shouldBuffer = await self.#waitForBufferSpace();
+            } catch (error) {
+              state.detached = true;
+              if (self.#deleteConsumer(state)) {
+                self.#tryTrimBuffer();
+              }
+              throw error;
+            }
             if (shouldBuffer === null) {
               state.detached = true;
               state.error = self.#cancelError;

test/parallel/test-stream-iter-share-from.js

@@ -221,23 +221,29 @@ async function testShareDropNewest() {
 // =============================================================================
 
 async function testShareStrictBackpressure() {
-  async function* source() {
-    for (let i = 0; i < 10; i++) {
-      yield [new Uint8Array(16384)];
+  for (const transformed of [false, true]) {
+    async function* source() {
+      for (let i = 0; i < 10; i++) {
+        yield [new Uint8Array(16384)];
+      }
     }
-  }
-  const shared = share(source(), { budget: 32768, backpressure: 'strict' });
-  const fast = shared.pull();
-  // Create a second consumer that never reads — this prevents buffer trimming
-  shared.pull();
+    const shared = share(source(), {
+      budget: 32768,
+      backpressure: 'strict',
+    });
+    const consumer = transformed ?
+      shared.pull((chunks) => chunks) : shared.pull();
+    const fast = consumer[Symbol.asyncIterator]();
+    // This consumer prevents the buffer from being trimmed.
+    shared.pull();
 
-  // The fast consumer's pulls will eventually cause the buffer to exceed
-  // the budget (since the slow consumer prevents trimming),
-  // triggering an ERR_OUT_OF_RANGE error.
-  await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of fast) { /* consume */ }
-  }, { code: 'ERR_OUT_OF_RANGE' });
+    await fast.next();
+    await fast.next();
+    await assert.rejects(fast.next(), { code: 'ERR_OUT_OF_RANGE' });
+    assert.strictEqual(shared.consumerCount, 1);
+    assert.strictEqual((await fast.next()).done, true);
+    shared.cancel();
+  }
 }
 
 Promise.all([