stream: guard shareSync() against re-entrant source reads

If a shareSync() source read from a consumer of the same share while
producing a value, the nested read re-entered the source iterator.
For generators this threw "Generator is already running" from inside
the nested read, which recorded that error as the share's source
error while the outer read was still in progress, leaving every
consumer in an error state.

Fail the nested read with ERR_INVALID_STATE before touching the
share's state. The source sees the error and may handle it; if it lets
it escape, it becomes the source error as with any other source
failure.

Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: #66483
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

lib/internal/streams/iter/share.js

@@ -60,6 +60,7 @@ const {
     ERR_INVALID_ARG_TYPE,
     ERR_INVALID_ARG_VALUE,
     ERR_INVALID_RETURN_VALUE,
+    ERR_INVALID_STATE,
     ERR_OUT_OF_RANGE,
   },
 } = require('internal/errors');
@@ -548,6 +549,7 @@ class SyncShareImpl {
   #sourceExhausted = false;
   #sourceError = kNoShareError;
   #cancelled = false;
+  #pulling = false;
   #cachedMinCursor = 0;
   #cachedMinCursorConsumers = 0;
   /** Cumulative byte size of buffered entries */
@@ -747,6 +749,15 @@ class SyncShareImpl {
   #pullFromSource() {
     if (this.#sourceExhausted || this.#cancelled) return;
 
+    // A source that reads from its own share would re-enter its own
+    // iterator. Fail that read without touching the share's state; the
+    // source sees the error and may handle it.
+    if (this.#pulling) {
+      throw new ERR_INVALID_STATE(
+        'shareSync() source cannot be read while it is producing a value');
+    }
+
+    this.#pulling = true;
     try {
       this.#sourceIterator ||= this.#source[SymbolIterator]();
 
@@ -760,6 +771,8 @@ class SyncShareImpl {
     } catch (error) {
       this.#sourceError = error;
       this.#sourceExhausted = true;
+    } finally {
+      this.#pulling = false;
     }
   }
 

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

@@ -264,6 +264,50 @@ function testShareSyncDropOldestSplitsOversizedBatches() {
     Array.from({ length: slowSeen.length }, (_, i) => 50 - slowSeen.length + i));
 }
 
+function testShareSyncReentrantSourceRead() {
+  const enc = new TextEncoder();
+  let sibling;
+  let reentrantError;
+  function* source() {
+    yield [enc.encode('a')];
+    try {
+      sibling.next();
+    } catch (err) {
+      reentrantError = err;
+    }
+    yield [enc.encode('b')];
+  }
+  const shared = shareSync(source());
+  const c1 = shared.pull()[Symbol.iterator]();
+  sibling = shared.pull()[Symbol.iterator]();
+
+  assert.deepStrictEqual(c1.next().value, [enc.encode('a')]);
+  assert.deepStrictEqual(sibling.next().value, [enc.encode('a')]);
+  // Pulling 'b' runs the source, which tries to read its own share.
+  assert.deepStrictEqual(c1.next().value, [enc.encode('b')]);
+  assert.strictEqual(reentrantError?.code, 'ERR_INVALID_STATE');
+  // The failed re-entrant read left the share intact.
+  assert.deepStrictEqual(sibling.next().value, [enc.encode('b')]);
+  assert.strictEqual(c1.next().done, true);
+  assert.strictEqual(sibling.next().done, true);
+}
+
+function testShareSyncReentrantSourceReadUncaught() {
+  let sibling;
+  function* source() {
+    yield [new Uint8Array(1)];
+    sibling.next();
+  }
+  const shared = shareSync(source());
+  const c1 = shared.pull()[Symbol.iterator]();
+  sibling = shared.pull()[Symbol.iterator]();
+  c1.next();
+  sibling.next();
+  // The error escapes the source, so it becomes the share's source error.
+  assert.throws(() => c1.next(), { code: 'ERR_INVALID_STATE' });
+  assert.throws(() => sibling.next(), { code: 'ERR_INVALID_STATE' });
+}
+
 function testShareSyncStringSource() {
   const shared = shareSync('hello-sync-share');
   const result = textSync(shared.pull());
@@ -281,6 +325,8 @@ Promise.all([
   testShareSyncRejectsUnbounded(),
   testShareSyncRejectsDropNewest(),
   testShareSyncDropOldestSplitsOversizedBatches(),
+  testShareSyncReentrantSourceRead(),
+  testShareSyncReentrantSourceReadUncaught(),
   testShareSyncStringSource(),
   testShareSyncRetainsBufferWhenAllConsumersDetach(),
   testShareSyncStrictBackpressureDetaches(),