diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 97efe7f3065b..3825dbf42487 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -181,6 +181,9 @@ class ShareImpl { } await self.#pullFromSource(!shouldBuffer); + if (!shouldBuffer) { + await self.#waitForBufferSpaceAfterDrop(); + } } }; @@ -299,6 +302,17 @@ class ShareImpl { return true; } + async #waitForBufferSpaceAfterDrop() { + while (this.#bufferedBytes >= this.#options.budget && + !this.#cancelled && + this.#sourceError === undefined && + !this.#sourceExhausted) { + const { promise, resolve } = PromiseWithResolvers(); + ArrayPrototypePush(this.#pullWaiters, resolve); + await promise; + } + } + #pullFromSource(discard = false) { if (this.#sourceExhausted || this.#cancelled) { return PromiseResolve(); diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js index 0b7e22ee5cee..806e30876302 100644 --- a/test/parallel/test-stream-iter-share-from.js +++ b/test/parallel/test-stream-iter-share-from.js @@ -170,40 +170,50 @@ async function testShareDropOldest() { } async function testShareDropNewest() { - // With drop-newest and a stalled consumer, the async path allows the - // buffer to grow beyond budget (the "drop" applies to the - // backpressure signal, not the buffer contents). Both consumers - // ultimately see all items. + let pulls = 0; + let secondPull; + const secondPullStarted = new Promise((resolve) => { + secondPull = resolve; + }); + async function* source() { - for (let i = 0; i < 4; i++) { + for (let i = 0; i < 7; i++) { + pulls++; + if (pulls === 2) secondPull(); const chunk = new Uint8Array(16384); chunk[0] = i; yield [chunk]; } } - const shared = share(source(), { budget: 32768, backpressure: 'drop-newest' }); - const fast = shared.pull(); - const slow = shared.pull(); + const shared = share(source(), { + budget: 16384, + backpressure: 'drop-newest', + }); + const fast = shared.pull()[Symbol.asyncIterator](); + const slow = shared.pull()[Symbol.asyncIterator](); - // Fast consumer reads all items - const fastIndices = []; - for await (const batch of fast) { - for (const chunk of batch) { - fastIndices.push(chunk[0]); - } - } - assert.strictEqual(fastIndices.length, 2); + const first = await fast.next(); + assert.strictEqual(first.value[0][0], 0); - // Slow consumer also sees all items (buffer grew past budget) - const slowIndices = []; - for await (const batch of slow) { - for (const chunk of batch) { - slowIndices.push(chunk[0]); - } - } - assert.strictEqual(slowIndices.length, 2); - assert.strictEqual(slowIndices[0], 0); - assert.strictEqual(slowIndices[1], 1); + let nextSettled = false; + const next = fast.next().then((result) => { + nextSettled = true; + return result; + }); + + await secondPullStarted; + await new Promise(setImmediate); + assert.strictEqual(pulls, 2); + assert.strictEqual(nextSettled, false); + + const slowResult = await slow.next(); + assert.strictEqual(slowResult.value[0][0], 0); + + const nextResult = await next; + assert.strictEqual(nextResult.value[0][0], 2); + assert.strictEqual(pulls, 3); + + shared.cancel(); } // =============================================================================