stream: preserve toReadableSync batch after backpressure

lib/internal/streams/iter/classic.js

@@ -363,23 +363,37 @@ function toReadableSync(source, options = kNullPrototype) {
 
   const ReadableCtor = lazyReadable();
   const iterator = source[SymbolIterator]();
+  let hasBatch = false;
+  let batch;
+  let batchIndex = 0;
 
   return new ReadableCtor({
     __proto__: null,
     highWaterMark,
     read() {
       for (;;) {
-        const { value: batch, done } = iterator.next();
+        if (hasBatch) {
+          while (batchIndex < batch.length) {
+            if (!this.push(batch[batchIndex++])) return;
+          }
+          batch = undefined;
+          hasBatch = false;
+          batchIndex = 0;
+        }
+
+        const result = iterator.next();
+        const { done } = result;
         if (done) {
           this.push(null);
           return;
         }
-        for (let i = 0; i < batch.length; i++) {
-          if (!this.push(batch[i])) return;
-        }
+        batch = result.value;
+        hasBatch = true;
       }
     },
     destroy(err, cb) {
+      batch = undefined;
+      hasBatch = false;
       if (typeof iterator.return === 'function') iterator.return();
       cb(err);
     },

test/parallel/test-stream-iter-to-readable.js

@@ -439,6 +439,21 @@ async function testBackpressureSync() {
   assert.strictEqual(chunks.length, 10);
 }
 
+// =============================================================================
+// fromStreamIterSync: backpressure within a batch
+// =============================================================================
+
+async function testBackpressureSyncMultiChunkBatch() {
+  function* gen() {
+    yield [Buffer.from('a'), Buffer.from('b'), Buffer.from('c')];
+  }
+
+  const readable = toReadableSync(gen(), { highWaterMark: 1 });
+  const result = await collect(readable);
+
+  assert.strictEqual(result.toString(), 'abc');
+}
+
 // =============================================================================
 // fromStreamIterSync: source error
 // =============================================================================
@@ -613,6 +628,7 @@ Promise.all([
   testWithTransformAsync(),
   testBasicSync(),
   testBackpressureSync(),
+  testBackpressureSyncMultiChunkBatch(),
   testErrorSync(),
   testDestroySync(),
   testRoundTrip(),