From baae906a12b5fc50def40766c49eeb008f3a1202 Mon Sep 17 00:00:00 2001
From: "Kamat, Trivikram" <16024985+trivikr@users.noreply.github.com>
Date: Fri, 10 Jul 2026 19:16:54 -0700
Subject: [PATCH] stream: fix drop-newest behavior in share()

Discard upstream pull results when the shared buffer has reached its
highWaterMark instead of allowing the buffer to grow beyond its limit.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
---
 lib/internal/streams/iter/share.js           | 14 +++++++-------
 test/parallel/test-stream-iter-share-from.js | 17 +++++++----------
 2 files changed, 14 insertions(+), 17 deletions(-)

diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js
index e65a5eb648b620..199ddf978e5566 100644
--- a/lib/internal/streams/iter/share.js
+++ b/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) {
diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js
index 362bfef78f1a2e..f7a9336bbc2691 100644
--- a/test/parallel/test-stream-iter-share-from.js
+++ b/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']);
 }
 
 // =============================================================================