nodejs
diff --git a/‎lib/internal/streams/iter/duplex.js‎
Lines changed: 2 additions & 6 deletions b/‎lib/internal/streams/iter/duplex.js‎
Lines changed: 2 additions & 6 deletions
diff --git a/‎lib/internal/streams/iter/push.js‎
Lines changed: 8 additions & 1 deletion b/‎lib/internal/streams/iter/push.js‎
Lines changed: 8 additions & 1 deletion
diff --git a/‎test/parallel/test-stream-iter-push-writer.js‎
Lines changed: 20 additions & 0 deletions b/‎test/parallel/test-stream-iter-push-writer.js‎
Lines changed: 20 additions & 0 deletions
@@ -74,9 +74,7 @@ function duplex(options = { __proto__: null }) {
       if (aClosed) return;
       aClosed = true;
       // End the writer (signals end-of-stream to B's readable)
-      if (aWriter.endSync() < 0) {
-        await aWriter.end();
-      }
+      aWriter.endSync();
       // Stop iteration of this channel's readable
       if (aReadableIterator?.return) {
         await aReadableIterator.return();
@@ -104,9 +102,7 @@ function duplex(options = { __proto__: null }) {
     async close() {
       if (bClosed) return;
       bClosed = true;
-      if (bWriter.endSync() < 0) {
-        await bWriter.end();
-      }
+      bWriter.endSync();
       if (bReadableIterator?.return) {
         await bReadableIterator.return();
         bReadableIterator = null;
 
@@ -274,7 +274,10 @@ class PushQueue {
     if (this.#writerState === 'errored') {
       return -2; // Signal to reject with stored error
     }
-    if (this.#writerState === 'closing' || this.#writerState === 'closed') {
+    if (this.#writerState === 'closing') {
+      return -3; // Signal to PushWriter: wait for drain to complete
+    }
+    if (this.#writerState === 'closed') {
       return this.#bytesWritten; // Idempotent
     }
 
@@ -636,6 +639,10 @@ class PushWriter {
     if (result === -3) {
       // Closing: buffer has data, create deferred promise that resolves
       // when consumer drains past the end sentinel
+      const pendingEndPromise = this.#queue.pendingEndPromise;
+      if (pendingEndPromise !== null) {
+        return pendingEndPromise;
+      }
       const { promise, resolve, reject } = PromiseWithResolvers();
       this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
       return promise;
 
@@ -232,6 +232,25 @@ async function testEndAsyncReturnValue() {
   await consume;
 }
 
+async function testEndAfterEndSyncWaitsForDrain() {
+  const { writer, readable } = push();
+  writer.writeSync('hello');
+  assert.strictEqual(writer.endSync(), -1);
+
+  let ended = false;
+  const end = writer.end().then((n) => {
+    ended = true;
+    return n;
+  });
+
+  await Promise.resolve();
+  assert.strictEqual(ended, false);
+
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of readable) { /* drain */ }
+  assert.strictEqual(await end, 5);
+}
+
 async function testWriteUint8Array() {
   const { writer, readable } = push();
   writer.write(new Uint8Array([72, 73])); // 'HI'
@@ -413,6 +432,7 @@ Promise.all([
   testOndrainProtocolErrorPropagates(),
   testFail(),
   testEndAsyncReturnValue(),
+  testEndAfterEndSyncWaitsForDrain(),
   testWriteUint8Array(),
   testOndrainWaitsForDrain(),
   testConsumerThrowRejectsWrites(),