Revert "stream: revert fix cloned webstreams not being unref'd"

marco-ippolito · marco-ippolito · commit cffd2cc0c9ed · 2024-05-27T22:21:51.000+02:00
This reverts commit 8d20b64. PR-URL: #53144 Fixes: #53143 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rafael Gonzaga <rafael.nunu@hotmail.com>
diff --git a/lib/internal/webstreams/transfer.js b/lib/internal/webstreams/transfer.js
@@ -146,6 +146,8 @@ class CrossRealmTransformReadableSource {
         error);
       port.close();
     };
+
+    port.unref();
   }
 
   start(controller) {
@@ -219,6 +221,7 @@ class CrossRealmTransformWritableSink {
       port.close();
     };
 
+    port.unref();
   }
 
   start(controller) {
diff --git a/test/parallel/test-webstreams-clone-unref.js 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
@@ -454,12 +454,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();
     });

Original file line number	Diff line number	Diff line change
`@@ -146,6 +146,8 @@ class CrossRealmTransformReadableSource {`
`146`	`146`	`error);`
`147`	`147`	`port.close();`
`148`	`148`	`};`
	`149`	`+`
	`150`	`+ port.unref();`
`149`	`151`	`}`
`150`	`152`
`151`	`153`	`start(controller) {`
`@@ -219,6 +221,7 @@ class CrossRealmTransformWritableSink {`
`219`	`221`	`port.close();`
`220`	`222`	`};`
`221`	`223`
	`224`	`+ port.unref();`
`222`	`225`	`}`
`223`	`226`
`224`	`227`	`start(controller) {`