nodejs
diff --git a/‎doc/api/stream_iter.md‎
Lines changed: 6 additions & 4 deletions b/‎doc/api/stream_iter.md‎
Lines changed: 6 additions & 4 deletions
diff --git a/‎lib/internal/streams/iter/broadcast.js‎
Lines changed: 92 additions & 34 deletions b/‎lib/internal/streams/iter/broadcast.js‎
Lines changed: 92 additions & 34 deletions
@@ -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();
 
@@ -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);