stream: drain pending writes before broadcast end

Keep processing writes that were queued before end(). Signal the
broadcast end only after those writes enter the shared buffer.

Resolve end() after all consumers reach end-of-stream. This prevents
backpressured writes from remaining pending forever.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: codex:gpt-5.6-sol
PR-URL: #65334
Fixes: #65333
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

doc/api/stream_iter.md

@@ -431,14 +431,16 @@ the write. Use [`ondrain()`][] to wait for capacity rather than polling.
     the pending `end()` call; it does not fail the writer itself.
 * Returns: {Promise} Fulfills with the total number of bytes written.
 
-Signal that no more data will be written.
+Signals that no more data will be written and waits for buffered data to drain.
 
 #### `writer.endSync()`
 
-* Returns: {number} Total bytes written, or `-1` if the writer is not open.
+* Returns: {number} Total bytes written, or `-1` if ending cannot complete
+  synchronously.
 
-Synchronous variant of `writer.end()`. Returns `-1` if the writer is already
-closed or errored. Can be used as a try-fallback pattern:
+Synchronous variant of `writer.end()`. A return value of `-1` means closing has
+started but requires asynchronous draining. Use the try-fallback pattern to
+await completion:
 
 ```cjs
 const result = writer.endSync();

lib/internal/streams/iter/broadcast.js

@@ -14,6 +14,8 @@ const {
   PromiseReject,
   PromiseResolve,
   PromiseWithResolvers,
+  SafePromisePrototypeFinally,
+  SafePromiseRace,
   SafeSet,
   Symbol,
   SymbolAsyncDispose,
@@ -77,6 +79,22 @@ const kEnd = Symbol('kEnd');
 const kAbort = Symbol('kAbort');
 const kCanWrite = Symbol('kCanWrite');
 const kOnBufferDrained = Symbol('kOnBufferDrained');
+const kOnEndDrained = Symbol('kOnEndDrained');
+const kPendingWriteRemoved = Symbol('kPendingWriteRemoved');
+
+function raceEndWithSignal(promise, signal) {
+  if (!signal) return promise;
+
+  const { promise: aborted, reject } = PromiseWithResolvers();
+  const onAbort = () => reject(signal.reason);
+  signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
+  if (signal.aborted) onAbort();
+
+  return SafePromisePrototypeFinally(
+    SafePromiseRace([promise, aborted]),
+    () => signal.removeEventListener('abort', onAbort),
+  );
+}
 
 // =============================================================================
 // Broadcast Implementation
@@ -100,6 +118,7 @@ class BroadcastImpl {
   constructor(options) {
     this.#options = options;
     this[kOnBufferDrained] = null;
+    this[kOnEndDrained] = null;
   }
 
   setWriter(writer) {
@@ -183,6 +202,7 @@ class BroadcastImpl {
       if (self.#deleteConsumer(state)) {
         self.#tryTrimBuffer();
       }
+      self.#notifyEndDrained();
     }
 
     return {
@@ -358,10 +378,11 @@ class BroadcastImpl {
         }
       }
     }
+    this.#notifyEndDrained();
   }
 
   [kAbort](reason) {
-    if (this.#ended || this.#error !== undefined) return;
+    if (this.#error !== undefined) return;
     this.#error = reason;
     this.#ended = true;
 
@@ -396,6 +417,12 @@ class BroadcastImpl {
 
   // Private methods
 
+  #notifyEndDrained() {
+    if (this.#ended && this.#consumers.size === 0) {
+      this[kOnEndDrained]?.();
+    }
+  }
+
   #recomputeMinCursor() {
     const { minCursor, minCursorConsumers } = getMinCursor(
       this.#consumers, this.#bufferStart + this.#buffer.length);
@@ -516,8 +543,9 @@ let getBroadcastPendingWrites;
 class BroadcastWriter {
   #broadcast;
   #totalBytes = 0;
-  #closed;
-  #aborted = false;
+  #state = 'open';
+  #error;
+  #pendingEnd;
   #pendingWrites = new RingBuffer();
   #pendingDrains = [];
 
@@ -532,8 +560,11 @@ class BroadcastWriter {
 
     this.#broadcast[kOnBufferDrained] = () => {
       this.#resolvePendingWrites();
-      this.#resolvePendingDrains(true);
+      if (this.#state === 'open') {
+        this.#resolvePendingDrains(true);
+      }
     };
+    this.#broadcast[kOnEndDrained] = () => this.#endDrained();
   }
 
   // The drainable protocol works with Stream.ondrain to provide a notification
@@ -547,20 +578,12 @@ class BroadcastWriter {
     return promise;
   }
 
-  #isClosed() {
-    return this.#closed !== undefined;
-  }
-
-  #isClosedOrAborted() {
-    return this.#isClosed() || this.#aborted;
-  }
-
   get canWrite() {
-    return this.#isClosedOrAborted() ? null : this.#broadcast[kCanWrite]();
+    return this.#state === 'open' ? this.#broadcast[kCanWrite]() : null;
   }
 
   #canUseWriteFastPath(signal) {
-    return !signal && !this.#isClosed() && !this.#aborted &&
+    return !signal && this.#state === 'open' &&
         this.#broadcast[kCanWrite]();
   }
 
@@ -592,13 +615,15 @@ class BroadcastWriter {
   }
 
   async #writevSlow(chunks, signal) {
-    // Check for pre-aborted
-    signal?.throwIfAborted();
-
-    if (this.#isClosedOrAborted()) {
+    if (this.#state === 'errored') {
+      throw this.#error;
+    }
+    if (this.#state !== 'open') {
       throw new ERR_INVALID_STATE.TypeError('Writer is closed');
     }
 
+    signal?.throwIfAborted();
+
     const converted = convertChunks(chunks);
 
     if (this.#broadcast[kWrite](converted)) {
@@ -624,7 +649,7 @@ class BroadcastWriter {
   }
 
   writeSync(chunk) {
-    if (this.#isClosedOrAborted()) return false;
+    if (this.#state !== 'open') return false;
     if (!this.#broadcast[kCanWrite]()) return false;
     const converted =
       toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
   writevSync(chunks) {
     validateArray(chunks, 'chunks');
-    if (this.#isClosedOrAborted()) return false;
+    if (this.#state !== 'open') return false;
     if (!this.#broadcast[kCanWrite]()) return false;
     const converted = convertChunks(chunks);
     if (this.#broadcast[kWrite](converted)) {
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
   end(options) {
     const signal = getWriterSignal(options);
+    if (this.#state === 'errored') return PromiseReject(this.#error);
+    if (this.#state === 'closed') return PromiseResolve(this.#totalBytes);
     if (signal?.aborted) return PromiseReject(signal.reason);
 
-    if (this.#isClosed()) return this.#closed;
-    this.#closed = PromiseResolve(this.#totalBytes);
-    this.#broadcast[kEnd]();
-    this.#resolvePendingDrains(false);
-    return this.#closed;
+    const endPromise = this.#getEndPromise();
+    if (this.#state === 'open') {
+      this.#state = 'closing';
+      this.#resolvePendingDrains(false);
+      this.#finishEndIfReady();
+    }
+
+    return raceEndWithSignal(endPromise, signal);
   }
 
   endSync() {
-    if (this.#closed) return this.#totalBytes;
-    this.#closed = PromiseResolve(this.#totalBytes);
-    this.#broadcast[kEnd]();
+    if (this.#state === 'closed') return this.#totalBytes;
+    if (this.#state === 'errored' || this.#state === 'closing') return -1;
+
+    this.#state = 'closing';
     this.#resolvePendingDrains(false);
-    return this.#totalBytes;
+    this.#finishEndIfReady();
+    return this.#state === 'closed' ? this.#totalBytes : -1;
   }
 
   fail(reason) {
-    if (this.#isClosedOrAborted()) return;
-    this.#aborted = true;
-    this.#closed = PromiseResolve(this.#totalBytes);
+    if (this.#state === 'errored' || this.#state === 'closed') return;
+    this.#state = 'errored';
     const error = reason ?? new ERR_INVALID_STATE.TypeError('Failed');
+    this.#error = error;
     this.#rejectPendingWrites(error);
     this.#rejectPendingDrains(error);
+    this.#pendingEnd?.reject(error);
     this.#broadcast[kAbort](error);
   }
 
   [SymbolAsyncDispose]() {
+    if (this.#state === 'closing') return this.#getEndPromise();
     this.fail();
     return PromiseResolve();
   }
@@ -688,11 +722,33 @@ class BroadcastWriter {
   }
 
   [kCancelWriter]() {
-    if (this.#isClosed()) return;
-    this.#closed = PromiseResolve(this.#totalBytes);
+    if (this.#state === 'closed' || this.#state === 'errored') return;
+    this.#state = 'closed';
     this.#rejectPendingWrites(
       lazyDOMException('Broadcast cancelled', 'AbortError'));
     this.#resolvePendingDrains(false);
+    this.#pendingEnd?.resolve(this.#totalBytes);
+  }
+
+  #getEndPromise() {
+    this.#pendingEnd ??= PromiseWithResolvers();
+    return this.#pendingEnd.promise;
+  }
+
+  #finishEndIfReady() {
+    if (this.#state === 'closing' && this.#pendingWrites.length === 0) {
+      this.#broadcast[kEnd]();
+    }
+  }
+
+  #endDrained() {
+    if (this.#state !== 'closing') return;
+    this.#state = 'closed';
+    this.#pendingEnd?.resolve(this.#totalBytes);
+  }
+
+  [kPendingWriteRemoved]() {
+    this.#finishEndIfReady();
   }
 
   /**
@@ -724,6 +780,7 @@ class BroadcastWriter {
         break;
       }
     }
+    this.#finishEndIfReady();
   }
 
   #rejectPendingWrites(error) {
@@ -756,6 +813,7 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) {
     if (idx !== -1) pendingWrites.removeAt(idx);
     entry.chunk = null;
     reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
+    if (idx !== -1) self[kPendingWriteRemoved]();
   };
   entry.resolve = function() {
     signal.removeEventListener('abort', onAbort);