From 926e8150ed8d422e326d53a8daea1dcdf6509731 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Oct 2023 14:25:51 +0200
Subject: [PATCH] stream: avoid unnecessary drain for sync stream

---
 lib/internal/streams/writable.js              | 19 +++++++++++--------
 .../test-stream-duplex-readable-end.js        | 13 +++++--------
 2 files changed, 16 insertions(+), 16 deletions(-)

diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
index 5800c9df171ff2..d06f9e4e88b240 100644
--- a/lib/internal/streams/writable.js
+++ b/lib/internal/streams/writable.js
@@ -108,6 +108,7 @@ const kWriteCb = 1 << 26;
 const kExpectWriteCb = 1 << 27;
 const kAfterWriteTickInfo = 1 << 28;
 const kAfterWritePending = 1 << 29;
+const kIsDuplex = 1 << 30;
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {
@@ -286,6 +287,7 @@ function WritableState(options, stream, isDuplex) {
 
   if (options && options.objectMode) this.state |= kObjectMode;
   if (isDuplex && options && options.writableObjectMode) this.state |= kObjectMode;
+  if (isDuplex) this.state |= kIsDuplex;
 
   // The point at which write() starts returning false
   // Note: 0 is a valid value, means that we always return false if
@@ -513,14 +515,6 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
 
   state.length += len;
 
-  // stream._write resets state.length
-  const ret = state.length < state.highWaterMark;
-
-  // We must ensure that previous needDrain will not be reset to false.
-  if (!ret) {
-    state.state |= kNeedDrain;
-  }
-
   if ((state.state & (kWriting | kErrored | kCorked | kConstructed)) !== kConstructed) {
     state.buffered.push({ chunk, encoding, callback });
     if ((state.state & kAllBuffers) !== 0 && encoding !== 'buffer') {
@@ -539,6 +533,15 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
     state.state &= ~kSync;
   }
 
+  const ret = (
+    state.length < state.highWaterMark &&
+    ((state.state & kIsDuplex === 0) || stream._readableState?.ended !== true)
+  );
+
+  if (!ret) {
+    state.state |= kNeedDrain;
+  }
+
   // Return false if errored or destroyed in order to break
   // any synchronous while(stream.write(data)) loops.
   return ret && (state.state & (kDestroyed | kErrored)) === 0;
diff --git a/test/parallel/test-stream-duplex-readable-end.js b/test/parallel/test-stream-duplex-readable-end.js
index 0e3e62aacb14bc..5a6f8d3bb99893 100644
--- a/test/parallel/test-stream-duplex-readable-end.js
+++ b/test/parallel/test-stream-duplex-readable-end.js
@@ -4,12 +4,12 @@ const common = require('../common');
 const assert = require('assert');
 const stream = require('stream');
 
-let loops = 5;
+let loops = 0;
 
 const src = new stream.Readable({
   read() {
-    if (loops--)
-      this.push(Buffer.alloc(20000));
+    this.push(Buffer.alloc(20000));
+    assert(loops++ < 2);
   }
 });
 
@@ -22,8 +22,5 @@ const dst = new stream.Transform({
 
 src.pipe(dst);
 
-dst.on('data', () => { });
-dst.on('end', common.mustCall(() => {
-  assert.strictEqual(loops, 3);
-  assert.ok(src.isPaused());
-}));
+dst.on('data', common.mustNotCall());
+dst.on('end', common.mustCall());