stream: wait for push writer end fallback to drain

When endSync() returns -1 after buffered writes, a follow-up end()
should stay pending until the readable side drains the queued data.

Do not make duplex channel close() wait for that drain, since close()
only needs to signal EOF to the peer. Waiting there can deadlock when
the peer starts reading only after close() resolves.

Fixes: #63502

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5

Author: trivikr

lib/internal/streams/iter/duplex.js

@@ -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;

lib/internal/streams/iter/push.js

@@ -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;

test/parallel/test-stream-iter-push-writer.js

@@ -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(),