@@ -6,6 +6,7 @@
 // channel's writer appears in the other channel's readable.
 
 const {
+  SafePromiseAllReturnVoid,
   SymbolAsyncDispose,
   SymbolAsyncIterator,
 } = primordials;
@@ -50,68 +51,8 @@ function duplex(options = { __proto__: null }) {
     backpressure: b?.backpressure ?? backpressure,
   });
 
-  let aClosed = false;
-  let bClosed = false;
-  // Track active iterators so close() can call .return() on them
-  let aReadableIterator = null;
-  let bReadableIterator = null;
-
-  const channelA = {
-    __proto__: null,
-    get writer() { return aWriter; },
-    // Wrap readable to track the iterator for cleanup on close()
-    get readable() {
-      return {
-        __proto__: null,
-        [SymbolAsyncIterator]() {
-          const iter = aReadable[SymbolAsyncIterator]();
-          aReadableIterator = iter;
-          return iter;
-        },
-      };
-    },
-    async close() {
-      if (aClosed) return;
-      aClosed = true;
-      // End the writer (signals end-of-stream to B's readable)
-      aWriter.endSync();
-      // Stop iteration of this channel's readable
-      if (aReadableIterator?.return) {
-        await aReadableIterator.return();
-        aReadableIterator = null;
-      }
-    },
-    [SymbolAsyncDispose]() {
-      return this.close();
-    },
-  };
-
-  const channelB = {
-    __proto__: null,
-    get writer() { return bWriter; },
-    get readable() {
-      return {
-        __proto__: null,
-        [SymbolAsyncIterator]() {
-          const iter = bReadable[SymbolAsyncIterator]();
-          bReadableIterator = iter;
-          return iter;
-        },
-      };
-    },
-    async close() {
-      if (bClosed) return;
-      bClosed = true;
-      bWriter.endSync();
-      if (bReadableIterator?.return) {
-        await bReadableIterator.return();
-        bReadableIterator = null;
-      }
-    },
-    [SymbolAsyncDispose]() {
-      return this.close();
-    },
-  };
+  const channelA = createDuplexChannel(aWriter, aReadable);
+  const channelB = createDuplexChannel(bWriter, bReadable);
 
   // Signal handler: fail both writers with the abort reason so consumers
   // see the error. This is an error-path shutdown, not a clean close.
@@ -132,6 +73,38 @@ function duplex(options = { __proto__: null }) {
   return [channelA, channelB];
 }
 
+function createDuplexChannel(writer, readable) {
+  // A push readable has one shared consumer state. Keeping an iterator from
+  // creation lets close() terminate that state even if no caller has iterated.
+  const closeIterator = readable[SymbolAsyncIterator]();
+  let closePromise;
+
+  return {
+    __proto__: null,
+    get writer() { return writer; },
+    get readable() { return readable; },
+    close() {
+      closePromise ??= closeDuplexChannel(writer, closeIterator);
+      return closePromise;
+    },
+    [SymbolAsyncDispose]() {
+      return this.close();
+    },
+  };
+}
+
+async function closeDuplexChannel(writer, closeIterator) {
+  const result = writer.endSync();
+  const endPromise = result < 0 ? writer.end() : undefined;
+  const returnPromise = closeIterator.return();
+
+  if (endPromise !== undefined) {
+    await SafePromiseAllReturnVoid([endPromise, returnPromise]);
+  } else {
+    await returnPromise;
+  }
+}
+
 module.exports = {
   duplex,
 };
@@ -14,32 +14,26 @@ async function testBasicDuplex() {
 
   // A writes, B reads
   await channelA.writer.write('hello from A');
-  await channelA.close();
-
+  const closing = channelA.close();
   const dataAtB = await text(channelB.readable);
+  await closing;
   assert.strictEqual(dataAtB, 'hello from A');
 }
 
 async function testBidirectional() {
   const [channelA, channelB] = duplex();
 
-  // A writes to B, B writes to A concurrently
-  const writeA = (async () => {
-    await channelA.writer.write('A to B');
-    await channelA.close();
-  })();
-
-  const writeB = (async () => {
-    await channelB.writer.write('B to A');
-    await channelB.close();
-  })();
-
-  const readAtB = text(channelB.readable);
-  const readAtA = text(channelA.readable);
+  await channelA.writer.write('A to B');
+  await channelB.writer.write('B to A');
 
-  await Promise.all([writeA, writeB]);
-
-  const [dataAtA, dataAtB] = await Promise.all([readAtA, readAtB]);
+  const endA = channelA.writer.end();
+  const endB = channelB.writer.end();
+  const [dataAtA, dataAtB] = await Promise.all([
+    text(channelA.readable),
+    text(channelB.readable),
+  ]);
+  await Promise.all([endA, endB]);
+  await Promise.all([channelA.close(), channelB.close()]);
 
   assert.strictEqual(dataAtB, 'A to B');
   assert.strictEqual(dataAtA, 'B to A');
@@ -51,19 +45,29 @@ async function testMultipleWrites() {
   await channelA.writer.write('one');
   await channelA.writer.write('two');
   await channelA.writer.write('three');
-  await channelA.close();
-
+  const closing = channelA.close();
   const data = await text(channelB.readable);
+  await closing;
   assert.strictEqual(data, 'onetwothree');
 }
 
 async function testChannelClose() {
   const [channelA, channelB] = duplex();
-
-  await channelA.close();
-
-  // Should be able to close twice without error
-  await channelA.close();
+  const iteratorA = channelA.readable[Symbol.asyncIterator]();
+  const otherIteratorA = channelA.readable[Symbol.asyncIterator]();
+  const pendingRead = iteratorA.next();
+
+  const closing = channelA.close();
+  assert.strictEqual(channelA.close(), closing);
+  await closing;
+
+  assert.strictEqual((await pendingRead).done, true);
+  assert.strictEqual((await otherIteratorA.next()).done, true);
+  assert.strictEqual(
+    (await channelA.readable[Symbol.asyncIterator]().next()).done, true);
+  await assert.rejects(channelB.writer.write('late'), {
+    code: 'ERR_INVALID_STATE',
+  });
 
   // B's readable should end (A -> B direction is closed)
   const batches = [];
@@ -80,9 +84,9 @@ async function testWithOptions() {
   });
 
   await channelA.writer.write('msg');
-  await channelA.close();
-
+  const closing = channelA.close();
   const data = await text(channelB.readable);
+  await closing;
   assert.strictEqual(data, 'msg');
 }
 
@@ -95,15 +99,17 @@ async function testPerChannelOptions() {
   // Channel A -> B direction uses A's options
   // Channel B -> A direction uses B's options
   await channelA.writer.write('from-a');
-  await channelA.close();
-
   await channelB.writer.write('from-b');
-  await channelB.close();
+
+  const endA = channelA.writer.end();
+  const endB = channelB.writer.end();
 
   const [dataAtA, dataAtB] = await Promise.all([
     text(channelA.readable),
     text(channelB.readable),
   ]);
+  await Promise.all([endA, endB]);
+  await Promise.all([channelA.close(), channelB.close()]);
 
   assert.strictEqual(dataAtB, 'from-a');
   assert.strictEqual(dataAtA, 'from-b');
@@ -146,17 +152,39 @@ async function testWriterEndWithPreAbortedSignal() {
 async function testEmptyDuplex() {
   const [channelA, channelB] = duplex();
 
-  // Close without writing
-  await channelA.close();
-  await channelB.close();
+  await channelA.writer.end();
+  await channelB.writer.end();
 
   const dataAtA = await bytes(channelA.readable);
   const dataAtB = await bytes(channelB.readable);
+  await Promise.all([channelA.close(), channelB.close()]);
 
   assert.strictEqual(dataAtA.byteLength, 0);
   assert.strictEqual(dataAtB.byteLength, 0);
 }
 
+async function testCloseWaitsForDrain() {
+  const [channelA, channelB] = duplex();
+  await channelA.writer.write('buffered');
+
+  let closed = false;
+  const closing = channelA.close().then(common.mustCall(() => {
+    closed = true;
+  }));
+  await new Promise(setImmediate);
+  assert.strictEqual(closed, false);
+
+  assert.strictEqual(await text(channelB.readable), 'buffered');
+  await closing;
+}
+
+async function testClosePropagatesWriterFailure() {
+  const [channelA] = duplex();
+  const reason = new Error('writer failed');
+  channelA.writer.fail(reason);
+  await assert.rejects(channelA.close(), (error) => error === reason);
+}
+
 // Channel fail propagation
 async function testChannelFail() {
   const [a, b] = duplex();
@@ -200,6 +228,8 @@ Promise.all([
   testAbortSignal(),
   testWriterEndWithPreAbortedSignal(),
   testEmptyDuplex(),
+  testCloseWaitsForDrain(),
+  testClosePropagatesWriterFailure(),
   testChannelFail(),
   testAbortSignalBothChannels(),
 ]).then(common.mustCall());
@@ -99,16 +99,16 @@ assert.throws(() => duplex({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' });
   const [a, b] = duplex({ budget: Number.MAX_SAFE_INTEGER });
   assert.strictEqual(a.writer.canWrite, true);
   assert.strictEqual(b.writer.canWrite, true);
-  a.close();
-  b.close();
+  a.writer.endSync();
+  b.writer.endSync();
 }
 // Per-direction overrides
 {
   const [a, b] = duplex({ a: { budget: 16384 }, b: { budget: 32768 } });
   assert.strictEqual(a.writer.canWrite, true);
   assert.strictEqual(b.writer.canWrite, true);
-  a.close();
-  b.close();
+  a.writer.endSync();
+  b.writer.endSync();
 }
 
 assert.throws(() => duplex({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });
@@ -397,8 +397,8 @@ async function testAsyncValidation() {
 // Duplex with valid options
 {
   const [a, b] = duplex({ budget: 16384 });
-  a.close();
-  b.close();
+  a.writer.endSync();
+  b.writer.endSync();
 }
 
 // Broadcast with valid options
Original file line number	Diff line number	Diff line change
`@@ -99,16 +99,16 @@ assert.throws(() => duplex({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' });`
`99`	`99`	`const [a, b] = duplex({ budget: Number.MAX_SAFE_INTEGER });`
`100`	`100`	`assert.strictEqual(a.writer.canWrite, true);`
`101`	`101`	`assert.strictEqual(b.writer.canWrite, true);`
`102`		`- a.close();`
`103`		`- b.close();`
	`102`	`+ a.writer.endSync();`
	`103`	`+ b.writer.endSync();`
`104`	`104`	`}`
`105`	`105`	`// Per-direction overrides`
`106`	`106`	`{`
`107`	`107`	`const [a, b] = duplex({ a: { budget: 16384 }, b: { budget: 32768 } });`
`108`	`108`	`assert.strictEqual(a.writer.canWrite, true);`
`109`	`109`	`assert.strictEqual(b.writer.canWrite, true);`
`110`		`- a.close();`
`111`		`- b.close();`
	`110`	`+ a.writer.endSync();`
	`111`	`+ b.writer.endSync();`
`112`	`112`	`}`
`113`	`113`
`114`	`114`	`assert.throws(() => duplex({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });`
`@@ -397,8 +397,8 @@ async function testAsyncValidation() {`
`397`	`397`	`// Duplex with valid options`
`398`	`398`	`{`
`399`	`399`	`const [a, b] = duplex({ budget: 16384 });`
`400`		`- a.close();`
`401`		`- b.close();`
	`400`	`+ a.writer.endSync();`
	`401`	`+ b.writer.endSync();`
`402`	`402`	`}`
`403`	`403`
`404`	`404`	`// Broadcast with valid options`