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>

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -630,9 +630,10 @@ class BroadcastWriter {
     return false;
   }
 
-  // end() is synchronous internally - signal accepted for interface compliance.
   end(options) {
-    getWriterSignal(options);
+    const signal = getWriterSignal(options);
+    if (signal?.aborted) return PromiseReject(signal.reason);
+
     if (this.#isClosed()) return this.#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 {
 
 const kNoFailReason = Symbol('kNoFailReason');
 
+function raceEndWithSignal(promise, signal) {
+  if (!signal) return promise;
+  if (signal.aborted) return PromiseReject(signal.reason);
+
+  const {
+    promise: signaledPromise,
+    resolve,
+    reject,
+  } = PromiseWithResolvers();
+  const onAbort = () => 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);
+    },
+  );
+  return signaledPromise;
+}
+
 // =============================================================================
 // PushQueue - Internal Queue with Chunk-Based Backpressure
 // =============================================================================
@@ -628,7 +658,9 @@ class PushWriter {
   }
 
   end(options) {
-    getWriterSignal(options);
+    const signal = getWriterSignal(options);
+    if (signal?.aborted) return PromiseReject(signal.reason);
+
     const result = this.#queue.end();
     if (result === -2) {
       // Errored: reject with stored error
@@ -639,11 +671,11 @@ class PushWriter {
       // when consumer drains past the end sentinel
       const pendingEndPromise = this.#queue.pendingEndPromise;
       if (pendingEndPromise !== null) {
-        return pendingEndPromise;
+        return raceEndWithSignal(pendingEndPromise, signal);
       }
       const { promise, resolve, reject } = PromiseWithResolvers();
       this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
-      return promise;
+      return raceEndWithSignal(promise, signal);
     }
     // >= 0: byte count (immediate close or idempotent)
     return PromiseResolve(result);

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

@@ -111,6 +111,22 @@ async function testWriterEnd() {
   assert.strictEqual(data, 'data');
 }
 
+async function testWriterEndWithPreAbortedSignal() {
+  const { writer, broadcast: bc } = broadcast();
+  const consumer = bc.push();
+  const reason = new Error('end aborted');
+
+  await assert.rejects(
+    writer.end({ signal: AbortSignal.abort(reason) }),
+    (error) => error === reason,
+  );
+
+  // A rejected end must leave the writer open.
+  await writer.write('data');
+  assert.strictEqual(await writer.end(), 4);
+  assert.strictEqual(await text(consumer), 'data');
+}
+
 async function testWriterFail() {
   const { writer, broadcast: bc } = broadcast();
   const consumer = 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() {
   );
 }
 
+async function testWriterEndWithPreAbortedSignal() {
+  const [channelA, channelB] = duplex();
+  const reason = new Error('end aborted');
+
+  await assert.rejects(
+    channelA.writer.end({ signal: AbortSignal.abort(reason) }),
+    (error) => error === reason,
+  );
+
+  await channelA.writer.write('still open');
+  const completedEnd = channelA.writer.end();
+  assert.strictEqual(await text(channelB.readable), 'still open');
+  assert.strictEqual(await completedEnd, 10);
+  await channelB.close();
+}
+
 async function testEmptyDuplex() {
   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() {
   await consume;
 }
 
+async function testEndWithPreAbortedSignal() {
+  const { writer, readable } = push();
+  const reason = new Error('end aborted');
+
+  writer.writeSync('hello');
+  await assert.rejects(
+    writer.end({ signal: AbortSignal.abort(reason) }),
+    (error) => error === reason,
+  );
+
+  // A rejected end must leave the writer open.
+  writer.writeSync(' world');
+  const consume = text(readable);
+  assert.strictEqual(await writer.end(), 11);
+  assert.strictEqual(await consume, 'hello world');
+}
+
+async function testEndSignalAbortWhileDraining() {
+  const { writer, readable } = push();
+  const controller = new AbortController();
+  const reason = new Error('end aborted while draining');
+
+  writer.writeSync('hello');
+  const abortedEnd = writer.end({ signal: controller.signal });
+  controller.abort(reason);
+
+  await assert.rejects(abortedEnd, (error) => error === reason);
+
+  // Aborting the operation does not undo the end-of-stream signal.
+  const completedEnd = writer.end();
+  assert.strictEqual(await text(readable), 'hello');
+  assert.strictEqual(await completedEnd, 5);
+}
+
 async function testEndAfterEndSyncWaitsForDrain() {
   const { writer, readable } = push();
   writer.writeSync('hello');
@@ -553,6 +587,8 @@ Promise.all([
   testOndrainProtocolErrorPropagates(),
   testFail(),
   testEndAsyncReturnValue(),
+  testEndWithPreAbortedSignal(),
+  testEndSignalAbortWhileDraining(),
   testEndAfterEndSyncWaitsForDrain(),
   testWriteUint8Array(),
   testOndrainWaitsForDrain(),