stream: align Readable.toWeb termination with eos

PR-URL: #62394
Backport-PR-URL: #64675
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: ikeyan

lib/internal/streams/end-of-stream.js

@@ -6,6 +6,8 @@
 const {
   Promise,
   PromisePrototypeThen,
+  ReflectApply,
+  Symbol,
   SymbolDispose,
 } = primordials;
 
@@ -62,6 +64,50 @@ function bindAsyncResource(fn, type) {
   };
 }
 
+/**
+ * Returns the current stream error tracked by eos(), if any.
+ * @param {import('stream').Stream} stream
+ * @returns {Error | null}
+ */
+function getEosErrored(stream) {
+  const errored = isWritableErrored(stream) || isReadableErrored(stream);
+  return typeof errored !== 'boolean' && errored || null;
+}
+
+/**
+ * Returns the error eos() would report from an immediate close, including
+ * premature close detection for unfinished readable or writable sides.
+ * @param {import('stream').Stream} stream
+ * @param {boolean} readable
+ * @param {boolean | null} readableFinished
+ * @param {boolean} writable
+ * @param {boolean | null} writableFinished
+ * @returns {Error | null}
+ */
+function getEosOnCloseError(stream, readable, readableFinished, writable, writableFinished) {
+  const errored = getEosErrored(stream);
+  if (errored) {
+    return errored;
+  }
+
+  if (readable && !readableFinished && isReadableNodeStream(stream, true)) {
+    if (!isReadableFinished(stream, false)) {
+      return new ERR_STREAM_PREMATURE_CLOSE();
+    }
+  }
+  if (writable && !writableFinished) {
+    if (!isWritableFinished(stream, false)) {
+      return new ERR_STREAM_PREMATURE_CLOSE();
+    }
+  }
+
+  return null;
+}
+
+// Internal only: if eos() can settle immediately, invoke the callback before
+// returning cleanup. Callers must tolerate cleanup yet to be assigned.
+const kEosNodeSynchronousCallback = Symbol('kEosNodeSynchronousCallback');
+
 function eos(stream, options, callback) {
   if (arguments.length === 2) {
     callback = options;
@@ -74,10 +120,6 @@ function eos(stream, options, callback) {
   validateFunction(callback, 'callback');
   validateAbortSignal(options.signal, 'options.signal');
 
-  // Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
-  // is a bottleneck here.
-  callback = once(bindAsyncResource(callback, 'STREAM_END_OF_STREAM'));
-
   if (isReadableStream(stream) || isWritableStream(stream)) {
     return eosWeb(stream, options, callback);
   }
@@ -89,15 +131,6 @@ function eos(stream, options, callback) {
   const readable = options.readable ?? isReadableNodeStream(stream);
   const writable = options.writable ?? isWritableNodeStream(stream);
 
-  const wState = stream._writableState;
-  const rState = stream._readableState;
-
-  const onlegacyfinish = () => {
-    if (!stream.writable) {
-      onfinish();
-    }
-  };
-
   // TODO (ronag): Improve soft detection to include core modules and
   // common ecosystem modules that do properly emit 'close' but fail
   // this generic check.
@@ -106,8 +139,83 @@ function eos(stream, options, callback) {
     isReadableNodeStream(stream) === readable &&
     isWritableNodeStream(stream) === writable
   );
-
   let writableFinished = isWritableFinished(stream, false);
+  let readableFinished = isReadableFinished(stream, false);
+
+  const wState = stream._writableState;
+  const rState = stream._readableState;
+
+  /**
+   * @type {Error | null | undefined}
+   * undefined: to be determined
+   * null: no error
+   * Error: an error occurred
+   */
+  let immediateResult;
+  if (isClosed(stream)) {
+    immediateResult = getEosOnCloseError(
+      stream,
+      readable,
+      readableFinished,
+      writable,
+      writableFinished,
+    );
+  } else if (wState?.errorEmitted || rState?.errorEmitted) {
+    if (!willEmitClose) {
+      immediateResult = getEosErrored(stream);
+    }
+  } else if (
+    !readable &&
+    (!willEmitClose || isReadable(stream)) &&
+    (writableFinished || isWritable(stream) === false) &&
+    (wState == null || wState.pendingcb === undefined || wState.pendingcb === 0)
+  ) {
+    immediateResult = getEosErrored(stream);
+  } else if (
+    !writable &&
+    (!willEmitClose || isWritable(stream)) &&
+    (readableFinished || isReadable(stream) === false)
+  ) {
+    immediateResult = getEosErrored(stream);
+  } else if ((rState && stream.req && stream.aborted)) {
+    immediateResult = getEosErrored(stream);
+  }
+  let cleanup = () => {
+    callback = nop;
+  };
+  if (immediateResult !== undefined) {
+    if (options.error !== false) {
+      stream.on('error', nop);
+      cleanup = () => {
+        callback = nop;
+        stream.removeListener('error', nop);
+      };
+    }
+  } else if (options.signal?.aborted) {
+    immediateResult = new AbortError(undefined, { cause: options.signal.reason });
+  }
+  if (immediateResult !== undefined && options[kEosNodeSynchronousCallback]) {
+    ReflectApply(callback, stream, immediateResult === null ? [] : [immediateResult]);
+    return cleanup;
+  }
+
+  // Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
+  // is a bottleneck here.
+  callback = bindAsyncResource(callback, 'STREAM_END_OF_STREAM');
+
+  if (immediateResult !== undefined) {
+    process.nextTick(() => ReflectApply(callback, stream, immediateResult === null ? [] : [immediateResult]));
+    return cleanup;
+  }
+
+  callback = once(callback);
+
+  const onlegacyfinish = () => {
+    if (!stream.writable) {
+      onfinish();
+    }
+  };
+
   const onfinish = () => {
     writableFinished = true;
     // Stream should not be destroyed here. If it is that
@@ -126,7 +234,6 @@ function eos(stream, options, callback) {
     }
   };
 
-  let readableFinished = isReadableFinished(stream, false);
   const onend = () => {
     readableFinished = true;
     // Stream should not be destroyed here. If it is that
@@ -149,41 +256,13 @@ function eos(stream, options, callback) {
     callback.call(stream, err);
   };
 
-  let closed = isClosed(stream);
-
   const onclose = () => {
-    closed = true;
-
-    const errored = isWritableErrored(stream) || isReadableErrored(stream);
-
-    if (errored && typeof errored !== 'boolean') {
-      return callback.call(stream, errored);
-    }
-
-    if (readable && !readableFinished && isReadableNodeStream(stream, true)) {
-      if (!isReadableFinished(stream, false))
-        return callback.call(stream,
-                             new ERR_STREAM_PREMATURE_CLOSE());
-    }
-    if (writable && !writableFinished) {
-      if (!isWritableFinished(stream, false))
-        return callback.call(stream,
-                             new ERR_STREAM_PREMATURE_CLOSE());
-    }
-
-    callback.call(stream);
-  };
-
-  const onclosed = () => {
-    closed = true;
-
-    const errored = isWritableErrored(stream) || isReadableErrored(stream);
-
-    if (errored && typeof errored !== 'boolean') {
-      return callback.call(stream, errored);
+    const error = getEosOnCloseError(stream, readable, readableFinished, writable, writableFinished);
+    if (error === null) {
+      callback.call(stream);
+    } else {
+      callback.call(stream, error);
     }
-
-    callback.call(stream);
   };
 
   const onrequest = () => {
@@ -217,30 +296,7 @@ function eos(stream, options, callback) {
   }
   stream.on('close', onclose);
 
-  if (closed) {
-    process.nextTick(onclose);
-  } else if (wState?.errorEmitted || rState?.errorEmitted) {
-    if (!willEmitClose) {
-      process.nextTick(onclosed);
-    }
-  } else if (
-    !readable &&
-    (!willEmitClose || isReadable(stream)) &&
-    (writableFinished || isWritable(stream) === false) &&
-    (wState == null || wState.pendingcb === undefined || wState.pendingcb === 0)
-  ) {
-    process.nextTick(onclosed);
-  } else if (
-    !writable &&
-    (!willEmitClose || isWritable(stream)) &&
-    (readableFinished || isReadable(stream) === false)
-  ) {
-    process.nextTick(onclosed);
-  } else if ((rState && stream.req && stream.aborted)) {
-    process.nextTick(onclosed);
-  }
-
-  const cleanup = () => {
+  cleanup = () => {
     callback = nop;
     stream.removeListener('aborted', onclose);
     stream.removeListener('complete', onfinish);
@@ -255,7 +311,7 @@ function eos(stream, options, callback) {
     stream.removeListener('close', onclose);
   };
 
-  if (options.signal && !closed) {
+  if (options.signal) {
     const abort = () => {
       // Keep it because cleanup removes it.
       const endCallback = callback;
@@ -264,23 +320,23 @@ function eos(stream, options, callback) {
         stream,
         new AbortError(undefined, { cause: options.signal.reason }));
     };
-    if (options.signal.aborted) {
-      process.nextTick(abort);
-    } else {
-      addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
-      const disposable = addAbortListener(options.signal, abort);
-      const originalCallback = callback;
-      callback = once((...args) => {
-        disposable[SymbolDispose]();
-        originalCallback.apply(stream, args);
-      });
-    }
+    addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
+    const disposable = addAbortListener(options.signal, abort);
+    const originalCallback = callback;
+    callback = once((...args) => {
+      disposable[SymbolDispose]();
+      ReflectApply(originalCallback, stream, args);
+    });
   }
 
   return cleanup;
 }
 
 function eosWeb(stream, options, callback) {
+  // Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
+  // is a bottleneck here.
+  callback = once(bindAsyncResource(callback, 'STREAM_END_OF_STREAM'));
+
   let isAborted = false;
   let abort = nop;
   if (options.signal) {
@@ -339,4 +395,5 @@ function finished(stream, opts) {
 module.exports = {
   eos,
   finished,
+  kEosNodeSynchronousCallback,
 };