stream: fix drop-newest behavior in share()

lib/internal/streams/iter/share.js

@@ -174,15 +174,15 @@ class ShareImpl {
             }
 
             // Need to pull from source - check buffer limit
-            const canPull = await self.#waitForBufferSpace();
-            if (!canPull) {
+            const shouldBuffer = await self.#waitForBufferSpace();
+            if (shouldBuffer === null) {
               state.detached = true;
               self.#deleteConsumer(state);
               if (self.#sourceError) throw self.#sourceError;
               return { __proto__: null, done: true, value: undefined };
             }
 
-            await self.#pullFromSource();
+            await self.#pullFromSource(!shouldBuffer);
           }
         };
 
@@ -263,7 +263,7 @@ class ShareImpl {
   async #waitForBufferSpace() {
     while (this.#buffer.length >= this.#options.highWaterMark) {
       if (this.#cancelled || this.#sourceError || this.#sourceExhausted) {
-        return !this.#cancelled;
+        return this.#cancelled ? null : true;
       }
 
       switch (this.#options.backpressure) {
@@ -289,13 +289,13 @@ class ShareImpl {
           this.#recomputeMinCursor();
           return true;
         case 'drop-newest':
-          return true;
+          return false;
       }
     }
     return true;
   }
 
-  #pullFromSource() {
+  #pullFromSource(discard = false) {
     if (this.#sourceExhausted || this.#cancelled) {
       return PromiseResolve();
     }
@@ -337,7 +337,7 @@ class ShareImpl {
 
         if (result.done) {
           this.#sourceExhausted = true;
-        } else {
+        } else if (!discard) {
           this.#buffer.push(result.value);
         }
       } catch (error) {

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

@@ -168,10 +168,8 @@ async function testShareDropOldest() {
 }
 
 async function testShareDropNewest() {
-  // With drop-newest and a stalled consumer, the async path allows the
-  // buffer to grow beyond highWaterMark (the "drop" applies to the
-  // backpressure signal, not the buffer contents). Both consumers
-  // ultimately see all items.
+  // With drop-newest and a stalled consumer, upstream results are discarded
+  // once the buffer reaches highWaterMark.
   async function* source() {
     for (let i = 0; i < 4; i++) {
       yield [new TextEncoder().encode(`${i}`)];
@@ -181,25 +179,24 @@ async function testShareDropNewest() {
   const fast = shared.pull();
   const slow = shared.pull();
 
-  // Fast consumer reads all items
+  // The fast consumer fills the buffer, then drives the source to completion.
   const fastItems = [];
   for await (const batch of fast) {
     for (const chunk of batch) {
       fastItems.push(new TextDecoder().decode(chunk));
     }
   }
-  assert.strictEqual(fastItems.length, 4);
+  assert.deepStrictEqual(fastItems, ['0', '1']);
+  assert.strictEqual(shared.bufferSize, 2);
 
-  // Slow consumer also sees all items (buffer grew past hwm)
+  // The stalled consumer sees the buffered items, but not the dropped results.
   const slowItems = [];
   for await (const batch of slow) {
     for (const chunk of batch) {
       slowItems.push(new TextDecoder().decode(chunk));
     }
   }
-  assert.strictEqual(slowItems.length, 4);
-  assert.strictEqual(slowItems[0], '0');
-  assert.strictEqual(slowItems[3], '3');
+  assert.deepStrictEqual(slowItems, ['0', '1']);
 }
 
 // =============================================================================