@@ -435,7 +435,11 @@ the write. Use [`ondrain()`][] to wait for capacity rather than polling.
     the pending `end()` call; it does not fail the writer itself.
 * Returns: {Promise} Fulfills with the total number of bytes written.
 
-Signals that no more data will be written and waits for buffered data to drain.
+Signals that no more data will be written. Writes already waiting for buffer
+space remain ordered before the end of the stream, while later writes fail. If
+data is outstanding, the returned promise fulfills after the consumer pulls
+`done: true` beyond the final batch. If no data is buffered or pending, the
+writer closes immediately.
 
 #### `writer.endSync()`
 
 
@@ -299,7 +299,10 @@ class PushQueue {
       const onAbort = () => {
         // Remove from queue so it doesn't occupy a slot
         const idx = this.#pendingWrites.indexOf(entry);
-        if (idx !== -1) this.#pendingWrites.removeAt(idx);
+        if (idx !== -1) {
+          this.#pendingWrites.removeAt(idx);
+          this.#resolvePendingReads();
+        }
         reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
       };
 
@@ -334,30 +337,31 @@ class PushQueue {
       return this.#bytesWritten; // Idempotent
     }
 
-    this.#rejectPendingWrites(
-      new ERR_INVALID_STATE.TypeError('Writer closed'));
     this.#resolvePendingDrains(false);
 
-    // If buffer is empty, close immediately
-    if (this.#slots.length === 0) {
+    // If there is no accepted work to drain, close immediately.
+    if (this.#slots.length === 0 && this.#pendingWrites.length === 0) {
       this.#writerState = 'closed';
       this.#cleanup();
       this.#resolvePendingReads();
       return this.#bytesWritten;
     }
 
-    // Buffer has data: transition to closing, defer completion until drained
+    // Accepted work remains: close after it drains and the consumer pulls EOF.
     this.#writerState = 'closing';
     return -3; // Signal to PushWriter: create deferred end promise
   }
 
   /**
-   * Called by the read path when the consumer has drained all data while
-   * the writer is in the 'closing' state. Transitions to 'closed' and
-   * resolves the pending end promise.
+   * Called when the consumer pulls past all accepted data while the writer is
+   * closing. Transitions to 'closed' and resolves the pending end promise.
    */
   endDrained() {
-    if (this.#writerState !== 'closing') return;
+    if (this.#writerState !== 'closing' ||
+        this.#slots.length > 0 ||
+        this.#pendingWrites.length > 0) {
+      return;
+    }
     this.#writerState = 'closed';
     this.#cleanup();
     if (this.#pendingEnd) {
@@ -384,15 +388,14 @@ class PushQueue {
     this.#cleanup();
     this.#rejectPendingReads(this.#error);
     this.#rejectPendingDrains(this.#error);
+    this.#rejectPendingWrites(this.#error);
 
     if (wasClosing) {
       // Short-circuit the graceful drain: reject the pending end promise
       if (this.#pendingEnd) {
         this.#pendingEnd.reject(this.#error);
         this.#pendingEnd = null;
       }
-    } else {
-      this.#rejectPendingWrites(this.#error);
     }
   }
 
@@ -446,10 +449,6 @@ class PushQueue {
     if (this.#slots.length > 0) {
       const result = this.#drain();
       this.#resolvePendingWrites();
-      // After draining, check if writer was closing and buffer is now empty
-      if (this.#writerState === 'closing' && this.#slots.length === 0) {
-        this.endDrained();
-      }
       return { __proto__: null, done: false, value: result };
     }
 
@@ -555,7 +554,9 @@ class PushQueue {
         } catch (error) {
           pending.reject(error);
         }
-      } else if (this.#writerState === 'closing' && this.#slots.length === 0) {
+      } else if (this.#writerState === 'closing' &&
+                 this.#slots.length === 0 &&
+                 this.#pendingWrites.length === 0) {
         this.endDrained();
         const pending = this.#pendingReads.shift();
         pending.resolve({ __proto__: null, done: true, value: undefined });
 
@@ -84,9 +84,9 @@ async function testWriteWithSignalRejects() {
   await assert.rejects(writePromise, { name: 'AbortError' });
 
   // Clean up
-  writer.end();
-  // eslint-disable-next-line no-unused-vars
-  for await (const _ of readable) { break; }
+  const end = writer.end();
+  await text(readable);
+  await end;
 }
 
 async function testWriteWithPreAbortedSignal() {
@@ -100,9 +100,10 @@ async function testWriteWithPreAbortedSignal() {
 
   // Writer should still be usable for other writes
   writer.write('ok');
-  writer.end();
+  const end = writer.end();
   const data = await text(readable);
   assert.strictEqual(data, 'ok');
+  await end;
 }
 
 async function testCancelledWriteRemovedFromQueue() {
@@ -127,7 +128,7 @@ async function testCancelledWriteRemovedFromQueue() {
   // The cancelled write should NOT occupy a pending slot.
   // A new write should succeed now that the buffer has room.
   await writer.write(kChunk);
-  writer.end();
+  const end = writer.end();
 
   const result = await iter.next();
   assert.ok(!result.done);
@@ -136,7 +137,8 @@ async function testCancelledWriteRemovedFromQueue() {
     totalBytes += chunk.byteLength;
   }
   assert.strictEqual(totalBytes, 16384);
-  await iter.return();
+  assert.strictEqual((await iter.next()).done, true);
+  await end;
 }
 
 async function testOndrainResolvesFalseOnConsumerBreak() {
@@ -505,28 +507,48 @@ async function testConsumerThrowRejectsPendingRead() {
   await readRejects;
 }
 
-// end() while writes are pending rejects those writes
-async function testEndRejectsPendingWrites() {
+// end() drains writes that were already pending, then waits for EOF to be read.
+async function testEndDrainsPendingWrites() {
   const kChunk = new Uint8Array(16384);
   const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' });
   writer.writeSync(kChunk); // fill budget
 
   // This write blocks on backpressure
   const writePromise = writer.write(kChunk);
+  const endPromise = writer.end();
+  await assert.rejects(writer.write(kChunk), { code: 'ERR_INVALID_STATE' });
 
-  await new Promise(setImmediate);
+  let ended = false;
+  endPromise.then(common.mustCall(() => { ended = true; }));
+  const iterator = readable[Symbol.asyncIterator]();
 
-  // Ending should reject the pending write
-  writer.endSync();
+  assert.strictEqual((await iterator.next()).done, false);
+  await writePromise;
+  assert.strictEqual((await iterator.next()).done, false);
+  await Promise.resolve();
+  assert.strictEqual(ended, false);
 
-  await assert.rejects(
-    () => writePromise,
-    { code: 'ERR_INVALID_STATE' },
-  );
+  assert.strictEqual((await iterator.next()).done, true);
+  assert.strictEqual(await endPromise, kChunk.byteLength * 2);
+  assert.strictEqual(ended, true);
+}
 
-  // Clean up: drain the readable
-  // eslint-disable-next-line no-unused-vars
-  for await (const _ of readable) { break; }
+async function testEndWaitsForEofPull() {
+  const { writer, readable } = push();
+  writer.writeSync('hello');
+  const endPromise = writer.end();
+  let ended = false;
+  endPromise.then(common.mustCall(() => { ended = true; }));
+  const iterator = readable[Symbol.asyncIterator]();
+
+  const data = await iterator.next();
+  assert.strictEqual(data.done, false);
+  await Promise.resolve();
+  assert.strictEqual(ended, false);
+
+  assert.strictEqual((await iterator.next()).done, true);
+  await endPromise;
+  assert.strictEqual(ended, true);
 }
 
 async function testEndIdempotentWhenClosed() {
@@ -679,7 +701,8 @@ Promise.all([
   testConsumerReturnResolvesPendingRead(),
   testEndRejectsAfterConsumerReturn(),
   testConsumerThrowRejectsPendingRead(),
-  testEndRejectsPendingWrites(),
+  testEndDrainsPendingWrites(),
+  testEndWaitsForEofPull(),
   testEndIdempotentWhenClosed(),
   testEndRejectsWhenErrored(),
   testAsyncDispose(),