stream: ensure async dispoal after endSync awaits for drain

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65658
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

doc/api/stream_iter.md

@@ -458,6 +458,12 @@ or errored, this is a no-op. Unlike `write()` and `end()`, `fail()` is
 unconditionally synchronous because failing a writer is a pure state
 transition with no async work to perform.
 
+#### `writer[Symbol.asyncDispose]()`
+
+If the writer is open, calls `writer.fail()`. If the writer is closing after
+`end()` or `endSync()`, waits for buffered data to drain. If the writer is
+already closed or errored, resolves immediately.
+
 #### `writer.write(chunk[, options])`
 
 * `chunk` {Uint8Array|string}

lib/internal/streams/iter/push.js

@@ -409,12 +409,12 @@ class PushQueue {
     return this.#writerState;
   }
 
-  get pendingEndPromise() {
-    return this.#pendingEnd?.promise ?? null;
-  }
-
-  setPendingEnd(pending) {
-    this.#pendingEnd = pending;
+  getPendingEndPromise() {
+    if (this.#pendingEnd === null) {
+      const { promise, resolve, reject } = PromiseWithResolvers();
+      this.#pendingEnd = { __proto__: null, promise, resolve, reject };
+    }
+    return this.#pendingEnd.promise;
   }
 
   /**
@@ -681,15 +681,7 @@ class PushWriter {
       return PromiseReject(this.#queue.error);
     }
     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 raceEndWithSignal(pendingEndPromise, signal);
-      }
-      const { promise, resolve, reject } = PromiseWithResolvers();
-      this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
-      return raceEndWithSignal(promise, signal);
+      return raceEndWithSignal(this.#queue.getPendingEndPromise(), signal);
     }
     // >= 0: byte count (immediate close or idempotent)
     return PromiseResolve(result);
@@ -709,8 +701,7 @@ class PushWriter {
   [SymbolAsyncDispose]() {
     const state = this.#queue.writerState;
     if (state === 'closing') {
-      // Wait for graceful drain
-      return this.#queue.pendingEndPromise ?? PromiseResolve();
+      return this.#queue.getPendingEndPromise();
     }
     if (state === 'open') {
       this.fail();

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

@@ -531,6 +531,23 @@ async function testAsyncDispose() {
   }
 }
 
+async function testAsyncDisposeWaitsAfterEndSync() {
+  const { writer, readable } = push({ budget: 16384 });
+  writer.writeSync('hello');
+  assert.strictEqual(writer.endSync(), -1);
+
+  let disposed = false;
+  const disposal = writer[Symbol.asyncDispose]().then(() => {
+    disposed = true;
+  });
+  await Promise.resolve();
+  assert.strictEqual(disposed, false);
+
+  assert.strictEqual(await text(readable), 'hello');
+  await disposal;
+  assert.strictEqual(disposed, true);
+}
+
 async function testSyncDispose() {
   const { writer, readable } = push({ budget: 16384 });
   writer.writeSync('hello');
@@ -635,5 +652,6 @@ Promise.all([
   testEndIdempotentWhenClosed(),
   testEndRejectsWhenErrored(),
   testAsyncDispose(),
+  testAsyncDisposeWaitsAfterEndSync(),
   testSyncDispose(),
 ]).then(common.mustCall());