diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 62dfd8a288ab4c..012f273151e94d 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -607,7 +607,11 @@ class ReadableStream { const transfer = lazyTransfer(); setupReadableStreamDefaultControllerFromSource( this, - new transfer.CrossRealmTransformReadableSource(port), + // The MessagePort is set to be referenced when reading. + // After two MessagePorts are closed, there is a problem with + // lingering commitments not being properly resolved. + // https://github.com/nodejs/node/issues/51486 + new transfer.CrossRealmTransformReadableSource(port, true), 0, () => 1); } } diff --git a/lib/internal/webstreams/transfer.js b/lib/internal/webstreams/transfer.js index 136b0d81a99464..07ccca1e682b83 100644 --- a/lib/internal/webstreams/transfer.js +++ b/lib/internal/webstreams/transfer.js @@ -102,10 +102,11 @@ function InternalCloneableDOMException() { InternalCloneableDOMException[kDeserialize] = () => {}; class CrossRealmTransformReadableSource { - constructor(port) { + constructor(port, unref) { this[kState] = { port, controller: undefined, + unref, }; port.onmessage = ({ data }) => { @@ -143,6 +144,8 @@ class CrossRealmTransformReadableSource { error); port.close(); }; + + port.unref(); } start(controller) { @@ -150,6 +153,10 @@ class CrossRealmTransformReadableSource { } async pull() { + if (this[kState].unref) { + this[kState].unref = false; + this[kState].port.ref(); + } this[kState].port.postMessage({ type: 'pull' }); } @@ -170,11 +177,12 @@ class CrossRealmTransformReadableSource { } class CrossRealmTransformWritableSink { - constructor(port) { + constructor(port, unref) { this[kState] = { port, controller: undefined, backpressurePromise: createDeferredPromise(), + unref, }; port.onmessage = ({ data }) => { @@ -211,6 +219,7 @@ class CrossRealmTransformWritableSink { port.close(); }; + port.unref(); } start(controller) { @@ -218,6 +227,10 @@ class CrossRealmTransformWritableSink { } async write(chunk) { + if (this[kState].unref) { + this[kState].unref = false; + this[kState].port.ref(); + } if (this[kState].backpressurePromise === undefined) { this[kState].backpressurePromise = { promise: PromiseResolve(), @@ -262,9 +275,12 @@ class CrossRealmTransformWritableSink { } function newCrossRealmReadableStream(writable, port) { + // The MessagePort is left unref. + // There is a problem with the process not terminating. + // https://github.com/nodejs/node/issues/44985 const readable = new ReadableStream( - new CrossRealmTransformReadableSource(port)); + new CrossRealmTransformReadableSource(port, false)); const promise = readableStreamPipeTo(readable, writable, false, false, false); @@ -278,9 +294,12 @@ function newCrossRealmReadableStream(writable, port) { } function newCrossRealmWritableSink(readable, port) { + // The MessagePort is left unref. + // There is a problem with the process not terminating. + // https://github.com/nodejs/node/issues/44985 const writable = new WritableStream( - new CrossRealmTransformWritableSink(port)); + new CrossRealmTransformWritableSink(port, false)); const promise = readableStreamPipeTo(readable, writable, false, false, false); setPromiseHandled(promise); diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index eea99f321d50c6..30785d798a4f00 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -283,7 +283,11 @@ class WritableStream { const transfer = lazyTransfer(); setupWritableStreamDefaultControllerFromSink( this, - new transfer.CrossRealmTransformWritableSink(port), + // The MessagePort is set to be referenced when reading. + // After two MessagePorts are closed, there is a problem with + // lingering commitments not being properly resolved. + // https://github.com/nodejs/node/issues/51486 + new transfer.CrossRealmTransformWritableSink(port, true), 1, () => 1); } diff --git a/test/parallel/test-webstreams-clone-unref.js b/test/parallel/test-webstreams-clone-unref.js new file mode 100644 index 00000000000000..88a9cebd9c3046 --- /dev/null +++ b/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) + +const rs1 = new ReadableStream(); +const ws1 = new WritableStream(); + +const [rs2, ws2] = structuredClone([rs1, ws1], { transfer: [rs1, ws1] }); + +ok(rs2 instanceof ReadableStream); +ok(ws2 instanceof WritableStream); diff --git a/test/parallel/test-whatwg-webstreams-transfer.js b/test/parallel/test-whatwg-webstreams-transfer.js index 01cfaa02ad075e..7be01c339652c0 100644 --- a/test/parallel/test-whatwg-webstreams-transfer.js +++ b/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(); });