@@ -21,7 +21,6 @@ const {
   SymbolAsyncDispose,
   SymbolAsyncIterator,
   SymbolDispose,
-  TypedArrayPrototypeGetByteLength,
 } = primordials;
 
 const { lazyDOMException } = require('internal/util');
@@ -59,6 +58,7 @@ const {
   kMultiConsumerDefaultBudget,
   kResolvedPromise,
   convertChunks,
+  createBatchEntry,
   getWriterSignal,
   getMinCursor,
   hasProtocol,
@@ -67,6 +67,7 @@ const {
   wrapError,
   toUint8Array,
   validateBackpressure,
+  validateBatchEntry,
 } = require('internal/streams/iter/utils');
 
 const {
@@ -215,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 &&
@@ -304,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.
@@ -322,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) {
@@ -338,7 +340,7 @@ class BroadcastImpl {
       }
     }
 
-    this.#buffer.push(chunk);
+    this.#buffer.push(entry);
     this.#bufferedBytes += batchSize;
     this.#notifyConsumers();
     return true;
@@ -352,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 &&
@@ -433,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;
@@ -445,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() {
@@ -464,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 &&
@@ -587,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);
@@ -600,10 +609,9 @@ 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);
+      this.#broadcast[kWrite](batch);
+      this.#totalBytes += batch.byteLength;
       return kResolvedPromise;
     }
     return this.#writevSlow(chunks, signal);
@@ -619,12 +627,10 @@ 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]);
-      }
+    if (this.#broadcast[kWrite](batch)) {
+      this.#totalBytes += batch.byteLength;
       return;
     }
 
@@ -636,20 +642,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;
@@ -660,10 +667,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;
@@ -752,9 +758,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);
@@ -765,14 +771,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();
@@ -806,18 +815,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 });
 
@@ -54,6 +54,8 @@ const {
 
 const {
   concatBytes,
+  createBatchEntry,
+  validateBatchEntry,
   yieldAbortable,
 } = require('internal/streams/iter/utils');
 
@@ -89,6 +91,17 @@ function isMergeOptions(value) {
 // Shared chunk collection helpers
 // =============================================================================
 
+function flattenBatchEntries(entries) {
+  const chunks = [];
+  for (let i = 0; i < entries.length; i++) {
+    const batch = validateBatchEntry(entries[i]);
+    for (let j = 0; j < batch.length; j++) {
+      ArrayPrototypePush(chunks, batch[j]);
+    }
+  }
+  return chunks;
+}
+
 /**
  * Collect chunks from a sync source into an array.
  * @param {Iterable<Uint8Array[]>} source
@@ -98,23 +111,23 @@ function isMergeOptions(value) {
 function collectSync(source, limit) {
   // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc.
   const normalized = fromSync(source);
-  const chunks = [];
+  const entries = [];
   let totalBytes = 0;
 
   for (const batch of normalized) {
-    for (let i = 0; i < batch.length; i++) {
-      const chunk = batch[i];
-      if (limit !== undefined) {
-        totalBytes += TypedArrayPrototypeGetByteLength(chunk);
+    const entry = createBatchEntry(batch);
+    if (limit !== undefined) {
+      for (let i = 0; i < entry.views.length; i++) {
+        totalBytes += entry.views[i].byteLength;
         if (totalBytes > limit) {
           throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
         }
       }
-      ArrayPrototypePush(chunks, chunk);
     }
+    ArrayPrototypePush(entries, entry);
   }
 
-  return chunks;
+  return flattenBatchEntries(entries);
 }
 
 /**
@@ -131,16 +144,14 @@ async function collectAsync(source, signal, limit) {
   const abortableSource = signal && isAsyncIterable(source) ?
     yieldAbortable(source, signal) : source;
   const normalized = from(abortableSource);
-  const chunks = [];
+  const entries = [];
 
   // Fast path: no signal and no limit
   if (!signal && limit === undefined) {
     for await (const batch of normalized) {
-      for (let i = 0; i < batch.length; i++) {
-        ArrayPrototypePush(chunks, batch[i]);
-      }
+      ArrayPrototypePush(entries, createBatchEntry(batch));
     }
-    return chunks;
+    return flattenBatchEntries(entries);
   }
 
   // Slow path: with signal or limit checks
@@ -149,19 +160,19 @@ async function collectAsync(source, signal, limit) {
 
   for await (const batch of iterable) {
     signal?.throwIfAborted();
-    for (let i = 0; i < batch.length; i++) {
-      const chunk = batch[i];
-      if (limit !== undefined) {
-        totalBytes += TypedArrayPrototypeGetByteLength(chunk);
+    const entry = createBatchEntry(batch);
+    if (limit !== undefined) {
+      for (let i = 0; i < entry.views.length; i++) {
+        totalBytes += entry.views[i].byteLength;
         if (totalBytes > limit) {
           throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
         }
       }
-      ArrayPrototypePush(chunks, chunk);
     }
+    ArrayPrototypePush(entries, entry);
   }
 
-  return chunks;
+  return flattenBatchEntries(entries);
 }
 
 /**