stream: defend against re-entrancy in writev

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65652
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -610,9 +610,11 @@ class BroadcastWriter {
     if (this.#canUseWriteFastPath(signal)) {
       const converted = convertChunks(chunks);
       const batch = createBatchEntry(converted);
-      this.#broadcast[kWrite](batch);
-      this.#totalBytes += batch.byteLength;
-      return kResolvedPromise;
+      if (this.#state === 'open' && this.#broadcast[kWrite](batch)) {
+        this.#totalBytes += batch.byteLength;
+        return kResolvedPromise;
+      }
+      return this.#writeBatchSlow(batch, signal);
     }
     return this.#writevSlow(chunks, signal);
   }
@@ -629,6 +631,19 @@ class BroadcastWriter {
 
     const batch = createBatchEntry(convertChunks(chunks));
 
+    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;

lib/internal/streams/iter/push.js

@@ -657,12 +657,10 @@ class PushWriter {
   writev(chunks, options) {
     validateArray(chunks, 'chunks');
     const signal = getWriterSignal(options);
-    if (!signal && this.#queue.canWriteSync()) {
-      const bytes = convertChunks(chunks);
-      this.#queue.writeSync(bytes);
+    const bytes = convertChunks(chunks);
+    if (!signal && this.#queue.writeSync(bytes)) {
       return kResolvedPromise;
     }
-    const bytes = convertChunks(chunks);
     return this.#queue.writeAsync(bytes, signal);
   }
 

lib/internal/streams/iter/utils.js

@@ -185,19 +185,6 @@ function toUint8Array(chunk) {
   return chunk;
 }
 
-/**
- * Check if all chunks in an array are already Uint8Array.
- * Short-circuits on the first non-Uint8Array chunk found.
- * @param {Array<Uint8Array|string>} chunks
- * @returns {boolean}
- */
-function allUint8Array(chunks) {
-  for (let i = 0; i < chunks.length; i++) {
-    if (!isUint8Array(chunks[i])) return false;
-  }
-  return true;
-}
-
 function snapshotByteView(value) {
   const buffer = TypedArrayPrototypeGetBuffer(value);
   const sharedBufferView = isSharedArrayBuffer(buffer) ?
@@ -317,9 +304,6 @@ function concatBytes(chunks) {
  * @returns {Uint8Array[]}
  */
 function convertChunks(chunks) {
-  if (allUint8Array(chunks)) {
-    return ArrayPrototypeSlice(chunks);
-  }
   const len = chunks.length;
   const result = new Array(len);
   for (let i = 0; i < len; i++) {
@@ -446,7 +430,6 @@ module.exports = {
   kMultiConsumerDefaultBudget,
   kPushDefaultBudget,
   kResolvedPromise,
-  allUint8Array,
   concatBytes,
   convertChunks,
   createBatchEntry,

test/parallel/test-stream-iter-writer-reentrancy.js

@@ -0,0 +1,65 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { broadcast, push } = require('stream/iter');
+
+const factories = [
+  () => push({ budget: 16384 }),
+  () => {
+    const { writer, broadcast: bc } = broadcast({ budget: 16384 });
+    return { __proto__: null, writer, readable: bc.push() };
+  },
+];
+
+async function testWritevReentrancy() {
+  for (const factory of factories) {
+    const { writer, readable } = factory();
+    const chunks = [];
+    Object.defineProperty(chunks, 0, {
+      __proto__: null,
+      enumerable: true,
+      get: common.mustCall(() => {
+        assert.strictEqual(
+          writer.writeSync(new Uint8Array(16384)), true);
+        return Uint8Array.of(42);
+      }),
+    });
+
+    let resolved = false;
+    const write = writer.writev(chunks).then(common.mustCall(() => {
+      resolved = true;
+    }));
+    await new Promise(setImmediate);
+    assert.strictEqual(resolved, false);
+
+    const iterator = readable[Symbol.asyncIterator]();
+    assert.strictEqual((await iterator.next()).value[0].byteLength, 16384);
+    await write;
+    assert.strictEqual((await iterator.next()).value[0][0], 42);
+    writer.endSync();
+    assert.strictEqual((await iterator.next()).done, true);
+  }
+
+  for (const factory of factories) {
+    const { writer, readable } = factory();
+    const chunks = [];
+    Object.defineProperty(chunks, 0, {
+      __proto__: null,
+      enumerable: true,
+      get: common.mustCall(() => {
+        writer.endSync();
+        return Uint8Array.of(42);
+      }),
+    });
+
+    await assert.rejects(writer.writev(chunks), {
+      code: 'ERR_INVALID_STATE',
+    });
+    assert.strictEqual(
+      (await readable[Symbol.asyncIterator]().next()).done, true);
+  }
+}
+
+testWritevReentrancy().then(common.mustCall());