stream: avoid duplicate writes in toWritable · nodejs/node@2e86855 · GitHub
Skip to content

Commit 2e86855

Browse files
trivikraduh95
authored andcommitted
stream: avoid duplicate writes in toWritable
PushWriter can return false after accepting a chunk when block backpressure is active. Teach the classic Writable adapter to treat that case as accepted backpressure instead of retrying through the async write path. Fixes: #63359 Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #63360 Backport-PR-URL: #64675 Fixes: #63359 Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 88a3392 commit 2e86855

4 files changed

Lines changed: 119 additions & 6 deletions

File tree

lib/internal/streams/iter/classic.js

Lines changed: 45 additions & 6 deletions

lib/internal/streams/iter/push.js

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ const {
3232

3333
const {
3434
drainableProtocol,
35+
kSyncWriteAccepted,
3536
kSyncWriteAcceptedOnFalse,
3637
} = require('internal/streams/iter/types');
3738

@@ -545,11 +546,16 @@ class PushQueue {
545546

546547
class PushWriter {
547548
#queue;
549+
#syncWriteAccepted = false;
548550

549551
constructor(queue) {
550552
this.#queue = queue;
551553
}
552554

555+
[kSyncWriteAccepted]() {
556+
return this.#syncWriteAccepted;
557+
}
558+
553559
[drainableProtocol]() {
554560
const desired = this.desiredSize;
555561
if (desired === null) return null;
@@ -589,19 +595,23 @@ class PushWriter {
589595
}
590596

591597
writeSync(chunk) {
598+
this.#syncWriteAccepted = false;
592599
const bytes = toUint8Array(chunk);
593600
const result = this.#queue.writeSync([bytes]);
594601
if (!result && this.#queue.backpressurePolicy === 'block' &&
595602
this.#queue.desiredSize === 0) {
596603
// Block policy: force-enqueue and return false as backpressure signal.
597604
// Data IS accepted; false tells caller to slow down.
598605
this.#queue.forceEnqueue([bytes]);
606+
this.#syncWriteAccepted = true;
599607
return false;
600608
}
609+
this.#syncWriteAccepted = result;
601610
return result;
602611
}
603612

604613
writevSync(chunks) {
614+
this.#syncWriteAccepted = false;
605615
if (!ArrayIsArray(chunks)) {
606616
throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
607617
}
@@ -610,8 +620,10 @@ class PushWriter {
610620
if (!result && this.#queue.backpressurePolicy === 'block' &&
611621
this.#queue.desiredSize === 0) {
612622
this.#queue.forceEnqueue(bytes);
623+
this.#syncWriteAccepted = true;
613624
return false;
614625
}
626+
this.#syncWriteAccepted = result;
615627
return result;
616628
}
617629

lib/internal/streams/iter/types.js

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
6464
*/
6565
const kValidatedSource = Symbol('kValidatedSource');
6666

67+
/**
68+
* Internal sentinel for writers whose sync write methods can return false
69+
* after accepting data as a backpressure signal.
70+
*/
71+
const kSyncWriteAccepted = Symbol('kSyncWriteAccepted');
72+
73+
/**
74+
* Internal sentinel for writers whose sync write methods may return false
75+
* after accepting data when backpressure is applied. Such writers must expose
76+
* desiredSize so callers can distinguish accepted backpressure from a sync
77+
* write that was not performed.
78+
*/
6779
const kSyncWriteAcceptedOnFalse = Symbol('kSyncWriteAcceptedOnFalse');
6880

6981
module.exports = {
7082
broadcastProtocol,
7183
drainableProtocol,
84+
kSyncWriteAccepted,
7285
kSyncWriteAcceptedOnFalse,
7386
kValidatedSource,
7487
kValidatedTransform,

test/parallel/test-stream-iter-writable-from.js

Lines changed: 49 additions & 0 deletions

0 commit comments

Comments
 (0)