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: #64727Fixes: #64726
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -630,9 +630,10 @@ class BroadcastWriter {
 returnfalse;
 }
 
-// end() is synchronous internally - signal accepted for interface compliance.
 end(options){
-getWriterSignal(options);
+constsignal=getWriterSignal(options);
+if(signal?.aborted)returnPromiseReject(signal.reason);
+
 if(this.#isClosed())returnthis.#closed;
 this.#closed =PromiseResolve(this.#totalBytes);
 this.#broadcast[kEnd]();

lib/internal/streams/iter/push.js

@@ -8,6 +8,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ PromisePrototypeThen,
  PromiseReject,
  PromiseResolve,
  PromiseWithResolvers,
@@ -55,6 +56,35 @@ const {
 
 constkNoFailReason=Symbol('kNoFailReason');
 
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+if(signal.aborted)returnPromiseReject(signal.reason);
+
+const{
+promise: signaledPromise,
+ resolve,
+ reject,
+}=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+
+signal.addEventListener('abort',onAbort,{
+__proto__: null,
+once: true,
+});
+PromisePrototypeThen(
+promise,
+(value)=>{
+signal.removeEventListener('abort',onAbort);
+resolve(value);
+},
+(reason)=>{
+signal.removeEventListener('abort',onAbort);
+reject(reason);
+},
+);
+returnsignaledPromise;
+}
+
 // =============================================================================
 // PushQueue - Internal Queue with Chunk-Based Backpressure
 // =============================================================================
@@ -628,7 +658,9 @@ class PushWriter {
 }
 
 end(options){
-getWriterSignal(options);
+constsignal=getWriterSignal(options);
+if(signal?.aborted)returnPromiseReject(signal.reason);
+
 constresult=this.#queue.end();
 if(result===-2){
 // Errored: reject with stored error
@@ -639,11 +671,11 @@ class PushWriter {
 // when consumer drains past the end sentinel
 constpendingEndPromise=this.#queue.pendingEndPromise;
 if(pendingEndPromise!==null){
-returnpendingEndPromise;
+returnraceEndWithSignal(pendingEndPromise,signal);
 }
 const{ promise, resolve, reject }=PromiseWithResolvers();
 this.#queue.setPendingEnd({__proto__: null, promise, resolve, reject });
-returnpromise;
+returnraceEndWithSignal(promise,signal);
 }
 // >= 0: byte count (immediate close or idempotent)
 returnPromiseResolve(result);

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

@@ -111,6 +111,22 @@ async function testWriterEnd() {
 assert.strictEqual(data,'data');
 }
 
+asyncfunctiontestWriterEndWithPreAbortedSignal(){
+const{ writer,broadcast: bc}=broadcast();
+constconsumer=bc.push();
+constreason=newError('end aborted');
+
+awaitassert.rejects(
+writer.end({signal: AbortSignal.abort(reason)}),
+(error)=>error===reason,
+);
+
+// A rejected end must leave the writer open.
+awaitwriter.write('data');
+assert.strictEqual(awaitwriter.end(),4);
+assert.strictEqual(awaittext(consumer),'data');
+}
+
 asyncfunctiontestWriterFail(){
 const{ writer,broadcast: bc}=broadcast();
 constconsumer=bc.push();
@@ -308,6 +324,7 @@ Promise.all([
 testWriteSync(),
 testWritevSync(),
 testWriterEnd(),
+testWriterEndWithPreAbortedSignal(),
 testWriterFail(),
 testCancelWithoutReason(),
 testCancelWithReason(),

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

@@ -127,6 +127,22 @@ async function testAbortSignal() {
 );
 }
 
+asyncfunctiontestWriterEndWithPreAbortedSignal(){
+const[channelA,channelB]=duplex();
+constreason=newError('end aborted');
+
+awaitassert.rejects(
+channelA.writer.end({signal: AbortSignal.abort(reason)}),
+(error)=>error===reason,
+);
+
+awaitchannelA.writer.write('still open');
+constcompletedEnd=channelA.writer.end();
+assert.strictEqual(awaittext(channelB.readable),'still open');
+assert.strictEqual(awaitcompletedEnd,10);
+awaitchannelB.close();
+}
+
 asyncfunctiontestEmptyDuplex(){
 const[channelA,channelB]=duplex();
 
@@ -182,6 +198,7 @@ Promise.all([
 testWithOptions(),
 testPerChannelOptions(),
 testAbortSignal(),
+testWriterEndWithPreAbortedSignal(),
 testEmptyDuplex(),
 testChannelFail(),
 testAbortSignalBothChannels(),

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

@@ -260,6 +260,40 @@ async function testEndAsyncReturnValue() {
 awaitconsume;
 }
 
+asyncfunctiontestEndWithPreAbortedSignal(){
+const{ writer, readable }=push();
+constreason=newError('end aborted');
+
+writer.writeSync('hello');
+awaitassert.rejects(
+writer.end({signal: AbortSignal.abort(reason)}),
+(error)=>error===reason,
+);
+
+// A rejected end must leave the writer open.
+writer.writeSync(' world');
+constconsume=text(readable);
+assert.strictEqual(awaitwriter.end(),11);
+assert.strictEqual(awaitconsume,'hello world');
+}
+
+asyncfunctiontestEndSignalAbortWhileDraining(){
+const{ writer, readable }=push();
+constcontroller=newAbortController();
+constreason=newError('end aborted while draining');
+
+writer.writeSync('hello');
+constabortedEnd=writer.end({signal: controller.signal});
+controller.abort(reason);
+
+awaitassert.rejects(abortedEnd,(error)=>error===reason);
+
+// Aborting the operation does not undo the end-of-stream signal.
+constcompletedEnd=writer.end();
+assert.strictEqual(awaittext(readable),'hello');
+assert.strictEqual(awaitcompletedEnd,5);
+}
+
 asyncfunctiontestEndAfterEndSyncWaitsForDrain(){
 const{ writer, readable }=push();
 writer.writeSync('hello');
@@ -553,6 +587,8 @@ Promise.all([
 testOndrainProtocolErrorPropagates(),
 testFail(),
 testEndAsyncReturnValue(),
+testEndWithPreAbortedSignal(),
+testEndSignalAbortWhileDraining(),
 testEndAfterEndSyncWaitsForDrain(),
 testWriteUint8Array(),
 testOndrainWaitsForDrain(),