Repository: denoland/deno
Pull request: fix(ext/node): add worker_threads stdin support

## Summary

- Adds `worker.stdin` support when `{ stdin: true }` is passed to the `Worker` constructor
- Parent creates a `Writable` stream (`worker.stdin`) that sends `WORKER_STDIN` / `WORKER_STDIN_END` messages to the worker via `postMessage`
- Worker replaces `process.stdin` with a `Readable` that receives data from those messages, using an early `addEventListener` with `stopImmediatePropagation()` to hide internal messages from user handlers
- Fixes stdout/stderr forwarding to preserve binary data (passes `chunk` directly instead of `String(chunk)`)
- Enables 3 new passing node_compat tests: `test-worker-stdio.js`, `test-worker-stdio-flush.js`, `test-worker-stdio-flush-inflight.js`

## Test plan

- [x] `cargo test --test node_compat parallel -- --filter "test-worker-stdio"` — 3 new tests pass
- [x] `cargo test --test node_compat parallel -- --filter "test-worker-no-stdin-stdout-interaction"` — existing test still passes
- [x] `tools/lint.js` passes
- [x] `tools/format.js` passes

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Changed files:
filename: ext/node/polyfills/worker_threads.ts
patch: @@ -38,7 +38,7 @@ import {
   validateObject,
 } from "ext:deno_node/internal/validators.mjs";
 import { EventEmitter } from "node:events";
-import { Readable } from "node:stream";
+import { Readable, Writable } from "node:stream";
 import {
   BroadcastChannel as WebBroadcastChannel,
   refBroadcastChannel,
@@ -100,7 +100,30 @@ function isWorkerOnlineMsg(data: unknown): data is WorkerOnlineMsg {
 
 interface WorkerStdioMsg {
   type: "WORKER_STDERR" | "WORKER_STDOUT";
-  data: string;
+  // deno-lint-ignore no-explicit-any
+  data: any;
+}
+
+interface WorkerStdinMsg {
+  type: "WORKER_STDIN";
+  // deno-lint-ignore no-explicit-any
+  data: any;
+}
+
+interface WorkerStdinEndMsg {
+  type: "WORKER_STDIN_END";
+}
+
+function isWorkerStdinMsg(data: unknown): data is WorkerStdinMsg {
+  return typeof data === "object" && data !== null &&
+    ObjectHasOwn(data, "type") &&
+    (data as { "type": unknown })["type"] === "WORKER_STDIN";
+}
+
+function isWorkerStdinEndMsg(data: unknown): data is WorkerStdinEndMsg {
+  return typeof data === "object" && data !== null &&
+    ObjectHasOwn(data, "type") &&
+    (data as { "type": unknown })["type"] === "WORKER_STDIN_END";
 }
 
 function isWorkerStderrMsg(data: unknown): data is WorkerStdioMsg {
@@ -184,7 +207,7 @@ class NodeWorker extends EventEmitter {
     stackSizeMb: 4,
   };
   // https://nodejs.org/api/worker_threads.html#workerstdin
-  stdin = null;
+  stdin: Writable | null = null;
   // https://nodejs.org/api/worker_threads.html#workerstdout
   stdout: Readable = new Readable({ read() {} });
   // https://nodejs.org/api/worker_threads.html#workerstderr
@@ -336,6 +359,7 @@ class NodeWorker extends EventEmitter {
       name: this.#name,
       isEval: !!options?.eval,
       isWorkerThread: true,
+      hasStdin: !!options?.stdin,
     }, options?.transferList ?? []);
 
     if (options?.eval) {
@@ -375,6 +399,35 @@ class NodeWorker extends EventEmitter {
     );
     this.#id = id;
     this.threadId = id;
+
+    if (options?.stdin) {
+      // deno-lint-ignore no-this-alias
+      const worker = this;
+      this.stdin = new Writable({
+        write(chunk, _encoding, callback) {
+          try {
+            worker.postMessage({
+              type: "WORKER_STDIN",
+              data: chunk,
+            });
+            callback();
+          } catch (err) {
+            callback(err);
+          }
+        },
+        final(callback) {
+          try {
+            worker.postMessage({
+              type: "WORKER_STDIN_END",
+            });
+            callback();
+          } catch (err) {
+            callback(err);
+          }
+        },
+      });
+    }
+
     this.#pollControl();
     this.#pollMessages();
     process.nextTick(() => process.emit("worker", this));
@@ -751,6 +804,31 @@ internals.__initWorkerThreads = (
           }
         }
 
+        // Replace process.stdin with a Readable that receives
+        // data from the parent via WORKER_STDIN messages.
+        if (metadata.hasStdin) {
+          const workerStdin = new Readable({ read() {} });
+          process.stdin = workerStdin;
+
+          // Register an early listener to intercept stdin messages
+          // before any user-registered handlers. Remove the listener
+          // once stdin ends so the worker can exit cleanly.
+          const stdinHandler = (ev) => {
+            const msg = ev.data;
+            if (isWorkerStdinMsg(msg)) {
+              // deno-lint-ignore prefer-primordials
+              workerStdin.push(msg.data);
+              ev.stopImmediatePropagation();
+            } else if (isWorkerStdinEndMsg(msg)) {
+              // deno-lint-ignore prefer-primordials
+              workerStdin.push(null);
+              parentPort.removeEventListener("message", stdinHandler);
+              ev.stopImmediatePropagation();
+            }
+          };
+          parentPort.addEventListener("message", stdinHandler);
+        }
+
         // Forward stdout writes to the parent so worker.stdout
         // is readable from the host side.
         const origStdoutWrite = FunctionPrototypeBind(
@@ -760,7 +838,7 @@ internals.__initWorkerThreads = (
         process.stdout.write = function (chunk, encoding, callback) {
           parentPort.postMessage({
             type: "WORKER_STDOUT",
-            data: typeof chunk === "string" ? chunk : String(chunk),
+            data: chunk,
           });
           return FunctionPrototypeCall(
             origStdoutWrite,
@@ -780,7 +858,7 @@ internals.__initWorkerThreads = (
         process.stderr.write = function (chunk, encoding, callback) {
           parentPort.postMessage({
             type: "WORKER_STDERR",
-            data: typeof chunk === "string" ? chunk : String(chunk),
+            data: chunk,
           });
           return FunctionPrototypeCall(
             origStderrWrite,

filename: tests/node_compat/config.jsonc
patch: @@ -1588,6 +1588,9 @@
     "parallel/test-worker-ref-onexit.js": {},
     "parallel/test-worker-relative-path-double-dot.js": {},
     "parallel/test-worker-relative-path.js": {},
+    "parallel/test-worker-stdio.js": {},
+    "parallel/test-worker-stdio-flush.js": {},
+    "parallel/test-worker-stdio-flush-inflight.js": {},
     "parallel/test-worker-terminate-unrefed.js": {},
     "parallel/test-worker-unref-from-message-during-exit.js": {},
     "parallel/test-worker-thread-name.js": {},