Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 25 additions & 8 deletions src/dotenv/env_loader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -1289,16 +1289,35 @@ 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<Item = Entry<'_, Box<[u8]>, 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<NullDelimitedEnvMap, AllocError> {
let envp_count = self.map.count();
let mut storage: Vec<Box<[u8]>> = 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();
Expand All @@ -1309,7 +1328,7 @@ impl Map {
envp_buf.push(env_buf.as_ptr().cast::<c_char>());
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 {
Expand All @@ -1330,17 +1349,15 @@ 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;
}
}

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;
Expand Down
2 changes: 1 addition & 1 deletion src/dotenv/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
5 changes: 5 additions & 0 deletions src/jsc/VirtualMachine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down
86 changes: 86 additions & 0 deletions test/js/node/cluster.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, string>) {
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 },
});
});
Loading