worker_threads: defer worker-side stdio/Console setup to first access

src/js/node/worker_threads.ts

@@ -5,8 +5,10 @@ type WebWorker = InstanceType<typeof globalThis.Worker>;
 
 const EventEmitter = require("node:events");
 const { SafeMap } = require("internal/primordials");
-const Readable = require("internal/streams/readable");
-const Writable = require("internal/streams/writable");
+// Readable/Writable are loaded lazily: this module is preloaded in every
+// node:worker_threads worker, and the streams graph is several ms cold.
+let Readable;
+let Writable;
 const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared");
 const {
   validateString,
@@ -26,8 +28,6 @@ function normalizeWorkerName(rawName) {
   return "";
 }
 
-const { isAbsolute: pathIsAbsolute } = require("node:path");
-
 // node's filename validation for non-eval workers: absolute or "./"/"../"-relative
 // paths and file: URL objects; bare specifiers and string URLs are rejected.
 function validateWorkerFilename(filename) {
@@ -41,7 +41,7 @@ function validateWorkerFilename(filename) {
     // throws the canonical ERR_INVALID_ARG_TYPE with the exact node message.
     return filename;
   }
-  if (pathIsAbsolute(filename) || /^\.\.?[\\/]/.test(filename)) {
+  if (require("node:path").isAbsolute(filename) || /^\.\.?[\\/]/.test(filename)) {
     return filename;
   }
   let message =
@@ -334,6 +334,7 @@ const BUN_WORKER_MESSAGING_KEY = "@@bunWorkerThreadsMessaging";
 // Readable fed by a control MessagePort (worker.stdout/stderr on the parent,
 // process.stdin in the worker). The peer posts arrays of Buffers; null signals EOF.
 function makePortReadable(port) {
+  Readable ??= require("internal/streams/readable");
   let attached = false;
   let ended = false;
   function onMessage(payload) {
@@ -380,6 +381,7 @@ function makePortReadable(port) {
 // Writable that forwards chunks over a control MessagePort (worker.stdin on the
 // parent, process.stdout/stderr in the worker). final() posts null as EOF.
 function makePortWritable(port) {
+  Writable ??= require("internal/streams/writable");
   // Reader-side acks complete the in-flight writev. The listener refs the
   // event loop; release that immediately — the port is re-ref'd only while a
   // batch is awaiting its ack, so unflushed data keeps the writer alive
@@ -435,44 +437,57 @@ function makePortWritable(port) {
   });
 }
 
+// Install a self-replacing accessor: the first get() builds the value and
+// rewrites the slot to a plain writable data property, so later reads are
+// direct and identity-stable; set() does the same rewrite so assignment works.
+function defineLazy(obj, name, enumerable, make) {
+  Object.defineProperty(obj, name, {
+    configurable: true,
+    enumerable,
+    get() {
+      const value = make();
+      Object.defineProperty(obj, name, { value, writable: true, configurable: true, enumerable });
+      return value;
+    },
+    set(value) {
+      Object.defineProperty(obj, name, { value, writable: true, configurable: true, enumerable });
+    },
+  });
+}
+
 function setupWorkerStdio(stdio) {
+  const proc: any = process;
   const { stdin, stdout, stderr } = stdio;
-  if (stdout) {
-    Object.defineProperty(process, "stdout", {
-      value: makePortWritable(stdout),
-      writable: true,
-      configurable: true,
-      enumerable: true,
-    });
-  }
-  if (stderr) {
-    Object.defineProperty(process, "stderr", {
-      value: makePortWritable(stderr),
-      writable: true,
-      configurable: true,
-      enumerable: true,
-    });
-  }
+  // Shadow the native PropertyCallback slots via plain [[Set]] first: defineProperty
+  // on one reifies the process static table and invokes constructStdout/Stderr/Stdin,
+  // cold-loading node:stream just to build fd streams we then discard.
+  proc.stdin = undefined;
+  if (stdout) proc.stdout = undefined;
+  if (stderr) proc.stderr = undefined;
+  // Lazy: the streams graph + Console constructor are several ms cold, and most
+  // workers never touch process.stdout/stderr/stdin directly.
+  if (stdout) defineLazy(proc, "stdout", true, () => makePortWritable(stdout));
+  if (stderr) defineLazy(proc, "stderr", true, () => makePortWritable(stderr));
   // node always replaces a worker's process.stdin: port-backed when { stdin: true },
   // otherwise an immediately-EOF'd stream — never the process-wide fd 0, which
   // would race the main thread (and hang on a TTY).
-  Object.defineProperty(process, "stdin", {
-    value: stdin
+  defineLazy(proc, "stdin", true, () =>
+    stdin
       ? makePortReadable(stdin)
-      : new Readable({
+      : new (Readable ??= require("internal/streams/readable"))({
           read() {
             this.push(null);
           },
         }),
-    writable: true,
-    configurable: true,
-    enumerable: true,
-  });
+  );
   // node routes console.log through process.stdout/stderr; Bun's global console
   // writes the fd directly, so rebind it to the captured streams when present.
+  // Capture via require("node:console") first so the module registry caches the
+  // native object (which carries .Console/.write); otherwise a later user require
+  // would evaluate `export default console` through this getter and cache a bare instance.
   if (stdout || stderr) {
-    const { Console } = require("node:console");
-    globalThis.console = new Console(process.stdout, process.stderr);
+    const nativeConsole = require("node:console") as typeof globalThis.console & { Console: any };
+    defineLazy(globalThis, "console", false, () => new nativeConsole.Console(process.stdout, process.stderr));
   }
 }
 
@@ -1199,7 +1214,7 @@ class Worker extends EventEmitter {
 
   getHeapSnapshot(options: unknown) {
     const stringPromise = this.#worker.getHeapSnapshot(options);
-    return stringPromise.then(s => new HeapSnapshotStream(s));
+    return stringPromise.then(s => makeHeapSnapshotStream(s));
   }
 
   getHeapStatistics() {
@@ -1351,21 +1366,24 @@ class Worker extends EventEmitter {
   }
 }
 
-class HeapSnapshotStream extends Readable {
-  #json: string | undefined;
-
-  constructor(json: string) {
-    super();
-    this.#json = json;
-  }
-
-  _read() {
-    if (this.#json !== undefined) {
-      this.push(this.#json);
-      this.push(null);
-      this.#json = undefined;
+let _HeapSnapshotStream;
+function makeHeapSnapshotStream(json: string) {
+  Readable ??= require("internal/streams/readable");
+  _HeapSnapshotStream ??= class HeapSnapshotStream extends Readable {
+    #json: string | undefined;
+    constructor(json: string) {
+      super();
+      this.#json = json;
     }
-  }
+    _read() {
+      if (this.#json !== undefined) {
+        this.push(this.#json);
+        this.push(null);
+        this.#json = undefined;
+      }
+    }
+  };
+  return new _HeapSnapshotStream(json);
 }
 
 export default {

test/js/node/worker_threads/worker-stdio-lazy.test.ts

@@ -0,0 +1,63 @@
+import { expect, test } from "bun:test";
+import { once } from "node:events";
+import { Worker } from "node:worker_threads";
+
+// Regression: the stdio rebind used to run eagerly in every worker's preload,
+// cold-loading node:stream + Console and reifying the process static table on
+// every spawn even when the worker never touched stdio.
+test("worker process.stdio and console are installed as lazy accessors", async () => {
+  const worker = new Worker(
+    `
+    const { parentPort } = require("worker_threads");
+    const d = (obj, k) => {
+      const desc = Object.getOwnPropertyDescriptor(obj, k);
+      return desc ? (desc.get ? "accessor" : "value") : "none";
+    };
+    const before = {
+      stdout: d(process, "stdout"),
+      stderr: d(process, "stderr"),
+      stdin: d(process, "stdin"),
+      console: d(globalThis, "console"),
+    };
+    // first read materializes the stream; second read must be the same object
+    const s1 = process.stdout;
+    const s2 = process.stdout;
+    parentPort.postMessage({
+      before,
+      afterStdout: d(process, "stdout"),
+      sameInstance: s1 === s2,
+      isWritable: typeof s1.write === "function",
+    });
+    `,
+    { eval: true },
+  );
+  const [msg] = await once(worker, "message");
+  expect(msg).toEqual({
+    before: { stdout: "accessor", stderr: "accessor", stdin: "accessor", console: "accessor" },
+    afterStdout: "value",
+    sameInstance: true,
+    isWritable: true,
+  });
+  await worker.terminate();
+});
+
+// node:console is `export default globalThis.console`; evaluating it after the
+// lazy getter is installed would cache a plain Console instance that lacks
+// .Console/.write, so the preload primes the module registry first.
+test("require('node:console') in a worker still exposes Console and write", async () => {
+  const worker = new Worker(
+    `
+    const { parentPort } = require("worker_threads");
+    const c = require("node:console");
+    parentPort.postMessage({
+      hasConsoleCtor: typeof c.Console === "function",
+      hasWrite: typeof c.write === "function",
+      hasAsyncIterator: typeof c[Symbol.asyncIterator] === "function",
+    });
+    `,
+    { eval: true },
+  );
+  const [msg] = await once(worker, "message");
+  expect(msg).toEqual({ hasConsoleCtor: true, hasWrite: true, hasAsyncIterator: true });
+  await worker.terminate();
+});