stream/iter: Share "drop-newest" can eagerly drain the upstream source

[trivikr]

Version

main

Platform

N/A

Subsystem

stream

What steps will reproduce the bug?

import { share } from 'node:stream/iter';

let batchesYielded = 0;

async function* source() {
  for (let i = 0; i < 7; i++) {
    batchesYielded++;
    console.log('source yielded', i);
    yield [new Uint8Array(16_384).fill(i)];
  }
}

const shared = share(source(), {
  budget: 16_384,
  backpressure: 'drop-newest',
});

const fast = shared.pull()[Symbol.asyncIterator]();
const stalled = shared.pull();

let result = await fast.next();
console.log('after first next', {
  done: result.done,
  firstByte: result.value?.[0][0],
  batchesYielded,
});

const second = fast.next();

await new Promise(setImmediate);
console.log('batches after one event-loop turn:', batchesYielded);

shared.cancel();
console.log('second next:', await second);

void stalled;

How often does it reproduce? Is there a required condition?

Always

What is the expected behavior? Why is that the expected behavior?

source yielded 0
after first next { done: false, firstByte: 0, batchesYielded: 1 }
source yielded 1
batches after one event-loop turn: 2
second next: [Object: null prototype] { done: true, value: undefined }

The second fast.next() should discard only one upstream result and then wait for buffer space. shared.cancel() subsequently settles that pending call so the repro exits cleanly.

From §13.2.2: Share buffering and backpressure:

“With "drop-newest", the upstream pull result is discarded.”

What do you see instead?

source yielded 0
after first next { done: false, firstByte: 0, batchesYielded: 1 }
source yielded 1
source yielded 2
source yielded 3
source yielded 4
source yielded 5
source yielded 6
batches after one event-loop turn: 7
second next: [Object: null prototype] { done: true, value: undefined }

One fast.next() pulls and discards all six remaining batches while the buffer is full.

Additional information

No response