Skip to content

Commit 13ab435

Browse files
committed
stream: reject pull on signal abort during flush
An asynchronous transform can abort the pipeline while its final flush resolves to null. With no final batch yielded, the pipeline previously completed without observing the abort. Check the signal after transform iteration completes so the consumer rejects with the abort reason. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: codex:gpt-5.6-sol
1 parent 977c20e commit 13ab435

2 files changed

Lines changed: 22 additions & 0 deletions

File tree

lib/internal/streams/iter/pull.js

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -764,6 +764,10 @@ async function* createAsyncPipeline(source, transforms, signal) {
764764
controller.signal.throwIfAborted();
765765
yield batch;
766766
}
767+
// A transform can abort while completing without producing a final batch,
768+
// for example when an async flush resolves to null. In that case the loop
769+
// body has no opportunity to observe the abort.
770+
controller.signal.throwIfAborted();
767771
completed = true;
768772
} catch (error) {
769773
if (!controller.signal.aborted) {

test/parallel/test-stream-iter-pull-async.js

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -295,6 +295,23 @@ async function testPullStatelessTransformFlushError() {
295295
}, { message: 'async flush boom' });
296296
}
297297

298+
// An abort during an async flush must not be swallowed when the flush resolves
299+
// to null and therefore produces no final batch.
300+
async function testPullSignalAbortDuringAsyncFlush() {
301+
const ac = new AbortController();
302+
const reason = new Error('aborted during flush');
303+
const transform = async (chunks) => {
304+
if (chunks !== null) return chunks;
305+
ac.abort(reason);
306+
return null;
307+
};
308+
309+
await assert.rejects(
310+
() => text(pull(from('x'), transform, { signal: ac.signal })),
311+
(error) => error === reason,
312+
);
313+
}
314+
298315
// Pull with a sync iterable source (not async)
299316
async function testPullWithSyncSource() {
300317
function* gen() {
@@ -409,6 +426,7 @@ async function testTransformOptionsNotShared() {
409426
testPullStatelessTransformFlush(),
410427
testPullConsecutiveStatelessTransformFlush(),
411428
testPullStatelessTransformFlushError(),
429+
testPullSignalAbortDuringAsyncFlush(),
412430
testPullWithSyncSource(),
413431
testPullStringSource(),
414432
testTransformReturnsSingleUint8Array(),

0 commit comments

Comments
 (0)