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: #65334Fixes: #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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);

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
 constresult=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');
 constkAbort=Symbol('kAbort');
 constkCanWrite=Symbol('kCanWrite');
 constkOnBufferDrained=Symbol('kOnBufferDrained');
+constkOnEndDrained=Symbol('kOnEndDrained');
+constkPendingWriteRemoved=Symbol('kPendingWriteRemoved');
+
+functionraceEndWithSignal(promise,signal){
+if(!signal)returnpromise;
+
+const{promise: aborted, reject }=PromiseWithResolvers();
+constonAbort=()=>reject(signal.reason);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted)onAbort();
+
+returnSafePromisePrototypeFinally(
+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;
 classBroadcastWriter{
  #broadcast;
  #totalBytes =0;
- #closed;
- #aborted =false;
+ #state ='open';
+ #error;
+ #pendingEnd;
  #pendingWrites =newRingBuffer();
  #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 {
 returnpromise;
 }
 
- #isClosed(){
-returnthis.#closed !==undefined;
-}
-
- #isClosedOrAborted(){
-returnthis.#isClosed()||this.#aborted;
-}
-
 getcanWrite(){
-returnthis.#isClosedOrAborted() ? null :this.#broadcast[kCanWrite]();
+returnthis.#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'){
+throwthis.#error;
+}
+if(this.#state !=='open'){
 thrownewERR_INVALID_STATE.TypeError('Writer is closed');
 }
 
+signal?.throwIfAborted();
+
 constconverted=convertChunks(chunks);
 
 if(this.#broadcast[kWrite](converted)){
@@ -624,7 +649,7 @@ class BroadcastWriter {
 }
 
 writeSync(chunk){
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=
 toUint8Array(chunk);
@@ -637,7 +662,7 @@ class BroadcastWriter {
 
 writevSync(chunks){
 validateArray(chunks,'chunks');
-if(this.#isClosedOrAborted())returnfalse;
+if(this.#state !=='open')returnfalse;
 if(!this.#broadcast[kCanWrite]())returnfalse;
 constconverted=convertChunks(chunks);
 if(this.#broadcast[kWrite](converted)){
@@ -651,34 +676,43 @@ class BroadcastWriter {
 
 end(options){
 constsignal=getWriterSignal(options);
+if(this.#state ==='errored')returnPromiseReject(this.#error);
+if(this.#state ==='closed')returnPromiseResolve(this.#totalBytes);
 if(signal?.aborted)returnPromiseReject(signal.reason);
 
-if(this.#isClosed())returnthis.#closed;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
-this.#resolvePendingDrains(false);
-returnthis.#closed;
+constendPromise=this.#getEndPromise();
+if(this.#state ==='open'){
+this.#state ='closing';
+this.#resolvePendingDrains(false);
+this.#finishEndIfReady();
+}
+
+returnraceEndWithSignal(endPromise,signal);
 }
 
 endSync(){
-if(this.#closed)returnthis.#totalBytes;
-this.#closed =PromiseResolve(this.#totalBytes);
-this.#broadcast[kEnd]();
+if(this.#state ==='closed')returnthis.#totalBytes;
+if(this.#state ==='errored'||this.#state ==='closing')return-1;
+
+this.#state ='closing';
 this.#resolvePendingDrains(false);
-returnthis.#totalBytes;
+this.#finishEndIfReady();
+returnthis.#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';
 consterror=reason??newERR_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')returnthis.#getEndPromise();
 this.fail();
 returnPromiseResolve();
 }
@@ -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();
+returnthis.#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);