streams: more iter streams fixes

doc/api/stream_iter.md

@@ -1497,11 +1497,19 @@ added:
 * `options` {Object}
   * `budget` {number} Must be >= 16384.
     **Default:** `65536`.
-  * `backpressure` {string} **Default:** `'strict'`.
+  * `backpressure` {string} `'strict'`, `'drop-oldest'`, or `'drop-newest'`.
+    **Default:** `'strict'`.
 * Returns: {SyncShare}
 
 Synchronous version of [`share()`][].
 
+Because there is no way to wait in a synchronous context, `'unbounded'` is not
+supported and throws `ERR_INVALID_ARG_VALUE`. With `'drop-newest'`, a consumer
+that reaches the end of the buffer while the budget is exhausted discards a
+single entry from the source and then returns `{ done: true }` without a
+value; the consumer is not detached, so it can resume once the slowest
+consumer advances and releases budget.
+
 ### Class: `SyncShare`
 
 #### Static method: `SyncShare.fromSync(input[, options])`

lib/internal/quic/quic.js

@@ -2640,15 +2640,10 @@ class QuicStream {
     error = inner.destroyError ?? error;
     if (error !== undefined) {
       inner.pendingClose.reject(error);
+      inner.pendingStream.reject(error);
     } else {
       inner.pendingClose.resolve();
-    }
-    if (inner.state.pending) {
-      if (error !== undefined) {
-        inner.pendingStream.reject(error);
-      } else {
-        inner.pendingStream.resolve(error);
-      }
+      inner.pendingStream.resolve();
     }
     debug('stream closed');
     if (onStreamClosedChannel.hasSubscribers) {

lib/internal/streams/iter/broadcast.js

@@ -24,6 +24,10 @@ const {
 } = primordials;
 
 const { lazyDOMException } = require('internal/util');
+const {
+  AbortController,
+  abortSignal,
+} = require('internal/abort_controller');
 
 const {
   codes: {
@@ -63,6 +67,7 @@ const {
   parsePullArgs,
   toWriterUint8Array,
   validateBatchEntry,
+  yieldAbortable,
 } = require('internal/streams/iter/utils');
 const {
   converters,
@@ -79,6 +84,7 @@ const kAbort = Symbol('kAbort');
 const kCanWrite = Symbol('kCanWrite');
 const kOnBufferDrained = Symbol('kOnBufferDrained');
 const kOnEndDrained = Symbol('kOnEndDrained');
+const kOnCancel = Symbol('kOnCancel');
 const kPendingWriteRemoved = Symbol('kPendingWriteRemoved');
 const kNoBroadcastError = Symbol('kNoBroadcastError');
 
@@ -120,6 +126,7 @@ class BroadcastImpl {
     this.#options = options;
     this[kOnBufferDrained] = null;
     this[kOnEndDrained] = null;
+    this[kOnCancel] = null;
   }
 
   setWriter(writer) {
@@ -307,6 +314,9 @@ class BroadcastImpl {
     }
     this.#consumers.clear();
     this.#cachedMinCursorConsumers = 0;
+    const onCancel = this[kOnCancel];
+    this[kOnCancel] = null;
+    onCancel?.(reason);
   }
 
   [SymbolDispose]() {
@@ -444,6 +454,8 @@ class BroadcastImpl {
   }
 
   #tryTrimBuffer() {
+    // Retain buffered data for consumers that attach while none are active.
+    if (this.#consumers.size === 0) return;
     if (this.#cachedMinCursorConsumers === 0) {
       this.#recomputeMinCursor();
     }
@@ -913,13 +925,23 @@ const Broadcast = {
     });
     const result = broadcast(options);
     const { signal } = options;
+    const controller = new AbortController();
+    if (signal?.aborted) {
+      abortSignal(controller.signal, signal.reason);
+    }
+    const onCancel = (reason) => {
+      if (!controller.signal.aborted) {
+        abortSignal(controller.signal, reason);
+      }
+    };
+    result.broadcast[kOnCancel] = onCancel;
 
     const pump = async () => {
       const w = result.writer;
       try {
         if (isAsyncIterable(source)) {
-          for await (const chunks of source) {
-            signal?.throwIfAborted();
+          for await (const chunks of yieldAbortable(source, controller.signal)) {
+            controller.signal.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
                 await w.writev(chunks, signal ? { signal } : undefined);
@@ -930,7 +952,7 @@ const Broadcast = {
           }
         } else if (isSyncIterable(source)) {
           for (const chunks of source) {
-            signal?.throwIfAborted();
+            controller.signal.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
                 await w.writev(chunks, signal ? { signal } : undefined);
@@ -944,7 +966,13 @@ const Broadcast = {
           await w.end(signal ? { signal } : undefined);
         }
       } catch (error) {
-        w.fail(error);
+        if (!controller.signal.aborted) {
+          w.fail(error);
+        }
+      } finally {
+        if (result.broadcast[kOnCancel] === onCancel) {
+          result.broadcast[kOnCancel] = null;
+        }
       }
     };
     PromisePrototypeThen(pump(), undefined, () => {});

lib/internal/streams/iter/share.js

@@ -56,6 +56,7 @@ const {
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_ARG_VALUE,
     ERR_INVALID_RETURN_VALUE,
     ERR_OUT_OF_RANGE,
   },
@@ -200,7 +201,16 @@ class ShareImpl {
             }
 
             // Need to pull from source - check buffer limit
-            const shouldBuffer = await self.#waitForBufferSpace();
+            let shouldBuffer;
+            try {
+              shouldBuffer = await self.#waitForBufferSpace();
+            } catch (error) {
+              state.detached = true;
+              if (self.#deleteConsumer(state)) {
+                self.#tryTrimBuffer();
+              }
+              throw error;
+            }
             if (shouldBuffer === null) {
               state.detached = true;
               state.error = self.#cancelError;
@@ -573,17 +583,13 @@ class SyncShareImpl {
             }
 
             // Check buffer limit
+            let dropped = false;
             if (self.#bufferedBytes >= self.#options.budget) {
               switch (self.#options.backpressure) {
                 case 'strict':
                   throw new ERR_OUT_OF_RANGE(
                     'buffered bytes', `< ${self.#options.budget}`,
                     self.#bufferedBytes);
-                case 'unbounded':
-                  throw new ERR_OUT_OF_RANGE(
-                    'buffered bytes', `< ${self.#options.budget} ` +
-                    '(unbounded not available in sync context)',
-                    self.#bufferedBytes);
                 case 'drop-oldest':
                   while (self.#bufferedBytes >= self.#options.budget &&
                          self.#buffer.length > 0) {
@@ -600,13 +606,20 @@ class SyncShareImpl {
                   self.#recomputeMinCursor();
                   break;
                 case 'drop-newest':
-                  state.detached = true;
-                  self.#deleteConsumer(state);
-                  return { __proto__: null, done: true, value: undefined };
+                  // Discarding does not reclaim budget, and the slowest
+                  // consumer cannot advance while this synchronous next() is
+                  // running, so at most one entry may be dropped per call.
+                  // Looping here would spin forever on an unbounded source
+                  // and drain a finite one in a single call.
+                  self.#pullFromSource(true);
+                  dropped = true;
+                  break;
               }
             }
 
-            self.#pullFromSource();
+            if (!dropped) {
+              self.#pullFromSource();
+            }
 
             if (self.#sourceError !== kNoShareError) {
               state.detached = true;
@@ -685,7 +698,7 @@ class SyncShareImpl {
     this.cancel();
   }
 
-  #pullFromSource() {
+  #pullFromSource(discard = false) {
     if (this.#sourceExhausted || this.#cancelled) return;
 
     try {
@@ -695,7 +708,7 @@ class SyncShareImpl {
 
       if (result.done) {
         this.#sourceExhausted = true;
-      } else {
+      } else if (!discard) {
         const entry = createBatchEntry(result.value);
         this.#buffer.push(entry);
         this.#bufferedBytes += entry.byteLength;
@@ -805,6 +818,11 @@ function shareSync(source, options = { __proto__: null }) {
     backpressure = 'strict',
   } = options;
   validateInteger(budget, 'options.budget', 16384);
+  if (backpressure === 'unbounded') {
+    throw new ERR_INVALID_ARG_VALUE(
+      'options.backpressure', backpressure,
+      'unbounded is not supported by shareSync()');
+  }
 
   const opts = {
     __proto__: null,

test/parallel/test-stream-iter-broadcast-basic.js

@@ -350,6 +350,18 @@ async function testLateJoinerSeesBufferedData() {
   assert.strictEqual(result, 'before-join');
 }
 
+async function testLateJoinerAfterDetachSeesBufferedData() {
+  const { writer, broadcast: bc } = broadcast({ budget: 16384 });
+  const first = bc.push()[Symbol.asyncIterator]();
+
+  writer.writeSync('before-detach');
+  await first.return();
+
+  const second = bc.push();
+  writer.endSync();
+  assert.strictEqual(await text(second), 'before-detach');
+}
+
 async function testOverlappingNextKeepsEarlierRead() {
   const { writer, broadcast: bc } = broadcast();
   const it = bc.push()[Symbol.asyncIterator]();
@@ -403,5 +415,6 @@ Promise.all([
   testFailDetachesConsumers(),
   testWriterFailIdempotent(),
   testLateJoinerSeesBufferedData(),
+  testLateJoinerAfterDetachSeesBufferedData(),
   testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

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

@@ -4,6 +4,7 @@
 const common = require('../common');
 const assert = require('assert');
 const { broadcast, Broadcast, from, text } = require('stream/iter');
+const { setImmediate } = require('timers/promises');
 
 // =============================================================================
 // Broadcast.from
@@ -117,34 +118,46 @@ async function testAlreadyAbortedSignal() {
 // =============================================================================
 
 async function testBroadcastFromCancelWhileBlocked() {
-  // Create a slow async source that blocks between yields
-  let sourceFinished = false;
-  async function* slowSource() {
-    const enc = new TextEncoder();
-    yield [enc.encode('chunk1')];
-    // Simulate a long delay without keeping the cancelled source alive.
-    await new Promise((resolve) => setTimeout(resolve, 10000).unref());
-    yield [enc.encode('chunk2')];
-    sourceFinished = true;
-  }
-
-  const { broadcast: bc } = Broadcast.from(slowSource());
-  const consumer = bc.push();
+  let resolveNext;
+  let sourceReturned = false;
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        next() {
+          const { promise, resolve } = Promise.withResolvers();
+          resolveNext = resolve;
+          return promise;
+        },
+        return() {
+          sourceReturned = true;
+          return Promise.resolve({ __proto__: null, done: true });
+        },
+      };
+    },
+  };
 
-  // Read the first chunk
-  const iter = consumer[Symbol.asyncIterator]();
-  const first = await iter.next();
-  assert.strictEqual(first.done, false);
+  const { writer, broadcast: bc } = Broadcast.from(source);
+  const iter = bc.push()[Symbol.asyncIterator]();
+  const pendingRead = iter.next();
+  await setImmediate();
 
-  // Cancel while the source is blocked waiting to yield the next chunk
+  let writesAfterCancel = 0;
+  writer.writevSync = () => { writesAfterCancel++; return true; };
   bc.cancel();
-
-  // The iteration should complete (not hang)
-  const next = await iter.next();
-  assert.strictEqual(next.done, true);
-
-  // Source should NOT have finished (we cancelled before chunk2)
-  assert.strictEqual(sourceFinished, false);
+  assert.deepStrictEqual(await pendingRead, {
+    __proto__: null,
+    done: true,
+    value: undefined,
+  });
+
+  resolveNext({
+    __proto__: null,
+    done: false,
+    value: [new TextEncoder().encode('late')],
+  });
+  await setImmediate();
+  assert.strictEqual(writesAfterCancel, 0);
+  assert.strictEqual(sourceReturned, true);
 }
 
 // =============================================================================

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

@@ -221,23 +221,29 @@ async function testShareDropNewest() {
 // =============================================================================
 
 async function testShareStrictBackpressure() {
-  async function* source() {
-    for (let i = 0; i < 10; i++) {
-      yield [new Uint8Array(16384)];
+  for (const transformed of [false, true]) {
+    async function* source() {
+      for (let i = 0; i < 10; i++) {
+        yield [new Uint8Array(16384)];
+      }
     }
+    const shared = share(source(), {
+      budget: 32768,
+      backpressure: 'strict',
+    });
+    const consumer = transformed ?
+      shared.pull((chunks) => chunks) : shared.pull();
+    const fast = consumer[Symbol.asyncIterator]();
+    // This consumer prevents the buffer from being trimmed.
+    shared.pull();
+
+    await fast.next();
+    await fast.next();
+    await assert.rejects(fast.next(), { code: 'ERR_OUT_OF_RANGE' });
+    assert.strictEqual(shared.consumerCount, 1);
+    assert.strictEqual((await fast.next()).done, true);
+    shared.cancel();
   }
-  const shared = share(source(), { budget: 32768, backpressure: 'strict' });
-  const fast = shared.pull();
-  // Create a second consumer that never reads — this prevents buffer trimming
-  shared.pull();
-
-  // The fast consumer's pulls will eventually cause the buffer to exceed
-  // the budget (since the slow consumer prevents trimming),
-  // triggering an ERR_OUT_OF_RANGE error.
-  await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of fast) { /* consume */ }
-  }, { code: 'ERR_OUT_OF_RANGE' });
 }
 
 Promise.all([