FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: honor AbortSignal in Writer.end() · nodejs/node@486cff4 · GitHub

/ node Public

Commit 486cff4

Browse files
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 dc30379 commit 486cff4

5 files changed

Lines changed: 108 additions & 5 deletions

File tree

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -630,9 +630,10 @@ class BroadcastWriter {
630630
return false;
631631
}
632632

633-
// end() is synchronous internally - signal accepted for interface compliance.
634633
end(options) {
635-
getWriterSignal(options);
634+
const signal = getWriterSignal(options);
635+
if (signal?.aborted) return PromiseReject(signal.reason);
636+
636637
if (this.#isClosed()) return this.#closed;
637638
this.#closed = PromiseResolve(this.#totalBytes);
638639
this.#broadcast[kEnd]();

‎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
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,40 @@ async function testEndAsyncReturnValue() {
260260
await consume;
261261
}
262262

263+
async function testEndWithPreAbortedSignal() {
264+
const { writer, readable } = push();
265+
const reason = new Error('end aborted');
266+
267+
writer.writeSync('hello');
268+
await assert.rejects(
269+
writer.end({ signal: AbortSignal.abort(reason) }),
270+
(error) => error === reason,
271+
);
272+
273+
// A rejected end must leave the writer open.
274+
writer.writeSync(' world');
275+
const consume = text(readable);
276+
assert.strictEqual(await writer.end(), 11);
277+
assert.strictEqual(await consume, 'hello world');
278+
}
279+
280+
async function testEndSignalAbortWhileDraining() {
281+
const { writer, readable } = push();
282+
const controller = new AbortController();
283+
const reason = new Error('end aborted while draining');
284+
285+
writer.writeSync('hello');
286+
const abortedEnd = writer.end({ signal: controller.signal });
287+
controller.abort(reason);
288+
289+
await assert.rejects(abortedEnd, (error) => error === reason);
290+
291+
// Aborting the operation does not undo the end-of-stream signal.
292+
const completedEnd = writer.end();
293+
assert.strictEqual(await text(readable), 'hello');
294+
assert.strictEqual(await completedEnd, 5);
295+
}
296+
263297
async function testEndAfterEndSyncWaitsForDrain() {
264298
const { writer, readable } = push();
265299
writer.writeSync('hello');
@@ -553,6 +587,8 @@ Promise.all([
553587
testOndrainProtocolErrorPropagates(),
554588
testFail(),
555589
testEndAsyncReturnValue(),
590+
testEndWithPreAbortedSignal(),
591+
testEndSignalAbortWhileDraining(),
556592
testEndAfterEndSyncWaitsForDrain(),
557593
testWriteUint8Array(),
558594
testOndrainWaitsForDrain(),

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL