stream: honor AbortSignal in Writer.end()

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(),