From bf47906157b93875df8c5672fb8be822adf397b0 Mon Sep 17 00:00:00 2001 From: "Kamat, Trivikram" <16024985+trivikr@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:27:52 -0700 Subject: [PATCH] stream: preserve falsy cancellation reasons Use undefined as the no-error sentinel when cancelling broadcast and share consumers. This ensures that 0, an empty string, false, and null are propagated instead of being converted to clean completion. Make sync share surface cancellation reasons before handling detached consumers, and add regression coverage for async and sync consumers. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: codex:gpt-5.6-sol --- lib/internal/streams/iter/broadcast.js | 10 +++--- lib/internal/streams/iter/share.js | 26 +++++++------- .../test-stream-iter-broadcast-basic.js | 16 ++++----- test/parallel/test-stream-iter-share-async.js | 12 +++++++ test/parallel/test-stream-iter-share-sync.js | 36 +++++++++---------- 5 files changed, 56 insertions(+), 44 deletions(-) diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 5eadbb0da49f..bd4d7b9779fd 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -87,7 +87,7 @@ class BroadcastImpl { #consumers = new SafeSet(); #waiters = []; // Consumers with pending resolve (subset of #consumers) #ended = false; - #error = null; + #error; #cancelled = false; #options; #writer = null; @@ -176,7 +176,9 @@ class BroadcastImpl { __proto__: null, next() { if (state.detached) { - if (self.#error) return PromiseReject(self.#error); + if (self.#error !== undefined) { + return PromiseReject(self.#error); + } return kDone; } @@ -193,7 +195,7 @@ class BroadcastImpl { { __proto__: null, done: false, value: chunk }); } - if (self.#error) { + if (self.#error !== undefined) { state.detached = true; self.#deleteConsumer(state); return PromiseReject(self.#error); @@ -343,7 +345,7 @@ class BroadcastImpl { } [kAbort](reason) { - if (this.#ended || this.#error) return; + if (this.#ended || this.#error !== undefined) return; this.#error = reason; this.#ended = true; diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index e4a312fdb41c..97efe7f3065b 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -73,7 +73,7 @@ class ShareImpl { #consumers = new SafeSet(); #sourceIterator = null; #sourceExhausted = false; - #sourceError = null; + #sourceError; #cancelled = false; #pulling = false; #pullWaiters = []; @@ -129,7 +129,7 @@ class ShareImpl { __proto__: null, [SymbolAsyncIterator]() { const getNext = async () => { - if (self.#sourceError) { + if (self.#sourceError !== undefined) { state.detached = true; self.#consumers.delete(state); throw self.#sourceError; @@ -141,7 +141,7 @@ class ShareImpl { // cursor must re-pull rather than terminating prematurely. for (;;) { if (state.detached) { - if (self.#sourceError) throw self.#sourceError; + if (self.#sourceError !== undefined) throw self.#sourceError; return { __proto__: null, done: true, value: undefined }; } @@ -167,7 +167,7 @@ class ShareImpl { if (self.#sourceExhausted) { state.detached = true; self.#deleteConsumer(state); - if (self.#sourceError) throw self.#sourceError; + if (self.#sourceError !== undefined) throw self.#sourceError; return { __proto__: null, done: true, value: undefined }; } @@ -176,7 +176,7 @@ class ShareImpl { if (shouldBuffer === null) { state.detached = true; self.#deleteConsumer(state); - if (self.#sourceError) throw self.#sourceError; + if (self.#sourceError !== undefined) throw self.#sourceError; return { __proto__: null, done: true, value: undefined }; } @@ -260,7 +260,9 @@ class ShareImpl { async #waitForBufferSpace() { while (this.#bufferedBytes >= this.#options.budget) { - if (this.#cancelled || this.#sourceError || this.#sourceExhausted) { + if (this.#cancelled || + this.#sourceError !== undefined || + this.#sourceExhausted) { return this.#cancelled ? null : true; } @@ -418,7 +420,7 @@ class SyncShareImpl { #consumers = new SafeSet(); #sourceIterator = null; #sourceExhausted = false; - #sourceError = null; + #sourceError; #cancelled = false; #cachedMinCursor = 0; #cachedMinCursorConsumers = 0; @@ -467,14 +469,14 @@ class SyncShareImpl { return { __proto__: null, next() { - if (state.detached) { - return { __proto__: null, done: true, value: undefined }; - } - if (self.#sourceError) { + if (self.#sourceError !== undefined) { state.detached = true; self.#deleteConsumer(state); throw self.#sourceError; } + if (state.detached) { + return { __proto__: null, done: true, value: undefined }; + } if (self.#cancelled) { state.detached = true; self.#deleteConsumer(state); @@ -535,7 +537,7 @@ class SyncShareImpl { self.#pullFromSource(); - if (self.#sourceError) { + if (self.#sourceError !== undefined) { state.detached = true; self.#deleteConsumer(state); throw self.#sourceError; diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js index 0644a28a462d..125f386210c9 100644 --- a/test/parallel/test-stream-iter-broadcast-basic.js +++ b/test/parallel/test-stream-iter-broadcast-basic.js @@ -260,15 +260,15 @@ async function testWriterFailIdempotent() { }, { message: 'fail!' }); } -// cancel() with falsy reason (0, "", false) should still treat as error async function testCancelWithFalsyReason() { - const { broadcast: bc } = broadcast(); - const consumer = bc.push(); - const resultPromise = text(consumer).catch((err) => err); - await new Promise((resolve) => setImmediate(resolve)); - bc.cancel(0); - const result = await resultPromise; - assert.strictEqual(result, 0); + for (const reason of [0, '', false, null]) { + const { broadcast: bc } = broadcast(); + const iterator = bc.push()[Symbol.asyncIterator](); + + bc.cancel(reason); + + await assert.rejects(iterator.next(), (error) => error === reason); + } } // Late-joining consumer should read from oldest buffered entry diff --git a/test/parallel/test-stream-iter-share-async.js b/test/parallel/test-stream-iter-share-async.js index f55ca35f6977..314e7bfcf01c 100644 --- a/test/parallel/test-stream-iter-share-async.js +++ b/test/parallel/test-stream-iter-share-async.js @@ -132,6 +132,17 @@ async function testShareCancelWithReason() { ); } +async function testShareCancelWithFalsyReason() { + for (const reason of [0, '', false, null]) { + const shared = share(from('data')); + const iterator = shared.pull()[Symbol.asyncIterator](); + + shared.cancel(reason); + + await assert.rejects(iterator.next(), (error) => error === reason); + } +} + async function testShareAbortSignal() { const ac = new AbortController(); const reason = new Error('share aborted'); @@ -357,6 +368,7 @@ Promise.all([ testShareCancel(), testShareCancelMidIteration(), testShareCancelWithReason(), + testShareCancelWithFalsyReason(), testShareAbortSignal(), testShareAbortSignalWhileSourcePullPending(), testSharePullAbortSignalRejectsPendingNext(), diff --git a/test/parallel/test-stream-iter-share-sync.js b/test/parallel/test-stream-iter-share-sync.js index 2b0b1944a7ff..20c98d134206 100644 --- a/test/parallel/test-stream-iter-share-sync.js +++ b/test/parallel/test-stream-iter-share-sync.js @@ -87,37 +87,32 @@ function testShareSyncCancelMidIteration() { } function testShareSyncCancelWithReason() { - // When cancel(reason) is called, a consumer that hasn't started - // iterating is already detached, so it sees done:true (not the error). - // But a consumer that is mid-iteration when another consumer cancels - // with a reason will see the error on the next pull after cancel. const enc = new TextEncoder(); function* gen() { yield [enc.encode('a')]; yield [enc.encode('b')]; - yield [enc.encode('c')]; } const shared = shareSync(gen(), { budget: 16384 }); - const c1 = shared.pull(); - const c2 = shared.pull(); + const iterator1 = shared.pull()[Symbol.iterator](); + const iterator2 = shared.pull()[Symbol.iterator](); + const reason = new Error('sync cancel reason'); + + iterator1.next(); + shared.cancel(reason); - // c1 reads one item, then c2 cancels with a reason - const iter1 = c1[Symbol.iterator](); - const first = iter1.next(); - assert.strictEqual(first.done, false); + assert.throws(() => iterator1.next(), (error) => error === reason); + assert.throws(() => iterator2.next(), (error) => error === reason); +} - shared.cancel(new Error('sync cancel reason')); +function testShareSyncCancelWithFalsyReason() { + for (const reason of [0, '', false, null]) { + const shared = shareSync(fromSync('data')); + const iterator = shared.pull()[Symbol.iterator](); - // c1 was already iterating, it's now detached → done - const next = iter1.next(); - assert.strictEqual(next.done, true); + shared.cancel(reason); - // c2 never started, also detached → done (not error) - const batches = []; - for (const batch of c2) { - batches.push(batch); + assert.throws(() => iterator.next(), (error) => error === reason); } - assert.strictEqual(batches.length, 0); } // ============================================================================= @@ -157,6 +152,7 @@ Promise.all([ testShareSyncCancel(), testShareSyncCancelMidIteration(), testShareSyncCancelWithReason(), + testShareSyncCancelWithFalsyReason(), testShareSyncSourceError(), testShareSyncStringSource(), ]).then(common.mustCall());