stream: ensure factory signals remain active through closing

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65658
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

doc/api/stream_iter.md

@@ -765,7 +765,9 @@ added: v24.20.0
     **Default:** `16384`.
   * `backpressure` {string} Backpressure policy: `'strict'`, `'unbounded'`,
     `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`.
-  * `signal` {AbortSignal} Abort the stream.
+  * `signal` {AbortSignal} Abort the stream. The signal remains active while
+    buffered data drains after `writer.end()`; aborting during that time fails
+    the writer and rejects the pending `end()` promise.
 * Returns: {Object}
   * `writer` {Writable} The writer side.
   * `readable` {AsyncIterable} whose chunks fulfill with {Uint8Array\[]}

lib/internal/streams/iter/push.js

@@ -331,14 +331,14 @@ class PushQueue {
       return this.#bytesWritten; // Idempotent
     }
 
-    this.#cleanup();
     this.#rejectPendingWrites(
       new ERR_INVALID_STATE.TypeError('Writer closed'));
     this.#resolvePendingDrains(false);
 
     // If buffer is empty, close immediately
     if (this.#slots.length === 0) {
       this.#writerState = 'closed';
+      this.#cleanup();
       this.#resolvePendingReads();
       return this.#bytesWritten;
     }
@@ -356,6 +356,7 @@ class PushQueue {
   endDrained() {
     if (this.#writerState !== 'closing') return;
     this.#writerState = 'closed';
+    this.#cleanup();
     if (this.#pendingEnd) {
       this.#pendingEnd.resolve(this.#bytesWritten);
       this.#pendingEnd = null;

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

@@ -294,6 +294,21 @@ async function testEndSignalAbortWhileDraining() {
   assert.strictEqual(await completedEnd, 5);
 }
 
+async function testFactorySignalAbortWhileDraining() {
+  const controller = new AbortController();
+  const reason = new Error('stream aborted while draining');
+  const { writer, readable } = push({ signal: controller.signal });
+
+  writer.writeSync('hello');
+  const end = writer.end();
+  const endRejected = assert.rejects(end, (error) => error === reason);
+  controller.abort(reason);
+
+  await endRejected;
+  await assert.rejects(text(readable), (error) => error === reason);
+  await assert.rejects(writer.end(), (error) => error === reason);
+}
+
 async function testEndAfterEndSyncWaitsForDrain() {
   const { writer, readable } = push();
   writer.writeSync('hello');
@@ -603,6 +618,7 @@ Promise.all([
   testEndAsyncReturnValue(),
   testEndWithPreAbortedSignal(),
   testEndSignalAbortWhileDraining(),
+  testFactorySignalAbortWhileDraining(),
   testEndAfterEndSyncWaitsForDrain(),
   testWriteUint8Array(),
   testOndrainWaitsForDrain(),