Skip to content

Commit 6d8e643

Browse files
jasnelladuh95
authored andcommitted
stream: ensure async dispoal after endSync awaits for drain
Signed-off-by: James M Snell <[email protected]> Assisted-by: Opencode PR-URL: #65658 Reviewed-By: Trivikram Kamat <[email protected]>
1 parent a082916 commit 6d8e643

3 files changed

Lines changed: 32 additions & 17 deletions

File tree

‎doc/api/stream_iter.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -458,6 +458,12 @@ or errored, this is a no-op. Unlike `write()` and `end()`, `fail()` is
458458
unconditionally synchronous because failing a writer is a pure state
459459
transition with no async work to perform.
460460

461+
#### `writer[Symbol.asyncDispose]()`
462+
463+
If the writer is open, calls `writer.fail()`. If the writer is closing after
464+
`end()` or `endSync()`, waits for buffered data to drain. If the writer is
465+
already closed or errored, resolves immediately.
466+
461467
#### `writer.write(chunk[, options])`
462468

463469
* `chunk` {Uint8Array|string}

‎lib/internal/streams/iter/push.js‎

Lines changed: 8 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -409,12 +409,12 @@ class PushQueue {
409409
return this.#writerState;
410410
}
411411

412-
get pendingEndPromise() {
413-
return this.#pendingEnd?.promise ?? null;
414-
}
415-
416-
setPendingEnd(pending) {
417-
this.#pendingEnd = pending;
412+
getPendingEndPromise() {
413+
if (this.#pendingEnd === null) {
414+
const { promise, resolve, reject } = PromiseWithResolvers();
415+
this.#pendingEnd = { __proto__: null, promise, resolve, reject };
416+
}
417+
return this.#pendingEnd.promise;
418418
}
419419

420420
/**
@@ -681,15 +681,7 @@ class PushWriter {
681681
return PromiseReject(this.#queue.error);
682682
}
683683
if (result === -3) {
684-
// Closing: buffer has data, create deferred promise that resolves
685-
// when consumer drains past the end sentinel
686-
const pendingEndPromise = this.#queue.pendingEndPromise;
687-
if (pendingEndPromise !== null) {
688-
return raceEndWithSignal(pendingEndPromise, signal);
689-
}
690-
const { promise, resolve, reject } = PromiseWithResolvers();
691-
this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
692-
return raceEndWithSignal(promise, signal);
684+
return raceEndWithSignal(this.#queue.getPendingEndPromise(), signal);
693685
}
694686
// >= 0: byte count (immediate close or idempotent)
695687
return PromiseResolve(result);
@@ -709,8 +701,7 @@ class PushWriter {
709701
[SymbolAsyncDispose]() {
710702
const state = this.#queue.writerState;
711703
if (state === 'closing') {
712-
// Wait for graceful drain
713-
return this.#queue.pendingEndPromise ?? PromiseResolve();
704+
return this.#queue.getPendingEndPromise();
714705
}
715706
if (state === 'open') {
716707
this.fail();

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -531,6 +531,23 @@ async function testAsyncDispose() {
531531
}
532532
}
533533

534+
async function testAsyncDisposeWaitsAfterEndSync() {
535+
const { writer, readable } = push({ budget: 16384 });
536+
writer.writeSync('hello');
537+
assert.strictEqual(writer.endSync(), -1);
538+
539+
let disposed = false;
540+
const disposal = writer[Symbol.asyncDispose]().then(() => {
541+
disposed = true;
542+
});
543+
await Promise.resolve();
544+
assert.strictEqual(disposed, false);
545+
546+
assert.strictEqual(await text(readable), 'hello');
547+
await disposal;
548+
assert.strictEqual(disposed, true);
549+
}
550+
534551
async function testSyncDispose() {
535552
const { writer, readable } = push({ budget: 16384 });
536553
writer.writeSync('hello');
@@ -635,5 +652,6 @@ Promise.all([
635652
testEndIdempotentWhenClosed(),
636653
testEndRejectsWhenErrored(),
637654
testAsyncDispose(),
655+
testAsyncDisposeWaitsAfterEndSync(),
638656
testSyncDispose(),
639657
]).then(common.mustCall());

0 commit comments

Comments
 (0)