stream: fix cloned webstreams not being unref correctly

PR-URL: #51526 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Vinícius Lourenço Claro Cardoso <[email protected]>

Author: tsctx

lib/internal/webstreams/readablestream.js

@@ -607,7 +607,11 @@ class ReadableStream{
 consttransfer=lazyTransfer();
 setupReadableStreamDefaultControllerFromSource(
 this,
-newtransfer.CrossRealmTransformReadableSource(port),
+// The MessagePort is set to be referenced when reading.
+// After two MessagePorts are closed, there is a problem with
+// lingering promise not being properly resolved.
+// https://github.com/nodejs/node/issues/51486
+newtransfer.CrossRealmTransformReadableSource(port,true),
 0,()=>1);
 }
 }

lib/internal/webstreams/transfer.js

@@ -102,10 +102,11 @@ function InternalCloneableDOMException(){
 InternalCloneableDOMException[kDeserialize]=()=>{};
 
 classCrossRealmTransformReadableSource{
-constructor(port){
+constructor(port,unref){
 this[kState]={
  port,
 controller: undefined,
+ unref,
 };
 
 port.onmessage=({ data })=>{
@@ -143,13 +144,19 @@ class CrossRealmTransformReadableSource{
 error);
 port.close();
 };
+
+port.unref();
 }
 
 start(controller){
 this[kState].controller=controller;
 }
 
 asyncpull(){
+if(this[kState].unref){
+this[kState].unref=false;
+this[kState].port.ref();
+}
 this[kState].port.postMessage({type: 'pull'});
 }
 
@@ -170,11 +177,12 @@ class CrossRealmTransformReadableSource{
 }
 
 classCrossRealmTransformWritableSink{
-constructor(port){
+constructor(port,unref){
 this[kState]={
  port,
 controller: undefined,
 backpressurePromise: createDeferredPromise(),
+ unref,
 };
 
 port.onmessage=({ data })=>{
@@ -211,13 +219,18 @@ class CrossRealmTransformWritableSink{
 port.close();
 };
 
+port.unref();
 }
 
 start(controller){
 this[kState].controller=controller;
 }
 
 asyncwrite(chunk){
+if(this[kState].unref){
+this[kState].unref=false;
+this[kState].port.ref();
+}
 if(this[kState].backpressurePromise===undefined){
 this[kState].backpressurePromise={
 promise: PromiseResolve(),
@@ -262,12 +275,12 @@ class CrossRealmTransformWritableSink{
 }
 
 functionnewCrossRealmReadableStream(writable,port){
-constreadable=
-newReadableStream(
-newCrossRealmTransformReadableSource(port));
+// MessagePort should always be unref.
+// There is a problem with the process not terminating.
+// https://github.com/nodejs/node/issues/44985
+constreadable=newReadableStream(newCrossRealmTransformReadableSource(port,false));
 
-constpromise=
-readableStreamPipeTo(readable,writable,false,false,false);
+constpromise=readableStreamPipeTo(readable,writable,false,false,false);
 
 setPromiseHandled(promise);
 
@@ -278,12 +291,15 @@ function newCrossRealmReadableStream(writable, port){
 }
 
 functionnewCrossRealmWritableSink(readable,port){
-constwritable=
-newWritableStream(
-newCrossRealmTransformWritableSink(port));
+// MessagePort should always be unref.
+// There is a problem with the process not terminating.
+// https://github.com/nodejs/node/issues/44985
+constwritable=newWritableStream(newCrossRealmTransformWritableSink(port,false));
 
 constpromise=readableStreamPipeTo(readable,writable,false,false,false);
+
 setPromiseHandled(promise);
+
 return{
  writable,
  promise,

lib/internal/webstreams/writablestream.js

@@ -263,8 +263,6 @@ class WritableStream{
 this[kState].transfer.readable=readable;
 this[kState].transfer.promise=promise;
 
-setPromiseHandled(this[kState].transfer.promise);
-
 return{
 data: {port: this[kState].transfer.port2},
 deserializeInfo:
@@ -283,7 +281,11 @@ class WritableStream{
 consttransfer=lazyTransfer();
 setupWritableStreamDefaultControllerFromSink(
 this,
-newtransfer.CrossRealmTransformWritableSink(port),
+// The MessagePort is set to be referenced when reading.
+// After two MessagePorts are closed, there is a problem with
+// lingering promise not being properly resolved.
+// https://github.com/nodejs/node/issues/51486
+newtransfer.CrossRealmTransformWritableSink(port,true),
 1,
 ()=>1);
 }

test/parallel/test-webstreams-clone-unref.js

@@ -0,0 +1,16 @@
+'use strict';
+
+require('../common');
+const{ ok }=require('node:assert');
+
+// This test verifies that cloned ReadableStream and WritableStream instances
+// do not keep the process alive. The test fails if it timesout (it should just
+// exit immediately)
+
+constrs1=newReadableStream();
+constws1=newWritableStream();
+
+const[rs2,ws2]=structuredClone([rs1,ws1],{transfer: [rs1,ws1]});
+
+ok(rs2instanceofReadableStream);
+ok(ws2instanceofWritableStream);

test/parallel/test-whatwg-webstreams-transfer.js

@@ -464,12 +464,23 @@ const theData = 'hello'
  tracker.verify();
  });
 
+ // We create an interval to keep the event loop alive while
+ // we wait for the stream read to complete. The reason this is needed is because there's
+ // otherwise nothing to keep the worker thread event loop alive long enough to actually
+ // complete the read from the stream. Under the covers the ReadableStream uses an
+ // unref'd MessagePort to communicate with the main thread. Because the MessagePort
+ // is unref'd, it's existence would not keep the thread alive on its own. There was previously
+ // a bug where this MessagePort was ref'd which would block the thread and main thread
+ // from terminating at all unless the stream was consumed/closed.
+ const i = setInterval(() =>{}, 1000);
+
  parentPort.onmessage = tracker.calls(({data }) =>{
  assert(isReadableStream(data));
  const reader = data.getReader();
  reader.read().then(tracker.calls((result) =>{
  assert(!result.done);
  assert(result.value instanceof Uint8Array);
+ clearInterval(i);
  }));
  parentPort.close();
  });