Commit 17e7b689961 for nodejs
commit 17e7b689961b6a8d90d6a54abc0728075ddd9ccc
Author: James M Snell <jasnell@gmail.com>
Date: Sun Oct 4 02:39:06 2026 +0000
stream: do not miss 'drain' in fromWritable()
When a write filled the classic Writable, fromWritable() recorded that
it needed to drain, but only listened for 'drain' once a later write
was queued or something waited for drain. If the Writable emitted
'drain' before that, for example because its write callback ran on a
microtask or with process.nextTick(), the event was missed and the flag
was never cleared: the next write() or writev() never settled, canWrite
stayed false and ondrain() never resolved.
Listen for 'drain' as soon as a write returns false, and keep listening
until it is emitted.
Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/66648
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js
index 2a7b818efd9..0eafba0182b 100644
--- a/lib/internal/streams/iter/classic.js
+++ b/lib/internal/streams/iter/classic.js
@@ -621,7 +621,9 @@ function fromWritable(writable, options = kNullPrototype) {
}
function removeDrainListenerIfIdle() {
+ // Keep listening while needsDrain is set: it is only cleared by 'drain'.
if (!drainListenerInstalled ||
+ needsDrain ||
pendingWrites.length !== 0 ||
drainWaiters.length !== 0) {
return;
@@ -701,7 +703,12 @@ function fromWritable(writable, options = kNullPrototype) {
for (let i = 0; i < chunks.length; i++) {
const bytes = chunks[i];
if (!writable.write(bytes)) {
+ // Listen for 'drain' now, even if nothing is waiting yet: the
+ // Writable can emit it before the next write or wait (for example
+ // when its write callback runs on a microtask), and needsDrain
+ // would never be cleared.
needsDrain = true;
+ installDrainListener();
ok = false;
}
totalBytes += TypedArrayPrototypeGetByteLength(bytes);
diff --git a/test/parallel/test-stream-iter-from-writable-lifecycle.js b/test/parallel/test-stream-iter-from-writable-lifecycle.js
index d5be23fdd03..450fe8d7359 100644
--- a/test/parallel/test-stream-iter-from-writable-lifecycle.js
+++ b/test/parallel/test-stream-iter-from-writable-lifecycle.js
@@ -277,7 +277,40 @@ async function testSignalAbortedByUnderlyingEnd() {
assert.strictEqual(await ending, 0);
}
+// A write that fills the Writable resolves before 'drain'. If the Writable's
+// write callback runs on a microtask or a tick, 'drain' can be emitted before
+// the next write or wait; the adapter must not miss it.
+async function testDrainBeforeNextWrite() {
+ for (const defer of [queueMicrotask, process.nextTick]) {
+ const writable = new Writable({
+ highWaterMark: 4,
+ write(chunk, encoding, callback) { defer(callback); },
+ });
+ const writer = fromWritable(writable);
+
+ for (let i = 0; i < 3; i++) {
+ await writer.write('abcd');
+ }
+ await writer.writev(['ab', 'cd']);
+ await writer.writev(['ab', 'cd']);
+
+ await setImmediate();
+ assert.strictEqual(writer.canWrite, true);
+ // Nothing is waiting any more, so the 'drain' listener is removed.
+ assert.strictEqual(writable.listenerCount('drain'), 0);
+
+ writer.write('abcd');
+ assert.strictEqual(writer.canWrite, false);
+ assert.strictEqual(await ondrain(writer), true);
+ assert.strictEqual(writer.canWrite, true);
+
+ await writer.end();
+ assert.strictEqual(writable.listenerCount('drain'), 0);
+ }
+}
+
Promise.all([
+ testDrainBeforeNextWrite(),
testQueuedWriteThrowsDuringFlush(),
testQueuedWriteErrorsDuringFlush(),
testWriteErrorsSynchronously(),