diff --git a/src/dotenv/env_loader.rs b/src/dotenv/env_loader.rs index a0fcafe596df..ce89c4a8120a 100644 --- a/src/dotenv/env_loader.rs +++ b/src/dotenv/env_loader.rs @@ -4,7 +4,7 @@ use std::sync::OnceLock; use std::sync::atomic::{AtomicBool, AtomicPtr, Ordering}; use bun_alloc::AllocError; -use bun_collections::{ArrayHashMapExt, GetOrPutResult, StringSet}; +use bun_collections::{ArrayHashMapExt, Entry, GetOrPutResult, StringSet}; use bun_core::{self, Output}; use bun_core::{ZStr, strings}; use bun_paths::{self, MAX_PATH_BYTES, PathBuffer}; @@ -1289,7 +1289,27 @@ pub struct Map { pub map: HashTable, } +/// A child that inherits `NODE_UNIQUE_ID` from a node:cluster worker takes itself for a worker. +static OMIT_CLUSTER_UNIQUE_ID_FROM_CHILD_ENV: AtomicBool = AtomicBool::new(false); + +pub fn omit_cluster_unique_id_from_child_env() { + OMIT_CLUSTER_UNIQUE_ID_FROM_CHILD_ENV.store(true, Ordering::SeqCst); +} + impl Map { + /// The entries of the env of a child process. + fn child_env_entries(&mut self) -> impl Iterator, HashTableValue>> { + let omitted = if OMIT_CLUSTER_UNIQUE_ID_FROM_CHILD_ENV.load(Ordering::SeqCst) { + self.map.get_index(b"NODE_UNIQUE_ID") + } else { + None + }; + self.map + .iterator() + .enumerate() + .filter_map(move |(index, entry)| (omitted != Some(index)).then_some(entry)) + } + /// Builds a NULL-terminated `K=V\0` envp array. Returns an owning struct so /// dropping it frees the joined buffers (PORTING.md §Forbidden: no Box::leak). pub fn create_null_delimited_env_map(&mut self) -> Result { @@ -1297,8 +1317,7 @@ impl Map { let mut storage: Vec> = Vec::with_capacity(envp_count); let mut envp_buf: Vec<*const c_char> = Vec::with_capacity(envp_count + 1); { - let mut it = self.map.iterator(); - while let Some(pair) = it.next() { + for pair in self.child_env_entries() { let klen = pair.key_ptr.len(); let vlen = pair.value_ptr.value.len(); let mut env_buf = vec![0u8; klen + vlen + 2].into_boxed_slice(); @@ -1309,7 +1328,7 @@ impl Map { envp_buf.push(env_buf.as_ptr().cast::()); storage.push(env_buf); } - debug_assert!(envp_buf.len() == envp_count); + debug_assert!(envp_buf.len() <= envp_count); } envp_buf.push(core::ptr::null()); // sentinel Ok(NullDelimitedEnvMap { @@ -1330,8 +1349,7 @@ impl Map { // from 4 bytes), so the UTF-8 byte length is a safe upper bound. let mut capacity: usize = 4; { - let mut it = self.map.iterator(); - while let Some(pair) = it.next() { + for pair in self.child_env_entries() { capacity += pair.key_ptr.len() + 1 + pair.value_ptr.value.len() + 1; } } @@ -1339,8 +1357,7 @@ impl Map { let mut result = vec![0u16; capacity]; let mut i: usize = 0; { - let mut it = self.map.iterator(); - while let Some(pair) = it.next() { + for pair in self.child_env_entries() { i += strings::convert_utf8_to_utf16_in_buffer(&mut result[i..], pair.key_ptr).len(); result[i] = b'=' as u16; i += 1; diff --git a/src/dotenv/lib.rs b/src/dotenv/lib.rs index b99816e76d72..c77fc4842bd3 100644 --- a/src/dotenv/lib.rs +++ b/src/dotenv/lib.rs @@ -8,7 +8,7 @@ pub use error::{Error, Result}; pub use env_loader::{ DirEntryKeys, DotEnvBehavior, DotEnvFileSuffix, HAS_NO_CLEAR_SCREEN_CLI_FLAG, HashTable, HashTableValue, INSTANCE, Loader, Map, NullDelimitedEnvMap, S3Credentials, instance, - set_instance, + omit_cluster_unique_id_from_child_env, set_instance, }; /// `dotenv::map::{HashTable, Entry}` namespace expected by `install_jsc::ini_jsc` et al. diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index b4873eb211ef..784ee483873e 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -4294,6 +4294,11 @@ impl VirtualMachine { } } + // Not removed from the map: `process.env` and `Bun.serve` read it there. + if map.get(b"NODE_UNIQUE_ID").is_some() { + bun_dotenv::omit_cluster_unique_id_from_child_env(); + } + // Node.js checks if this is set to "1" and no other value if let Some(value) = map.get(b"NODE_PRESERVE_SYMLINKS") { self.transpiler.resolver.opts.preserve_symlinks = value == b"1"; diff --git a/test/js/node/cluster.test.ts b/test/js/node/cluster.test.ts index 3248646bd3b0..2624b22519f7 100644 --- a/test/js/node/cluster.test.ts +++ b/test/js/node/cluster.test.ts @@ -1388,3 +1388,89 @@ test.concurrent("require('cluster') does not throw when NODE_UNIQUE_ID is set af expect({ stdout, stderr }).toEqual({ stdout: "loaded\n", stderr: "" }); expect(exitCode).toBe(0); }); + +// A worker is a process that starts with NODE_UNIQUE_ID, which cluster.fork() sets. +test.concurrent("a process that a worker starts with Bun.spawn and no env option is not a worker", async () => { + // Bun.spawn with no `env` passes the startup env on, not process.env, where the worker setup deletes NODE_UNIQUE_ID. + using dir = tempDir("cluster-spawn-default-env", { + "worker.js": ` +const cmd = [process.execPath, "child.js"]; +const spawn = () => Bun.spawn({ cmd, stdout: "pipe", stderr: "inherit" }).stdout.text(); +const beforeWorkerSetup = spawn(); +require("node:cluster"); +const afterWorkerSetup = spawn(); +const thread = new Worker(require.resolve("./thread.js")); +const fromThread = new Promise((resolve, reject) => { + thread.onmessage = event => resolve(event.data); + thread.onerror = reject; +}); +const spawnSync = Bun.spawnSync({ cmd, stderr: "inherit" }).stdout.toString(); +Promise.all([beforeWorkerSetup, afterWorkerSetup, fromThread]).then(([before, after, inThread]) => { + console.log("Bun.spawn before the worker setup:", before.trim()); + console.log("Bun.spawn:", after.trim()); + console.log("Bun.spawnSync:", spawnSync.trim()); + console.log("Bun.spawnSync in a worker thread:", inThread.trim()); + process.exit(0); +}); +`, + "thread.js": ` +postMessage(Bun.spawnSync({ cmd: [process.execPath, "child.js"], stderr: "inherit" }).stdout.toString()); +`, + "child.js": ` +// Read first: a process that takes itself for a worker deletes the variable when node:cluster loads. +const uniqueId = process.env.NODE_UNIQUE_ID ?? null; +console.log(JSON.stringify({ uniqueId, isWorker: require("node:cluster").isWorker })); +`, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "worker.js"], + env: { ...bunEnv, NODE_UNIQUE_ID: "1" }, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const notAWorker = '{"uniqueId":null,"isWorker":false}'; + expect({ stdout, stderr, exitCode }).toEqual({ + stdout: + `Bun.spawn before the worker setup: ${notAWorker}\n` + + `Bun.spawn: ${notAWorker}\n` + + `Bun.spawnSync: ${notAWorker}\n` + + `Bun.spawnSync in a worker thread: ${notAWorker}\n`, + stderr: "", + exitCode: 0, + }); +}); + +test.concurrent.skipIf(!isLinux)("Bun.serve in a worker defaults reusePort to true", async () => { + // docs/guides/http/cluster.mdx: reusePort is Linux only. + await using server = Bun.serve({ port: 0, reusePort: true, fetch: () => new Response("ok") }); + using dir = tempDir("cluster-serve-reuse-port", { + "serve.js": ` +require("node:cluster"); +try { + Bun.serve({ port: Number(process.env.SERVE_PORT), fetch: () => new Response("ok") }); + console.log("listening"); +} catch (error) { + console.log(error.code ?? String(error)); +} +process.exit(0); +`, + }); + async function serve(env: Record) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "serve.js"], + env: { ...bunEnv, SERVE_PORT: String(server.port), ...env }, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout, stderr, exitCode }; + } + const [notAWorker, worker] = await Promise.all([serve({}), serve({ NODE_UNIQUE_ID: "1" })]); + expect({ notAWorker, worker }).toEqual({ + notAWorker: { stdout: "EADDRINUSE\n", stderr: "", exitCode: 0 }, + worker: { stdout: "listening\n", stderr: "", exitCode: 0 }, + }); +});