stream: align Readable.toWeb termination with eos

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

Author: ikeyan

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

@@ -7,6 +7,7 @@ const {
  Promise,
  PromisePrototypeThen,
  ReflectApply,
+ Symbol,
  SymbolDispose,
 }=primordials;
 
@@ -66,6 +67,50 @@ function bindAsyncResource(fn, type) {
 };
 }
 
+/**
+ * Returns the current stream error tracked by eos(), if any.
+ * @param {import('stream').Stream} stream
+ * @returns {Error | null}
+ */
+functiongetEosErrored(stream){
+consterrored=isWritableErrored(stream)||isReadableErrored(stream);
+returntypeoferrored!=='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}
+ */
+functiongetEosOnCloseError(stream,readable,readableFinished,writable,writableFinished){
+consterrored=getEosErrored(stream);
+if(errored){
+returnerrored;
+}
+
+if(readable&&!readableFinished&&isReadableNodeStream(stream,true)){
+if(!isReadableFinished(stream,false)){
+returnnewERR_STREAM_PREMATURE_CLOSE();
+}
+}
+if(writable&&!writableFinished){
+if(!isWritableFinished(stream,false)){
+returnnewERR_STREAM_PREMATURE_CLOSE();
+}
+}
+
+returnnull;
+}
+
+// Internal only: if eos() can settle immediately, invoke the callback before
+// returning cleanup. Callers must tolerate cleanup yet to be assigned.
+constkEosNodeSynchronousCallback=Symbol('kEosNodeSynchronousCallback');
+
 functioneos(stream,options,callback){
 if(arguments.length===2){
 callback=options;
@@ -78,14 +123,6 @@ function eos(stream, options, callback) {
 validateFunction(callback,'callback');
 validateAbortSignal(options.signal,'options.signal');
 
-if(AsyncContextFrame.current()||enabledHooksExist()){
-// Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
-// is a bottleneck here.
-callback=once(bindAsyncResource(callback,'STREAM_END_OF_STREAM'));
-}else{
-callback=once(callback);
-}
-
 if(isReadableStream(stream)||isWritableStream(stream)){
 returneosWeb(stream,options,callback);
 }
@@ -97,15 +134,6 @@ function eos(stream, options, callback) {
 constreadable=options.readable??isReadableNodeStream(stream);
 constwritable=options.writable??isWritableNodeStream(stream);
 
-constwState=stream._writableState;
-constrState=stream._readableState;
-
-constonlegacyfinish=()=>{
-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.
@@ -114,8 +142,85 @@ function eos(stream, options, callback) {
 isReadableNodeStream(stream)===readable&&
 isWritableNodeStream(stream)===writable
 );
-
 letwritableFinished=isWritableFinished(stream,false);
+letreadableFinished=isReadableFinished(stream,false);
+
+constwState=stream._writableState;
+constrState=stream._readableState;
+
+/**
+ * @type {Error | null | undefined}
+ * undefined: to be determined
+ * null: no error
+ * Error: an error occurred
+ */
+letimmediateResult;
+if(isClosed(stream)){
+immediateResult=getEosOnCloseError(
+stream,
+readable,
+readableFinished,
+writable,
+writableFinished,
+);
+}elseif(wState?.errorEmitted||rState?.errorEmitted){
+if(!willEmitClose){
+immediateResult=getEosErrored(stream);
+}
+}elseif(
+!readable&&
+(!willEmitClose||isReadable(stream))&&
+(writableFinished||isWritable(stream)===false)&&
+(wState==null||wState.pendingcb===undefined||wState.pendingcb===0)
+){
+immediateResult=getEosErrored(stream);
+}elseif(
+!writable&&
+(!willEmitClose||isWritable(stream))&&
+(readableFinished||isReadable(stream)===false)
+){
+immediateResult=getEosErrored(stream);
+}elseif((rState&&stream.req&&stream.aborted)){
+immediateResult=getEosErrored(stream);
+}
+letcleanup=()=>{
+callback=nop;
+};
+if(immediateResult!==undefined){
+if(options.error!==false){
+stream.on('error',nop);
+cleanup=()=>{
+callback=nop;
+stream.removeListener('error',nop);
+};
+}
+}elseif(options.signal?.aborted){
+immediateResult=newAbortError(undefined,{cause: options.signal.reason});
+}
+if(immediateResult!==undefined&&options[kEosNodeSynchronousCallback]){
+ReflectApply(callback,stream,immediateResult===null ? [] : [immediateResult]);
+returncleanup;
+}
+
+if(AsyncContextFrame.current()||enabledHooksExist()){
+// 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]));
+returncleanup;
+}
+
+callback=once(callback);
+
+constonlegacyfinish=()=>{
+if(!stream.writable){
+onfinish();
+}
+};
+
 constonfinish=()=>{
 writableFinished=true;
 // Stream should not be destroyed here. If it is that
@@ -134,7 +239,6 @@ function eos(stream, options, callback) {
 }
 };
 
-letreadableFinished=isReadableFinished(stream,false);
 constonend=()=>{
 readableFinished=true;
 // Stream should not be destroyed here. If it is that
@@ -157,41 +261,13 @@ function eos(stream, options, callback) {
 callback.call(stream,err);
 };
 
-letclosed=isClosed(stream);
-
 constonclose=()=>{
-closed=true;
-
-consterrored=isWritableErrored(stream)||isReadableErrored(stream);
-
-if(errored&&typeoferrored!=='boolean'){
-returncallback.call(stream,errored);
-}
-
-if(readable&&!readableFinished&&isReadableNodeStream(stream,true)){
-if(!isReadableFinished(stream,false))
-returncallback.call(stream,
-newERR_STREAM_PREMATURE_CLOSE());
-}
-if(writable&&!writableFinished){
-if(!isWritableFinished(stream,false))
-returncallback.call(stream,
-newERR_STREAM_PREMATURE_CLOSE());
-}
-
-callback.call(stream);
-};
-
-constonclosed=()=>{
-closed=true;
-
-consterrored=isWritableErrored(stream)||isReadableErrored(stream);
-
-if(errored&&typeoferrored!=='boolean'){
-returncallback.call(stream,errored);
+consterror=getEosOnCloseError(stream,readable,readableFinished,writable,writableFinished);
+if(error===null){
+callback.call(stream);
+}else{
+callback.call(stream,error);
 }
-
-callback.call(stream);
 };
 
 constonrequest=()=>{
@@ -225,30 +301,7 @@ function eos(stream, options, callback) {
 }
 stream.on('close',onclose);
 
-if(closed){
-process.nextTick(onclose);
-}elseif(wState?.errorEmitted||rState?.errorEmitted){
-if(!willEmitClose){
-process.nextTick(onclosed);
-}
-}elseif(
-!readable&&
-(!willEmitClose||isReadable(stream))&&
-(writableFinished||isWritable(stream)===false)&&
-(wState==null||wState.pendingcb===undefined||wState.pendingcb===0)
-){
-process.nextTick(onclosed);
-}elseif(
-!writable&&
-(!willEmitClose||isWritable(stream))&&
-(readableFinished||isReadable(stream)===false)
-){
-process.nextTick(onclosed);
-}elseif((rState&&stream.req&&stream.aborted)){
-process.nextTick(onclosed);
-}
-
-constcleanup=()=>{
+cleanup=()=>{
 callback=nop;
 stream.removeListener('aborted',onclose);
 stream.removeListener('complete',onfinish);
@@ -263,7 +316,7 @@ function eos(stream, options, callback) {
 stream.removeListener('close',onclose);
 };
 
-if(options.signal&&!closed){
+if(options.signal){
 constabort=()=>{
 // Keep it because cleanup removes it.
 constendCallback=callback;
@@ -272,23 +325,27 @@ function eos(stream, options, callback) {
 stream,
 newAbortError(undefined,{cause: options.signal.reason}));
 };
-if(options.signal.aborted){
-process.nextTick(abort);
-}else{
-addAbortListener??=require('internal/events/abort_listener').addAbortListener;
-constdisposable=addAbortListener(options.signal,abort);
-constoriginalCallback=callback;
-callback=once((...args)=>{
-disposable[SymbolDispose]();
-ReflectApply(originalCallback,stream,args);
-});
-}
+addAbortListener??=require('internal/events/abort_listener').addAbortListener;
+constdisposable=addAbortListener(options.signal,abort);
+constoriginalCallback=callback;
+callback=once((...args)=>{
+disposable[SymbolDispose]();
+ReflectApply(originalCallback,stream,args);
+});
 }
 
 returncleanup;
 }
 
 functioneosWeb(stream,options,callback){
+if(AsyncContextFrame.current()||enabledHooksExist()){
+// Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
+// is a bottleneck here.
+callback=once(bindAsyncResource(callback,'STREAM_END_OF_STREAM'));
+}else{
+callback=once(callback);
+}
+
 letisAborted=false;
 letabort=nop;
 if(options.signal){
@@ -347,4 +404,5 @@ function finished(stream, opts) {
 module.exports={
  eos,
  finished,
+ kEosNodeSynchronousCallback,
 };