stream: apply multiple stream/iter fixes

lib/internal/streams/iter/broadcast.js

@@ -21,7 +21,6 @@ const {
   SymbolAsyncDispose,
   SymbolAsyncIterator,
   SymbolDispose,
-  TypedArrayPrototypeGetByteLength,
 } = primordials;
 
 const { lazyDOMException } = require('internal/util');
@@ -52,13 +51,14 @@ const {
 } = require('internal/streams/iter/from');
 
 const {
-  pull: pullWithTransforms,
+  pullWithConsumerCleanup,
 } = require('internal/streams/iter/pull');
 
 const {
   kMultiConsumerDefaultBudget,
   kResolvedPromise,
   convertChunks,
+  createBatchEntry,
   getWriterSignal,
   getMinCursor,
   hasProtocol,
@@ -67,6 +67,7 @@ const {
   wrapError,
   toUint8Array,
   validateBackpressure,
+  validateBatchEntry,
 } = require('internal/streams/iter/utils');
 
 const {
@@ -157,12 +158,7 @@ class BroadcastImpl {
     // When no transforms, return rawConsumer directly (controller elided
     // per PULL-02 optimization -- no transforms means no signal recipient).
     if (transforms.length > 0 || signal) {
-      const pullArgs = [...transforms];
-      if (signal) {
-        ArrayPrototypePush(pullArgs,
-                           { __proto__: null, signal });
-      }
-      return pullWithTransforms(rawConsumer, ...pullArgs);
+      return pullWithConsumerCleanup(rawConsumer, transforms, signal);
     }
     return rawConsumer;
   }
@@ -220,7 +216,8 @@ class BroadcastImpl {
 
             const bufferIndex = state.cursor - self.#bufferStart;
             if (bufferIndex < self.#buffer.length) {
-              const chunk = self.#buffer.get(bufferIndex);
+              const chunk = self.#readEntry(self.#buffer.get(bufferIndex));
+              if (chunk === null) return PromiseReject(self.#error);
               const cursor = state.cursor;
               state.cursor++;
               if (cursor === self.#cachedMinCursor &&
@@ -309,10 +306,10 @@ class BroadcastImpl {
 
   // Methods accessed by BroadcastWriter via symbol keys
 
-  [kWrite](chunk) {
+  [kWrite](entry) {
     if (this.#ended || this.#cancelled) return false;
 
-    const batchSize = this.#batchByteSize(chunk);
+    const batchSize = entry.byteLength;
 
     // Skip empty chunks -- zero-byte writes would accumulate infinitely
     // without ever triggering backpressure under a byte-budget model.
@@ -327,7 +324,7 @@ class BroadcastImpl {
           while (this.#bufferedBytes >= this.#options.budget &&
                  this.#buffer.length > 0) {
             const evicted = this.#buffer.shift();
-            this.#bufferedBytes -= this.#batchByteSize(evicted);
+            this.#bufferedBytes -= evicted.byteLength;
             this.#bufferStart++;
           }
           for (const consumer of this.#consumers) {
@@ -343,7 +340,7 @@ class BroadcastImpl {
       }
     }
 
-    this.#buffer.push(chunk);
+    this.#buffer.push(entry);
     this.#bufferedBytes += batchSize;
     this.#notifyConsumers();
     return true;
@@ -357,7 +354,8 @@ class BroadcastImpl {
       while (consumer.resolve) {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
-          const chunk = this.#buffer.get(bufferIndex);
+          const chunk = this.#readEntry(this.#buffer.get(bufferIndex));
+          if (chunk === null) return;
           const cursor = consumer.cursor;
           consumer.cursor++;
           if (cursor === this.#cachedMinCursor &&
@@ -438,7 +436,7 @@ class BroadcastImpl {
     if (trimCount > 0) {
       for (let i = 0; i < trimCount; i++) {
         const evicted = this.#buffer.get(i);
-        this.#bufferedBytes -= this.#batchByteSize(evicted);
+        this.#bufferedBytes -= evicted.byteLength;
       }
       this.#buffer.trimFront(trimCount);
       this.#bufferStart = this.#cachedMinCursor;
@@ -450,12 +448,16 @@ class BroadcastImpl {
     }
   }
 
-  #batchByteSize(batch) {
-    let size = 0;
-    for (let i = 0; i < batch.length; i++) {
-      size += TypedArrayPrototypeGetByteLength(batch[i]);
+  #readEntry(entry) {
+    try {
+      return validateBatchEntry(entry);
+    } catch (error) {
+      this.#writer.fail(error);
+      if (this.#error === undefined) this[kAbort](error);
+      this.#buffer.clear();
+      this.#bufferedBytes = 0;
+      return null;
     }
-    return size;
   }
 
   #notifyConsumers() {
@@ -469,7 +471,8 @@ class BroadcastImpl {
       if (consumer.resolve) {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
-          const chunk = this.#buffer.get(bufferIndex);
+          const chunk = this.#readEntry(this.#buffer.get(bufferIndex));
+          if (chunk === null) return;
           const cursor = consumer.cursor;
           consumer.cursor++;
           if (cursor === this.#cachedMinCursor &&
@@ -592,8 +595,9 @@ class BroadcastWriter {
     // Fast path: no signal, writer open, buffer has space
     if (this.#canUseWriteFastPath(signal)) {
       const converted = toUint8Array(chunk);
-      this.#broadcast[kWrite]([converted]);
-      this.#totalBytes += TypedArrayPrototypeGetByteLength(converted);
+      const batch = createBatchEntry([converted]);
+      this.#broadcast[kWrite](batch);
+      this.#totalBytes += batch.byteLength;
       return kResolvedPromise;
     }
     return this.#writevSlow([chunk], signal);
@@ -605,11 +609,12 @@ class BroadcastWriter {
     // Fast path: no signal, writer open, buffer has space
     if (this.#canUseWriteFastPath(signal)) {
       const converted = convertChunks(chunks);
-      this.#broadcast[kWrite](converted);
-      for (let i = 0; i < converted.length; i++) {
-        this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]);
+      const batch = createBatchEntry(converted);
+      if (this.#state === 'open' && this.#broadcast[kWrite](batch)) {
+        this.#totalBytes += batch.byteLength;
+        return kResolvedPromise;
       }
-      return kResolvedPromise;
+      return this.#writeBatchSlow(batch, signal);
     }
     return this.#writevSlow(chunks, signal);
   }
@@ -624,12 +629,23 @@ class BroadcastWriter {
 
     signal?.throwIfAborted();
 
-    const converted = convertChunks(chunks);
+    const batch = createBatchEntry(convertChunks(chunks));
 
-    if (this.#broadcast[kWrite](converted)) {
-      for (let i = 0; i < converted.length; i++) {
-        this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]);
-      }
+    return this.#writeBatchSlow(batch, signal);
+  }
+
+  async #writeBatchSlow(batch, signal) {
+    if (this.#state === 'errored') {
+      throw this.#error;
+    }
+    if (this.#state !== 'open') {
+      throw new ERR_INVALID_STATE.TypeError('Writer is closed');
+    }
+
+    signal?.throwIfAborted();
+
+    if (this.#broadcast[kWrite](batch)) {
+      this.#totalBytes += batch.byteLength;
       return;
     }
 
@@ -641,20 +657,21 @@ class BroadcastWriter {
           'Backpressure violation: too many pending writes. ' +
           'Await each write() call to respect backpressure.');
       }
-      return this.#createPendingWrite(converted, signal);
+      return this.#createPendingWrite(batch, signal);
     }
 
     // 'unbounded' policy
-    return this.#createPendingWrite(converted, signal);
+    return this.#createPendingWrite(batch, signal);
   }
 
   writeSync(chunk) {
     if (this.#state !== 'open') return false;
     if (!this.#broadcast[kCanWrite]()) return false;
     const converted =
       toUint8Array(chunk);
-    if (this.#broadcast[kWrite]([converted])) {
-      this.#totalBytes += TypedArrayPrototypeGetByteLength(converted);
+    const batch = createBatchEntry([converted]);
+    if (this.#broadcast[kWrite](batch)) {
+      this.#totalBytes += batch.byteLength;
       return true;
     }
     return false;
@@ -665,10 +682,9 @@ class BroadcastWriter {
     if (this.#state !== 'open') return false;
     if (!this.#broadcast[kCanWrite]()) return false;
     const converted = convertChunks(chunks);
-    if (this.#broadcast[kWrite](converted)) {
-      for (let i = 0; i < converted.length; i++) {
-        this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]);
-      }
+    const batch = createBatchEntry(converted);
+    if (this.#broadcast[kWrite](batch)) {
+      this.#totalBytes += batch.byteLength;
       return true;
     }
     return false;
@@ -757,9 +773,9 @@ class BroadcastWriter {
    * promise rejects. Signal listeners are cleaned up on normal resolution.
    * @returns {Promise<void>}
    */
-  #createPendingWrite(chunk, signal) {
+  #createPendingWrite(batch, signal) {
     const { promise, resolve, reject } = PromiseWithResolvers();
-    const entry = { __proto__: null, chunk, resolve, reject };
+    const entry = { __proto__: null, batch, resolve, reject };
     this.#pendingWrites.push(entry);
     if (signal) {
       wireBroadcastWriteSignal(entry, signal, resolve, reject, this);
@@ -770,14 +786,17 @@ class BroadcastWriter {
   #resolvePendingWrites() {
     while (this.#pendingWrites.length > 0 && this.#broadcast[kCanWrite]()) {
       const pending = this.#pendingWrites.shift();
-      if (this.#broadcast[kWrite](pending.chunk)) {
-        for (let i = 0; i < pending.chunk.length; i++) {
-          this.#totalBytes += TypedArrayPrototypeGetByteLength(pending.chunk[i]);
+      try {
+        validateBatchEntry(pending.batch);
+        if (this.#broadcast[kWrite](pending.batch)) {
+          this.#totalBytes += pending.batch.byteLength;
+          pending.resolve();
+        } else {
+          this.#pendingWrites.unshift(pending);
+          break;
         }
-        pending.resolve();
-      } else {
-        this.#pendingWrites.unshift(pending);
-        break;
+      } catch (error) {
+        pending.reject(error);
       }
     }
     this.#finishEndIfReady();
@@ -811,18 +830,18 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) {
     const pendingWrites = getBroadcastPendingWrites(self);
     const idx = pendingWrites.indexOf(entry);
     if (idx !== -1) pendingWrites.removeAt(idx);
-    entry.chunk = null;
+    entry.batch = null;
     reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
     if (idx !== -1) self[kPendingWriteRemoved]();
   };
   entry.resolve = function() {
     signal.removeEventListener('abort', onAbort);
-    entry.chunk = null;
+    entry.batch = null;
     resolve();
   };
   entry.reject = function(reason) {
     signal.removeEventListener('abort', onAbort);
-    entry.chunk = null;
+    entry.batch = null;
     reject(reason);
   };
   signal.addEventListener('abort', onAbort, { __proto__: null, once: true });