worker: only unref port for stdin if we ref’ed it before

We set the `kStartedReading` flag from `_read()` for Worker stdio,
and then `ref()` the port.

However, the `.on('end')` handler is also attached when `._read()`
is not called, e.g. when `process.stdin` inside a Worker is prematurely
ended because stdin was not enabled by the parent thread.

In that case, we should not call `.unref()` for stdin if we did not
also call `.ref()` for it before.

Fixes: #28144
PR-URL: #28153
Reviewed-By: Colin Ihrig <[email protected]>
Reviewed-By: Rich Trott <[email protected]>
Reviewed-By: Ruben Bridgewater <[email protected]>

Author: addaleax

lib/internal/worker/io.js

@@ -169,8 +169,10 @@ class ReadableWorkerStdio extends Readable {
     this[kIncrementsPortRef] = true;
     this[kStartedReading] = false;
     this.on('end', () => {
-      if (this[kIncrementsPortRef] && --this[kPort][kWaitingStreams] === 0)
-        this[kPort].unref();
+      if (this[kStartedReading] && this[kIncrementsPortRef]) {
+        if (--this[kPort][kWaitingStreams] === 0)
+          this[kPort].unref();
+      }
     });
   }
 

test/parallel/test-worker-no-stdin-stdout-interaction.js

@@ -0,0 +1,20 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { Worker, isMainThread } = require('worker_threads');
+
+// Regression test for https://github.com/nodejs/node/issues/28144.
+
+if (isMainThread) {
+  const w = new Worker(__filename);
+  w.on('exit', common.mustCall((status) => {
+    assert.strictEqual(status, 0);
+  }));
+  w.stdout.on('data', common.mustCall(10));
+} else {
+  process.stdin.on('data', () => {});
+
+  for (let i = 0; i < 10; ++i) {
+    process.stdout.write(`processing(${i})\n`, common.mustCall());
+  }
+}