Skip to content

Commit f090d13

Browse files
stream: detach sync share consumer before throwing
The `strict` backpressure branch of `SyncShareImpl` threw `ERR_OUT_OF_RANGE` while leaving the consumer attached, so its cursor stayed in the consumer set and kept pinning the buffer. Consumers that were still reading correctly then hit the same error a few entries later, because the buffer could no longer be trimmed past the stuck cursor. Build the error first, then detach and delete the consumer before throwing, so the reported byte count still describes the condition that triggered the failure. Every other terminal path of the same iterator already detaches, and the async `ShareImpl` was fixed the same way. Signed-off-by: Christian Aurich Zanettini Martins <christian.aurichzm@gmail.com>
1 parent c90c3dd commit f090d13

2 files changed

Lines changed: 45 additions & 2 deletions

File tree

‎lib/internal/streams/iter/share.js‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -586,10 +586,14 @@ class SyncShareImpl {
586586
let dropped = false;
587587
if (self.#bufferedBytes >= self.#options.budget) {
588588
switch (self.#options.backpressure) {
589-
case 'strict':
590-
throw new ERR_OUT_OF_RANGE(
589+
case 'strict': {
590+
const error = new ERR_OUT_OF_RANGE(
591591
'buffered bytes', `< ${self.#options.budget}`,
592592
self.#bufferedBytes);
593+
state.detached = true;
594+
self.#deleteConsumer(state);
595+
throw error;
596+
}
593597
case 'drop-oldest':
594598
while (self.#bufferedBytes >= self.#options.budget &&
595599
self.#buffer.length > 0) {

‎test/parallel/test-stream-iter-share-sync.js‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -211,6 +211,44 @@ function testShareSyncDropNewestUnboundedSource() {
211211
shared.cancel();
212212
}
213213

214+
// A strict budget failure detaches the consumer that threw so it cannot
215+
// later pin the buffer for consumers that continue reading.
216+
function testShareSyncStrictBackpressure() {
217+
for (const transformed of [false, true]) {
218+
function* source() {
219+
for (let i = 0; i < 10; i++) {
220+
yield [new Uint8Array(16384)];
221+
}
222+
}
223+
224+
const shared = shareSync(source(), {
225+
budget: 32768,
226+
backpressure: 'strict',
227+
});
228+
229+
const consumer = transformed ?
230+
shared.pull((chunks) => chunks) : shared.pull();
231+
const failed = consumer[Symbol.iterator]();
232+
const live = shared.pull()[Symbol.iterator]();
233+
234+
assert.strictEqual(failed.next().done, false);
235+
assert.strictEqual(failed.next().done, false);
236+
assert.throws(() => failed.next(), {
237+
code: 'ERR_OUT_OF_RANGE',
238+
});
239+
240+
assert.strictEqual(shared.consumerCount, 1);
241+
assert.strictEqual(failed.next().done, true);
242+
243+
let count = 0;
244+
while (!live.next().done) {
245+
count++;
246+
}
247+
assert.strictEqual(count, 10);
248+
assert.strictEqual(shared.consumerCount, 0);
249+
}
250+
}
251+
214252
// shareSync() accepts string source directly (normalized via fromSync())
215253
function testShareSyncStringSource() {
216254
const shared = shareSync('hello-sync-share');
@@ -229,5 +267,6 @@ Promise.all([
229267
testShareSyncRejectsUnbounded(),
230268
testShareSyncDropNewest(),
231269
testShareSyncDropNewestUnboundedSource(),
270+
testShareSyncStrictBackpressure(),
232271
testShareSyncStringSource(),
233272
]).then(common.mustCall());

0 commit comments

Comments
 (0)