nodejs
diff --git a/‎lib/internal/streams/iter/share.js‎
Lines changed: 7 additions & 7 deletions b/‎lib/internal/streams/iter/share.js‎
Lines changed: 7 additions & 7 deletions
diff --git a/‎test/parallel/test-stream-iter-share-from.js‎
Lines changed: 7 additions & 10 deletions b/‎test/parallel/test-stream-iter-share-from.js‎
Lines changed: 7 additions & 10 deletions
@@ -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) {
 
@@ -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']);
 }
 
 // =============================================================================
Original file line number	Diff line number	Diff line change
@@ -174,15 +174,15 @@ class ShareImpl {
174	174	}
175	175
176	176	// Need to pull from source - check buffer limit
177		- const canPull = await self.#waitForBufferSpace();
178		- if (!canPull) {
	177	+ const shouldBuffer = await self.#waitForBufferSpace();
	178	+ if (shouldBuffer === null) {
179	179	state.detached = true;
180	180	self.#deleteConsumer(state);
181	181	if (self.#sourceError) throw self.#sourceError;
182	182	return { __proto__: null, done: true, value: undefined };
183	183	}
184	184
185		- await self.#pullFromSource();
	185	+ await self.#pullFromSource(!shouldBuffer);
186	186	}
187	187	};
188	188
@@ -263,7 +263,7 @@ class ShareImpl {
263	263	async #waitForBufferSpace() {
264	264	while (this.#buffer.length >= this.#options.highWaterMark) {
265	265	if (this.#cancelled \|\| this.#sourceError \|\| this.#sourceExhausted) {
266		- return !this.#cancelled;
	266	+ return this.#cancelled ? null : true;
267	267	}
268	268
269	269	switch (this.#options.backpressure) {
@@ -289,13 +289,13 @@ class ShareImpl {
289	289	this.#recomputeMinCursor();
290	290	return true;
291	291	case 'drop-newest':
292		- return true;
	292	+ return false;
293	293	}
294	294	}
295	295	return true;
296	296	}
297	297
298		- #pullFromSource() {
	298	+ #pullFromSource(discard = false) {
299	299	if (this.#sourceExhausted \|\| this.#cancelled) {
300	300	return PromiseResolve();
301	301	}
@@ -337,7 +337,7 @@ class ShareImpl {
337	337
338	338	if (result.done) {
339	339	this.#sourceExhausted = true;
340		- } else {
	340	+ } else if (!discard) {
341	341	this.#buffer.push(result.value);
342	342	}
343	343	} catch (error) {