From 849b419b4189a3abb5ed3f324dd4ec3ca61f8fa5 Mon Sep 17 00:00:00 2001
From: "Kamat, Trivikram" <16024985+trivikr@users.noreply.github.com>
Date: Fri, 24 Jul 2026 19:36:29 -0700
Subject: [PATCH] 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
---
 lib/internal/streams/iter/broadcast.js        |  5 ++-
 lib/internal/streams/iter/push.js             | 38 +++++++++++++++++--
 .../test-stream-iter-broadcast-basic.js       | 17 +++++++++
 test/parallel/test-stream-iter-duplex.js      | 17 +++++++++
 test/parallel/test-stream-iter-push-writer.js | 36 ++++++++++++++++++
 5 files changed, 108 insertions(+), 5 deletions(-)

diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js
index 4f6bdc9f348a36..808c254425bdce 100644
--- a/lib/internal/streams/iter/broadcast.js
+++ b/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]();
diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js
index 52281b3e7990c6..1586ee6e743bbc 100644
--- a/lib/internal/streams/iter/push.js
+++ b/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);
diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js
index 6f1efe4625da19..0644a28a462d23 100644
--- a/test/parallel/test-stream-iter-broadcast-basic.js
+++ b/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(),
diff --git a/test/parallel/test-stream-iter-duplex.js b/test/parallel/test-stream-iter-duplex.js
index 502c1f88ebb6ac..0969b91e7d2161 100644
--- a/test/parallel/test-stream-iter-duplex.js
+++ b/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(),
diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js
index 9614ac93c46fdb..8ce555ca92512e 100644
--- a/test/parallel/test-stream-iter-push-writer.js
+++ b/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(),