From 1ad67bc66b2ed66540f14624749ebce39f373de2 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 15:09:17 +0000 Subject: [PATCH] stream: make share budget failures detach before throwing Signed-off-by: James M Snell Assisted-by: Opencode PR-URL: https://github.com/nodejs/node/pull/66028 Reviewed-By: Filip Skokan Reviewed-By: Tim Perry --- lib/internal/streams/iter/share.js | 11 +++++- test/parallel/test-stream-iter-share-from.js | 36 ++++++++++++-------- 2 files changed, 31 insertions(+), 16 deletions(-) diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 6154509a5b6..d02f9476271 100644 --- a/lib/internal/streams/iter/share.js +++ b/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; diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js index 806e3087630..69b782ce560 100644 --- a/test/parallel/test-stream-iter-share-from.js +++ b/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([