stream: limit iter from sync iterable batches

Bound sync iterable normalization in from() and fromSync() to
FROM_BATCH_SIZE. This avoids unbounded batches for from() sync
iterable fallbacks and lets fromSync() coalesce plain Uint8Array
values for writev paths.
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63324
Backport-PR-URL: #64675
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>

Author: trivikr

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

benchmark/streams/iter-from-batching.js

@@ -0,0 +1,100 @@
+// Measures batching behavior for stream/iter from() and fromSync()
+// with plain synchronous Uint8Array iterables.
+'use strict';
+
+constcommon=require('../common.js');
+const{ closeSync, openSync, writeSync, writevSync }=require('fs');
+const{ devNull }=require('os');
+
+constbench=common.createBenchmark(main,{
+method: ['from-first-batch','from-sync-writev'],
+chunks: [256,4096,16384],
+chunkSize: [16],
+n: [100,1000],
+},{
+flags: ['--experimental-stream-iter'],
+combinationFilter({ method, chunks, n }){
+if(n===1){
+returntrue;
+}
+if(method==='from-first-batch'){
+returnn===1000;
+}
+returnn===100&&chunks!==16384;
+},
+test: {
+chunks: 256,
+chunkSize: 16,
+n: 1,
+},
+});
+
+functionmain({ method, chunks, chunkSize, n }){
+switch(method){
+case'from-first-batch':
+returnbenchFromFirstBatch(chunks,chunkSize,n);
+case'from-sync-writev':
+returnbenchFromSyncWritev(chunks,chunkSize,n);
+}
+}
+
+function*source(chunks,chunk){
+for(leti=0;i<chunks;i++){
+yieldchunk;
+}
+}
+
+functionbenchFromFirstBatch(chunks,chunkSize,n){
+const{ from }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+letseen=0;
+
+(async()=>{
+bench.start();
+for(leti=0;i<n;i++){
+constiterator=from(source(chunks,chunk))[Symbol.asyncIterator]();
+const{ value, done }=awaititerator.next();
+if(done||value.length===0){
+thrownewError('expected a batch');
+}
+seen+=value.length;
+}
+bench.end(n);
+if(seen===0){
+thrownewError('expected chunks');
+}
+})();
+}
+
+functionbenchFromSyncWritev(chunks,chunkSize,n){
+const{ pipeToSync }=require('stream/iter');
+constchunk=newUint8Array(chunkSize);
+constexpected=chunks*chunkSize*n;
+letseen=0;
+lettotal=0;
+constfd=openSync(devNull,'w');
+constwriter={
+writeSync(chunk){
+writeSync(fd,chunk);
+seen++;
+},
+writevSync(batch){
+writevSync(fd,batch);
+seen+=batch.length;
+},
+};
+
+try{
+bench.start();
+for(leti=0;i<n;i++){
+total+=pipeToSync(source(chunks,chunk),writer);
+}
+bench.end(chunks*n);
+}finally{
+closeSync(fd);
+}
+
+if(total!==expected||seen!==chunks*n){
+thrownewError('unexpected chunk count');
+}
+}

lib/internal/streams/iter/from.js

@@ -47,7 +47,7 @@ const {
  toUint8Array,
 }=require('internal/streams/iter/utils');
 
-// Maximum number of chunks to yield per batch from from(Uint8Array[]).
+// Maximum number of chunks to yield per batch from from()/fromSync().
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 constFROM_BATCH_SIZE=128;
@@ -190,33 +190,66 @@ function isUint8ArrayBatch(value) {
 returntrue;
 }
 
+function*yieldBoundedBatch(batch){
+if(batch.length===0){
+return;
+}
+if(batch.length<=FROM_BATCH_SIZE){
+yieldbatch;
+return;
+}
+for(leti=0;i<batch.length;i+=FROM_BATCH_SIZE){
+yieldArrayPrototypeSlice(batch,i,i+FROM_BATCH_SIZE);
+}
+}
+
 /**
  * Normalize a sync streamable source, yielding batches of Uint8Array.
  * @param {Iterable} source
  * @yields {Uint8Array[]}
  */
 function*normalizeSyncSource(source){
+letbatch=[];
+
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
-if(value.length>0){
-yieldvalue;
+if(batch.length>0){
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
-yield[value];
+ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value
-constbatch=[];
-for(constchunkofnormalizeSyncValue(value)){
-ArrayPrototypePush(batch,chunk);
-}
 if(batch.length>0){
 yieldbatch;
+batch=[];
+}
+letvalueBatch=[];
+for(constchunkofnormalizeSyncValue(value)){
+ArrayPrototypePush(valueBatch,chunk);
+if(valueBatch.length===FROM_BATCH_SIZE){
+yieldvalueBatch;
+valueBatch=[];
+}
 }
+if(valueBatch.length>0){
+yieldvalueBatch;
+}
+}
+
+if(batch.length>0){
+yieldbatch;
 }
 }
 
@@ -329,36 +362,42 @@ async function* normalizeAsyncSource(source) {
 return;
 }
 
-// Fall back to sync iteration - batch all sync values together
+// Fall back to sync iteration - batch sync values together with a bound.
 if(isSyncIterable(source)){
-constbatch=[];
+letbatch=[];
 
 for(constvalueofsource){
 // Fast path 1: value is already a Uint8Array[] batch
 if(isUint8ArrayBatch(value)){
 // Flush any accumulated batch first
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
-}
-if(value.length>0){
-yieldvalue;
+yieldbatch;
+batch=[];
 }
+yield*yieldBoundedBatch(value);
 continue;
 }
 // Fast path 2: value is a single Uint8Array (very common)
 if(isUint8Array(value)){
 ArrayPrototypePush(batch,value);
+if(batch.length===FROM_BATCH_SIZE){
+yieldbatch;
+batch=[];
+}
 continue;
 }
 // Slow path: normalize the value - must flush and yield individually
 if(batch.length>0){
-yieldArrayPrototypeSlice(batch);
-batch.length=0;
+yieldbatch;
+batch=[];
 }
-constasyncBatch=[];
+letasyncBatch=[];
 forawait(constchunkofnormalizeAsyncValue(value)){
 ArrayPrototypePush(asyncBatch,chunk);
+if(asyncBatch.length===FROM_BATCH_SIZE){
+yieldasyncBatch;
+asyncBatch=[];
+}
 }
 if(asyncBatch.length>0){
 yieldasyncBatch;

test/parallel/test-stream-iter-from-coverage.js

@@ -31,6 +31,22 @@ async function testFromSyncSubBatching() {
 assert.strictEqual(totalChunks,200);
 }
 
+// fromSync: generic sync iterables of Uint8Array use bounded batches
+asyncfunctiontestFromSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+for(constbatchoffromSync(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // from: Uint8Array[] with > 128 elements triggers sub-batching (async)
 asyncfunctiontestFromAsyncSubBatching(){
 constbigBatch=Array.from({length: 200},
@@ -44,6 +60,22 @@ async function testFromAsyncSubBatching() {
 assert.strictEqual(batches[1].length,72);
 }
 
+// from: sync iterables use bounded batches instead of one unbounded batch
+asyncfunctiontestFromAsyncSyncIterableSubBatching(){
+function*gen(){
+for(leti=0;i<200;i++){
+yieldnewUint8Array([i&0xFF]);
+}
+}
+constbatches=[];
+forawait(constbatchoffrom(gen())){
+batches.push(batch);
+}
+assert.strictEqual(batches.length,2);
+assert.strictEqual(batches[0].length,128);
+assert.strictEqual(batches[1].length,72);
+}
+
 // Exact boundary: 128 elements → single batch (no split)
 asyncfunctiontestFromSubBatchingBoundary(){
 constexactBatch=Array.from({length: 128},
@@ -133,7 +165,9 @@ async function testFromSyncInvalidYield() {
 
 Promise.all([
 testFromSyncSubBatching(),
+testFromSyncIterableSubBatching(),
 testFromAsyncSubBatching(),
+testFromAsyncSyncIterableSubBatching(),
 testFromSubBatchingBoundary(),
 testFromSubBatchingBoundaryPlus1(),
 testFromSyncDataViewInGenerator(),

test/parallel/test-stream-iter-from-sync.js

@@ -66,9 +66,10 @@ function testFromSyncGenerator() {
 for(constbatchofreadable){
 batches.push(batch);
 }
-assert.strictEqual(batches.length,2);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,2);
 assert.deepStrictEqual(batches[0][0],newUint8Array([1,2]));
-assert.deepStrictEqual(batches[1][0],newUint8Array([3,4]));
+assert.deepStrictEqual(batches[0][1],newUint8Array([3,4]));
 }
 
 functiontestFromSyncNestedIterables(){

test/parallel/test-stream-iter-pipeto-writev.js

@@ -151,6 +151,27 @@ async function testPipeToSyncWritev() {
 assert.ok(batches.some((b)=>b.length>1));
 }
 
+// pipeToSync batches plain Uint8Array chunks for writevSync
+asyncfunctiontestPipeToSyncPlainChunksWritev(){
+constbatches=[];
+constwrites=[];
+constwriter={
+writevSync(chunks){batches.push(chunks);},
+writeSync(chunk){writes.push(chunk);returntrue;},
+endSync(){return0;},
+};
+function*source(){
+yieldnewUint8Array([1]);
+yieldnewUint8Array([2]);
+yieldnewUint8Array([3]);
+}
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,3);
+assert.strictEqual(batches.length,1);
+assert.strictEqual(batches[0].length,3);
+assert.strictEqual(writes.length,0);
+}
+
 // pipeToSync with writer that has write() and writeSync() — writeSync preferred
 asyncfunctiontestPipeToSyncWriteFallback(){
 constsyncWrites=[];
@@ -174,5 +195,6 @@ Promise.all([
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
+testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());