#[cfg(feature = "deno_core")] use std::time::Instant; use std::{ collections::{HashMap, HashSet}, fs, process::Stdio, sync::Arc, }; use base64::Engine; use itertools::Itertools; use serde_json::value::RawValue; use uuid::Uuid; use windmill_parser_ts::{remove_pinned_import_specifiers, remove_pinned_imports}; use windmill_queue::{append_logs, CanceledBy, MiniPulledJob, PrecomputedAgentInfo}; use crate::{ check_no_network_jailed, common::{ build_command_with_isolation, create_args_and_out_file, get_reserved_variables, parse_npm_config, read_file, read_file_content, read_result, resolve_nsjail_timeout, resolve_nsjail_tmp_mount_block, resolve_nsjail_tmpfs_size_bytes, script_envars, start_child_process, write_file_binary, MaybeLock, OccupancyMetrics, StreamNotifier, DEV_CONF_NSJAIL, }, get_proxy_envs_for_lang, get_run_proxy_envs, handle_child::handle_child, is_sandboxing_enabled, nsjail_clone_newnet, read_ee_registry_with_workspace_override, BUNFIG_INSTALL_SCOPES, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR, BUN_INSTALL_MIN_RELEASE_AGE, BUN_NO_CACHE, BUN_PATH, DISABLE_BUN_INSTALL_SANDBOX, DISABLE_NUSER, HOME_ENV, NODE_BIN_PATH, NODE_PATH, NPMRC, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_AVAILABLE, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, TRACING_PROXY_CA_CERT_PATH, TZ_ENV, }; use windmill_common::{ cache, client::AuthedClient, jobs::JobKind, min_version::MIN_VERSION_SUPPORTS_BUN_LOCKFILE_V2, scripts::{id_to_codebase_info, CodebaseInfo, ScriptHash, ScriptLang}, utils::WarnAfterExt, workspace_dependencies::{clean_lock_from_annotations, WorkspaceDependenciesPrefetched}, }; use windmill_types::s3::BundleFormat; #[cfg(windows)] use crate::SYSTEM_ROOT; use tokio::{fs::File, process::Command}; use tokio::io::AsyncReadExt; use windmill_common::{ error::{self, Result}, worker::{write_file, Connection, DISABLE_BUNDLING}, DB, }; use crate::global_cache::{exists_in_cache, save_cache}; #[cfg(all(feature = "enterprise", feature = "parquet"))] use windmill_object_store::attempt_fetch_bytes; use windmill_parser::Typ; // The Windows loader uses a virtual "windmill-url" namespace instead of writing .url // files to disk, which avoids Windows path issues. The virtual namespace approach is // likely better on all fronts but we keep the original .url-file loader on Linux to // avoid breaking back-compat. #[cfg(not(windows))] pub const RELATIVE_BUN_LOADER: &str = include_str!("../loader.bun.js"); #[cfg(windows)] pub const RELATIVE_BUN_LOADER: &str = include_str!("../loader.bun.windows.js"); pub const RELATIVE_BUN_BUILDER: &str = include_str!("../loader_builder.bun.js"); const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.config.proto"); const NSJAIL_CONFIG_INSTALL_BUN_CONTENT: &str = include_str!("../nsjail/install.bun.config.proto"); pub const BUN_LOCK_SPLIT: &str = "\n//bun.lock\n"; pub const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n"; pub const BUN_LOCK_SPLIT_WINDOWS: &str = "\r\n//bun.lock\r\n"; pub const BUN_LOCKB_SPLIT_WINDOWS: &str = "\r\n//bun.lockb\r\n"; pub const EMPTY_FILE: &str = ""; /// Bun args for dedicated worker (without the script path) pub const BUN_DEDICATED_WORKER_ARGS: &[&str] = &["run", "-i", "--prefer-offline"]; /// Pre-computed codegen data for a TypeScript/Bun/Deno script. /// Computed in Rust from the parsed signature, then baked into the wrapper template. #[cfg(any(feature = "private", test))] pub struct TsScriptCodegen { pub spread: String, pub date_conversions: String, pub preprocessor_spread: Option, pub preprocessor_date_conversions: Option, } /// Parse a TS script and compute the codegen data (arg spread, date conversions, preprocessor). /// This is the same logic that was used on main in `start_worker`. #[cfg(any(feature = "private", test))] pub fn compute_ts_codegen(content: &str) -> TsScriptCodegen { let sig = windmill_parser_ts::parse_deno_signature(content, true, false, None).unwrap_or_default(); let arg_names: Vec<&str> = sig.args.iter().map(|a| a.name.as_str()).collect(); let spread = arg_names.join(", "); let dates = sig .args .iter() .filter(|a| matches!(a.typ, Typ::Datetime)) .map(|a| { format!( "{name} = {name} ? new Date({name}) : undefined", name = a.name ) }) .join("\n "); let pre_sig = windmill_parser_ts::parse_deno_signature( content, true, false, Some("preprocessor".to_string()), ) .ok() .filter(|s| !s.args.is_empty()); let preprocessor_spread = pre_sig .as_ref() .map(|s| s.args.iter().map(|a| a.name.as_str()).join(", ")); let preprocessor_date_conversions = pre_sig.as_ref().map(|s| { s.args .iter() .filter(|a| matches!(a.typ, Typ::Datetime)) .map(|a| { format!( "{name} = {name} ? new Date({name}) : undefined", name = a.name ) }) .join("\n ") }); TsScriptCodegen { spread, date_conversions: dates, preprocessor_spread, preprocessor_date_conversions, } } /// Script entry for the unified wrapper generator. /// `import_name`: the file stem used in the import path (e.g., "main" → `./main.ts`, or "f__script" → `./f__script.ts`) #[cfg(any(feature = "private", test))] pub struct TsScriptEntry<'a> { pub import_name: &'a str, pub original_path: &'a str, pub codegen: &'a TsScriptCodegen, } /// Generate a wrapper for dedicated workers and runner groups. /// All scripts are baked in at codegen time with static imports and inline arg handling. /// Protocol: /// execd: -> execute the single registered script (non-runner-group) /// execd_preprocess: -> preprocess + execute the single registered script /// exec:: -> execute script by path (runner groups with multiple scripts) /// exec_preprocess:: -> preprocess + execute script by path /// end -> exit #[cfg(any(feature = "private", test))] pub fn generate_multi_script_wrapper(scripts: &[TsScriptEntry<'_>], ext: &str) -> String { let is_debug = std::env::var("RUST_LOG").is_ok_and(|x| x == "windmill=debug"); let print_lines = if is_debug { r#"console.log("[debug] " + line);"# } else { "" }; let imports: String = scripts .iter() .enumerate() .map(|(i, e)| { format!( "import * as _s{i} from \"./{import_name}.{ext}\";", import_name = e.import_name ) }) .collect::>() .join("\n"); // Generate per-script getArgs / getPreArgs functions let mut functions = String::new(); let mut registrations = String::new(); for (i, entry) in scripts.iter().enumerate() { let cg = entry.codegen; let spread = &cg.spread; let dates = &cg.date_conversions; functions.push_str(&format!( r#" function getArgs_{i}(line) {{ let {{ {spread} }} = JSON.parse(line); {dates} return [ {spread} ]; }} "# )); let pre_fn = if let Some(ref pre_spread) = cg.preprocessor_spread { let pre_dates = cg.preprocessor_date_conversions.as_deref().unwrap_or(""); functions.push_str(&format!( r#" function getPreArgs_{i}(line) {{ let {{ {pre_spread} }} = JSON.parse(line); {pre_dates} return [ {pre_spread} ]; }} "# )); format!("getPreArgs_{i}") } else { "null".to_string() }; registrations.push_str(&format!( "scripts.set(\"{path}\", {{ module: _s{i}, getArgs: getArgs_{i}, getPreArgs: {pre_fn} }});\n", path = entry.original_path, )); } format!( r#" {imports} import * as Readline from "node:readline" BigInt.prototype.toJSON = function () {{ return this.toString(); }}; const scripts = new Map(); {functions} {registrations} console.log('start'); for await (const line of Readline.createInterface({{ input: process.stdin }})) {{ {print_lines} if (line === "end") {{ process.exit(0); }} // Direct execution: single-script dedicated workers (no path needed) if (line.startsWith("execd_preprocess:")) {{ const argsJson = line.slice("execd_preprocess:".length); const entry = scripts.values().next().value; try {{ if (!entry.getPreArgs) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }})); continue; }} const preArgs = entry.getPreArgs(argsJson); const preprocessedArgs = await entry.module.preprocessor(...preArgs); console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value)); const mainArgs = entry.getArgs(JSON.stringify(preprocessedArgs ?? {{}})); const res = await entry.module.main(...mainArgs); console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); }} catch (e) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }})); }} continue; }} if (line.startsWith("execd:")) {{ const argsJson = line.slice("execd:".length); const entry = scripts.values().next().value; try {{ const args = entry.getArgs(argsJson); const res = await entry.module.main(...args); console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); }} catch (e) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }})); }} continue; }} // Path-based execution: runner groups with multiple scripts if (line.startsWith("exec_preprocess:")) {{ const rest = line.slice("exec_preprocess:".length); const colonIdx = rest.indexOf(":"); if (colonIdx === -1) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "Malformed exec_preprocess command: missing colon separator", name: "Error" }})); continue; }} const scriptPath = rest.slice(0, colonIdx); const argsJson = rest.slice(colonIdx + 1); const entry = scripts.get(scriptPath); if (!entry) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "Script not found: " + scriptPath, name: "Error" }})); continue; }} try {{ if (!entry.getPreArgs) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }})); continue; }} const preArgs = entry.getPreArgs(argsJson); const preprocessedArgs = await entry.module.preprocessor(...preArgs); console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value)); const mainArgs = entry.getArgs(JSON.stringify(preprocessedArgs ?? {{}})); const res = await entry.module.main(...mainArgs); console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); }} catch (e) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }})); }} continue; }} if (line.startsWith("exec:")) {{ const rest = line.slice("exec:".length); const colonIdx = rest.indexOf(":"); if (colonIdx === -1) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "Malformed exec command: missing colon separator", name: "Error" }})); continue; }} const scriptPath = rest.slice(0, colonIdx); const argsJson = rest.slice(colonIdx + 1); const entry = scripts.get(scriptPath); if (!entry) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "Script not found: " + scriptPath, name: "Error" }})); continue; }} try {{ const args = entry.getArgs(argsJson); const res = await entry.module.main(...args); console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); }} catch (e) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }})); }} continue; }} console.error("Unknown command:", line); }} "# ) } /// Returns (package.json, bun.lock(b), is_empty, is_binary) pub(crate) fn split_lockfile(lockfile: &str) -> (&str, Option<&str>, bool, bool) { if let Some(index) = lockfile.find(BUN_LOCK_SPLIT) { // Split using "\n//bun.lock\n" let (before, after_with_sep) = lockfile.split_at(index); let after = &after_with_sep[BUN_LOCK_SPLIT.len()..]; (before, Some(after), after == EMPTY_FILE, false) } else if let Some(index) = lockfile.find(BUN_LOCKB_SPLIT) { // Split using "\n//bun.lockb\n" let (before, after_with_sep) = lockfile.split_at(index); let after = &after_with_sep[BUN_LOCKB_SPLIT.len()..]; (before, Some(after), after == EMPTY_FILE, true) } else if let Some(index) = lockfile.find(BUN_LOCK_SPLIT_WINDOWS) { // Split using "\r\n//bun.lock\r\n" let (before, after_with_sep) = lockfile.split_at(index); let after = &after_with_sep[BUN_LOCK_SPLIT_WINDOWS.len()..]; (before, Some(after), after == EMPTY_FILE, false) } else if let Some(index) = lockfile.find(BUN_LOCKB_SPLIT_WINDOWS) { // Split using "\r\n//bun.lockb\r\n" let (before, after_with_sep) = lockfile.split_at(index); let after = &after_with_sep[BUN_LOCKB_SPLIT_WINDOWS.len()..]; (before, Some(after), after == EMPTY_FILE, true) } else { (lockfile, None, false, false) } } /// An empty `lockfileVersion: 1` lockfile, planted before `bun install` so bun writes v1. /// /// bun keeps whichever version the lockfile it found already had, and only raises it when the /// dependencies genuinely need newer syntax. Seeding therefore gets a real v1 lockfile — written /// by bun, not rewritten by us — whenever v1 can express the resolution, and lets bun escalate /// when it cannot. [`bun_lockfile_version`] catches the escalation afterwards. const EMPTY_V1_BUN_LOCK: &str = r#"{ "lockfileVersion": 1, "configVersion": 1, "workspaces": { "": {}, }, "packages": {} }"#; async fn seed_v1_bun_lockfile(job_dir: &str) -> Result<()> { let path = format!("{job_dir}/bun.lock"); if tokio::fs::metadata(&path).await.is_ok() { return Ok(()); } write_file(job_dir, "bun.lock", EMPTY_V1_BUN_LOCK)?; Ok(()) } /// The `lockfileVersion` a bun text lockfile declares, if it declares one. fn bun_lockfile_version(lockfile: &str) -> Option { let i = lockfile.find("\"lockfileVersion\"")?; let rest = &lockfile[i + "\"lockfileVersion\"".len()..]; let digits = rest .trim_start() .strip_prefix(':')? .trim_start() .chars() .take_while(char::is_ascii_digit) .collect::(); digits.parse().ok() } pub async fn gen_bun_lockfile( mem_peak: &mut i32, canceled_by: &mut Option, job_id: &Uuid, w_id: &str, db: Option<&Connection>, token: &str, script_path: &str, job_dir: &str, base_internal_url: &str, worker_name: &str, export_pkg: bool, workspace_dependencies: &WorkspaceDependenciesPrefetched, npm_mode: bool, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, temp_script_refs: &Option>, quiet: bool, ) -> Result> { let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(None).await; let mut empty_deps = false; if let Some(package_json_content) = workspace_dependencies.get_bun()? { gen_bunfig(job_dir, job_id, w_id, db).await?; write_file(job_dir, "package.json", package_json_content.as_str())?; } else { let temp_refs_json = temp_script_refs .as_ref() .and_then(|m| serde_json::to_string(m).ok()) .unwrap_or_else(|| "null".to_string()); let loader = RELATIVE_BUN_LOADER .replace("W_ID", w_id) .replace("BASE_INTERNAL_URL", base_internal_url) .replace("TOKEN", token) .replace( "CURRENT_PATH", &crate::common::use_flow_root_path(script_path), ) .replace("RAW_GET_ENDPOINT", "raw") .replace("TEMP_SCRIPT_REFS_PLACEHOLDER", &temp_refs_json); write_file( &job_dir, "build.js", &format!( r#" {loader} {RELATIVE_BUN_BUILDER} "# ), )?; gen_bunfig(job_dir, job_id, w_id, db).await?; let mut child_cmd = Command::new(&*BUN_PATH); child_cmd .current_dir(job_dir) .env_clear() .envs(common_bun_proc_envs.clone()) .args(vec!["run", "build.js"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); #[cfg(windows)] child_cmd.env("SystemRoot", SYSTEM_ROOT.as_str()); let child_process = start_child_process(child_cmd, &*BUN_PATH, false).await?; if let Some(db) = db { let mut quiet_buf = String::new(); let result = handle_child( job_id, db, mem_peak, canceled_by, child_process, false, worker_name, w_id, "bun build", None, false, occupancy_metrics, if quiet { Some(&mut quiet_buf) } else { None }, None, ) .await; if quiet && result.is_err() { append_logs( job_id, w_id, format!("\n--- BUN BUILD (failed) ---\n{quiet_buf}"), db, ) .await; } result?; } else { let output = Box::into_pin(child_process.wait_with_output()).await?; if !output.status.success() { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); return Err(error::Error::ExecutionErr(format!( "bun build exited with non-zero status: {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}", output.status ))); } } let new_package_json = read_file_content(&format!("{job_dir}/package.json")).await?; empty_deps = new_package_json == r#"{ "dependencies": {} }"#; } if !empty_deps { if !npm_mode && !MIN_VERSION_SUPPORTS_BUN_LOCKFILE_V2.met_conservatively() { seed_v1_bun_lockfile(job_dir).await?; } install_bun_lockfile( mem_peak, canceled_by, job_id, w_id, db, job_dir, worker_name, common_bun_proc_envs, npm_mode, occupancy_metrics, quiet, ) .await?; } else { if !quiet { if let Some(db) = db { append_logs(job_id, w_id, "\nempty dependencies, skipping install", db).await; } } } if export_pkg { let mut content; { let mut file = File::open(format!("{job_dir}/package.json")).await?; let mut buf = String::default(); file.read_to_string(&mut buf).await?; content = buf; } if !npm_mode { #[cfg(any(target_os = "linux", target_os = "macos"))] content.push_str(BUN_LOCK_SPLIT); #[cfg(target_os = "windows")] content.push_str(BUN_LOCK_SPLIT_WINDOWS); { let file = format!("{job_dir}/bun.lock"); if !empty_deps && tokio::fs::metadata(&file).await.is_ok() { let mut file = File::open(&file).await?; let mut buf = String::default(); file.read_to_string(&mut buf).await?; if !MIN_VERSION_SUPPORTS_BUN_LOCKFILE_V2.met_conservatively() { // Seeding asked bun for v1; a higher version back means these // dependencies cannot be expressed in one. Storing it anyway would not // fail on an older worker — it would install from package.json alone and // silently resolve different versions. match bun_lockfile_version(&buf) { Some(1) => {} Some(v) => { return Err(error::Error::ExecutionErr(format!( "bun produced a v{v} lockfile, which workers older than {} \ cannot read. Finish upgrading every worker before deploying \ dependencies that need it (overrides, catalogs).", MIN_VERSION_SUPPORTS_BUN_LOCKFILE_V2.version() ))); } // The guard cannot classify this one, so it must not pass silently: // a bun that stops writing the header would reopen the drift hole // with nothing in the logs. None => tracing::warn!( "bun wrote a lockfile with no readable lockfileVersion; storing \ it unchecked for job {job_id}" ), } } content.push_str(&buf); } else { content.push_str(&EMPTY_FILE); } } } Ok(Some(content)) } else { Ok(None) } } async fn gen_bunfig( job_dir: &str, job_id: &Uuid, w_id: &str, db: Option<&Connection>, ) -> Result<()> { let npmrc = if let Some(conn) = db { read_ee_registry_with_workspace_override( NPMRC.read().await.clone(), "npmrc", "npmrc", job_id, w_id, conn, ) .await } else { NPMRC.read().await.clone() }; if npmrc.as_ref().is_some_and(|c| !c.trim().is_empty()) { return write_bun_registry_config(job_dir, npmrc, None, None); } let (registry, bunfig_install_scopes) = if let Some(conn) = db { ( read_ee_registry_with_workspace_override( NPM_CONFIG_REGISTRY.read().await.clone(), "npm_config_registry", "npm registry", job_id, w_id, conn, ) .await, read_ee_registry_with_workspace_override( BUNFIG_INSTALL_SCOPES.read().await.clone(), "bunfig_install_scopes", "bunfig install scopes", job_id, w_id, conn, ) .await, ) } else { ( NPM_CONFIG_REGISTRY.read().await.clone(), BUNFIG_INSTALL_SCOPES.read().await.clone(), ) }; write_bun_registry_config(job_dir, None, registry, bunfig_install_scopes) } /// The files [`write_bun_registry_config`] may create in the directory bun installs from. /// Both can hold a registry auth token, so `prepare-deps` deletes them by these names once /// the install is over. pub(crate) const BUN_NPMRC_FILE: &str = ".npmrc"; pub(crate) const BUN_CONFIG_FILE: &str = "bunfig.toml"; /// Write the registry configuration `bun install` picks up from its working directory: the /// `npmrc` setting verbatim as `.npmrc` when set, otherwise a `bunfig.toml` holding the /// registry URL, its auth token and the install scopes. /// /// Shared with the debugger's `prepare-deps`, which resolves the same settings without a /// database (see `prepare_deps.rs`), so the two install paths configure bun identically. pub(crate) fn write_bun_registry_config( job_dir: &str, npmrc: Option, registry: Option, bunfig_install_scopes: Option, ) -> Result<()> { if let Some(ref npmrc_content) = npmrc { if !npmrc_content.trim().is_empty() { tracing::debug!("Writing .npmrc for bun from npmrc setting"); write_file(job_dir, BUN_NPMRC_FILE, npmrc_content)?; return Ok(()); } } if registry.is_some() || bunfig_install_scopes.is_some() { let (url, token_opt) = if let Some(ref s) = registry { let url = s.trim(); if url.is_empty() { ("https://registry.npmjs.org/".to_string(), None) } else { parse_npm_config(s) } } else { ("https://registry.npmjs.org/".to_string(), None) }; let registry_toml_string = if let Some(token) = token_opt { format!("{{ url = \"{url}\", token = \"{token}\" }}") } else { format!("\"{url}\"") }; let bunfig_toml = format!( r#" [install] registry = {} {} "#, registry_toml_string, bunfig_install_scopes .map(|x| format!("[install.scopes]\n{x}")) .unwrap_or("".to_string()) ); tracing::debug!("Writing following bunfig.toml: {bunfig_toml}"); let _ = write_file(&job_dir, BUN_CONFIG_FILE, &bunfig_toml)?; } Ok(()) } /// An nsjail bind mount of `path` onto itself, skipped when the host path is missing. /// Empty for a path that is not absolute or that the proto string cannot carry verbatim. fn optional_bind_mount(path: &str, rw: bool) -> String { if !path.starts_with('/') || path.contains(['"', '\\', '\n', '\r']) { return String::new(); } format!( "mount {{\n src: \"{path}\"\n dst: \"{path}\"\n is_bind: true\n rw: {rw}\n mandatory: false\n}}\n" ) } /// node-gyp's header cache under HOME. Kept across jailed installs like the package /// caches: without it every native build downloads the headers again, which a worker /// with no internet access cannot. const NODE_GYP_CACHE: &str = ".cache/node-gyp"; /// The install jail's HOME: a private tmpfs at the worker's HOME path, with the /// package manager's user config read-only and the caches kept there writable (npm's /// `~/.npm` in npm mode, node-gyp's headers in both). fn install_home_mount_block(home: &str, npm_mode: bool, tmpfs_size_bytes: &str) -> String { let home = home.trim_end_matches('/'); let mut block = String::new(); // A HOME under /tmp is already private: the jail's /tmp is. if home.starts_with('/') && home != "/tmp" && !home.starts_with("/tmp/") && !home.contains(['"', '\\', '\n', '\r']) { block.push_str(&format!( "mount {{\n dst: \"{home}\"\n fstype: \"tmpfs\"\n rw: true\n options: \"size={tmpfs_size_bytes}\"\n}}\n" )); } block.push_str(&optional_bind_mount(&format!("{home}/.npmrc"), false)); block.push_str(&optional_bind_mount(&format!("{home}/.bunfig.toml"), false)); block.push_str(&optional_bind_mount( &format!("{home}/{NODE_GYP_CACHE}"), true, )); if npm_mode { block.push_str(&optional_bind_mount(&format!("{home}/.npm"), true)); } block } /// An `.npmrc` value as npm reads it: `${VAR}` from the install env, and a leading `~/` /// as HOME. Without it, a `cafile=~/…` or `${HOME}/…` is never mounted and the jail's /// private HOME hides the certificate. fn expand_npmrc_value(value: &str, home: &str, envs: &HashMap) -> String { let mut out = String::new(); let mut rest = value; while let Some(start) = rest.find("${") { let Some(len) = rest[start..].find('}') else { break; }; out.push_str(&rest[..start]); let name = rest[start + 2..start + len].trim_end_matches('?'); out.push_str(envs.get(name).map(String::as_str).unwrap_or_default()); rest = &rest[start + len + 1..]; } out.push_str(rest); match out.strip_prefix("~/") { Some(p) => format!("{}/{p}", home.trim_end_matches('/')), None => out, } } /// TLS files the install is pointed at (CA bundles, registry client certificates) live /// outside the jail's mounts: bind them in, read-only, so private registries keep working. fn install_ca_mounts(job_dir: &str, home: &str, envs: &HashMap) -> String { const CA_ENVS: [&str; 4] = [ "NODE_EXTRA_CA_CERTS", "SSL_CERT_FILE", "SSL_CERT_DIR", "NPM_CONFIG_CAFILE", ]; // Also registry-scoped (`//host/:certfile=`), hence the match on the last segment. const TLS_FILE_KEYS: [&str; 3] = ["cafile", "certfile", "keyfile"]; let from_env = crate::worker::WHITELIST_ENVS .iter() .filter(|(k, _)| CA_ENVS.iter().any(|c| c.eq_ignore_ascii_case(k))) .map(|(_, v)| v.trim().to_string()); // `.npmrc` lines and bunfig's `[install] cafile = "…"` share the `key = value` shape. let from_config = [ format!("{job_dir}/{BUN_NPMRC_FILE}"), format!("{home}/.npmrc"), format!("{job_dir}/{BUN_CONFIG_FILE}"), format!("{home}/.bunfig.toml"), ] .into_iter() .filter_map(|f| fs::read_to_string(f).ok()) .flat_map(|content| { content .lines() .filter_map(|l| { let (k, v) = l.split_once('=')?; let key = k.trim().rsplit(':').next()?; TLS_FILE_KEYS .iter() .any(|t| t.eq_ignore_ascii_case(key)) .then(|| expand_npmrc_value(v.trim().trim_matches(['"', '\'']), home, envs)) }) .collect::>() }); from_env .chain(from_config) .filter(|p| !p.is_empty()) .unique() .map(|p| optional_bind_mount(&p, false)) .collect() } /// The files `bun install` reads from, and writes to, the directory it installs in. #[cfg(unix)] const BUN_INSTALL_INPUTS: [&str; 3] = ["package.json", "bun.lock", "bun.lockb"]; const BUN_INSTALL_OUTPUTS: [&str; 3] = ["node_modules", "bun.lock", "bun.lockb"]; /// What the worker reads or mounts from the job dir after an install that ran in it. const BUN_JOB_DIR_OUTPUTS: [&str; 5] = [ "node_modules", "package.json", "bun.lock", "bun.lockb", "package-lock.json", ]; /// Whether an install left `name` at `path`, refusing anything but a real directory /// (node_modules) or regular file there: the install jail could write it, and the worker /// follows what it then reads, renames or mounts. fn check_install_output(path: &str, name: &str) -> Result { let expect_dir = name == "node_modules"; match fs::symlink_metadata(path) { Ok(m) if (expect_dir && m.is_dir()) || (!expect_dir && m.is_file()) => Ok(true), Ok(_) => Err(error::Error::ExecutionErr(format!( "dependency install left an unexpected {name}: expected a regular {}", if expect_dir { "directory" } else { "file" } ))), Err(_) => Ok(false), } } /// Registry configuration bun reads from the install dir. It can hold an auth token, so /// a staged install gets it bind-mounted read-only from the job dir instead of a copy in /// the cache, which every later install jail mounts. const BUN_REGISTRY_FILES: [&str; 2] = [BUN_CONFIG_FILE, BUN_NPMRC_FILE]; /// Open a directory without following a symlink at `path`. #[cfg(unix)] fn open_dir_nofollow(path: &str) -> std::io::Result { use std::os::unix::fs::OpenOptionsExt; fs::OpenOptions::new() .read(true) .custom_flags(nix::libc::O_DIRECTORY | nix::libc::O_NOFOLLOW) .open(path) } /// Delete entry `name` of the directory open as `dir`, off the async worker thread: a /// staging dir can hold a whole node_modules. A symlink there is removed, never followed. #[cfg(unix)] fn remove_entry_detached(dir: Arc, name: String) { use std::os::fd::AsRawFd; let remove = move || { let path = format!("/proc/self/fd/{}/{name}", dir.as_raw_fd()); let _ = match fs::symlink_metadata(&path) { Ok(m) if m.is_dir() => fs::remove_dir_all(&path), Ok(_) => fs::remove_file(&path), Err(e) => Err(e), }; }; match tokio::runtime::Handle::try_current() { Ok(rt) => { rt.spawn_blocking(remove); } Err(_) => remove(), } } /// A directory for one jailed bun install inside the bun cache, removed when dropped. /// /// The jail binds the job dir and the cache as two mounts, and link(2) fails across /// mounts, so bun would copy every file out of the cache instead of hardlinking it. /// Installing here and renaming node_modules into the job dir keeps the hardlinks. /// /// Every install jail can write the cache, so any path through it may since have become a /// symlink. The worker reaches this dir only through descriptors opened without following /// links before the jail runs ([`Self::entry`]); `path` is for the jail config alone. struct BunStagingDir { #[cfg(unix)] root: Arc, #[cfg(unix)] dir: fs::File, #[cfg(unix)] name: String, path: String, } impl BunStagingDir { /// Entry `f` of the staging dir, resolved through its descriptor. fn entry(&self, f: &str) -> String { #[cfg(unix)] { use std::os::fd::AsRawFd; format!("/proc/self/fd/{}/{f}", self.dir.as_raw_fd()) } #[cfg(not(unix))] format!("{}/{f}", self.path) } } #[cfg(unix)] impl Drop for BunStagingDir { fn drop(&mut self) { remove_entry_detached(self.root.clone(), std::mem::take(&mut self.name)); } } /// How long a staging dir may outlive its install before the next one deletes it: a /// worker killed mid-install leaves it behind. Well past the longest install. #[cfg(unix)] const BUN_STAGING_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(6 * 60 * 60); /// Whether a bun install needs nothing of the job dir beyond [`BUN_INSTALL_INPUTS`]: no /// dependency by path, no workspaces or patches, and no root lifecycle script, which bun /// runs next to the manifest and which may read the job's other files. #[cfg(any(unix, test))] fn bun_install_can_stage(package_json: &str, lock: Option<&str>) -> bool { const DEP_SECTIONS: [&str; 6] = [ "dependencies", "devDependencies", "optionalDependencies", "peerDependencies", "overrides", "resolutions", ]; const ROOT_SCRIPTS: [&str; 6] = [ "preinstall", "install", "postinstall", "preprepare", "prepare", "postprepare", ]; const LOCAL_SPECS: [&str; 8] = [ "file:", "link:", "workspace:", "portal:", "./", "../", "/", "~/", ]; fn has_local_spec(v: &serde_json::Value) -> bool { match v { serde_json::Value::String(s) => LOCAL_SPECS.iter().any(|p| s.trim().starts_with(p)), serde_json::Value::Object(m) => m.values().any(has_local_spec), _ => false, } } let Ok(serde_json::Value::Object(pkg)) = serde_json::from_str(package_json) else { return false; }; let root_script = pkg .get("scripts") .and_then(|s| s.as_object()) .is_some_and(|s| ROOT_SCRIPTS.iter().any(|k| s.contains_key(*k))); if root_script || pkg.contains_key("workspaces") || pkg.contains_key("patchedDependencies") || DEP_SECTIONS .iter() .any(|k| pkg.get(*k).is_some_and(has_local_spec)) { return false; } // Every text lockfile has a root `"workspaces"` entry, so only these mark local // packages: as a spec (`"file:x"`) or in a resolution (`"x@file:x"`). const LOCK_LOCAL_PROTOCOLS: [&str; 4] = ["file:", "link:", "workspace:", "portal:"]; !lock.is_some_and(|l| { l.contains("\"patchedDependencies\"") || LOCK_LOCAL_PROTOCOLS .iter() .any(|p| l.contains(&format!("\"{p}")) || l.contains(&format!("@{p}"))) }) } /// None when node_modules could not be renamed from the cache into the job dir (another /// filesystem, or another mount of the same one): moving it back would be a copy, which /// is what installing in the job dir costs anyway. Also None when /// [`bun_install_can_stage`] says the install needs the job dir. fn bun_staging_dir(cache_dir: &str, job_dir: &str) -> Option { #[cfg(unix)] { use std::os::fd::AsRawFd; let package_json = fs::read_to_string(format!("{job_dir}/package.json")).ok()?; let lock = fs::read_to_string(format!("{job_dir}/bun.lock")).ok(); if !bun_install_can_stage(&package_json, lock.as_deref()) { return None; } // `cache_dir` is the jail's mount point, which no jail can replace; what is under // it can be anything. let root_path = format!("{cache_dir}/.wm-staging"); if fs::symlink_metadata(&root_path).is_ok_and(|m| !m.is_dir()) { fs::remove_file(&root_path).ok()?; } let _ = fs::create_dir(&root_path); let root = Arc::new(open_dir_nofollow(&root_path).ok()?); let root_fd = format!("/proc/self/fd/{}", root.as_raw_fd()); for e in fs::read_dir(&root_fd).ok()?.flatten() { let stale = e .metadata() .and_then(|m| m.modified()) .ok() .and_then(|t| t.elapsed().ok()) .is_some_and(|age| age > BUN_STAGING_MAX_AGE); if stale { remove_entry_detached(root.clone(), e.file_name().to_string_lossy().into_owned()); } } let name = Uuid::new_v4().to_string(); fs::create_dir(format!("{root_fd}/{name}")).ok()?; let dir = open_dir_nofollow(&format!("{root_fd}/{name}")).ok()?; let path = format!("{root_path}/{name}"); let staging = BunStagingDir { root, dir, name, path }; // A jail sharing the cache can list the new dir and plant a symlink in it before // the worker writes there: every file is created new, never opened as found. let create_new = |path: &str| { fs::OpenOptions::new() .write(true) .create_new(true) .open(path) }; // The move back is a rename, which is the only test that holds: two mounts of // one filesystem share st_dev and still refuse it. let probe = staging.entry(".rename-probe"); let probe_dst = format!("{job_dir}/.wm-staging-probe-{}", Uuid::new_v4()); create_new(&probe).ok()?; let renamed = fs::rename(&probe, &probe_dst).is_ok(); let _ = fs::remove_file(&probe_dst); if !renamed { return None; } for f in BUN_INSTALL_INPUTS { let src = format!("{job_dir}/{f}"); if fs::symlink_metadata(&src).is_ok_and(|m| m.is_file()) { let mut out = create_new(&staging.entry(f)).ok()?; std::io::copy(&mut fs::File::open(&src).ok()?, &mut out).ok()?; } } Some(staging) } #[cfg(not(unix))] { let _ = (cache_dir, job_dir); None } } /// A fetch npm 12 refuses by default but npm 11 made. Windmill still allows it, for now, when /// the install's config does not set the key either way. #[derive(Clone, Copy)] enum NpmFetchFallback { Git, Remote, } impl NpmFetchFallback { fn key(self) -> &'static str { match self { Self::Git => "allow-git", Self::Remote => "allow-remote", } } fn error_code(self) -> &'static str { match self { Self::Git => "EALLOWGIT", Self::Remote => "EALLOWREMOTE", } } fn what(self) -> &'static str { match self { Self::Git => "git dependencies", Self::Remote => "dependency tarballs hosted outside the configured registry", } } } /// The job-dir folder npm writes its debug log to (`--logs-dir`). The fallback reads a refusal /// from there, so the install's output keeps streaming to the job log instead of being held. const NPM_LOGS_DIR: &str = ".wm-npm-logs"; const NPM_LOG_READ_LIMIT: u64 = 16 * 1024 * 1024; const NPM_REFUSED_SPECS_SHOWN: usize = 20; /// The fetch a failed `npm install` was refused, read from the debug log npm wrote to /// `logs_dir`. Called once the install has exited, so no jailed process can still swap these /// paths; a symlink is never followed. fn npm_refused_fetch(logs_dir: &str) -> Option<(NpmFetchFallback, Vec)> { use std::io::{BufRead, Read}; if !fs::symlink_metadata(logs_dir).ok()?.is_dir() { return None; } let log = fs::read_dir(logs_dir) .ok()? .filter_map(|e| e.ok()) .filter(|e| e.file_type().is_ok_and(|t| t.is_file())) .map(|e| e.path()) .filter(|p| p.extension().is_some_and(|x| x == "log")) .max()?; let log = fs::File::open(log).ok()?.take(NPM_LOG_READ_LIMIT); parse_npm_refused_fetch(std::io::BufReader::new(log).lines().map_while(|l| l.ok())) } /// npm debug log lines read ` error `. fn parse_npm_refused_fetch( lines: impl Iterator, ) -> Option<(NpmFetchFallback, Vec)> { let mut fallback = None; let mut refused = vec![]; for line in lines { let Some(msg) = line .split_once(' ') .and_then(|(_, m)| m.strip_prefix("error ")) else { continue; }; if let Some(code) = msg.strip_prefix("code ") { fallback = fallback.or([NpmFetchFallback::Git, NpmFetchFallback::Remote] .into_iter() .find(|f| f.error_code() == code.trim())); } else if let Some(spec) = msg.strip_prefix("Refusing to fetch ") { if refused.len() < NPM_REFUSED_SPECS_SHOWN { refused.push(spec.trim().to_string()); } } } fallback.map(|f| (f, refused)) } /// Whether npm config sets `key` explicitly, from any source npm reads (global, user, project, /// env, including files redirected by `npm_config_userconfig` / `globalconfig`): `npm config /// ls` lists only those. A check that fails counts as set, so it never relaxes the policy. async fn npm_config_sets(key: &str, job_dir: &str, envs: &HashMap) -> bool { let output = Command::new(&*NPM_PATH) .current_dir(job_dir) .env_clear() .envs(envs) // The config can set `json` or `long`: JSON output hides an explicit value from // `npm_config_ls_sets`, and `long` also lists npm's defaults. .args(["config", "ls", "--json=false", "--long=false"]) .stdin(Stdio::null()) .kill_on_drop(true) .output(); match tokio::time::timeout(std::time::Duration::from_secs(30), output).await { Ok(Ok(o)) if o.status.success() => { npm_config_ls_sets(&String::from_utf8_lossy(&o.stdout), key) } _ => true, } } /// An overridden value is listed commented out (`; key = "v" ; overridden by …`): still set. fn npm_config_ls_sets(config_ls: &str, key: &str) -> bool { let prefix = format!("{key} = "); config_ls .lines() .any(|l| l.trim_start_matches("; ").starts_with(&prefix)) } pub async fn install_bun_lockfile( mem_peak: &mut i32, canceled_by: &mut Option, job_id: &Uuid, w_id: &str, db: Option<&Connection>, job_dir: &str, worker_name: &str, common_bun_proc_envs: HashMap, npm_mode: bool, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, quiet: bool, ) -> Result<()> { let package_manager = if npm_mode { &*NPM_PATH } else { &*BUN_PATH }; // The install runs the lifecycle scripts of dependencies (npm >= 12: those its allowScripts // policy covers; bun: its default trusted list and `trustedDependencies`), so when // sandboxing is on it is confined under nsjail like the run step. The env is the same // either way. let sandboxed = is_sandboxing_enabled() && !*DISABLE_BUN_INSTALL_SANDBOX; if is_sandboxing_enabled() && *DISABLE_BUN_INSTALL_SANDBOX { static WARNED: std::sync::Once = std::sync::Once::new(); WARNED.call_once(|| { tracing::warn!( "DISABLE_BUN_INSTALL_SANDBOX is set: Bun and npm dependency installs run \ outside nsjail, with their lifecycle scripts unconfined" ) }); } let home = common_bun_proc_envs .get("HOME") .cloned() .unwrap_or_else(|| HOME_ENV.clone()); let npm_logs_dir = format!("{job_dir}/{NPM_LOGS_DIR}"); let npm_logs_dir_flag = format!("--logs-dir={npm_logs_dir}"); let mut args = vec!["install"]; // A bun flag: npm 12 rejects unknown cli flags instead of warning. if !npm_mode { args.push("--save-text-lockfile"); } else { args.push(&npm_logs_dir_flag); } let no_cache = !npm_mode && *BUN_NO_CACHE; if no_cache { args.push("--no-cache"); } let mut registry_env = None; let mut npm_logs = if npm_mode { "NPM mode\n".to_string() } else if no_cache { "Bun install with --no-cache flag (BUN_NO_CACHE=true)\n".to_string() } else { "".to_string() }; let has_file = if npm_mode { let npmrc = if let Some(conn) = db { read_ee_registry_with_workspace_override( NPMRC.read().await.clone(), "npmrc", "npmrc", job_id, w_id, conn, ) .await } else { NPMRC.read().await.clone() }; if let Some(ref npmrc_content) = npmrc { if !npmrc_content.trim().is_empty() { npm_logs.push_str("Using .npmrc from instance settings\n"); write_file(job_dir, ".npmrc", npmrc_content)?; true } else { false } } else { let registry = if let Some(conn) = db { read_ee_registry_with_workspace_override( NPM_CONFIG_REGISTRY.read().await.clone(), "npm_config_registry", "npm registry", job_id, w_id, conn, ) .await } else { NPM_CONFIG_REGISTRY.read().await.clone() }; if let Some(registry) = registry { let content = registry .trim_start_matches("https:") .trim_start_matches("http:"); let mut splitted = registry.split(":_authToken="); let custom_registry = splitted.next().unwrap_or_default(); npm_logs.push_str(&format!( "Using custom npm registry: {custom_registry} {}\n", if splitted.next().is_some() { "with authToken" } else { "without authToken" } )); registry_env = Some(custom_registry.to_string()); write_file(job_dir, ".npmrc", content)?; true } else { false } } } else { false }; if !quiet && (npm_mode || no_cache) { if let Some(db) = db { append_logs(&job_id.clone(), w_id, npm_logs, db).await; } } if !has_file { gen_bunfig(job_dir, job_id, w_id, db).await?; } let make_cmd = |extra_args: &[String]| { let mut cmd = if sandboxed { let mut cmd = Command::new(NSJAIL_PATH.as_str()); cmd.args(["--config", "install.config.proto", "--", package_manager]); cmd } else { Command::new(package_manager) }; cmd.current_dir(job_dir) .env_clear() .envs(PROXY_ENVS.clone()) .envs(&common_bun_proc_envs) .envs(&*crate::worker::WHITELIST_ENVS); if let Some(registry) = registry_env.as_ref() { cmd.env("NPM_CONFIG_REGISTRY", registry); } cmd.args(&args) .args(extra_args) .stdout(Stdio::piped()) .stderr(Stdio::piped()); #[cfg(windows)] cmd.env("SystemRoot", SYSTEM_ROOT.as_str()); cmd }; let install_envs: HashMap = make_cmd(&[]) .as_std() .get_envs() .filter_map(|(k, v)| Some((k.to_str()?.to_string(), v?.to_str()?.to_string()))) .collect(); let mut staging = None; if sandboxed { // Bind-mounted into the jail, so they must exist before nsjail builds the mount // tree. let _ = std::fs::create_dir_all(format!("{home}/{NODE_GYP_CACHE}")); if npm_mode { let _ = std::fs::create_dir_all(format!("{home}/.npm")); } else { std::fs::create_dir_all(&*BUN_CACHE_DIR)?; staging = bun_staging_dir(&BUN_CACHE_DIR, job_dir); } let install_dir = staging.as_ref().map_or(job_dir, |s| s.path.as_str()); let bind = |src: &str, dst: &str, rw: bool| { format!( "mount {{\n src: \"{src}\"\n dst: \"{dst}\"\n is_bind: true\n rw: {rw}\n}}\n" ) }; // npm never reads the bun cache, and its install scripts get no write to it. let mut install_mounts = if npm_mode { String::new() } else { bind(&BUN_CACHE_DIR, &BUN_CACHE_DIR, true) }; match staging.as_ref() { // After the cache bind: these land inside it. Some(s) => { for f in BUN_REGISTRY_FILES { let src = format!("{job_dir}/{f}"); if fs::symlink_metadata(&src).is_ok_and(|m| m.is_file()) { install_mounts.push_str(&bind(&src, &format!("{}/{f}", s.path), false)); } } } None => install_mounts.push_str(&bind(job_dir, job_dir, true)), } let nsjail_timeout = match db { Some(conn) => resolve_nsjail_timeout(conn, w_id, *job_id, None).await, // no job to bound it: the caller waits for the install unbounded too None => "0".to_string(), }; write_file( job_dir, "install.config.proto", &NSJAIL_CONFIG_INSTALL_BUN_CONTENT .replace("{TIMEOUT}", &nsjail_timeout) .replace("{INSTALL_MOUNTS}", &install_mounts) .replace("{INSTALL_DIR}", install_dir) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace( "{HOME_MOUNT_BLOCK}", &install_home_mount_block( &home, npm_mode, &resolve_nsjail_tmpfs_size_bytes().await, ), ) .replace("{TRACING_PROXY_CA_CERT_PATH}", &*TRACING_PROXY_CA_CERT_PATH) .replace( "{CA_MOUNT}", &install_ca_mounts(job_dir, &home, &install_envs), ) .replace("#{DEV}", DEV_CONF_NSJAIL) .replace( "{TMP_MOUNT_BLOCK}", &resolve_nsjail_tmp_mount_block(job_dir).await, ), )?; } let child_executable = if sandboxed { NSJAIL_PATH.as_str() } else { package_manager.as_str() }; let mut fallback_flags: Vec = vec![]; let result = loop { if npm_mode { // Only this attempt's debug log may decide the fallback. let _ = tokio::fs::remove_dir_all(&npm_logs_dir).await; } let child_process = start_child_process(make_cmd(&fallback_flags), child_executable, false).await?; let result = if let Some(db) = db { let mut quiet_buf = String::new(); let result = handle_child( job_id, db, mem_peak, canceled_by, child_process, // jailed when sandboxing is on, so mem sampling finds the real process sandboxed, worker_name, w_id, "bun install", None, false, occupancy_metrics, if quiet { Some(&mut quiet_buf) } else { None }, None, ) .warn_after_seconds_for(10, "bun install") .await; if quiet && result.is_err() { // On failure, flush suppressed install output so the user can diagnose append_logs( job_id, w_id, format!("\n--- BUN INSTALL (failed) ---\n{quiet_buf}"), db, ) .await; } result.map(|_| ()) } else { let output = Box::into_pin(child_process.wait_with_output()).await?; if output.status.success() { Ok(()) } else { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); Err(error::Error::ExecutionErr(format!( "bun install exited with non-zero status: {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}", output.status ))) } }; if npm_mode && result.is_err() { if let Some((fallback, refused)) = npm_refused_fetch(&npm_logs_dir) { let flag = format!("--{}=all", fallback.key()); if !fallback_flags.contains(&flag) && !npm_config_sets(fallback.key(), job_dir, &install_envs).await { let key = fallback.key(); let msg = format!( "\nWARNING: npm refused {}:\n {}\nRetrying with {key}=all, which Windmill \ sets by default for now. This default will be removed in a future \ release: to keep fetching them, set `{key}=all` in the npm config (the \ npmrc instance or workspace setting on Enterprise Edition, or the \ worker's ~/.npmrc); set `{key}=none` to stop now.\n", fallback.what(), refused.join("\n ") ); tracing::warn!(%job_id, w_id, "{}", msg.trim()); if let Some(db) = db { append_logs(job_id, w_id, msg, db).await; } fallback_flags.push(flag); continue; } } } break result; }; if npm_mode { let _ = tokio::fs::remove_dir_all(&npm_logs_dir).await; } result?; if let Some(staging) = staging { // Leave the job dir as an install in place would have: what bun wrote or kept // replaces the job dir's copy, what it removed is removed there too. for f in BUN_INSTALL_OUTPUTS { let src = staging.entry(f); let dst = format!("{job_dir}/{f}"); match fs::symlink_metadata(&dst) { Ok(m) if m.is_dir() => tokio::fs::remove_dir_all(&dst).await?, Ok(_) => tokio::fs::remove_file(&dst).await?, Err(_) => {} } if fs::symlink_metadata(&src).is_ok() { fs::rename(&src, &dst)?; // Checked once out of the cache, where no other jail can swap it. if let Err(e) = check_install_output(&dst, f) { let _ = fs::remove_file(&dst); return Err(e); } } } } else if sandboxed { // The jail could write the job dir: what the worker reads or mounts from it next // must still be what an install leaves there. for f in BUN_JOB_DIR_OUTPUTS { check_install_output(&format!("{job_dir}/{f}"), f)?; } } if has_file { tokio::fs::remove_file(format!("{job_dir}/.npmrc")).await?; } Ok(()) } #[derive(PartialEq)] pub enum LoaderMode { Node, Bun, BunBundle, NodeBundle, BrowserBundle, } pub async fn build_loader( job_dir: &str, base_internal_url: &str, token: &str, w_id: &str, current_path: &str, mode: LoaderMode, temp_script_refs: &Option>, ) -> Result<()> { // Use forward slashes in JS strings to avoid backslash escape issues on Windows let job_dir_js = job_dir.replace('\\', "/"); let temp_refs_json = temp_script_refs .as_ref() .and_then(|m| serde_json::to_string(m).ok()) .unwrap_or_else(|| "null".to_string()); let loader = RELATIVE_BUN_LOADER .replace("W_ID", w_id) .replace("BASE_INTERNAL_URL", base_internal_url) .replace("TOKEN", token) .replace( "CURRENT_PATH", &crate::common::use_flow_root_path(current_path), ) .replace("RAW_GET_ENDPOINT", "raw_unpinned") .replace("TEMP_SCRIPT_REFS_PLACEHOLDER", &temp_refs_json); if mode == LoaderMode::Node { write_file( &job_dir, "node_builder.ts", &format!( r#" {loader} import {{ readdir }} from "node:fs/promises"; let fileNames = [] try {{ fileNames = await readdir("{job_dir_js}/node_modules") }} catch (e) {{ }} let result; try {{ result = await Bun.build({{ entrypoints: ["{job_dir_js}/wrapper.mjs"], outdir: "./", target: "node", plugins: [p], external: fileNames, minify: true, }}); }} catch(err) {{ console.log(err); console.log("Failed to build node bundle"); process.exit(1); }} if (!result?.success || !(result.outputs?.length > 0)) {{ for (const log of result?.logs ?? []) console.log(log); console.log("Failed to build node bundle: success=" + result?.success + ", outputs=" + (result?.outputs?.length ?? 0)); process.exit(1); }} "# ), )?; } else if mode == LoaderMode::Bun { write_file( &job_dir, "loader.bun.js", &format!( r#" import {{ plugin }} from "bun"; {loader} plugin(p) "# ), )?; } else if mode == LoaderMode::BunBundle || mode == LoaderMode::NodeBundle || mode == LoaderMode::BrowserBundle { write_file( &job_dir, "node_builder.ts", &format!( r#" {loader} let result; try {{ result = await Bun.build({{ entrypoints: ["{job_dir_js}/main.ts"], outdir: "./", target: "{}", plugins: [p], external: ["electron"], minify: {{ identifiers: false, syntax: true, whitespace: false }}, }}); }} catch(err) {{ console.log(err) console.log("Failed to build node bundle"); process.exit(1); }} if (!result?.success || !(result.outputs?.length > 0)) {{ for (const log of result?.logs ?? []) console.log(log); console.log("Failed to build node bundle: success=" + result?.success + ", outputs=" + (result?.outputs?.length ?? 0)); process.exit(1); }} "#, if mode == LoaderMode::BunBundle { "bun" } else if mode == LoaderMode::NodeBundle { "node" } else { "browser" } ), )?; } Ok(()) } pub async fn generate_wrapper_mjs( job_dir: &str, w_id: &str, job_id: &Uuid, worker_name: &str, db: &Connection, timeout: Option, mem_peak: &mut i32, canceled_by: &mut Option, common_bun_proc_envs: &HashMap, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> Result<()> { let mut child = Command::new(&*BUN_PATH); child .current_dir(job_dir) .env_clear() .envs(common_bun_proc_envs.clone()) .env("PATH", PATH_ENV.as_str()) .args(vec!["run", "node_builder.ts"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); #[cfg(windows)] child.env("SystemRoot", SYSTEM_ROOT.as_str()); let child_process = start_child_process(child, &*BUN_PATH, false).await?; handle_child( job_id, db, mem_peak, canceled_by, child_process, false, worker_name, w_id, "bun build", timeout, false, occupancy_metrics, None, None, ) .await?; fs::rename( format!("{job_dir}/wrapper.js"), format!("{job_dir}/wrapper.mjs"), ) .map_err(|e| error::Error::internal_err(format!("Could not move wrapper to mjs: {e:#}")))?; Ok(()) } pub async fn generate_bun_bundle( job_dir: &str, w_id: &str, job_id: &Uuid, worker_name: &str, db: Option<&Connection>, timeout: Option, mem_peak: &mut i32, canceled_by: &mut Option, common_bun_proc_envs: &HashMap, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> Result<()> { let mut child = Command::new(&*BUN_PATH); child .current_dir(job_dir) .env_clear() .envs(common_bun_proc_envs.clone()) .env("PATH", PATH_ENV.as_str()) .args(vec!["run", "node_builder.ts"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); #[cfg(windows)] child.env("SystemRoot", SYSTEM_ROOT.as_str()); let child_process = start_child_process(child, &*BUN_PATH, false).await?; if let Some(db) = db { handle_child( job_id, &db, mem_peak, canceled_by, child_process, false, worker_name, w_id, "bun build", timeout, false, occupancy_metrics, None, None, ) .warn_after_seconds_for(60, "bun build") .await?; } else { let output = Box::into_pin(child_process.wait_with_output()) .warn_after_seconds_for(60, "bun build") .await?; if !output.status.success() { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); return Err(error::Error::ExecutionErr(format!( "bun build exited with non-zero status: {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}", output.status ))); } } Ok(()) } /// [`generate_bun_bundle`], built once more with the version pins dropped from the import /// specifiers of `main.ts` if it fails. The lockfile pins those versions, but bun fails on a /// pinned specifier except where it tolerates a failed import (in a `try`, under a `.catch`, in /// dead code). Such a script builds as written and must keep that bundle, so only failures retry. async fn generate_bun_bundle_unpinning_imports( job_dir: &str, w_id: &str, job_id: &Uuid, worker_name: &str, db: Option<&Connection>, timeout: Option, mem_peak: &mut i32, canceled_by: &mut Option, common_bun_proc_envs: &HashMap, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> Result<()> { let built = generate_bun_bundle( job_dir, w_id, job_id, worker_name, db, timeout, mem_peak, canceled_by, common_bun_proc_envs, occupancy_metrics, ) .await; // Without a job, a failed build comes back as an `ExecutionErr`; with one, that variant is a // cancellation or timeout, which must not be retried. let build_failed = match &built { Err(error::Error::ExitStatus(..)) => true, Err(_) => db.is_none(), Ok(()) => false, }; if !build_failed { return built; } let Some(unpinned) = read_file_content(&format!("{job_dir}/main.ts")) .await .ok() .and_then(|main| { remove_pinned_import_specifiers(&main) .ok() .filter(|u| *u != main) }) else { return built; }; write_file(job_dir, "main.ts", &unpinned)?; if let Some(db) = db { append_logs( job_id, w_id, "\nbundling again with the imports' versions taken from the lockfile\n", db, ) .await; } generate_bun_bundle( job_dir, w_id, job_id, worker_name, db, timeout, mem_peak, canceled_by, common_bun_proc_envs, occupancy_metrics, ) .await } struct PulledCodebase { is_esm: bool, } async fn pull_codebase(w_id: &str, id: &str, job_dir: &str) -> Result { let path = windmill_object_store::bundle(&w_id, &id); let CodebaseInfo { is_tar, is_esm } = id_to_codebase_info(id); let bun_cache_path = format!( "{}/{}.{}", *windmill_common::worker::ROOT_CACHE_NOMOUNT_DIR, path, if is_tar { "tar" } else { "js" } ); let dst = format!( "{job_dir}/{}", if is_tar { "codebase.tar" } else { "main.js" } ); if std::fs::metadata(&bun_cache_path).is_ok() { tracing::info!("loading {bun_cache_path} from cache"); extract_saved_codebase(job_dir, &bun_cache_path, is_tar, &dst, false)?; } else { #[cfg(all(feature = "enterprise", feature = "parquet"))] let object_store = windmill_object_store::get_object_store().await; #[cfg(not(all(feature = "enterprise", feature = "parquet")))] let object_store: Option<()> = None; if &windmill_common::utils::MODE_AND_ADDONS.mode == &windmill_common::utils::Mode::Standalone && object_store.is_none() { let bun_cache_path = format!( "{}/{}", *windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR, path ); if std::fs::metadata(&bun_cache_path).is_ok() { tracing::info!("loading {bun_cache_path} from standalone bundle cache"); extract_saved_codebase(job_dir, &bun_cache_path, is_tar, &dst, true)?; } else { return Err(error::Error::ExecutionErr(format!( "(standalone bundle test mode) could not find codebase at {bun_cache_path}" ))); } } else { #[cfg(not(all(feature = "enterprise", feature = "parquet")))] return Err(error::Error::ExecutionErr( "codebase is an EE feature".to_string(), )); #[cfg(all(feature = "enterprise", feature = "parquet"))] if let Some(os) = object_store { let dirs_splitted = bun_cache_path.split("/").collect_vec(); std::fs::create_dir_all(dirs_splitted[..dirs_splitted.len() - 1].join("/"))?; let bytes = crate::global_cache::bounded_cache_io( "downloading", &path, attempt_fetch_bytes(os, &path), ) .await?; tracing::info!("loading {bun_cache_path} from object store"); windmill_common::worker::atomic_write_file_bytes( &bun_cache_path, bytes.as_ref(), false, )?; extract_saved_codebase(job_dir, &bun_cache_path, is_tar, &dst, false)?; } } } Ok(PulledCodebase { is_esm }) } fn extract_saved_codebase( job_dir: &str, bun_cache_path: &String, is_tar: bool, dst: &str, copy: bool, ) -> Result<()> { use crate::global_cache::extract_tar; Ok(if is_tar { extract_tar(fs::read(bun_cache_path)?.into(), job_dir)?; } else { if copy { std::fs::copy(bun_cache_path, dst)?; } else { #[cfg(unix)] std::os::unix::fs::symlink(bun_cache_path, dst)?; #[cfg(windows)] std::os::windows::fs::symlink_dir(bun_cache_path, dst)?; } }) } pub async fn prebundle_bun_script( inner_content: &str, lock: &str, script_path: &str, job_id: &Uuid, w_id: &str, db: Option<&DB>, job_dir: &str, base_internal_url: &str, worker_name: &str, token: &str, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, temp_script_refs: &Option>, modules: Option<&HashMap>, ) -> Result<()> { let (local_path, remote_path) = compute_bundle_local_and_remote_path( inner_content, lock, script_path, db, w_id, temp_script_refs, modules, ) .await; if exists_in_cache(&local_path, &remote_path).await { return Ok(()); } let annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content); if annotation.nobundling || *DISABLE_BUNDLING { return Ok(()); } let origin = format!("{job_dir}/main.js"); let mut content = remove_pinned_imports(inner_content)?; if crate::wac_executor::is_wac_v2_ts(inner_content) { content = crate::wac_executor::inject_wac_task_names(&content); content = format!("export {{ WorkflowCtx, StepSuspend, setWorkflowCtx }} from \"windmill-client\";\n{content}"); } write_file(job_dir, "main.ts", &content)?; // Remove any stale main.js so we never confuse a leftover (e.g. unbundled TS source // a caller dropped at this path) with a fresh Bun bundle output. let _ = std::fs::remove_file(&origin); build_loader( job_dir, base_internal_url, &token, w_id, script_path, if annotation.nodejs { LoaderMode::NodeBundle } else if annotation.native { LoaderMode::BrowserBundle } else { LoaderMode::BunBundle }, temp_script_refs, ) .await?; let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(None).await; generate_bun_bundle_unpinning_imports( job_dir, w_id, job_id, worker_name, db.map(|x| Connection::from(x.clone())).as_ref(), None, &mut 0, &mut None, &common_bun_proc_envs, occupancy_metrics, ) .await?; ensure_bundle_output_exists(&origin)?; save_cache(&local_path, &remote_path, &origin, false) .warn_after_seconds_for(60, "bundle cache save") .await?; Ok(()) } /// Refuse to cache a bundle if `Bun.build` finished without producing the /// expected output file. Belt-and-suspenders for any silent-failure mode the /// upstream wait-status / `result.success` checks don't already trip on. pub fn ensure_bundle_output_exists(bundle_path: &str) -> Result<()> { if !std::path::Path::new(bundle_path).exists() { return Err(error::Error::ExecutionErr(format!( "bun bundle output missing at {bundle_path} after Bun.build — refusing to cache" ))); } Ok(()) } pub const BUN_BUNDLE_OBJECT_STORE_PREFIX: &str = "bun_bundle/"; // A script version's relative-import list never changes (content is immutable // per hash), so parses are memoized without any invalidation. lazy_static::lazy_static! { static ref RELATIVE_IMPORTS_PER_HASH: quick_cache::sync::Cache>> = quick_cache::sync::Cache::new(1000); } const MAX_TRANSITIVE_IMPORT_PATHS: usize = 256; /// `(path, latest hash)` for the whole transitive closure of relative imports /// of `inner_content` — the set of scripts whose code gets inlined into the /// bundle, so all of them must key the bundle cache. Resolution goes through /// `IMPORTED_SCRIPT_HASH_CACHE` (notify-evicted, 60s TTL fallback, same /// version-selection predicate as the loader's content endpoint) and the /// per-hash script/parse caches, so steady state costs no DB queries. async fn collect_transitive_import_versions( db: &DB, w_id: &str, script_path: &str, inner_content: &str, ) -> Vec<(String, i64)> { let conn = Connection::from(db.clone()); let mut queue = crate::worker_lockfiles::extract_relative_imports( inner_content, script_path, &Some(ScriptLang::Bun), ) .unwrap_or_default(); let mut visited: HashSet = HashSet::new(); let mut versions: Vec<(String, i64)> = vec![]; while let Some(path) = queue.pop() { if !visited.insert(path.clone()) { continue; } if visited.len() > MAX_TRANSITIVE_IMPORT_PATHS { tracing::warn!( "transitive relative-import closure of {script_path} exceeds \ {MAX_TRANSITIVE_IMPORT_PATHS} scripts; bundle cache key covers only the first \ {MAX_TRANSITIVE_IMPORT_PATHS}" ); break; } let hash = match windmill_common::get_latest_script_hash_for_import_cached(db, w_id, &path) .await { Ok(Some(hash)) => hash, // Not a deployed script at this path (deleted, or not a script): // excluded from the key, matching what the bundler can inline. Ok(None) => continue, Err(e) => { tracing::warn!( "could not resolve import {path} while computing bundle cache key for \ {script_path}: {e:#}" ); continue; } }; versions.push((path.clone(), hash)); let imports = match RELATIVE_IMPORTS_PER_HASH.get(&hash) { Some(imports) => imports, None => match cache::script::fetch(&conn, ScriptHash(hash)).await { Ok((data, meta)) => { let imports = Arc::new(match meta.language { Some(ScriptLang::Bun) | Some(ScriptLang::Bunnative) | Some(ScriptLang::Deno) => { crate::worker_lockfiles::extract_relative_imports( &data.code, &path, &meta.language, ) .unwrap_or_default() } _ => vec![], }); RELATIVE_IMPORTS_PER_HASH.insert(hash, imports.clone()); imports } // A fetch error is transient, not a property of the (immutable) // content — memoizing it would drop this subtree from the key // until worker restart. Skip caching and retry next run. Err(e) => { tracing::warn!( "could not fetch import {path} (hash {hash}) while computing bundle \ cache key for {script_path}: {e:#}" ); Arc::new(vec![]) } }, }; queue.extend(imports.iter().cloned()); } // deterministic key regardless of traversal order versions.sort(); versions } pub async fn compute_bundle_local_and_remote_path( inner_content: &str, lock: &str, script_path: &str, db: Option<&DB>, w_id: &str, temp_script_refs: &Option>, modules: Option<&HashMap>, ) -> (String, String) { let mut input_src = format!("{inner_content}{lock}",); if let Some(db) = db { // The bundle inlines the whole transitive relative-import closure, so a // new deployed version of ANY script in it must change the key. for (path, hash) in collect_transitive_import_versions(db, w_id, script_path, inner_content).await { input_src.push_str(&path); input_src.push_str(&hash.to_string()); } // Modules are inlined too, and so is every deployed script they import: those are // reached through the module files, never through the entrypoint's own imports. let dir = script_path .rsplit_once('/') .map(|(dir, _)| dir) .unwrap_or(""); let mut module_entries = modules.into_iter().flatten().collect_vec(); module_entries.sort_by(|a, b| a.0.cmp(b.0)); for (key, module) in module_entries { let Some(stem) = key.strip_suffix(".ts") else { continue; }; let module_path = format!("{dir}/{stem}"); for (path, hash) in collect_transitive_import_versions(db, w_id, &module_path, &module.content).await { input_src.push_str(&path); input_src.push_str(&hash.to_string()); } } }; // Keep temp-script-ref (preview) bundles in a distinct cache slot: their // imports come from not-yet-deployed local content, so they must neither // reuse a deployed-content bundle nor be saved under the deployed key. if let Some(refs) = temp_script_refs { let mut entries: Vec<(&String, &String)> = refs.iter().collect(); entries.sort(); for (path, hash) in entries { input_src.push_str(path); input_src.push_str(hash); } } let ws_suffix = crate::workspace_registry_cache_suffix(w_id).await; input_src.push_str(&ws_suffix); // The loader resolves relative imports against the module files in the job dir, so // their content is inlined into the bundle this name covers. let hash = crate::worker::artifact_cache_name(input_src, modules); let local_path = format!("{}/{hash}", *BUN_BUNDLE_CACHE_DIR); #[cfg(windows)] let local_path = local_path.replace("/tmp", r"C:\tmp").replace("/", r"\"); let remote_path = format!("{BUN_BUNDLE_OBJECT_STORE_PREFIX}{hash}"); (local_path, remote_path) } pub async fn prepare_job_dir(reqs: &str, job_dir: &str) -> Result<()> { let (pkg, lock, empty, is_binary) = split_lockfile(reqs); let _ = write_file(job_dir, "package.json", pkg)?; if !empty { if let Some(lock) = lock { let _ = write_lock(lock, job_dir, is_binary).await?; } } Ok(()) } async fn write_lock(splitted_lockb_2: &str, job_dir: &str, is_binary: bool) -> Result<()> { if is_binary { write_file_binary( job_dir, "bun.lockb", &base64::engine::general_purpose::STANDARD .decode(splitted_lockb_2) .map_err(|_| error::Error::internal_err(format!("Could not decode bun.lockb")))?, ) .await?; } else { write_file(job_dir, "bun.lock", splitted_lockb_2)?; }; Ok(()) } #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_bun_job( maybe_lock: MaybeLock, codebase: Option<&String>, mem_peak: &mut i32, canceled_by: &mut Option, job: &MiniPulledJob, conn: &Connection, client: &AuthedClient, parent_runnable_path: Option, job_dir: &str, inner_content: &String, base_internal_url: &str, worker_name: &str, envs: HashMap, shared_mount: &str, new_args: &mut Option>>, occupancy_metrics: &mut OccupancyMetrics, precomputed_agent_info: Option, has_stream: &mut bool, modules: &Option>, ) -> error::Result> { let mut annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content); // Preview jobs may carry _TEMP_SCRIPT_REFS so relative imports resolve from // not-yet-deployed local content uploaded to raw_script_temp. Extracted up // front so it reaches both lockfile generation and the runtime loader. // Gated on JobKind::Preview because job.args includes caller-controlled // request args; honoring this key on deployed runs would let a caller swap // import resolution targets in deployed code. let temp_script_refs: Option> = if matches!(job.kind, JobKind::Preview) { job.args .as_ref() .and_then(|x| x.get("_TEMP_SCRIPT_REFS")) .and_then(|v| serde_json::from_str(v.get()).ok()) } else { None }; if annotation.sandbox && NSJAIL_AVAILABLE.is_none() { return Err(error::Error::ExecutionErr( "Script has //sandbox annotation but nsjail is not available on this worker. \ Please ensure nsjail is installed or remove the //sandbox annotation." .to_string(), )); } let jailed = is_sandboxing_enabled() || annotation.sandbox; // Native scripts take their network away inside the isolate instead. let jail_no_network = annotation.no_network && !annotation.native; check_no_network_jailed(jail_no_network, jailed)?; // Unbundled, Bun's run-time loader fetches workspace imports; Node's are bundled on the host. if jail_no_network && codebase.is_none() && !annotation.nodejs && (annotation.nobundling || *DISABLE_BUNDLING) { let script_path = crate::common::use_flow_root_path(job.runnable_path()); let script_dir = script_path .rsplit_once('/') .map(|(dir, _)| dir) .unwrap_or(""); let imports = std::iter::once((script_path.clone(), inner_content.as_str())) .chain(modules.iter().flatten().map(|(key, module)| { let stem = key.rsplit_once('.').map(|(stem, _)| stem).unwrap_or(key); (format!("{script_dir}/{stem}"), module.content.as_str()) })) .flat_map(|(path, code)| { windmill_parser_ts::parse_runtime_relative_imports(code, &path).unwrap_or_default() }); crate::check_no_network_imports( imports, &script_path, modules.as_ref(), "Bun without bundling (`//nobundling` or `DISABLE_BUNDLING`)", )?; } let (mut has_bundle_cache, cache_logs, local_path, remote_path) = if let (Some(lock), true) = ( maybe_lock.get_lock(), !annotation.nobundling && !*DISABLE_BUNDLING && codebase.is_none(), ) { let (local_path, remote_path) = match conn { Connection::Sql(db) => { compute_bundle_local_and_remote_path( inner_content, lock, job.runnable_path(), Some(db), &job.workspace_id, &temp_script_refs, modules.as_ref(), ) .await } Connection::Http(_) => { let (local_path, remote_path) = match precomputed_agent_info { Some(PrecomputedAgentInfo::Bun { local, remote }) => (local, remote), _ => { return Err(error::Error::ExecutionErr( "bun bundle is missing the precomputed agent info".to_string(), )) } }; (local_path, remote_path) } }; let (cache, logs) = crate::global_cache::load_cache(&local_path, &remote_path, false) .warn_after_seconds_for(60, "bundle cache load") .await; (cache, logs, local_path, remote_path) } else { (false, "".to_string(), "".to_string(), "".to_string()) }; if !codebase.is_some() && !has_bundle_cache { let _ = write_file(job_dir, "main.ts", inner_content)?; } else if !annotation.native && codebase.is_none() { let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#)?; } else if codebase.is_some() { // Write a valid fallback package.json for codebase mode. Without this, // nsjail creates an empty 0-byte file (from the mandatory: false mount) // which Node.js fails to parse as JSON (ERR_INVALID_PACKAGE_CONFIG). // If the codebase TAR includes a package.json, it will overwrite this. let _ = write_file(job_dir, "package.json", "{}")?; }; let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(Some(&base_internal_url)).await; let main_override = job.script_entrypoint_override.as_deref(); let apply_preprocessor = job.flow_step_id.as_deref() != Some("preprocessor") && job.preprocessed == Some(false); let is_wac_v2 = main_override.is_none() && crate::wac_executor::is_wac_v2_ts(inner_content); // Detect WAC v2 replay (resumed from suspend) to suppress verbose logs. // The actual step name is logged later by handle_wac_v2_output. let wac_replay_info: Option = if is_wac_v2 { if let Connection::Sql(db) = conn { let checkpoint = crate::wac_executor::load_checkpoint(db, &job.id).await?; if !checkpoint.completed_steps.is_empty() { Some(String::new()) } else { None } } else { None } } else { None }; // For WAC v2, inject variable names into unnamed task() calls so the // runtime can use them for step naming (timeline, graph). // `const double = task(async ...` → `const double = task("double", async ...` // Also handles: export const, let, var, and optional generic type parameters. // Skips calls that already have a string argument: `task("path", async ...` let inner_content = if is_wac_v2 { crate::wac_executor::inject_wac_task_names(inner_content) } else { inner_content.to_string() }; let inner_content = inner_content.as_str(); let mut format = BundleFormat::Cjs; if has_bundle_cache { let target; let symlink; #[cfg(unix)] { target = format!("{job_dir}/main.js"); symlink = std::os::unix::fs::symlink(&local_path, &target); } #[cfg(windows)] { target = format!("{job_dir}\\main.js"); symlink = std::fs::hard_link(&local_path, &target); } symlink.map_err(|e| { error::Error::ExecutionErr(format!( "could not copy cached binary from {local_path} to {job_dir}/main: {e:?}" )) })?; } else if let Some(codebase) = codebase.as_ref() { let pulled_codebase = pull_codebase(&job.workspace_id, codebase, job_dir).await?; if pulled_codebase.is_esm { format = BundleFormat::Esm; } } else { match &maybe_lock { MaybeLock::Resolved { lock } => { let (package_json, bun_lock, empty, is_binary) = split_lockfile(lock); if bun_lock.is_none() && !annotation.npm { return Err(error::Error::ExecutionErr( format!("Invalid requirements, expected to find //bun.lock{} split pattern in reqs. Found: |{lock}|", if is_binary {"b"} else {""}) )); } if annotation.npm { // The stored lock leads with `// workspace-dependencies` header lines: bun // reads package.json leniently, npm refuses it as invalid JSON. write_file( job_dir, "package.json", &clean_lock_from_annotations(package_json, ScriptLang::Bun), )?; } else { write_file(job_dir, "package.json", package_json)?; } let bun_lock = if annotation.npm { "" } else { bun_lock.unwrap() }; if !empty { if !annotation.npm { write_lock(bun_lock, job_dir, is_binary).await?; } install_bun_lockfile( mem_peak, canceled_by, &job.id, &job.workspace_id, Some(conn), job_dir, worker_name, common_bun_proc_envs.clone(), annotation.npm, &mut Some(occupancy_metrics), wac_replay_info.is_some(), ) .await?; } } MaybeLock::Unresolved { ref workspace_dependencies } => { // if is_sandboxing_enabled() || !empty_trusted_deps || has_custom_config_registry { if wac_replay_info.is_none() { let logs1 = "\n\n--- BUN INSTALL ---\n".to_string(); append_logs(&job.id, &job.workspace_id, logs1, conn).await; } gen_bun_lockfile( mem_peak, canceled_by, &job.id, &job.workspace_id, Some(conn), &client.token, job.runnable_path(), job_dir, base_internal_url, worker_name, false, workspace_dependencies, annotation.npm, &mut Some(occupancy_metrics), &temp_script_refs, wac_replay_info.is_some(), ) .await?; // } } } } if codebase.is_some() && format == BundleFormat::Cjs { annotation.nodejs = true } let mut init_logs = if let Some(ref replay_header) = wac_replay_info { // WAC v2 replay: use concise header, but still write main.ts if needed if !annotation.native && !has_bundle_cache && codebase.is_none() { write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?)?; } replay_header.clone() } else if annotation.native { "\n\n--- NATIVE CODE EXECUTION ---\n".to_string() } else if has_bundle_cache { if annotation.nodejs { "\n\n--- NODE BUNDLE SNAPSHOT EXECUTION ---\n".to_string() } else { "\n\n--- BUN BUNDLE SNAPSHOT EXECUTION ---\n".to_string() } } else if codebase.is_some() { if format == BundleFormat::Esm { "\n\n--- ESM CODEBASE SNAPSHOT EXECUTION ---\n".to_string() } else { "\n\n--- CJS CODEBASE SNAPSHOT EXECUTION ---\n".to_string() } } else if annotation.native { "\n\n--- NATIVE CODE EXECUTION ---\n".to_string() } else if annotation.nodejs { write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?)?; "\n\n--- NODE CODE EXECUTION ---\n".to_string() } else { write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?)?; // Module inlining has two phases: // 1. BUILD phase: loader.bun.js checks for local module files on disk (written by // write_module_files) and resolves them directly, so they get inlined into the bundle. // 2. RUN phase: overwrite main.ts with the bundled output below. The runtime wrapper // imports main.ts, which now contains the inlined modules from the build step. if modules.as_ref().is_some_and(|m| !m.is_empty()) { let bundle_path = std::path::Path::new(job_dir).join("out").join("main.js"); if bundle_path.exists() { // The lock-generation build kept every `pkg@version` specifier, and bun resolves // a pinned specifier outside node_modules, loading a second copy of the package. // The bundle holds the user's code too, so only the specifiers are rewritten, and // a bundle the parser rejects still runs as built, pins and all. let bundled = std::fs::read_to_string(&bundle_path)?; let unpinned = remove_pinned_import_specifiers(&bundled).unwrap_or_else(|e| { tracing::warn!( job_id = %job.id, "could not unpin the modules bundle, running it as built: {e:#}" ); bundled }); write_file(job_dir, "main.ts", &unpinned)?; } } "\n\n--- BUN CODE EXECUTION ---\n".to_string() }; if has_bundle_cache { init_logs = format!("\n{}{}", cache_logs, init_logs); } if annotation.sandbox { init_logs.push_str("sandbox mode (nsjail)\n"); } if jail_no_network { init_logs.push_str("no network\n"); } let write_wrapper_f = async { if !has_bundle_cache && annotation.native { return Ok(()) as error::Result<()>; } // let mut start = Instant::now(); let args = if is_wac_v2 { // For WAC v2, try to parse "main" args; if that fails, try the default export windmill_parser_ts::parse_deno_signature(inner_content, true, false, None) .unwrap_or_default() .args } else { windmill_parser_ts::parse_deno_signature( inner_content, true, false, main_override.map(ToString::to_string), )? .args }; let pre_args = if apply_preprocessor { Some( windmill_parser_ts::parse_deno_signature( inner_content, true, false, Some("preprocessor".to_string()), )? .args, ) } else { None }; let dates = pre_args .as_ref() .unwrap_or(&args) .iter() .filter_map(|x| { if matches!(x.typ, Typ::Datetime) { Some(x.name.as_str()) } else { None } }) .map(|x| { return format!(r#"args["{x}"] = args["{x}"] ? new Date(args["{x}"]) : undefined"#); }) .join("\n "); let spread = args.into_iter().map(|x| x.name).join(","); // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); // we cannot use Bun.read and Bun.write because it results in an EBADF error on cloud let main_name = main_override.unwrap_or("main"); // For WAC child jobs where the parser can't find params (task-wrapped consts), // fall back to passing arg values directly (filtering out internal fields) let child_spread = if spread.is_empty() && main_override.is_some() { "Object.values(Object.fromEntries(Object.entries(args).filter(([k]) => !k.startsWith('_'))))".to_string() } else { "argsObjToArr(args)".to_string() }; let main_import = if codebase.is_some() || has_bundle_cache { "./main.js" } else { "./main.ts" }; let wac_client_import = if has_bundle_cache { "./main.js" } else { "windmill-client" }; let preprocessor = if let Some(pre_args) = pre_args { let pre_spread = pre_args.into_iter().map(|x| x.name).join(","); format!( r#"if (Main.preprocessor === undefined || typeof Main.preprocessor !== 'function') {{ throw new Error("preprocessor function is missing"); }} function preArgsObjToArr({{ {pre_spread} }}) {{ return [ {pre_spread} ]; }} args = await Main.preprocessor(...preArgsObjToArr(args)); const args_json = JSON.stringify(args ?? null, (key, value) => typeof value === 'undefined' ? null : value); await fs.writeFile('args.json', args_json, {{ encoding: 'utf8' }})"# ) } else { "".to_string() }; let wac_spread = if spread.is_empty() { "Object.values(args)".to_string() } else { format!("argsObjToArr(args)") }; // Kept comment-free — this string is written out per job. // `_takePendingStepFailure` / `_takePendingSuspend` hand back what the body // caught and swallowed; honour them instead of reporting a bare `complete` // (see client.ts). Optional: npm clients may predate them. // `_warnUnobservedTaskFailures` reports what the body never awaited, and so // belongs only on the paths that end the round for good. A round that // dispatches, sleeps, checkpoints or waits for approval replays later and // re-registers the same failures from the checkpoint — keep those quiet. let wrapper_content = if is_wac_v2 { format!( r#" import * as Main from "{main_import}"; import {{ WorkflowCtx, StepSuspend, setWorkflowCtx }} from "{wac_client_import}"; import * as fs from "fs/promises"; let args = await fs.readFile('args.json', {{ encoding: 'utf8' }}).then(JSON.parse); const checkpoint = JSON.parse(await fs.readFile('checkpoint.json', {{ encoding: 'utf8' }})); function argsObjToArr({{ {spread} }}) {{ return [ {spread} ]; }} BigInt.prototype.toJSON = function () {{ return this.toString(); }}; // Find the workflow entrypoint (export default) let workflowFn = Main.default; if (!workflowFn || !workflowFn._is_workflow) {{ for (const key of Object.keys(Main)) {{ if (Main[key]?._is_workflow) {{ workflowFn = Main[key]; break; }} }} }} if (!workflowFn) {{ throw new Error("No workflow() entrypoint found. Wrap your main function with workflow()."); }} async function run() {{ {dates} {preprocessor} const argsArr = {wac_spread}; const ctx = new WorkflowCtx(checkpoint); setWorkflowCtx(ctx); try {{ const result = await workflowFn(...argsArr); setWorkflowCtx(null); const failed = ctx._takePendingStepFailure?.(); if (failed) {{ throw failed.error; }} const swallowed = ctx._takePendingSuspend?.(); if (swallowed) {{ throw swallowed; }} // Flush any unawaited tasks (e.g. forgotten await on last statement) const trailing = ctx._flushPending(); if (trailing.length > 0) {{ return {{ type: "dispatch", mode: trailing.length > 1 ? "parallel" : "sequential", steps: trailing }}; }} ctx._warnUnobservedTaskFailures?.(); return {{ type: "complete", result: result ?? null }}; }} catch (e) {{ setWorkflowCtx(null); if (e?.name === "StepSuspend" || e instanceof StepSuspend) {{ const dispatch = e.dispatchInfo ?? e.dispatch_info ?? {{}}; if (dispatch.mode === "step_complete") {{ return {{ type: "complete", result: dispatch.result ?? null }}; }} if (dispatch.mode === "inline_checkpoint") {{ return {{ type: "inline_checkpoint", key: dispatch.key, result: dispatch.result ?? null, started_at: dispatch.started_at, duration_ms: dispatch.duration_ms }}; }} if (dispatch.mode === "approval") {{ return {{ type: "approval", key: dispatch.key, timeout: dispatch.timeout, form: dispatch.form, self_approval_disabled: dispatch.self_approval_disabled, skin: dispatch.skin, description: dispatch.description }}; }} if (dispatch.mode === "sleep") {{ return {{ type: "sleep", key: dispatch.key, seconds: dispatch.seconds }}; }} return {{ type: "dispatch", mode: dispatch.mode ?? "sequential", steps: dispatch.steps ?? [] }}; }} ctx._warnUnobservedTaskFailures?.(); const failed = ctx._takePendingStepFailure?.(); if (failed) {{ throw failed.error; }} throw e; }} }} try {{ const output = await run(); if (output.type === "complete") {{ console.log(`\n--- WAC: complete ---`); }} const output_json = JSON.stringify(output, (key, value) => typeof value === 'undefined' ? null : value ); await fs.writeFile("result.json", output_json); process.exit(0); }} catch(e) {{ console.error(e); let err = {{ message: e.message, name: e.name, stack: e.stack }}; let step_id = process.env.WM_FLOW_STEP_ID; if (step_id) {{ err["step_id"] = step_id; }} const extra = {{}}; Object.getOwnPropertyNames(e).forEach((key) => {{ if (['line', 'name', 'stack', 'column', 'message', 'sourceURL', 'originalLine', 'originalColumn'].includes(key)) {{ return; }} extra[key] = e[key]; }}); if (Object.keys(extra).length > 0) {{ err["extra"] = extra; }} await fs.writeFile("result.json", JSON.stringify(err)); process.exit(1); }} "#, ) } else { format!( r#" import * as Main from "{main_import}"; import * as fs from "fs/promises"; let args = await fs.readFile('args.json', {{ encoding: 'utf8' }}).then(JSON.parse); function argsObjToArr({{ {spread} }}) {{ return [ {spread} ]; }} function isAsyncIterable(obj) {{ return obj != null && typeof obj[Symbol.asyncIterator] === 'function'; }} BigInt.prototype.toJSON = function () {{ return this.toString(); }}; async function run() {{ {dates} {preprocessor} // If the entrypoint has no parsed params (spread is empty), pass values directly // This handles WAC child jobs where tasks are const-wrapped functions const argsArr = {child_spread}; if (Main.{main_name} === undefined || typeof Main.{main_name} !== 'function') {{ throw new Error("{main_name} function is missing"); }} let entrypoint = Main.{main_name}; let res = await entrypoint(...argsArr); if (isAsyncIterable(res)) {{ for await (const chunk of res) {{ console.log("WM_STREAM: " + chunk.replace(/\n/g, '\\n')); }} res = null; }} const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); await fs.writeFile("result.json", res_json); process.exit(0); }} try {{ await run(); }} catch(e) {{ console.error(e); let err = {{ message: e.message, name: e.name, stack: e.stack }}; let step_id = process.env.WM_FLOW_STEP_ID; if (step_id) {{ err["step_id"] = step_id; }} const extra = {{}}; Object.getOwnPropertyNames(e).forEach((key) => {{ if (['line', 'name', 'stack', 'column', 'message', 'sourceURL', 'originalLine', 'originalColumn'].includes(key)) {{ return; }} extra[key] = e[key]; }}); if (Object.keys(extra).length > 0) {{ err["extra"] = extra; }} await fs.writeFile("result.json", JSON.stringify(err)); process.exit(1); }} "#, ) }; write_file(job_dir, "wrapper.mjs", &wrapper_content)?; Ok(()) as error::Result<()> }; let reserved_variables_args_out_f = async { let args_and_out_f = async { if !annotation.native { create_args_and_out_file(&client, job, job_dir, conn).await?; } Ok(()) as Result<()> }; let reserved_variables_f = async { let vars = get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; Ok(vars) as Result> }; let (_, reserved_variables) = tokio::try_join!(args_and_out_f, reserved_variables_f)?; Ok(reserved_variables) as error::Result> }; let build_cache = !has_bundle_cache && !annotation.nobundling && !*DISABLE_BUNDLING && !codebase.is_some() // A `no_network` run step cannot fetch relative imports, so they resolve while bundling. && (maybe_lock.get_lock().is_some() || annotation.native || jail_no_network); let write_loader_f = async { if build_cache { build_loader( job_dir, base_internal_url, &client.token, &job.workspace_id, job.runnable_path(), if annotation.nodejs { LoaderMode::NodeBundle } else if annotation.native { LoaderMode::BrowserBundle } else { LoaderMode::BunBundle }, &temp_script_refs, ) .await?; Ok(()) } else if !codebase.is_some() && !has_bundle_cache { build_loader( job_dir, base_internal_url, // Bun's run step reads this loader (Node's builds on the host), and could hand // the token out through its result. if jail_no_network && !annotation.nodejs { "" } else { &client.token }, &job.workspace_id, job.runnable_path(), if annotation.nodejs { LoaderMode::Node } else { LoaderMode::Bun }, &temp_script_refs, ) .await } else { Ok(()) } }; let (mut reserved_variables, _, _) = tokio::try_join!( reserved_variables_args_out_f, write_wrapper_f, write_loader_f )?; crate::strip_no_network_token(&mut reserved_variables, jail_no_network); // For WAC v2, write checkpoint.json before bun runs if is_wac_v2 { if let Connection::Sql(db) = conn { let checkpoint = crate::wac_executor::load_checkpoint(db, &job.id).await?; let checkpoint = crate::wac_executor::prepare_checkpoint_for_resume(db, &job.id, checkpoint).await?; let checkpoint_json = serde_json::to_string(&checkpoint).map_err(|e| { error::Error::internal_err(format!("Failed to serialize checkpoint: {e}")) })?; write_file(job_dir, "checkpoint.json", &checkpoint_json)?; } else { write_file(job_dir, "checkpoint.json", r#"{"completed_steps":{}}"#)?; } } // Prepend WAC re-exports to main.ts so the bundle includes WorkflowCtx etc. if build_cache && is_wac_v2 { let main_path = format!("{job_dir}/main.ts"); let current = read_file_content(&main_path).await?; write_file( job_dir, "main.ts", &format!("export {{ WorkflowCtx, StepSuspend, setWorkflowCtx }} from \"windmill-client\";\n{current}"), )?; } if !codebase.is_some() && !has_bundle_cache { if build_cache { generate_bun_bundle_unpinning_imports( job_dir, &job.workspace_id, &job.id, worker_name, Some(conn), job.timeout, mem_peak, canceled_by, &common_bun_proc_envs, &mut Some(occupancy_metrics), ) .await?; let bundle_path = format!("{job_dir}/main.js"); ensure_bundle_output_exists(&bundle_path)?; if !local_path.is_empty() { match save_cache(&local_path, &remote_path, &bundle_path, false) .warn_after_seconds_for(60, "bundle cache save") .await { Err(e) => { let em = format!("could not save {local_path} to bundle cache: {e:?}"); tracing::error!(em) } Ok(logs) => { init_logs.push_str(&"\n"); init_logs.push_str(&logs); init_logs.push_str(&"\n"); tracing::info!("saved bun bundle cache: {logs}") } } } if !annotation.native { let ex_wrapper = read_file_content(&format!("{job_dir}/wrapper.mjs")).await?; let mut rewritten = ex_wrapper.replace( "import * as Main from \"./main.ts\"", "import * as Main from \"./main.js\"", ); if is_wac_v2 { rewritten = rewritten.replace("from \"windmill-client\"", "from \"./main.js\""); } write_file(job_dir, "wrapper.mjs", &rewritten)?; write_file(job_dir, "package.json", r#"{ "type": "module" }"#)?; } fs::remove_file(format!("{job_dir}/main.ts"))?; has_bundle_cache = true; } else if annotation.nodejs { generate_wrapper_mjs( job_dir, &job.workspace_id, &job.id, worker_name, conn, job.timeout, mem_peak, canceled_by, &common_bun_proc_envs, &mut Some(occupancy_metrics), ) .await?; } } if annotation.native { #[cfg(not(feature = "deno_core"))] { tracing::error!( r#""deno_core" feature is not activated, but "//native" annotation used. Returning empty value..."# ); return Err(error::Error::internal_err("deno_core feature is not activated, but //native annotation used. Returning empty value...".to_string())); } #[cfg(feature = "deno_core")] { let env_code = build_nativets_env_code( base_internal_url, &reserved_variables, annotation.no_network, ); let js_code = read_file_content(&format!("{job_dir}/main.js")).await?; let started_at = Instant::now(); let args = crate::common::build_args_map(job, client, conn) .await? .map(sqlx::types::Json); let job_args = if args.is_some() { args.as_ref() } else { job.args.as_ref() }; append_logs(&job.id, &job.workspace_id, format!("{init_logs}\n"), conn).await; if apply_preprocessor { // First pass: run preprocessor function let pre_result = crate::js_eval::eval_fetch_timeout( env_code.clone(), inner_content.to_string(), js_code.clone(), job_args, Some("preprocessor".to_string()), job.id, job.timeout, conn, mem_peak, canceled_by, worker_name, &job.workspace_id, false, occupancy_metrics, None, has_stream, ) .await?; let preprocessed: HashMap> = serde_json::from_str(pre_result.get()).map_err(|e| { error::Error::internal_err(format!( "error deserializing preprocessed args: {e:#}" )) })?; *new_args = Some(preprocessed.clone()); // Second pass: run main with preprocessed args let preprocessed_json = sqlx::types::Json(preprocessed); let stream_notifier = StreamNotifier::new(conn, job); let result = crate::js_eval::eval_fetch_timeout( env_code, inner_content.to_string(), js_code, Some(&preprocessed_json), job.script_entrypoint_override.clone(), job.id, job.timeout, conn, mem_peak, canceled_by, worker_name, &job.workspace_id, false, occupancy_metrics, stream_notifier, has_stream, ) .await?; tracing::info!( "Executed native code (with preprocessor) in {}ms", started_at.elapsed().as_millis() ); return Ok(result); } let stream_notifier = StreamNotifier::new(conn, job); let result = crate::js_eval::eval_fetch_timeout( env_code, inner_content.to_string(), js_code, job_args, job.script_entrypoint_override.clone(), job.id, job.timeout, conn, mem_peak, canceled_by, worker_name, &job.workspace_id, false, occupancy_metrics, stream_notifier, has_stream, ) .await?; tracing::info!( "Executed native code in {}ms", started_at.elapsed().as_millis() ); return Ok(result); } } append_logs(&job.id, &job.workspace_id, init_logs, conn).await; //do not cache local dependencies let child = if jailed { let nsjail_timeout = resolve_nsjail_timeout(conn, &job.workspace_id, job.id, job.timeout).await; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); let args = if annotation.nodejs { vec![ "--config", "run.config.proto", "--", &NODE_BIN_PATH, "/tmp/nodejs/wrapper.mjs", ] } else if codebase.is_some() || has_bundle_cache { vec![ "--config", "run.config.proto", "--", &BUN_PATH, "run", "--preserve-symlinks", "/tmp/bun/wrapper.mjs", ] } else { vec![ "--config", "run.config.proto", "--", &BUN_PATH, "run", "--preserve-symlinks", "-i", "--prefer-offline", "-r", "/tmp/bun/loader.bun.js", "/tmp/bun/wrapper.mjs", ] }; nsjail_cmd .current_dir(job_dir) .env_clear() .envs(reserved_variables) .envs( get_run_proxy_envs( &ScriptLang::Bun, job.kind, &job.id, &job.workspace_id, conn, jail_no_network, ) .await?, ) .envs(common_bun_proc_envs) .env("PATH", PATH_ENV.as_str()) .args(args) .stdout(Stdio::piped()) .stderr(Stdio::piped()); let _ = write_file( job_dir, "run.config.proto", &NSJAIL_CONFIG_RUN_BUN_CONTENT .replace("{LANG}", if annotation.nodejs { "nodejs" } else { "bun" }) .replace("{JOB_DIR}", job_dir) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{CLONE_NEWNET}", nsjail_clone_newnet(jail_no_network)) .replace("{UIDGIDMAP}", if *DISABLE_NUSER { "" } else { "uidmap {\n inside_id: \"1000\"\n outside_id: \"\"\n count: 1\n}\n\ngidmap {\n inside_id: \"1000\"\n outside_id: \"\"\n count: 1\n}" }) .replace( "{SHARED_MOUNT}", &shared_mount.replace( "/tmp/shared", if annotation.nodejs { "/tmp/nodejs/shared" } else { "/tmp/bun/shared" }, ), ) .replace("{TRACING_PROXY_CA_CERT_PATH}", &*TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL) .replace( "{TMP_MOUNT_BLOCK}", &resolve_nsjail_tmp_mount_block(job_dir).await, ) .replace("{TIMEOUT}", &nsjail_timeout) .replace("{ENVARS}", &script_envars(&envs, &nsjail_cmd, &[])), )?; start_child_process(nsjail_cmd, NSJAIL_PATH.as_str(), false).await? } else { let cmd = if annotation.nodejs { let script_path = format!("{job_dir}/wrapper.mjs"); let args = vec!["--preserve-symlinks", script_path.as_str()]; let mut bun_cmd = build_command_with_isolation(&*NODE_BIN_PATH, &args); bun_cmd .current_dir(job_dir) .env_clear() .envs(envs) .envs(reserved_variables) .envs( get_proxy_envs_for_lang( &ScriptLang::Bun, job.kind, &job.id, &job.workspace_id, conn, ) .await?, ) .envs(common_bun_proc_envs) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); #[cfg(windows)] bun_cmd.env("SystemRoot", SYSTEM_ROOT.as_str()); bun_cmd } else { let script_path = format!("{job_dir}/wrapper.mjs"); let args: Vec<&str> = if codebase.is_some() || has_bundle_cache { vec!["run", &script_path] } else { vec![ "run", "--preserve-symlinks", "-i", "--prefer-offline", "-r", "./loader.bun.js", &script_path, ] }; let mut bun_cmd = build_command_with_isolation(&*BUN_PATH, &args); bun_cmd .current_dir(job_dir) .env_clear() .envs(envs) .envs(reserved_variables) .envs( get_proxy_envs_for_lang( &ScriptLang::Bun, job.kind, &job.id, &job.workspace_id, conn, ) .await?, ) .envs(common_bun_proc_envs) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); #[cfg(windows)] bun_cmd.env("SystemRoot", SYSTEM_ROOT.as_str()); bun_cmd }; let executable = if annotation.nodejs { &*NODE_BIN_PATH } else { &*BUN_PATH }; start_child_process(cmd, executable, false).await? }; let stream_notifier = StreamNotifier::new(conn, job); let handle_result = handle_child( &job.id, conn, mem_peak, canceled_by, child, is_sandboxing_enabled(), worker_name, &job.workspace_id, "bun run", job.timeout, false, &mut Some(occupancy_metrics), None, stream_notifier, ) .await?; *has_stream = handle_result.result_stream.is_some(); if apply_preprocessor { let args = read_file(&format!("{job_dir}/args.json")) .await .map_err(|e| { error::Error::internal_err(format!( "error while reading args from preprocessing: {e:#}" )) })?; let args: HashMap> = serde_json::from_str(args.get()).map_err(|e| { error::Error::internal_err(format!( "error while deserializing args from preprocessing: {e:#}" )) })?; *new_args = Some(args.clone()); } let result = read_result(job_dir, handle_result.result_stream).await?; // WAC v2 post-execution: parse output and handle dispatch/suspend if is_wac_v2 { return handle_wac_v2_output(result, job, conn, canceled_by, modules, new_args.as_ref()) .await; } Ok(result) } /// Resolve a module file from the parent script's modules map. /// For Script jobs, fetches from the `script` table by hash. /// For Preview jobs, fetches from `v2_job.raw_code` (modules stored inline). fn resolve_parent_module<'a>( modules: &'a Option>, module_key: &str, ) -> error::Result<&'a windmill_common::scripts::ScriptModule> { if let Some(modules) = modules { if let Some(module) = modules.get(module_key) { return Ok(module); } } Err(error::Error::ExecutionErr(format!( "Module '{}' not found in script modules", module_key ))) } /// Handle WAC v2 output after bun/python exits. Parse result as WacOutput, /// dispatch child jobs on suspend, or return the final result. pub async fn handle_wac_v2_output( result: Box, job: &MiniPulledJob, conn: &Connection, canceled_by: &mut Option, modules: &Option>, preprocessed_args: Option<&HashMap>>, ) -> error::Result> { use crate::wac_executor::{ load_checkpoint, parse_wac_output, update_checkpoint_for_dispatch, wac_cancelled_mid_segment, WacOutput, WacPark, WacStepDispatch, }; use serde_json::Value; use windmill_common::get_latest_flow_version_info_for_path; use windmill_common::jobs::{script_path_to_payload, JobKind, JobPayload, RawCode}; use windmill_common::runnable_settings::{ ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, }; use windmill_queue::{check_tag_available_for_push, push, PushArgs, PushIsolationLevel}; let output = parse_wac_output(&result)?; match output { WacOutput::Complete { result: value } => { // Workflow completed — return the inner result value let raw = serde_json::value::to_raw_value(&value).map_err(|e| { error::Error::internal_err(format!("Failed to serialize WAC result: {e}")) })?; Ok(raw) } WacOutput::Dispatch { mode, steps } => { if steps.is_empty() { return Err(error::Error::internal_err( "WAC v2 dispatch with no steps — this is a bug in the workflow SDK".to_string(), )); } let db = match conn { Connection::Sql(db) => db, _ => { return Err(error::Error::internal_err( "WAC v2 dispatch requires SQL connection".to_string(), )) } }; let mut checkpoint = load_checkpoint(db, &job.id).await?; // Source hash validation: detect if code changed between replays let current_hash = job.runnable_id.map(|h| h.0.to_string()).unwrap_or_default(); if !current_hash.is_empty() { if checkpoint.source_hash.is_empty() { checkpoint.source_hash = current_hash.clone(); } else if checkpoint.source_hash != current_hash { return Err(error::Error::ExecutionErr( "Workflow source code changed between replays. \ Cannot safely resume from checkpoint — step keys may have shifted. \ Please restart this workflow." .to_string(), )); } } let num_steps = steps.len(); tracing::info!( job_id = %job.id, mode = %mode, num_steps = num_steps, steps = ?steps.iter().map(|s| &s.name).collect::>(), "WAC v2 dispatching child jobs" ); // Create child jobs for each step. // Each child re-runs the full workflow with a checkpoint containing // _executing_key = step_key, so only that step runs its inner function. // // IMPORTANT: To prevent a race condition where a fast child completes // before the parent is suspended, we: // 1. Pre-generate child UUIDs // 2. Save checkpoint + suspend parent + seed child checkpoints // 3. THEN push the child jobs (making them visible to workers) // Read the parent's original args for the child jobs. // If preprocessed_args is Some, the wrapper just ran the preprocessor // and the DB row hasn't been updated yet — use those instead so child // re-runs of the parent see the post-preprocessor args. let parent_args: HashMap> = if let Some(pre) = preprocessed_args { // Single pass: parse each raw value into a serde_json::Value for // checkpoint.input_args, and clone the Box for the // returned HashMap. A parse failure surfaces as an error rather // than being silently coerced to null and persisted. let mut map = serde_json::Map::with_capacity(pre.len()); let mut owned = HashMap::with_capacity(pre.len()); for (k, v) in pre.iter() { let parsed = serde_json::from_str::(v.get()).map_err(|e| { error::Error::internal_err(format!( "Failed to parse preprocessed arg '{k}': {e}" )) })?; map.insert(k.clone(), parsed); owned.insert(k.clone(), v.clone()); } checkpoint.input_args = map; owned } else { let stored: serde_json::Map = checkpoint.input_args.clone(); if stored.is_empty() { // First dispatch — read from the parent job's args let row: Option = sqlx::query_scalar( "SELECT args FROM v2_job WHERE id = $1 AND workspace_id = $2", ) .bind(&job.id) .bind(&job.workspace_id) .fetch_optional(db) .await?; let args_val = row.unwrap_or(Value::Object(Default::default())); if let Value::Object(map) = args_val { // Store for future re-runs checkpoint.input_args = map.clone(); map.into_iter() .map(|(k, v)| { let raw = serde_json::value::to_raw_value(&v).map_err(|e| { error::Error::internal_err(format!( "Failed to serialize arg '{k}': {e}" )) })?; Ok((k, raw)) }) .collect::>>()? } else { HashMap::new() } } else { stored .into_iter() .map(|(k, v)| { let raw = serde_json::value::to_raw_value(&v).map_err(|e| { error::Error::internal_err(format!( "Failed to serialize arg '{k}': {e}" )) })?; Ok((k, raw)) }) .collect::>>()? } }; // Pre-generate child UUIDs so we can save them in the checkpoint // before the children become visible to workers. // Validate key uniqueness — duplicate keys would cause one child's // UUID to be overwritten in the job_ids map, making it unmappable // on completion (the parent would hang). { let mut seen_keys = std::collections::HashSet::new(); for s in &steps { if !seen_keys.insert(&s.key) { return Err(error::Error::internal_err(format!( "WAC v2 duplicate step key '{}' — each task call must produce a unique key", s.key ))); } } } let job_ids: Vec<(String, Uuid)> = steps .iter() .map(|s| (s.key.clone(), ulid::Ulid::new().into())) .collect(); // Resolve job_payload once (same for all children since they re-run // the parent script). The step's cache setting is for the workflow's // result; a task is cached only through its own `cache_ttl` option, // under a key of its own (see `cached_result_path`). let job_payload_template = match job.kind { JobKind::Script => { if let Some(hash) = job.runnable_id { Ok(JobPayload::ScriptHash { hash, path: job.runnable_path.clone().unwrap_or_default(), cache_ttl: None, cache_ignore_s3_path: None, dedicated_worker: None, language: job.script_lang.unwrap_or(ScriptLang::Bun), priority: job.priority, apply_preprocessor: false, concurrency_settings: ConcurrencySettings::default(), debouncing_settings: DebouncingSettings::default(), labels: None, // Capped by the parent at push, which already holds this // script's setting. job_token_scopes: None, }) } else { Err(error::Error::internal_err( "WAC v2 Script job missing runnable_id".to_string(), )) } } // A deployed flow runs an inline step as the `flow_node` its deploy // rewrote it into; the child re-runs that node the way a `Script` // child re-runs its hash, so `runnable_id` (the checkpoint's source // hash) stays the same across parent and children. JobKind::FlowScript => { if let Some(id) = job.runnable_id { Ok(JobPayload::FlowScript { id: windmill_common::flows::FlowNodeId(id.0), path: job.runnable_path.clone().unwrap_or_default(), language: job.script_lang.unwrap_or(ScriptLang::Bun), cache_ttl: None, cache_ignore_s3_path: None, dedicated_worker: None, concurrency_settings: ConcurrencySettings::default(), }) } else { Err(error::Error::internal_err( "WAC v2 FlowScript job missing runnable_id".to_string(), )) } } JobKind::Preview => { let row: Option<(Option, Option)> = sqlx::query_as( "SELECT raw_code, raw_lock FROM v2_job WHERE id = $1 AND workspace_id = $2", ) .bind(&job.id) .bind(&job.workspace_id) .fetch_optional(db) .await?; let (code, lock) = row.unwrap_or_default(); Ok(JobPayload::Code(RawCode { content: code.unwrap_or_default(), path: job.runnable_path.clone(), hash: None, language: job.script_lang.unwrap_or(ScriptLang::Bun), lock: lock, cache_ttl: None, cache_ignore_s3_path: None, dedicated_worker: None, concurrency_settings: ConcurrencySettingsWithCustom::default(), debouncing_settings: DebouncingSettings::default(), // `push` drops the `_MODULES` the children inherit with the // parent's args, so the preview's own modules are handed over here. modules: modules.clone(), tag: None, })) } _ => Err(error::Error::internal_err(format!( "WAC v2 unsupported job kind: {:?}", job.kind ))), }?; // Resolve what every task runs, and check any tag it picks, before the parent parks: // a task that cannot be pushed then fails the dispatch with no sibling queued. The // parent's code and input are built per child at push time, so a fan-out never holds // a copy of them for each child at once. enum ChildRunnable<'a> { // Re-runs the parent with `_executing_key`. Parent, // A `./` module of the parent script. Module(&'a windmill_common::scripts::ScriptModule), // A deployed script or flow. Deployed(JobPayload), } struct ResolvedChild<'a> { runnable: ChildRunnable<'a>, email: String, permissioned_as: String, } // The input a task with a runnable of its own is pushed with; the parent re-run gets // the parent's. Both the tag check and the push build it here, so the check resolves // `$args[...]` against what the child actually receives. let own_args = |step: &WacStepDispatch| -> HashMap> { step.args .iter() .map(|(k, v)| { let raw = serde_json::value::to_raw_value(v).unwrap(); (k.clone(), raw) }) .collect() }; let mut children: Vec = Vec::with_capacity(num_steps); for step in &steps { let (runnable, on_behalf_of) = match step.dispatch_type.as_str() { "script" if step.script.starts_with("./") => { // Module-relative path: resolve from parent script's modules let module_key = step.script.strip_prefix("./").unwrap(); let module = resolve_parent_module(modules, module_key)?; // Inline module code, not a separate runnable: it has no // identity of its own and runs as the parent. (ChildRunnable::Module(module), None) } "script" => { // Resolve script path to job payload (handles hash, lang, etc.) let (payload, _, _, _, _, on_behalf_of) = script_path_to_payload( &step.script, None, // no authed db for background workers db.clone(), &job.workspace_id, Some(true), // skip preprocessor ) .await?; (ChildRunnable::Deployed(payload), on_behalf_of) } "flow" => { let flow_info = get_latest_flow_version_info_for_path( None, db, &job.workspace_id, &step.script, true, ) .await?; let payload = JobPayload::Flow { path: step.script.clone(), dedicated_worker: flow_info.dedicated_worker, apply_preprocessor: false, version: flow_info.version, labels: flow_info.labels.clone(), job_token_scopes: flow_info.job_token_scopes.clone(), }; let on_behalf_of = flow_info.on_behalf_of(&job.workspace_id, db).await?; (ChildRunnable::Deployed(payload), on_behalf_of) } // "inline" — re-run parent with _executing_key _ => (ChildRunnable::Parent, None), }; // A target runnable that opts into on-behalf-of runs under its own // identity, never the caller's, so a step that reaches it through a // workflow cannot widen or narrow its permissions. `created_by` still // credits the caller, matching how the run API pushes these jobs. let (email, permissioned_as) = match on_behalf_of { Some(on_behalf_of) => (on_behalf_of.email, on_behalf_of.permissioned_as), None => ( job.permissioned_as_email.clone(), job.permissioned_as.clone(), ), }; // A task inheriting the parent's tag skips the check: that tag was // checked when the parent was pushed, and a dedicated worker's tag is // never in CUSTOM_TAGS. if let Some(tag) = step .tag .as_deref() .filter(|t| !t.is_empty() && *t != job.tag.as_str()) { let is_super_admin = windmill_common::auth::is_super_admin_email(db, &email).await?; let step_args = (!matches!(runnable, ChildRunnable::Parent)).then(|| own_args(step)); let args = PushArgs { args: step_args.as_ref().unwrap_or(&parent_args), extra: None }; check_tag_available_for_push( db, &job.workspace_id, tag, &args, is_super_admin, None, ) .warn_after_seconds_with_sql(1, "check_tag_available_for_push".to_string()) .await .map_err(|e| match e { error::Error::BadRequest(msg) => error::Error::BadRequest(format!( "task '{}' cannot run on tag '{tag}': {msg}", step.name )), e => e, })?; } children.push(ResolvedChild { runnable, email, permissioned_as }); } // Step 1: Save checkpoint, suspend parent, and seed child checkpoints // in a single transaction — all BEFORE children become visible. let segment_ms; { let mut tx = db.begin().await?; // Park before writing the checkpoint. This locks the queue row ahead of the // status row, the order `record_child_completion` takes, so a stale child // finishing while the parent re-dispatches cannot deadlock this transaction. // A cancel already on the row is also seen before anything is written, and // every child pushed after the commit finds a parked parent to decrement. match crate::wac_executor::suspend_wac_parent( &mut tx, &job.id, &job.workspace_id, num_steps as i32, 14.0 * 24.0 * 3600.0, ) .await? { WacPark::Parked(ms) => segment_ms = ms, WacPark::Cancelled(cancel) => { return Err(wac_cancelled_mid_segment(cancel, canceled_by)) } } // Update checkpoint with pending steps update_checkpoint_for_dispatch(&mut checkpoint, &steps, &mode, &job_ids); let status_json = serde_json::to_value(&checkpoint).map_err(|e| { error::Error::internal_err(format!("Failed to serialize checkpoint: {e}")) })?; sqlx::query( "INSERT INTO v2_job_status (id, workflow_as_code_status) VALUES ($1, jsonb_build_object('_checkpoint', $2::jsonb)) ON CONFLICT (id) DO UPDATE SET workflow_as_code_status = jsonb_set( COALESCE(v2_job_status.workflow_as_code_status, '{}'::jsonb), '{_checkpoint}', $2::jsonb )", ) .bind(&job.id) .bind(&status_json) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!("Failed to save WAC checkpoint: {e}")) })?; // Store per-child-job info for the WorkflowTimeline UI for (step, (_, child_id)) in steps.iter().zip(job_ids.iter()) { let child_id_str = child_id.to_string(); let timeline_val = serde_json::json!({ "scheduled_for": chrono::Utc::now().to_rfc3339(), "name": step.key, }); sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( COALESCE(workflow_as_code_status, '{}'::jsonb), ARRAY[$2], $3 ) WHERE id = $1", ) .bind(&job.id) .bind(&child_id_str) .bind(&timeline_val) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!( "Failed to update WAC timeline status: {e}" )) })?; } tx.commit().await?; } // Step 2: Push child jobs (now visible to workers). // Parent is already suspended, so child completions are safe. // Track successfully pushed children so we can cancel them on // partial failure (e.g. pushing child 3 of 5 fails). let mut pushed_ids: Vec = Vec::with_capacity(num_steps); let push_result: error::Result<()> = async { for ((step, (_, child_uuid)), child) in steps.iter().zip(job_ids.iter()).zip(children) { // A task with a runnable of its own (a deployed script or flow) queues // at that runnable's priority; any other task is the parent's code and // queues at the parent's. let own_runnable = matches!(step.dispatch_type.as_str(), "script" | "flow") && !step.script.starts_with("./"); let is_external = !matches!(child.runnable, ChildRunnable::Parent); let job_payload = match child.runnable { ChildRunnable::Parent => job_payload_template.clone(), ChildRunnable::Module(module) => JobPayload::Code(RawCode { content: module.content.clone(), path: job.runnable_path.clone(), hash: None, language: module.language, lock: module.lock.clone(), cache_ttl: None, cache_ignore_s3_path: None, dedicated_worker: None, concurrency_settings: ConcurrencySettingsWithCustom::default(), debouncing_settings: DebouncingSettings::default(), modules: None, tag: None, }), ChildRunnable::Deployed(payload) => payload, }; let step_args = is_external.then(|| own_args(step)); let push_args = PushArgs { args: step_args.as_ref().unwrap_or(&parent_args), extra: None }; // Apply step-level overrides to payload (cache, concurrency) let mut job_payload = job_payload; if let Some(cache_ttl) = step.cache_ttl { match &mut job_payload { JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } | JobPayload::FlowScript { cache_ttl: ref mut ct, .. } => { *ct = Some(cache_ttl) } JobPayload::Code(ref mut code) => code.cache_ttl = Some(cache_ttl), _ => {} } } if step.concurrent_limit.is_some() || step.concurrency_key.is_some() || step.concurrency_time_window_s.is_some() { match &mut job_payload { JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. } | JobPayload::FlowScript { concurrency_settings: ref mut cs, .. } => { if let Some(limit) = step.concurrent_limit { cs.concurrent_limit = Some(limit); } if let Some(ref key) = step.concurrency_key { cs.concurrency_key = Some(key.clone()); } if let Some(window) = step.concurrency_time_window_s { cs.concurrency_time_window_s = Some(window); } } JobPayload::Code(ref mut code) => { if let Some(limit) = step.concurrent_limit { code.concurrency_settings.concurrent_limit = Some(limit); } if let Some(ref key) = step.concurrency_key { code.concurrency_settings.custom_concurrency_key = Some(key.clone()); } if let Some(window) = step.concurrency_time_window_s { code.concurrency_settings.concurrency_time_window_s = Some(window); } } _ => {} } } let (_, mut tx) = push( db, PushIsolationLevel::IsolatedRoot(db.clone()), &job.workspace_id, job_payload, push_args, &job.created_by, &child.email, child.permissioned_as, None, None, None, None, Some(job.id), // parent_job job.root_job.or(Some(job.id)), // root_job job.flow_innermost_root_job, Some(*child_uuid), // pre-generated job_id false, // is_flow_step false, // same_worker None, // pre_run_error job.visible_to_owner, step.tag.clone().or_else(|| Some(job.tag.clone())), step.timeout.or(job.timeout), None, // flow_step_id step.priority .or(if own_runnable { None } else { job.priority }), None, // authed false, // running job.permissioned_as_end_user_email.clone(), None, // trigger None, // suspended_mode job.job_token_scopes.as_deref(), ) .await?; // Seed child checkpoint only for inline tasks (they need // _executing_key to know which step to run). External // scripts/flows don't need a WAC checkpoint. if !is_external { let child_checkpoint_json = serde_json::json!({ "completed_steps": &checkpoint.completed_steps, "_executing_key": &step.key, }); sqlx::query( "INSERT INTO v2_job_status (id, workflow_as_code_status) VALUES ($1, jsonb_build_object('_checkpoint', $2::jsonb)) ON CONFLICT (id) DO UPDATE SET workflow_as_code_status = jsonb_set( COALESCE(v2_job_status.workflow_as_code_status, '{}'::jsonb), '{_checkpoint}', $2::jsonb )", ) .bind(child_uuid) .bind(&child_checkpoint_json) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!( "Failed to seed child checkpoint: {e}" )) })?; } tx.commit().await.map_err(|e| { error::Error::internal_err(format!("Failed to commit child push: {e}")) })?; pushed_ids.push(*child_uuid); tracing::info!( parent_job = %job.id, child_job = %child_uuid, step_name = %step.name, step_key = %step.key, "WAC v2 dispatched child job" ); } Ok(()) } .await; if let Err(e) = push_result { tracing::error!( job_id = %job.id, error = %e, pushed_count = pushed_ids.len(), total_count = num_steps, "WAC v2 failed to push child jobs, cleaning up" ); // Cancel already-pushed children so they don't complete and // corrupt the checkpoint (they'd decrement suspend on a parent // that's about to be unsuspended and re-run). for child_id in &pushed_ids { let _ = sqlx::query!( "UPDATE v2_job_queue SET canceled_by = $2, canceled_reason = $3 WHERE id = $1", child_id, "system", "WAC dispatch failed: not all children could be pushed", ) .execute(db) .await; } // Clear pending_steps from checkpoint so the parent doesn't // think children are outstanding when it re-runs. let _ = sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = \ workflow_as_code_status #- '{_checkpoint,pending_steps}' \ WHERE id = $1", ) .bind(&job.id) .execute(db) .await; // Unsuspend parent so the error propagates instead of a 14-day hang. // Unlike the other suspend exits this one completes the job for real, so // it needs its segment start back — the in-memory copy is what the pull // stamped, before the suspend cleared the column. let _ = sqlx::query!( "UPDATE v2_job_queue SET suspend = 0, suspend_until = NULL, started_at = coalesce(started_at, $2, now()) WHERE id = $1", job.id, job.started_at, ) .execute(db) .await; return Err(e); } tracing::info!( job_id = %job.id, num_steps = num_steps, "WAC v2 parent job suspended" ); crate::wac_executor::end_wac_segment(conn, job, segment_ms); Err(error::Error::WacSuspended(format!( "WAC v2 job {} suspended waiting for {} child job(s)", job.id, num_steps ))) } WacOutput::Approval { key, timeout, form, self_approval_disabled, skin, description } => { let db = match conn { Connection::Sql(db) => db, _ => { return Err(error::Error::internal_err( "WAC v2 approval requires SQL connection".to_string(), )) } }; let mut checkpoint = load_checkpoint(db, &job.id).await?; let timeout_secs = timeout.unwrap_or(1800) as f64; // Mark this step as pending approval checkpoint.pending_steps = Some(crate::wac_executor::WacPendingSteps { mode: "approval".to_string(), keys: vec![key.clone()], job_ids: serde_json::Map::new(), }); let mut tx = db.begin().await?; // Park first: the queue row is locked before the status row, the order every // child completion takes. let segment_ms = match crate::wac_executor::suspend_wac_parent( &mut tx, &job.id, &job.workspace_id, 1, timeout_secs, ) .await? { WacPark::Parked(ms) => ms, WacPark::Cancelled(cancel) => { return Err(wac_cancelled_mid_segment(cancel, canceled_by)) } }; // Save checkpoint let status_json = serde_json::to_value(&checkpoint).map_err(|e| { error::Error::internal_err(format!("Failed to serialize checkpoint: {e}")) })?; sqlx::query( "INSERT INTO v2_job_status (id, workflow_as_code_status) VALUES ($1, jsonb_build_object('_checkpoint', $2::jsonb)) ON CONFLICT (id) DO UPDATE SET workflow_as_code_status = jsonb_set( COALESCE(v2_job_status.workflow_as_code_status, '{}'::jsonb), '{_checkpoint}', $2::jsonb )", ) .bind(&job.id) .bind(&status_json) .execute(&mut *tx) .await .map_err(|e| error::Error::internal_err(format!("Failed to save checkpoint: {e}")))?; // Store approval_conditions in flow_status for resume endpoint auth checks let sad = self_approval_disabled.unwrap_or(false); if sad { #[cfg(not(feature = "enterprise"))] return Err(error::Error::ExecutionErr( "Disabling self-approval is an enterprise only feature".to_string(), )); #[cfg(feature = "enterprise")] { use windmill_common::flow_status::ApprovalConditions; let approval_conditions = ApprovalConditions { user_auth_required: true, user_groups_required: vec![], self_approval_disabled: true, }; sqlx::query( "UPDATE v2_job_status SET flow_status = JSONB_SET( COALESCE(flow_status, '{}'::jsonb), '{approval_conditions}', $2::jsonb ) WHERE id = $1", ) .bind(&job.id) .bind(&serde_json::json!(approval_conditions)) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!( "Failed to save approval conditions: {e}" )) })?; } } // Generate resume URLs for the inline approval buttons. The resume_id // is derived from the step key so each waitForApproval() in the same // workflow gets a unique resume_job record, and so URLs the workflow // minted for this step ahead of time (getApprovalUrls) address the // same one. let resume_id: u32 = windmill_common::wac::approval_resume_id(&key); // Generate stateless approval token using shared utility let approval_token = windmill_common::variables::generate_approval_token(&job.workspace_id, job.id, db) .await?; let (resume_url, cancel_url, approval_page_url) = { use hmac::{Hmac, Mac}; use sha2::Sha256; use windmill_common::variables::get_workspace_key; let wkey = get_workspace_key(&job.workspace_id, db).await?; let mut mac = Hmac::::new_from_slice(wkey.as_bytes()) .map_err(|e| error::Error::internal_err(format!("HMAC key error: {e}")))?; mac.update(job.id.as_bytes()); mac.update(resume_id.to_be_bytes().as_ref()); let signature = hex::encode(mac.finalize().into_bytes()); let base_url = (**windmill_common::BASE_URL.load()).clone(); let w_id = &job.workspace_id; let job_id = &job.id; let resume = format!( "{base_url}/api/w/{w_id}/jobs_u/resume/{job_id}/{resume_id}/{signature}" ); let cancel = format!( "{base_url}/api/w/{w_id}/jobs_u/cancel/{job_id}/{resume_id}/{signature}" ); let approval_page = format!("{base_url}/approve/{w_id}/{job_id}?token={approval_token}"); (resume, cancel, approval_page) }; // Store approval form metadata for the approval page endpoint let mut approval_meta = serde_json::json!({ "key": key, "form": form, "timeout": timeout_secs as u32, "self_approval_disabled": sad, "skin": skin.unwrap_or_default(), "resume": resume_url, "cancel": cancel_url, "approvalPage": approval_page_url, }); if let Some(description) = description.filter(|d| !d.is_null()) { approval_meta["description"] = description; } sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( COALESCE(workflow_as_code_status, '{}'::jsonb), '{_approval}', $2::jsonb ) WHERE id = $1", ) .bind(&job.id) .bind(&approval_meta) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!("Failed to save approval meta: {e}")) })?; // Write timeline entry for the approval step { let now_str = chrono::Utc::now().to_rfc3339(); let timeline_val = serde_json::json!({ "scheduled_for": &now_str, "started_at": &now_str, "name": key, "approval": true, "self_approval_disabled": sad, "form": form, "resume": &resume_url, "cancel": &cancel_url, "approvalPage": &approval_page_url, }); let step_timeline_key = format!("_step/{}", key); sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( COALESCE(workflow_as_code_status, '{}'::jsonb), ARRAY[$2], $3 ) WHERE id = $1", ) .bind(&job.id) .bind(&step_timeline_key) .bind(&timeline_val) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!("Failed to write approval timeline: {e}")) })?; } tx.commit().await?; crate::wac_executor::end_wac_segment(conn, job, segment_ms); tracing::info!( job_id = %job.id, approval_key = %key, timeout_secs = timeout_secs, "WAC v2 parent job suspended waiting for approval" ); Err(error::Error::WacSuspended(format!( "WAC v2 job {} suspended waiting for approval (key: {})", job.id, key ))) } WacOutput::Sleep { key, seconds } => { let db = match conn { Connection::Sql(db) => db, _ => { return Err(error::Error::internal_err( "WAC v2 sleep requires SQL connection".to_string(), )) } }; let mut checkpoint = load_checkpoint(db, &job.id).await?; let sleep_secs = seconds.max(1) as f64; // Mark this step as pending sleep checkpoint.pending_steps = Some(crate::wac_executor::WacPendingSteps { mode: "sleep".to_string(), keys: vec![key.clone()], job_ids: serde_json::Map::new(), }); let mut tx = db.begin().await?; // Park first: the queue row is locked before the status row, the order every // child completion takes. suspend=1 (not 0) so the suspended pull query only // picks it up when `suspend_until <= now()`, not via `suspend <= 0`. let segment_ms = match crate::wac_executor::suspend_wac_parent( &mut tx, &job.id, &job.workspace_id, 1, sleep_secs, ) .await? { WacPark::Parked(ms) => ms, WacPark::Cancelled(cancel) => { return Err(wac_cancelled_mid_segment(cancel, canceled_by)) } }; // Save checkpoint let status_json = serde_json::to_value(&checkpoint).map_err(|e| { error::Error::internal_err(format!("Failed to serialize checkpoint: {e}")) })?; sqlx::query( "INSERT INTO v2_job_status (id, workflow_as_code_status) VALUES ($1, jsonb_build_object('_checkpoint', $2::jsonb)) ON CONFLICT (id) DO UPDATE SET workflow_as_code_status = jsonb_set( COALESCE(v2_job_status.workflow_as_code_status, '{}'::jsonb), '{_checkpoint}', $2::jsonb )", ) .bind(&job.id) .bind(&status_json) .execute(&mut *tx) .await .map_err(|e| error::Error::internal_err(format!("Failed to save checkpoint: {e}")))?; // Write a "sleep" marker in the timeline. Unlike real steps it // carries no execution bar — the frontend renders it as a // minimal label row (e.g. "sleep (2s)"). { let now_str = chrono::Utc::now().to_rfc3339(); let timeline_val = serde_json::json!({ "scheduled_for": &now_str, "name": key, "sleep_duration_s": seconds, }); let step_timeline_key = format!("_step/{}", key); sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( COALESCE(workflow_as_code_status, '{}'::jsonb), ARRAY[$2], $3 ) WHERE id = $1", ) .bind(&job.id) .bind(&step_timeline_key) .bind(&timeline_val) .execute(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!("Failed to write sleep timeline: {e}")) })?; } tx.commit().await?; crate::wac_executor::end_wac_segment(conn, job, segment_ms); tracing::info!( job_id = %job.id, sleep_key = %key, sleep_secs = sleep_secs, "WAC v2 parent job sleeping for {}s", sleep_secs ); Err(error::Error::WacSuspended(format!( "WAC v2 job {} sleeping for {}s (key: {})", job.id, seconds, key ))) } WacOutput::InlineCheckpoint { key, result: value, started_at, duration_ms } => { let db = match conn { Connection::Sql(db) => db, _ => { return Err(error::Error::internal_err( "WAC v2 inline checkpoint requires SQL connection".to_string(), )) } }; // All-or-nothing: the checkpoint save, the `_step/` timeline // write, and the `running = false` queue reset must commit // together. If we split them, a crash or failure in the middle // would leave the job queued with `running = true` but a // checkpoint that already contains the current step — any retry // would then skip the step entirely. Passing the caller's `tx` // into `persist_inline_checkpoint_delta` preserves the original // atomicity from before the shared-helper refactor. let source_hash = job.runnable_id.map(|h| h.0.to_string()); let mut tx = db.begin().await?; // Reset running=false so the job is immediately eligible for pickup. // Unlike dispatch (which sets suspend>0), inline checkpoints don't suspend — // the job should be re-run right away to continue past the cached step. // `prev` holds the pre-update row: RETURNING would see the cleared column. // Runs before the checkpoint write so the queue row is locked ahead of the // status row, the order every child completion takes. let segment_ms = sqlx::query_scalar!( "WITH prev AS (SELECT started_at FROM v2_job_queue WHERE id = $1) UPDATE v2_job_queue q SET running = false, started_at = null, extras = CASE WHEN q.started_at IS NULL OR q.extras ? 'wac_first_started_at' THEN q.extras ELSE jsonb_set(COALESCE(q.extras, '{}'::jsonb), '{wac_first_started_at}', to_jsonb(q.started_at)) END FROM prev WHERE q.id = $1 RETURNING (extract(epoch FROM now() - prev.started_at) * 1000)::bigint", job.id, ) .fetch_optional(&mut *tx) .await .map_err(|e| { error::Error::internal_err(format!( "Failed to reset running state for inline checkpoint: {e}" )) })? .flatten(); crate::wac_executor::persist_inline_checkpoint_delta( &mut tx, &job.id, source_hash.as_deref(), &key, value, started_at.as_deref(), duration_ms, ) .await?; tx.commit().await?; crate::wac_executor::end_wac_segment(conn, job, segment_ms); Err(error::Error::WacSuspended(format!( "WAC v2 job {} inline checkpoint for step {}", job.id, key ))) } } } pub async fn get_common_bun_proc_envs(base_internal_url: Option<&str>) -> HashMap { let mut bun_envs: HashMap = HashMap::from([ (String::from("PATH"), PATH_ENV.clone()), (String::from("HOME"), HOME_ENV.clone()), (String::from("TZ"), TZ_ENV.clone()), (String::from("FORCE_COLOR"), "1".to_string()), (String::from("DO_NOT_TRACK"), "1".to_string()), ( String::from("BUN_INSTALL_CACHE_DIR"), BUN_CACHE_DIR.to_string(), ), ( String::from("BUN_RUNTIME_TRANSPILER_CACHE_PATH"), "0".to_string(), ), ]); if let Some(base_url) = base_internal_url { bun_envs.insert( String::from("BASE_URL"), base_url.to_string().replace("localhost", "127.0.0.1"), ); } if let Some(ref node_path) = NODE_PATH.as_ref() { bun_envs.insert(String::from("NODE_PATH"), node_path.to_string()); } if let Some(secs) = *BUN_INSTALL_MIN_RELEASE_AGE.read().await { bun_envs.insert( String::from("BUN_INSTALL_MINIMUM_RELEASE_AGE"), secs.to_string(), ); } #[cfg(windows)] { bun_envs.insert("SystemRoot".to_string(), crate::SYSTEM_ROOT.to_string()); bun_envs.insert( "USERPROFILE".to_string(), crate::USERPROFILE_ENV.to_string(), ); } return bun_envs; } #[cfg(any(feature = "deno_core", feature = "private"))] pub fn build_nativets_env_code( base_internal_url: &str, reserved_variables: &HashMap, no_network: bool, ) -> String { format!( "const process = {{ env: {{}} }};\nconst BASE_URL = '{base_internal_url}';\nconst BASE_INTERNAL_URL = '{base_internal_url}';\nprocess.env['BASE_URL'] = BASE_URL;process.env['BASE_INTERNAL_URL'] = BASE_INTERNAL_URL;\n{}", reserved_variables .iter() // Useless without network, and a `//no_network` script could still // hand it out through its result to whoever authored it. .filter(|(k, _)| !(no_network && k.as_str() == "WM_TOKEN")) .map(|(k, v)| { // The key is attacker-controllable (custom workspace env vars), so // escape it as a string literal too, not just the value. let key_literal = windmill_common::variables::escape_js_single_quoted(k); let escaped = windmill_common::variables::escape_js_single_quoted(v); format!("process.env['{key_literal}'] = '{escaped}';") }) .collect::>() .join("\n") ) } #[cfg(feature = "private")] use crate::{ common::build_envs_map, dedicated_worker_oss::handle_dedicated_process, JobCompletedSender, }; #[cfg(feature = "private")] use tokio::sync::mpsc::Receiver; #[cfg(feature = "private")] use windmill_common::variables; #[cfg(feature = "private")] use windmill_queue::DedicatedWorkerJob; #[cfg(feature = "private")] async fn handle_dedicated_bunnative( inner_content: &str, js_code: &str, env_code: &str, token: &str, worker_name: &str, _w_id: &str, script_path: &str, db: &DB, jobs_rx: Receiver, killpill_rx: tokio::sync::broadcast::Receiver<()>, job_completed_tx: JobCompletedSender, client: &windmill_common::client::AuthedClient, ) -> Result<()> { #[cfg(not(feature = "deno_core"))] { let _ = ( inner_content, js_code, env_code, token, worker_name, script_path, db, jobs_rx, killpill_rx, job_completed_tx, client, ); return Err(error::Error::internal_err( "deno_core feature is not activated but native dedicated worker was started" .to_string(), )); } #[cfg(feature = "deno_core")] { use std::sync::Arc; use crate::common::transform_json; use windmill_common::worker::to_raw_value; use windmill_queue::{append_logs, JobCompleted, MiniCompletedJob}; use windmill_runtime_nativets::PrewarmedIsolate; let ann = windmill_runtime_nativets::get_annotation(inner_content); let parsed_args = windmill_parser_ts::parse_deno_signature(inner_content, true, false, None)?.args; let arg_names: Vec = parsed_args.into_iter().map(|x| x.name).collect(); let env_code = env_code.to_string(); let js_code = js_code.to_string(); let pre_arg_names: Option> = windmill_parser_ts::parse_deno_signature( inner_content, true, false, Some("preprocessor".to_string()), ) .ok() .filter(|sig| !sig.args.is_empty()) .map(|sig| sig.args.into_iter().map(|x| x.name).collect()); let mut warm = PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), arg_names.clone(), None, ); // Pre-warm preprocessor isolate if the script has a preprocessor let mut pre_warm = pre_arg_names.as_ref().map(|pre_names| { PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), pre_names.clone(), Some("preprocessor".to_string()), ) }); let init_log = format!("dedicated worker nativets: {worker_name}\n\n"); let alive = true; let mut killpill_rx = killpill_rx; let mut jobs_rx = jobs_rx; loop { tokio::select! { biased; _ = killpill_rx.recv(), if alive => { tracing::info!("received killpill for nativets dedicated worker"); break; }, job = jobs_rx.recv(), if alive => { if let Some(DedicatedWorkerJob { job, flow_runners, done_tx }) = job { let id = job.id; tracing::info!( "received job on nativets dedicated worker for {script_path}: {id}" ); let args = if let Some(args) = job.args.as_ref() { if let Some(x) = transform_json( client, &job.workspace_id, &args.0, &job, &db.into(), ).await? { serde_json::to_string(&x) .unwrap_or_else(|_| "{}".to_string()) } else { serde_json::to_string(&args) .unwrap_or_else(|_| "{}".to_string()) } } else { "{}".to_string() }; // Run the job: preprocess if needed, then execute main. // Uses a labeled block to unify error handling with a single JobCompleted send. let (result, success, preprocessed_args, logs) = 'job: { if let Err(e) = warm.wait_ready().await { break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": format!("isolate init failed: {e}"), "name": "Error"}))), false, None, init_log.clone(), ); } let needs_preprocessing = job.preprocessed == Some(false); let (main_args, preprocessed) = if needs_preprocessing { let Some(ref pre_names) = pre_arg_names else { break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": "preprocessor function is missing", "name": "Error"}))), false, None, init_log.clone(), ); }; let mut pre_isolate = pre_warm.take().unwrap_or_else(|| { PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), pre_names.clone(), Some("preprocessor".to_string()), ) }); if let Err(e) = pre_isolate.wait_ready().await { break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": format!("preprocessor isolate init failed: {e}"), "name": "Error"}))), false, None, init_log.clone(), ); } let pre_executing = pre_isolate.start_execution(args.clone()); // Pipeline: start pre-warming the next preprocessor isolate pre_warm = Some(PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), pre_names.clone(), Some("preprocessor".to_string()), )); let pre_result = match pre_executing.wait().await { Ok(r) => r, Err(e) => break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": format!("preprocessor failed: {e}"), "name": "Error"}))), false, None, init_log.clone(), ), }; if !pre_result.logs.is_empty() { append_logs(&id, &job.workspace_id, pre_result.logs, &db.into()).await; } let raw = match pre_result.result { Ok(r) => r, Err(e) => break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": format!("preprocessor failed: {e}"), "name": "Error"}))), false, None, init_log.clone(), ), }; let preprocessed: HashMap> = match serde_json::from_str(raw.get()) { Ok(v) => v, Err(e) => break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": format!("error deserializing preprocessed args: {e:#}"), "name": "Error"}))), false, None, init_log.clone(), ), }; let main_args = serde_json::to_string(&preprocessed) .unwrap_or_else(|_| "{}".to_string()); (main_args, Some(preprocessed)) } else { (args, None) }; let executing = warm.start_execution(main_args); // Pipeline: start pre-warming the next main isolate warm = PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), arg_names.clone(), None, ); let main_result = match executing.wait().await { Ok(r) => r, Err(e) => break 'job ( Arc::new(to_raw_value(&serde_json::json!({"message": format!("{e}"), "name": "Error"}))), false, preprocessed, init_log.clone(), ), }; let mut logs = init_log.clone(); if !main_result.logs.is_empty() { logs.push_str(&main_result.logs); } let (result, success) = match main_result.result { Ok(raw) => (Arc::new(raw), true), Err(e) => ( Arc::new(to_raw_value(&serde_json::json!({ "message": e, "name": "Error", }))), false, ), }; (result, success, preprocessed, logs) }; append_logs(&id, &job.workspace_id, logs, &db.into()).await; job_completed_tx.send_job(JobCompleted { job: MiniCompletedJob::from(job), result, result_columns: None, mem_peak: 0, resource_usage: None, canceled_by: None, success, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args, has_stream: Some(false), from_cache: None, flow_runners, done_tx, }, true).await?; } else { tracing::debug!("job channel closed for nativets dedicated worker"); break; } } } } Ok(()) } } #[cfg(feature = "private")] pub async fn start_worker( requirements_o: Option, codebase: Option, db: &sqlx::Pool, inner_content: &str, base_internal_url: &str, job_dir: &str, worker_name: &str, envs: HashMap, w_id: &str, script_path: &str, token: &str, job_completed_tx: JobCompletedSender, jobs_rx: Receiver, killpill_rx: tokio::sync::broadcast::Receiver<()>, client: windmill_common::client::AuthedClient, concurrency_semaphore: Option>, ) -> Result<()> { let mut logs = "".to_string(); let mut mem_peak: i32 = 0; let mut canceled_by: Option = None; tracing::info!("Starting worker {w_id};{script_path} (codebase: {codebase:?}"); if !codebase.is_some() { let _ = write_file(job_dir, "main.ts", inner_content)?; } let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(Some(&base_internal_url)).await; let mut annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content); let context = variables::get_reserved_variables( &Connection::from(db.clone()), w_id, &token, "dedicated_worker@windmill.dev", "dedicated_worker", "NOT_AVAILABLE", "dedicated_worker", Some(script_path.to_string()), None, None, None, None, None, None, None, None, None, None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; if annotation.native { // Native (V8) dedicated worker: bundle the code and dispatch to V8 instead of a subprocess. let main_code = remove_pinned_imports(inner_content)?; write_file(job_dir, "main.ts", &main_code)?; if let Some(reqs) = requirements_o.as_ref() { let (pkg, lock, empty, is_binary) = split_lockfile(reqs); write_file(job_dir, "package.json", pkg)?; if let Some(lock) = lock { if !empty { write_lock(lock, job_dir, is_binary).await?; install_bun_lockfile( &mut mem_peak, &mut canceled_by, &Uuid::nil(), w_id, Some(&Connection::from(db.clone())), job_dir, worker_name, common_bun_proc_envs.clone(), annotation.npm, &mut None, false, ) .await?; } } } build_loader( job_dir, base_internal_url, token, w_id, script_path, LoaderMode::BrowserBundle, &None, ) .await?; generate_bun_bundle( job_dir, w_id, &Uuid::nil(), worker_name, Some(&Connection::from(db.clone())), None, &mut mem_peak, &mut canceled_by, &common_bun_proc_envs, &mut None, ) .await?; let js_code = read_file_content(&format!("{job_dir}/main.js")).await?; let reserved_variables: HashMap = context .iter() .map(|x| (x.name.clone(), x.value.clone())) .collect(); let env_code = build_nativets_env_code( base_internal_url, &reserved_variables, annotation.no_network, ); return handle_dedicated_bunnative( inner_content, &js_code, &env_code, token, worker_name, w_id, script_path, db, jobs_rx, killpill_rx, job_completed_tx, &client, ) .await; } let mut format = BundleFormat::Cjs; if let Some(codebase) = codebase.as_ref() { let pulled_codebase = pull_codebase(w_id, codebase, job_dir).await?; if pulled_codebase.is_esm { format = BundleFormat::Esm; } } else if let Some(reqs) = requirements_o { let (pkg, lock, empty, is_binary) = split_lockfile(&reqs); if lock.is_none() { return Err(error::Error::ExecutionErr( format!("Invalid requirements, expected to find //bun.lockb split pattern in reqs. Found: |{reqs}|") )); } let _ = write_file(job_dir, "package.json", pkg)?; let lock = lock.unwrap(); if !empty { if is_binary { let _ = write_file_binary( job_dir, "bun.lockb", &base64::engine::general_purpose::STANDARD .decode(lock) .map_err(|_| { error::Error::internal_err("Could not decode bun.lockb".to_string()) })?, ) .await?; } else { write_file(job_dir, "bun.lock", lock)?; } install_bun_lockfile( &mut mem_peak, &mut canceled_by, &Uuid::nil(), &w_id, Some(&Connection::from(db.clone())), job_dir, worker_name, common_bun_proc_envs.clone(), annotation.npm, &mut None, false, ) .await?; tracing::info!("dedicated worker requirements installed: {reqs}"); } } else if is_sandboxing_enabled() { logs.push_str("\n\n--- BUN INSTALL ---\n"); let _ = gen_bun_lockfile( &mut mem_peak, &mut canceled_by, &Uuid::nil(), &w_id, Some(&Connection::from(db.clone())), token, &script_path, job_dir, base_internal_url, worker_name, false, &WorkspaceDependenciesPrefetched::extract( inner_content, ScriptLang::Bun, w_id, &None, &script_path, db.into(), ) .await?, annotation.npm, &mut None, &None, false, ) .await?; } let main_code = remove_pinned_imports(inner_content)?; let _ = write_file(job_dir, "main.ts", &main_code)?; let codegen = compute_ts_codegen(inner_content); let wrapper_ext = if codebase.is_some() { "js" } else { "ts" }; { let scripts = [ TsScriptEntry { import_name: "main", original_path: script_path, codegen: &codegen, }, ]; let wrapper_content = generate_multi_script_wrapper(&scripts, wrapper_ext); write_file(job_dir, "wrapper.mjs", &wrapper_content)?; } if format == BundleFormat::Esm { annotation.nodejs = false; } if !codebase.is_some() || format == BundleFormat::Esm { build_loader( job_dir, base_internal_url, token, w_id, script_path, if annotation.nodejs { LoaderMode::Node } else { LoaderMode::Bun }, &None, ) .await?; } if annotation.nodejs && !codebase.is_some() { generate_wrapper_mjs( job_dir, w_id, &Uuid::nil(), worker_name, &Connection::from(db.clone()), None, &mut mem_peak, &mut canceled_by, &common_bun_proc_envs, &mut None, ) .await?; } if annotation.nodejs { let wrapper_path = format!("{job_dir}/wrapper.mjs"); handle_dedicated_process( &*NODE_BIN_PATH, job_dir, context_envs, envs, context, common_bun_proc_envs, vec![&wrapper_path], killpill_rx, job_completed_tx, token, jobs_rx, worker_name, db, script_path, "nodejs", client, false, concurrency_semaphore, ) .await } else { handle_dedicated_process( &*BUN_PATH, job_dir, context_envs, envs, context, common_bun_proc_envs, vec![ "run", "--preserve-symlinks", "-i", "--prefer-offline", "-r", "./loader.bun.js", &format!("{job_dir}/wrapper.mjs"), ], killpill_rx, job_completed_tx, token, jobs_rx, worker_name, db, script_path, "bun", client, false, concurrency_semaphore, ) .await } } #[cfg(test)] mod tests { use super::*; #[test] fn bun_install_stages_only_self_contained_manifests() { let lock = r#"{"lockfileVersion": 1, "workspaces": {"": {"dependencies": {"zod": "^4"}}}}"#; assert!(bun_install_can_stage( r#"{"dependencies": {"zod": "^4"}}"#, Some(lock) )); for manifest in [ r#"{"dependencies": {"x": "file:./x"}}"#, r#"{"dependencies": {"x": "../x"}}"#, r#"{"overrides": {"y": {"x": "link:x"}}}"#, r#"{"workspaces": ["packages/*"]}"#, r#"{"patchedDependencies": {"zod@4.0.0": "patches/zod.patch"}}"#, r#"{"scripts": {"postinstall": "test -f main.ts"}}"#, "not json", ] { assert!(!bun_install_can_stage(manifest, None), "{manifest}"); } let local_lock = r#"{"packages": {"x": ["x@file:x", {}]}}"#; assert!(!bun_install_can_stage( r#"{"dependencies": {}}"#, Some(local_lock) )); } // Install jails can write the bun cache, so the worker must never follow a path // through it: neither a staging root nor a staging dir swapped for a symlink may // lead its cleanup outside the cache. #[cfg(target_os = "linux")] #[test] fn bun_staging_cleanup_does_not_follow_symlinks_out_of_the_cache() { let tmp = tempfile::tempdir().unwrap(); let (cache, job, outside) = ( tmp.path().join("cache"), tmp.path().join("job"), tmp.path().join("outside"), ); for d in [&cache, &job, &outside.join("victim")] { fs::create_dir_all(d).unwrap(); } fs::write(job.join("package.json"), r#"{"dependencies": {}}"#).unwrap(); // Old enough for the stale sweep if it were reached through the symlink. fs::File::open(outside.join("victim")) .unwrap() .set_modified(std::time::SystemTime::UNIX_EPOCH) .unwrap(); std::os::unix::fs::symlink(&outside, cache.join(".wm-staging")).unwrap(); let staging = bun_staging_dir(cache.to_str().unwrap(), job.to_str().unwrap()).expect("staging dir"); assert!(outside.join("victim").exists()); assert!(fs::symlink_metadata(cache.join(".wm-staging")) .unwrap() .is_dir()); let staged = cache.join(".wm-staging").join(&staging.name); fs::remove_dir_all(&staged).unwrap(); std::os::unix::fs::symlink(&outside, &staged).unwrap(); drop(staging); assert!(outside.join("victim").exists()); assert!(fs::symlink_metadata(&staged).is_err()); } #[test] fn test_bun_lockfile_version() { assert_eq!(bun_lockfile_version(EMPTY_V1_BUN_LOCK), Some(1)); assert_eq!( bun_lockfile_version("{\n \"lockfileVersion\": 2,\n \"packages\": {}\n}"), Some(2) ); // bun raises the version on its own for overrides/catalogs, so any version has to parse assert_eq!(bun_lockfile_version("{\"lockfileVersion\":3}"), Some(3)); assert_eq!( bun_lockfile_version("{ \"lockfileVersion\" : 42 }"), Some(42) ); assert_eq!(bun_lockfile_version("{\"packages\":{}}"), None); } #[test] fn test_split_lockfile_text_unix() { let lockfile = r#"{"dependencies":{"lodash":"^4.17.21"}} //bun.lock lockfile-content-here"#; let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!(pkg, r#"{"dependencies":{"lodash":"^4.17.21"}}"#); assert_eq!(lock, Some("lockfile-content-here")); assert!(!is_empty); assert!(!is_binary); } #[test] fn test_split_lockfile_text_windows() { let lockfile = "{\"dependencies\":{}}\r\n//bun.lock\r\nlockfile-content"; let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!(pkg, "{\"dependencies\":{}}"); assert_eq!(lock, Some("lockfile-content")); assert!(!is_empty); assert!(!is_binary); } #[test] fn test_split_lockfile_binary_unix() { let lockfile = r#"{"dependencies":{}} //bun.lockb YmluYXJ5LWNvbnRlbnQ="#; // base64 encoded "binary-content" let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!(pkg, r#"{"dependencies":{}}"#); assert_eq!(lock, Some("YmluYXJ5LWNvbnRlbnQ=")); assert!(!is_empty); assert!(is_binary); } #[test] fn test_split_lockfile_binary_windows() { let lockfile = "{\"dependencies\":{}}\r\n//bun.lockb\r\nYmluYXJ5LWNvbnRlbnQ="; let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!(pkg, "{\"dependencies\":{}}"); assert_eq!(lock, Some("YmluYXJ5LWNvbnRlbnQ=")); assert!(!is_empty); assert!(is_binary); } #[test] fn test_split_lockfile_empty() { let lockfile = r#"{"dependencies":{}} //bun.lock "#; let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!(pkg, r#"{"dependencies":{}}"#); assert_eq!(lock, Some(EMPTY_FILE)); assert!(is_empty); assert!(!is_binary); } #[test] fn test_split_lockfile_no_lock() { let lockfile = r#"{"dependencies":{"lodash":"^4.17.21"}}"#; let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!(pkg, r#"{"dependencies":{"lodash":"^4.17.21"}}"#); assert!(lock.is_none()); assert!(!is_empty); assert!(!is_binary); } // Samples are npm 12.2.0 output: a format change there silently turns the fetch fallback // off, or makes it override an explicit `none`. #[test] fn npm_fetch_fallback_reads_npm_output() { let log = r#"22 verbose exit 1 23 error code EALLOWGIT 24 error Fetching packages of type "git" have been disabled 25 error Refusing to fetch "wm-git-test@git+file:///work/git/wm-git-test.git" 26 verbose exit 1"#; let (fallback, refused) = parse_npm_refused_fetch(log.lines().map(str::to_string)) .expect("EALLOWGIT is a fetch refusal"); assert_eq!(fallback.key(), "allow-git"); assert_eq!( refused, [r#""wm-git-test@git+file:///work/git/wm-git-test.git""#] ); assert!( parse_npm_refused_fetch(["23 error code EJSONPARSE".to_string()].into_iter()).is_none() ); let config_ls = r#"; "global" config from /usr/etc/npmrc ; allow-git = "none" ; overridden by project ; "project" config from /work/j/.npmrc allow-remote = "all" registry = "http://127.0.0.1:4999/" ; npm version = 12.2.0"#; assert!(npm_config_ls_sets(config_ls, "allow-git")); assert!(npm_config_ls_sets(config_ls, "allow-remote")); assert!(!npm_config_ls_sets(config_ls, "allow-scripts")); } #[test] fn test_split_lockfile_multiline_package_json() { let lockfile = r#"{ "dependencies": { "lodash": "^4.17.21" } } //bun.lock lockfile-content"#; let (pkg, lock, is_empty, is_binary) = split_lockfile(lockfile); assert_eq!( pkg, r#"{ "dependencies": { "lodash": "^4.17.21" } }"# ); assert_eq!(lock, Some("lockfile-content")); assert!(!is_empty); assert!(!is_binary); } #[test] fn test_compute_ts_codegen_basic_args() { let code = r#"export function main(x: string, y: number) { return x; }"#; let cg = compute_ts_codegen(code); assert_eq!(cg.spread, "x, y"); assert!(cg.date_conversions.is_empty()); assert!(cg.preprocessor_spread.is_none()); } #[test] fn test_compute_ts_codegen_with_datetime() { let code = r#"export function main(name: string, created_at: Date, count: number) { return name; }"#; let cg = compute_ts_codegen(code); assert_eq!(cg.spread, "name, created_at, count"); assert!(cg.date_conversions.contains("created_at")); assert!(cg.date_conversions.contains("new Date")); } #[test] fn test_compute_ts_codegen_with_preprocessor() { let code = r#" export function main(x: string, ts: Date) { return x; } export function preprocessor(input: string, when: Date) { return { x: input, ts: when }; } "#; let cg = compute_ts_codegen(code); assert_eq!(cg.spread, "x, ts"); assert!(cg.date_conversions.contains("ts")); assert_eq!(cg.preprocessor_spread.as_deref(), Some("input, when")); assert!(cg .preprocessor_date_conversions .as_ref() .unwrap() .contains("when")); } #[test] fn test_compute_ts_codegen_no_args() { let code = r#"export function main() { return 42; }"#; let cg = compute_ts_codegen(code); assert!(cg.spread.is_empty()); assert!(cg.date_conversions.is_empty()); assert!(cg.preprocessor_spread.is_none()); } #[test] fn test_wrapper_contains_execd_protocol() { let code = r#"export function main(x: number) { return x; }"#; let cg = compute_ts_codegen(code); let scripts = vec![TsScriptEntry { import_name: "main", original_path: "test/script", codegen: &cg }]; let wrapper = generate_multi_script_wrapper(&scripts, "ts"); // Single-script wrapper must support execd: (direct, no path) assert!(wrapper.contains(r#"line.startsWith("execd:")"#)); // Must also support exec: for backward compat / runner groups assert!(wrapper.contains(r#"line.startsWith("exec:")"#)); // Must register the script in the map assert!(wrapper.contains(r#"scripts.set("test/script""#)); } #[test] fn test_wrapper_contains_execd_preprocess_protocol() { let code = r#"export function preprocessor(x: number) { return { x }; } export function main(x: number) { return x; }"#; let cg = compute_ts_codegen(code); let scripts = vec![TsScriptEntry { import_name: "main", original_path: "test/script", codegen: &cg }]; let wrapper = generate_multi_script_wrapper(&scripts, "ts"); assert!(wrapper.contains(r#"line.startsWith("execd_preprocess:")"#)); assert!(wrapper.contains(r#"line.startsWith("execd:")"#)); assert!(wrapper.contains(r#"line.startsWith("exec_preprocess:")"#)); assert!(wrapper.contains(r#"line.startsWith("exec:")"#)); } /// The bundle cache is global and content-keyed, so a key that ignores the inline /// modules hands one workspace's bundle — attacker helper code and all — to the next /// job whose main content and lockfile happen to match. #[tokio::test] async fn bundle_cache_key_separates_inline_module_content() { use windmill_common::scripts::ScriptModule; async fn key_for(modules: Option<&HashMap>) -> String { compute_bundle_local_and_remote_path( "import { h } from './helper.ts';\nexport async function main() { return h(); }", "{}\n//bun.lock\n", "u/alice/script", None, "w1", &None, modules, ) .await .1 } fn modules(content: &str) -> HashMap { HashMap::from([( "helper.ts".to_string(), ScriptModule { content: content.to_string(), language: ScriptLang::Bun, lock: None, }, )]) } let attacker = key_for(Some(&modules("export const h = () => 'attacker'"))).await; let victim = key_for(Some(&modules("export const h = () => 'victim'"))).await; assert_ne!(attacker, victim); assert_eq!( attacker, key_for(Some(&modules("export const h = () => 'attacker'"))).await, "same modules must still share a cache slot" ); // An absent map and an empty one are the same script, so they share a slot. assert_eq!(key_for(None).await, key_for(Some(&HashMap::new())).await); assert_ne!(key_for(None).await, attacker); } }