stream: honor AbortSignal in Writer.end() · nodejs/node@796c896 · GitHub
Skip to content

Commit 796c896

Browse files
trivikraduh95
trivikr
authored and
aduh95
committed
stream: honor AbortSignal in Writer.end()
Reject Writer.end() when its signal is already aborted without closing the writer. For push writers, reject the pending operation if the signal aborts while buffered data drains, while allowing the graceful close to continue. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: codex:gpt-5.6-sol PR-URL: #64727 Fixes: #64726 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
1 parent fde8523 commit 796c896

5 files changed

Lines changed: 108 additions & 5 deletions

File tree

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

Lines changed: 3 additions & 2 deletions

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

Lines changed: 35 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
const {
99
ArrayIsArray,
1010
ArrayPrototypePush,
11+
PromisePrototypeThen,
1112
PromiseReject,
1213
PromiseResolve,
1314
PromiseWithResolvers,
@@ -55,6 +56,35 @@ const {
5556

5657
const kNoFailReason = Symbol('kNoFailReason');
5758

59+
function raceEndWithSignal(promise, signal) {
60+
if (!signal) return promise;
61+
if (signal.aborted) return PromiseReject(signal.reason);
62+
63+
const {
64+
promise: signaledPromise,
65+
resolve,
66+
reject,
67+
} = PromiseWithResolvers();
68+
const onAbort = () => reject(signal.reason);
69+
70+
signal.addEventListener('abort', onAbort, {
71+
__proto__: null,
72+
once: true,
73+
});
74+
PromisePrototypeThen(
75+
promise,
76+
(value) => {
77+
signal.removeEventListener('abort', onAbort);
78+
resolve(value);
79+
},
80+
(reason) => {
81+
signal.removeEventListener('abort', onAbort);
82+
reject(reason);
83+
},
84+
);
85+
return signaledPromise;
86+
}
87+
5888
// =============================================================================
5989
// PushQueue - Internal Queue with Chunk-Based Backpressure
6090
// =============================================================================
@@ -628,7 +658,9 @@ class PushWriter {
628658
}
629659

630660
end(options) {
631-
getWriterSignal(options);
661+
const signal = getWriterSignal(options);
662+
if (signal?.aborted) return PromiseReject(signal.reason);
663+
632664
const result = this.#queue.end();
633665
if (result === -2) {
634666
// Errored: reject with stored error
@@ -639,11 +671,11 @@ class PushWriter {
639671
// when consumer drains past the end sentinel
640672
const pendingEndPromise = this.#queue.pendingEndPromise;
641673
if (pendingEndPromise !== null) {
642-
return pendingEndPromise;
674+
return raceEndWithSignal(pendingEndPromise, signal);
643675
}
644676
const { promise, resolve, reject } = PromiseWithResolvers();
645677
this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
646-
return promise;
678+
return raceEndWithSignal(promise, signal);
647679
}
648680
// >= 0: byte count (immediate close or idempotent)
649681
return PromiseResolve(result);

‎test/parallel/test-stream-iter-broadcast-basic.js‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,22 @@ async function testWriterEnd() {
111111
assert.strictEqual(data, 'data');
112112
}
113113

114+
async function testWriterEndWithPreAbortedSignal() {
115+
const { writer, broadcast: bc } = broadcast();
116+
const consumer = bc.push();
117+
const reason = new Error('end aborted');
118+
119+
await assert.rejects(
120+
writer.end({ signal: AbortSignal.abort(reason) }),
121+
(error) => error === reason,
122+
);
123+
124+
// A rejected end must leave the writer open.
125+
await writer.write('data');
126+
assert.strictEqual(await writer.end(), 4);
127+
assert.strictEqual(await text(consumer), 'data');
128+
}
129+
114130
async function testWriterFail() {
115131
const { writer, broadcast: bc } = broadcast();
116132
const consumer = bc.push();
@@ -308,6 +324,7 @@ Promise.all([
308324
testWriteSync(),
309325
testWritevSync(),
310326
testWriterEnd(),
327+
testWriterEndWithPreAbortedSignal(),
311328
testWriterFail(),
312329
testCancelWithoutReason(),
313330
testCancelWithReason(),

‎test/parallel/test-stream-iter-duplex.js‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,22 @@ async function testAbortSignal() {
127127
);
128128
}
129129

130+
async function testWriterEndWithPreAbortedSignal() {
131+
const [channelA, channelB] = duplex();
132+
const reason = new Error('end aborted');
133+
134+
await assert.rejects(
135+
channelA.writer.end({ signal: AbortSignal.abort(reason) }),
136+
(error) => error === reason,
137+
);
138+
139+
await channelA.writer.write('still open');
140+
const completedEnd = channelA.writer.end();
141+
assert.strictEqual(await text(channelB.readable), 'still open');
142+
assert.strictEqual(await completedEnd, 10);
143+
await channelB.close();
144+
}
145+
130146
async function testEmptyDuplex() {
131147
const [channelA, channelB] = duplex();
132148

@@ -182,6 +198,7 @@ Promise.all([
182198
testWithOptions(),
183199
testPerChannelOptions(),
184200
testAbortSignal(),
201+
testWriterEndWithPreAbortedSignal(),
185202
testEmptyDuplex(),
186203
testChannelFail(),
187204
testAbortSignalBothChannels(),

‎test/parallel/test-stream-iter-push-writer.js‎

Lines changed: 36 additions & 0 deletions

0 commit comments

Comments
 (0)