stream: avoid premature close when will not emit close

Fixes: #45281 PR-URL: #45301 Reviewed-By: Luigi Pinca <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: Minwoo Jung <[email protected]>
nodejs · Nov 21, 2022 · 5274a8f · 5274a8f
1 parent 43e002e
commit 5274a8f
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 4 deletions.
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
@@ -151,6 +151,18 @@ function eos(stream, options, callback) {
     callback.call(stream);
   };
 
+  const onclosed = () => {
+    closed = true;
+
+    const errored = isWritableErrored(stream) || isReadableErrored(stream);
+
+    if (errored && typeof errored !== 'boolean') {
+      return callback.call(stream, errored);
+    }
+
+    callback.call(stream);
+  };
+
   const onrequest = () => {
     stream.req.on('finish', onfinish);
   };
@@ -186,22 +198,22 @@ function eos(stream, options, callback) {
     process.nextTick(onclose);
   } else if (wState?.errorEmitted || rState?.errorEmitted) {
     if (!willEmitClose) {
-      process.nextTick(onclose);
+      process.nextTick(onclosed);
     }
   } else if (
     !readable &&
     (!willEmitClose || isReadable(stream)) &&
     (writableFinished || isWritable(stream) === false)
   ) {
-    process.nextTick(onclose);
+    process.nextTick(onclosed);
   } else if (
     !writable &&
     (!willEmitClose || isWritable(stream)) &&
     (readableFinished || isReadable(stream) === false)
   ) {
-    process.nextTick(onclose);
+    process.nextTick(onclosed);
   } else if ((rState && stream.req && stream.aborted)) {
-    process.nextTick(onclose);
+    process.nextTick(onclosed);
   }
 
   const cleanup = () => {

diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js
@@ -667,3 +667,17 @@ testClosed((opts) => new Writable({ write() {}, ...opts }));
     ).end();
   });
 }
+
+{
+  const stream = new Duplex({
+    write(chunk, enc, cb) {
+      setImmediate(cb);
+    }
+  });
+
+  stream.end('foo');
+
+  finished(stream, { readable: false }, common.mustCall((err) => {
+    assert(!err);
+  }));
+}