From c28314f424ea0e04b86565ce88e6c91e0df1a0cf Mon Sep 17 00:00:00 2001 From: hugocasa Date: Wed, 25 Mar 2026 16:13:04 +0100 Subject: [PATCH] feat: runner groups for shared-process multi-script dedicated workers (#8434) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add runner groups for shared-process multi-script dedicated workers Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: unify dedicated worker and runner group wrappers into single multi-script wrapper Replace per-language single-script wrappers with the unified load/exec/exec_preprocess/end protocol. Each start_worker() now writes scripts to scripts// and uses generate_multi_script_wrapper(). handle_dedicated_process() sends load: on start and exec: per job instead of raw JSON args. Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: merge runner groups into dedicated workers with inline arg metadata Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref to match EE branch Co-Authored-By: Claude Opus 4.6 (1M context) * fix: gate EE-only functions behind cfg(feature = "private") to fix OSS dead_code errors Co-Authored-By: Claude Opus 4.6 (1M context) * feat: auto-detect runner groups from workspace dependency annotations - New endpoint GET /scripts/list_dedicated_with_deps: returns dedicated scripts with parsed workspace dependency names from content annotations - Frontend: show dep badges in DedicatedWorkersSelector with links to workspace settings, warn when referenced dep doesn't exist, group scripts sharing deps into "Shared runner" sections - Remove manual "Runner groups" tab and RunnerGroupSelector component - Remove runner_groups from WorkerConfigOpt/WorkerConfig (auto-detected) - Fix Node.js single dedicated workers: transpile main.ts -> main.js via Bun.build so the multi-script wrapper's dynamic import() works under Node - Add package.json with type:module in scripts dir to silence Node warning Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: unify dedicated worker wrappers with baked-in codegen and routing Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * test: add e2e tests for multi-script dedicated worker routing (bun, deno, python) Co-Authored-By: Claude Opus 4.6 (1M context) * chore: remove dead generate_dedicated_worker_wrapper function Co-Authored-By: Claude Opus 4.6 (1M context) * feat: add dependency installation to runner groups + make dep functions pub(crate) Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * fix: prevent bun loader from intercepting absolute paths within cwd When a plugin's onResolve returns an absolute path, Bun re-invokes the resolver with that path. The loader was then routing it through the remote URL resolver, breaking runner group script imports. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use _wm_ prefix for runner group scripts to avoid bun loader interception Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: extract DENO_UNSTABLE_ARGS constant to avoid repeating flags Co-Authored-By: Claude Opus 4.6 (1M context) * chore: regenerate system prompts Co-Authored-By: Claude Opus 4.6 (1M context) * fix: gate private-only exports behind cfg(feature = "private") for OSS build Co-Authored-By: Claude Opus 4.6 (1M context) * fix: move format strings before handle_dedicated_process to fix lifetime Co-Authored-By: Claude Opus 4.6 (1M context) * chore: regenerate sqlx offline cache Co-Authored-By: Claude Opus 4.6 (1M context) * fix sqlx * fix: skip empty lines in deno e2e tests (double newline from console.log + '\n') Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use dict() instead of {{}} in python wrapper to avoid set literal {{{{}}}} in format!() produces {{}} which Python interprets as an empty set, not a dict. Use dict() which is unambiguous. Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: remove deno from runner groups and associated tests Deno resolves dependencies at runtime via URLs/import maps, so there's no shared node_modules/pip install to benefit from runner groups. Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: revert deno wrapper to inline old-style with exec: protocol Since deno doesn't support runner groups, the unified multi-script wrapper is unnecessary. Reverted to the old inline wrapper from main but adapted to use the exec:: protocol. Co-Authored-By: Claude Opus 4.6 (1M context) * refactor: extract deno wrapper into reusable function and add e2e tests Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use codebase presence (not nodejs annotation) to determine wrapper import extension On main, codebase scripts import ./main.js (pre-bundled JS). The wrapper_ext was incorrectly based on annotation.nodejs. Co-Authored-By: Claude Opus 4.6 (1M context) * feat: improve dedicated workers UI - combine lists, better badges, tooltips - Merge shared runners section with selected tags into one unified list - Move language tag to right side of selector for alignment - Change dep badge color from dark-gray to indigo - Add tooltip on yellow warning badge explaining missing workspace dep Co-Authored-By: Claude Opus 4.6 (1M context) * feat: group shared runners visually in dedicated workers list - Runner groups shown with a header (Shared runner · language · dep badge) - Scripts in the same group nested under the header - Standalone scripts/flows shown after groups - Used Svelte snippet for reusable tag row rendering Co-Authored-By: Claude Opus 4.6 (1M context) * fix: improve visual separation between shared runner groups and standalone items Co-Authored-By: Claude Opus 4.6 (1M context) * feat: give standalone runners same header style as shared runners - Each standalone script/flow gets its own header row with bg-surface-secondary - Header shows "Dedicated runner" / "Flow runner" label, dep link, language badge - Shared runner header: swapped language and dep badge positions - Dep shown as inline link instead of badge in headers for cleaner look Co-Authored-By: Claude Opus 4.6 (1M context) * feat: inline standalone runner path in header, language badge on right edge, no max height - Standalone items: path shown directly in header row (no sub-row) - Language badge placed after flex-1 spacer (right-aligned) - Removed max-h-64 overflow constraint from the list Co-Authored-By: Claude Opus 4.6 (1M context) * feat: consistent badges across runner list - dep+language on right, depBadge snippet - Shared runner scripts: show (workspace) and language badge on right - Standalone items: dep badges and language badge on right (after flex-1) - Shared runner header: dep badge and language badge on right - Extract depBadge snippet to deduplicate dep badge rendering - Picker selector also uses depBadge snippet Co-Authored-By: Claude Opus 4.6 (1M context) * fix: show language badge on standalone items, hide from shared runner sub-items - Fetch script language from API when not available from workspace deps - Hide dep+language badges from tagRow when script is inside a runner group (already shown in the group header) - Standalone items now always show language badge Co-Authored-By: Claude Opus 4.6 (1M context) * fix: differentiate badge colors - gray for language, indigo for workspace deps Matches codebase convention: gray for metadata (like script hashes), indigo for linkable features/entities. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use transparent (bordered) badge for language - visible on all backgrounds Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use gray badge for language everywhere Co-Authored-By: Claude Opus 4.6 (1M context) * fix: revert skills.ts and AI files, add _wm_ exclusion to Windows loader - Revert cli/src/guidance/skills.ts to main (not our change) - Revert AI provider formatting changes (not our change) - Add _wm_ prefix exclusion to loader.bun.windows.js filterResolve Co-Authored-By: Claude Opus 4.6 (1M context) * fix: update ee-repo-ref and regenerate system prompts after merge Co-Authored-By: Claude Opus 4.6 (1M context) * perf: use DISTINCT ON in list_dedicated_with_deps to dedup at DB level Avoids fetching all script versions and deduplicating in Rust. Addresses PR review feedback. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use sqlx query! macro for list_dedicated_with_deps and regenerate cache Co-Authored-By: Claude Opus 4.6 (1M context) * fix: dedicated worker review fixes and test coverage - Fix Python relative imports in dedicated workers (write loader.py, add import loader to wrapper when needed) - Move Python colon parsing inside try/except to prevent crashes on malformed stdin - Add indexOf guard in Bun/Deno wrappers for malformed protocol messages - Add stderr logging for unrecognized stdin commands in all wrappers - Remove asyncio handling from Python wrapper (consistent with normal path) - Add exec_preprocess protocol tests for Bun, Deno, and Python - Add argument transformation tests (dates, bytes, kwargs, sentinel) - Add relative import detection test for Python wrapper - Add PreprocessedArgs variant to DedicatedWorkerResult test helper Co-Authored-By: Claude Opus 4.6 (1M context) * fix: remove symlink from git and gate has_relative_imports behind private feature Co-Authored-By: Claude Opus 4.6 (1M context) * fix: update ee-repo-ref for dedicated_worker_ee.rs changes Co-Authored-By: Claude Opus 4.6 (1M context) * fix: add mixed exec+preprocess test to use ProtocolCmd::Exec variant Co-Authored-By: Claude Opus 4.6 (1M context) * fix: remove hanging deno missing-preprocessor test The Deno wrapper only generates the exec_preprocess handler when the script has a preprocessor function. Without one, the message is unrecognized and the test hangs reading stdout. Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref to 182943e5ad9bf2a905ccdf07d4e346437fb329a9 This commit updates the EE repository reference after PR #466 was merged in windmill-ee-private. Previous ee-repo-ref: 995f701fe3754be6260fc6b679e5de8fc636e68a New ee-repo-ref: 182943e5ad9bf2a905ccdf07d4e346437fb329a9 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 4.6 (1M context) Co-authored-by: windmill-internal-app[bot] --- ...6be1bceb79c82e7b5542f17b23b6d70cc02d6.json | 104 +++ ...7d5ee9490240a627b20a1037444845e39c5f.json} | 4 +- backend/ee-repo-ref.txt | 2 +- backend/tests/bun_jobs.rs | 769 +++++++++++++++++- backend/tests/python_jobs.rs | 461 +++++++++++ backend/windmill-api-scripts/src/scripts.rs | 69 +- backend/windmill-api/openapi.yaml | 54 ++ backend/windmill-common/src/ai_google.rs | 40 +- backend/windmill-common/src/worker.rs | 12 +- backend/windmill-test-utils/src/lib.rs | 10 + backend/windmill-worker/loader.bun.js | 2 +- backend/windmill-worker/loader.bun.windows.js | 2 +- backend/windmill-worker/src/bun_executor.rs | 360 +++++--- backend/windmill-worker/src/deno_executor.rs | 284 +++---- backend/windmill-worker/src/lib.rs | 19 +- .../windmill-worker/src/python_executor.rs | 499 +++++++++--- .../DedicatedWorkersSelector.svelte | 412 ++++++++-- 17 files changed, 2590 insertions(+), 513 deletions(-) create mode 100644 backend/.sqlx/query-f2fa27ed5020aa9c085176b25466be1bceb79c82e7b5542f17b23b6d70cc02d6.json rename backend/.sqlx/{query-d05f20431cd08f737bfbf904efedfdf104e3d77b0725c5355305d19f67359e90.json => query-fd4c5391107af34a3bf9b83b0c3f7d5ee9490240a627b20a1037444845e39c5f.json} (50%) diff --git a/backend/.sqlx/query-f2fa27ed5020aa9c085176b25466be1bceb79c82e7b5542f17b23b6d70cc02d6.json b/backend/.sqlx/query-f2fa27ed5020aa9c085176b25466be1bceb79c82e7b5542f17b23b6d70cc02d6.json new file mode 100644 index 0000000000..a0d7f223d0 --- /dev/null +++ b/backend/.sqlx/query-f2fa27ed5020aa9c085176b25466be1bceb79c82e7b5542f17b23b6d70cc02d6.json @@ -0,0 +1,104 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT ON (path) path, language AS \"language: ScriptLang\", content FROM script\n WHERE workspace_id = $1\n AND archived = false\n AND dedicated_worker = true\n AND language = ANY($2::SCRIPT_LANG[])\n ORDER BY path, created_at DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + }, + { + "ordinal": 2, + "name": "content", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + { + "Custom": { + "name": "script_lang[]", + "kind": { + "Array": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + } + } + } + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "f2fa27ed5020aa9c085176b25466be1bceb79c82e7b5542f17b23b6d70cc02d6" +} diff --git a/backend/.sqlx/query-d05f20431cd08f737bfbf904efedfdf104e3d77b0725c5355305d19f67359e90.json b/backend/.sqlx/query-fd4c5391107af34a3bf9b83b0c3f7d5ee9490240a627b20a1037444845e39c5f.json similarity index 50% rename from backend/.sqlx/query-d05f20431cd08f737bfbf904efedfdf104e3d77b0725c5355305d19f67359e90.json rename to backend/.sqlx/query-fd4c5391107af34a3bf9b83b0c3f7d5ee9490240a627b20a1037444845e39c5f.json index e778c17bf6..50cf586c79 100644 --- a/backend/.sqlx/query-d05f20431cd08f737bfbf904efedfdf104e3d77b0725c5355305d19f67359e90.json +++ b/backend/.sqlx/query-fd4c5391107af34a3bf9b83b0c3f7d5ee9490240a627b20a1037444845e39c5f.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO token\n (token_hash, token_prefix, token, label, super_admin, email)\n VALUES ($1, $2, $3, $4, $5, $6)", + "query": "INSERT INTO token\n (token_hash, token_prefix, token, label, super_admin, email)\n VALUES ($1, $2, $3, $4, $5, $6)", "describe": { "columns": [], "parameters": { @@ -15,5 +15,5 @@ }, "nullable": [] }, - "hash": "d05f20431cd08f737bfbf904efedfdf104e3d77b0725c5355305d19f67359e90" + "hash": "fd4c5391107af34a3bf9b83b0c3f7d5ee9490240a627b20a1037444845e39c5f" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 7d86a6114e..263ec1de9a 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -b5c8af4df9ba2c39fdd494d7a40f9a92fbff8abc +182943e5ad9bf2a905ccdf07d4e346437fb329a9 diff --git a/backend/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index 2edfef8989..7183cf6e10 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -891,25 +891,34 @@ mod dedicated_worker_protocol { use std::process::{Command, Stdio}; use windmill_test_utils::{parse_dedicated_worker_line, DedicatedWorkerResult}; use windmill_worker::{ - build_loader, generate_dedicated_worker_wrapper, LoaderMode, BUN_DEDICATED_WORKER_ARGS, - BUN_PATH, NODE_BIN_PATH, + build_loader, compute_ts_codegen, generate_multi_script_wrapper, LoaderMode, TsScriptEntry, + BUN_DEDICATED_WORKER_ARGS, BUN_PATH, NODE_BIN_PATH, }; + const TEST_SCRIPT_PATH: &str = "f/test/script"; + /// Creates test worker files and optionally bundles for Node.js (like production) /// Returns the path to the wrapper file to execute fn create_test_worker_files( dir: &std::path::Path, script: &str, - arg_names: &[&str], bundle_for_node: bool, ) -> std::path::PathBuf { let dir_str = dir.to_str().unwrap(); + // Write main.ts at root (like production single-script) std::fs::write(dir.join("main.ts"), script).unwrap(); + let codegen = compute_ts_codegen(script); + let ext = if bundle_for_node { "js" } else { "ts" }; + let scripts = [TsScriptEntry { + import_name: "main", + original_path: TEST_SCRIPT_PATH, + codegen: &codegen, + }]; + let wrapper = generate_multi_script_wrapper(&scripts, ext); + if bundle_for_node { - // For Node.js: bundle to JavaScript first (like production's build_loader with LoaderMode::Node) - let wrapper = generate_dedicated_worker_wrapper(arg_names, "./main.js", None, None); - std::fs::write(dir.join("wrapper.mjs"), wrapper).unwrap(); + std::fs::write(dir.join("wrapper.mjs"), &wrapper).unwrap(); // Use the exact same build_loader function as production tokio::runtime::Runtime::new() @@ -919,7 +928,7 @@ mod dedicated_worker_protocol { "http://localhost:8000", "test_token", "test-workspace", - "f/test/script", + TEST_SCRIPT_PATH, LoaderMode::Node, &None, )) @@ -945,10 +954,8 @@ mod dedicated_worker_protocol { std::fs::rename(&bundled_path, &output_path).unwrap(); output_path } else { - // For Bun: use TypeScript directly (like production) - let wrapper = generate_dedicated_worker_wrapper(arg_names, "./main.ts", None, None); let wrapper_path = dir.join("wrapper.mjs"); - std::fs::write(&wrapper_path, wrapper).unwrap(); + std::fs::write(&wrapper_path, &wrapper).unwrap(); wrapper_path } } @@ -957,14 +964,12 @@ mod dedicated_worker_protocol { fn run_worker_test( runtime: &str, script: &str, - arg_names: &[&str], jobs: Vec, ) -> Vec> { let temp_dir = tempfile::tempdir().unwrap(); // Create files and get the wrapper path (bundled for node, raw for bun) - let wrapper_path = - create_test_worker_files(temp_dir.path(), script, arg_names, runtime == "node"); + let wrapper_path = create_test_worker_files(temp_dir.path(), script, runtime == "node"); let wrapper_str = wrapper_path.to_str().unwrap(); // Build args matching production behavior @@ -1008,7 +1013,8 @@ mod dedicated_worker_protocol { let mut results = Vec::new(); for job_args in jobs { - writeln!(stdin, "{}", job_args.to_string()).unwrap(); + // Protocol: exec:: + writeln!(stdin, "exec:{}:{}", TEST_SCRIPT_PATH, job_args.to_string()).unwrap(); stdin.flush().unwrap(); let mut response = String::new(); @@ -1043,12 +1049,7 @@ export function main(x: number, y: number): number { return x + y; } "#; - let results = run_worker_test( - "node", - script, - &["x", "y"], - vec![serde_json::json!({"x": 5, "y": 3})], - ); + let results = run_worker_test("node", script, vec![serde_json::json!({"x": 5, "y": 3})]); assert_eq!(results.len(), 1); assert_eq!(results[0], Ok(serde_json::json!(8))); @@ -1062,7 +1063,7 @@ export function main(n: number): number { } "#; let jobs: Vec = (1..=5).map(|i| serde_json::json!({"n": i})).collect(); - let results = run_worker_test("node", script, &["n"], jobs); + let results = run_worker_test("node", script, jobs); assert_eq!(results.len(), 5); for (i, result) in results.iter().enumerate() { @@ -1081,7 +1082,6 @@ export function main(msg: string): never { let results = run_worker_test( "node", script, - &["msg"], vec![serde_json::json!({"msg": "test error"})], ); @@ -1099,12 +1099,7 @@ export function main(x: number, y: number): number { return x + y; } "#; - let results = run_worker_test( - "bun", - script, - &["x", "y"], - vec![serde_json::json!({"x": 5, "y": 3})], - ); + let results = run_worker_test("bun", script, vec![serde_json::json!({"x": 5, "y": 3})]); assert_eq!(results.len(), 1); assert_eq!(results[0], Ok(serde_json::json!(8))); @@ -1118,7 +1113,7 @@ export function main(n: number): number { } "#; let jobs: Vec = (1..=5).map(|i| serde_json::json!({"n": i})).collect(); - let results = run_worker_test("bun", script, &["n"], jobs); + let results = run_worker_test("bun", script, jobs); assert_eq!(results.len(), 5); for (i, result) in results.iter().enumerate() { @@ -1137,7 +1132,6 @@ export function main(msg: string): never { let results = run_worker_test( "bun", script, - &["msg"], vec![serde_json::json!({"msg": "test error"})], ); @@ -1145,6 +1139,721 @@ export function main(msg: string): never { assert!(results[0].is_err()); assert_eq!(results[0], Err("test error".to_string())); } + + // ==================== Multi-Script (Runner Group) Tests ==================== + + /// Job to send to a specific script in a multi-script wrapper + struct MultiScriptJob { + script_path: String, + args: serde_json::Value, + } + + /// Creates a multi-script wrapper with multiple scripts as flat files, returns the wrapper path + fn create_multi_script_worker_files( + dir: &std::path::Path, + scripts: &[(&str, &str)], // (original_path, script_content) + ) -> std::path::PathBuf { + let mut entries_data = Vec::new(); + for (path, content) in scripts { + let safe_name = format!("_wm_{}", path.replace('/', "__")); + std::fs::write(dir.join(format!("{safe_name}.ts")), content).unwrap(); + entries_data.push((safe_name, path.to_string(), compute_ts_codegen(content))); + } + + let entries: Vec> = entries_data + .iter() + .map(|(safe, path, cg)| TsScriptEntry { + import_name: safe.as_str(), + original_path: path.as_str(), + codegen: cg, + }) + .collect(); + + let wrapper = generate_multi_script_wrapper(&entries, "ts"); + let wrapper_path = dir.join("wrapper.mjs"); + std::fs::write(&wrapper_path, &wrapper).unwrap(); + wrapper_path + } + + /// Helper to run a multi-script dedicated worker test + fn run_multi_script_worker_test( + scripts: &[(&str, &str)], + jobs: Vec, + ) -> Vec> { + let temp_dir = tempfile::tempdir().unwrap(); + let wrapper_path = create_multi_script_worker_files(temp_dir.path(), scripts); + let wrapper_str = wrapper_path.to_str().unwrap(); + + let mut cmd_args: Vec<&str> = BUN_DEDICATED_WORKER_ARGS.to_vec(); + cmd_args.push(wrapper_str); + + let mut child = Command::new(BUN_PATH.as_str()) + .args(cmd_args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .current_dir(temp_dir.path()) + .spawn() + .expect("Failed to spawn worker process"); + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + + // Wait for "start" signal + let mut start_line = String::new(); + reader.read_line(&mut start_line).unwrap(); + assert_eq!( + parse_dedicated_worker_line(start_line.trim()), + DedicatedWorkerResult::Start, + "Expected 'start', got: {}", + start_line.trim() + ); + + let mut results = Vec::new(); + + for job in &jobs { + writeln!(stdin, "exec:{}:{}", job.script_path, job.args.to_string()).unwrap(); + stdin.flush().unwrap(); + + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + + match parse_dedicated_worker_line(response.trim()) { + DedicatedWorkerResult::Success(value) => results.push(Ok(value)), + DedicatedWorkerResult::Error(err) => { + let msg = err["message"] + .as_str() + .unwrap_or("Unknown error") + .to_string(); + results.push(Err(msg)); + } + other => panic!("Unexpected response: {:?}", other), + } + } + + writeln!(stdin, "end").unwrap(); + stdin.flush().unwrap(); + let _ = child.wait().expect("Worker process failed to exit"); + + results + } + + #[test] + fn test_multi_script_routing_basic() { + let script_add = r#" +export function main(a: number, b: number): number { + return a + b; +} +"#; + let script_mul = r#" +export function main(x: number, y: number): number { + return x * y; +} +"#; + let results = run_multi_script_worker_test( + &[("f/math/add", script_add), ("f/math/mul", script_mul)], + vec![ + MultiScriptJob { + script_path: "f/math/add".to_string(), + args: serde_json::json!({"a": 3, "b": 4}), + }, + MultiScriptJob { + script_path: "f/math/mul".to_string(), + args: serde_json::json!({"x": 5, "y": 6}), + }, + // Route back to add + MultiScriptJob { + script_path: "f/math/add".to_string(), + args: serde_json::json!({"a": 10, "b": 20}), + }, + ], + ); + + assert_eq!(results.len(), 3); + assert_eq!(results[0], Ok(serde_json::json!(7))); // 3 + 4 + assert_eq!(results[1], Ok(serde_json::json!(30))); // 5 * 6 + assert_eq!(results[2], Ok(serde_json::json!(30))); // 10 + 20 + } + + #[test] + fn test_multi_script_interleaved_jobs() { + let script_upper = r#" +export function main(s: string): string { + return s.toUpperCase(); +} +"#; + let script_len = r#" +export function main(s: string): number { + return s.length; +} +"#; + let results = run_multi_script_worker_test( + &[("f/str/upper", script_upper), ("f/str/len", script_len)], + vec![ + MultiScriptJob { + script_path: "f/str/upper".to_string(), + args: serde_json::json!({"s": "hello"}), + }, + MultiScriptJob { + script_path: "f/str/len".to_string(), + args: serde_json::json!({"s": "hello"}), + }, + MultiScriptJob { + script_path: "f/str/upper".to_string(), + args: serde_json::json!({"s": "world"}), + }, + MultiScriptJob { + script_path: "f/str/len".to_string(), + args: serde_json::json!({"s": "ab"}), + }, + ], + ); + + assert_eq!(results.len(), 4); + assert_eq!(results[0], Ok(serde_json::json!("HELLO"))); + assert_eq!(results[1], Ok(serde_json::json!(5))); + assert_eq!(results[2], Ok(serde_json::json!("WORLD"))); + assert_eq!(results[3], Ok(serde_json::json!(2))); + } + + #[test] + fn test_multi_script_unknown_path_error() { + let script = r#" +export function main(x: number): number { + return x; +} +"#; + let results = run_multi_script_worker_test( + &[("f/known", script)], + vec![MultiScriptJob { + script_path: "f/unknown".to_string(), + args: serde_json::json!({"x": 1}), + }], + ); + + assert_eq!(results.len(), 1); + assert!(results[0].is_err()); + assert!(results[0] + .as_ref() + .unwrap_err() + .contains("Script not found")); + } + + #[test] + fn test_multi_script_error_doesnt_break_other_scripts() { + let script_ok = r#" +export function main(x: number): number { + return x * 2; +} +"#; + let script_err = r#" +export function main(msg: string): never { + throw new Error(msg); +} +"#; + let results = run_multi_script_worker_test( + &[("f/ok", script_ok), ("f/err", script_err)], + vec![ + MultiScriptJob { + script_path: "f/ok".to_string(), + args: serde_json::json!({"x": 5}), + }, + MultiScriptJob { + script_path: "f/err".to_string(), + args: serde_json::json!({"msg": "boom"}), + }, + // Should still work after error in other script + MultiScriptJob { + script_path: "f/ok".to_string(), + args: serde_json::json!({"x": 10}), + }, + ], + ); + + assert_eq!(results.len(), 3); + assert_eq!(results[0], Ok(serde_json::json!(10))); + assert!(results[1].is_err()); + assert_eq!(results[1], Err("boom".to_string())); + assert_eq!(results[2], Ok(serde_json::json!(20))); + } + + // ==================== exec_preprocess Tests ==================== + + /// Raw protocol command to send to a dedicated worker + enum ProtocolCmd { + Exec { path: String, args: serde_json::Value }, + ExecPreprocess { path: String, args: serde_json::Value }, + } + + /// Run a multi-script worker test with raw protocol commands, returning all protocol lines + fn run_raw_protocol_test( + scripts: &[(&str, &str)], + commands: Vec, + ) -> Vec { + let temp_dir = tempfile::tempdir().unwrap(); + let wrapper_path = create_multi_script_worker_files(temp_dir.path(), scripts); + let wrapper_str = wrapper_path.to_str().unwrap(); + + let mut cmd_args: Vec<&str> = BUN_DEDICATED_WORKER_ARGS.to_vec(); + cmd_args.push(wrapper_str); + + let mut child = Command::new(BUN_PATH.as_str()) + .args(cmd_args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .current_dir(temp_dir.path()) + .spawn() + .expect("Failed to spawn worker process"); + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + + let mut start_line = String::new(); + reader.read_line(&mut start_line).unwrap(); + assert_eq!( + parse_dedicated_worker_line(start_line.trim()), + DedicatedWorkerResult::Start, + ); + + let mut results = Vec::new(); + + for cmd in &commands { + let line = match cmd { + ProtocolCmd::Exec { path, args } => format!("exec:{}:{}", path, args), + ProtocolCmd::ExecPreprocess { path, args } => { + format!("exec_preprocess:{}:{}", path, args) + } + }; + writeln!(stdin, "{}", line).unwrap(); + stdin.flush().unwrap(); + + // exec_preprocess produces 2 response lines (preprocessed_args + success/error) + // exec produces 1 response line (success/error) + let expected_lines = match cmd { + ProtocolCmd::ExecPreprocess { .. } => 2, + ProtocolCmd::Exec { .. } => 1, + }; + + for _ in 0..expected_lines { + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + let parsed = parse_dedicated_worker_line(response.trim()); + // If it's an error, stop reading more lines for this command + if matches!(parsed, DedicatedWorkerResult::Error(_)) { + results.push(parsed); + break; + } + results.push(parsed); + } + } + + writeln!(stdin, "end").unwrap(); + stdin.flush().unwrap(); + let _ = child.wait().expect("Worker process failed to exit"); + + results + } + + #[test] + fn test_bun_exec_preprocess() { + let script = r#" +export function preprocessor(x: number) { + return { x: x * 10 }; +} +export function main(x: number): number { + return x + 1; +} +"#; + let results = run_raw_protocol_test( + &[("f/test/pre", script)], + vec![ProtocolCmd::ExecPreprocess { + path: "f/test/pre".to_string(), + args: serde_json::json!({"x": 5}), + }], + ); + // Should get preprocessed_args then success + assert_eq!(results.len(), 2); + assert_eq!( + results[0], + DedicatedWorkerResult::PreprocessedArgs(serde_json::json!({"x": 50})) + ); + // main(50) => 51 + assert_eq!( + results[1], + DedicatedWorkerResult::Success(serde_json::json!(51)) + ); + } + + #[test] + fn test_bun_exec_preprocess_missing_preprocessor() { + let script = r#" +export function main(x: number): number { + return x; +} +"#; + let results = run_raw_protocol_test( + &[("f/test/nopre", script)], + vec![ProtocolCmd::ExecPreprocess { + path: "f/test/nopre".to_string(), + args: serde_json::json!({"x": 5}), + }], + ); + assert_eq!(results.len(), 1); + assert!(matches!(results[0], DedicatedWorkerResult::Error(_))); + } + + #[test] + fn test_bun_exec_preprocess_then_exec() { + let script = r#" +export function preprocessor(x: number) { + return { x: x * 2 }; +} +export function main(x: number): number { + return x + 100; +} +"#; + let results = run_raw_protocol_test( + &[("f/test/mixed", script)], + vec![ + ProtocolCmd::ExecPreprocess { + path: "f/test/mixed".to_string(), + args: serde_json::json!({"x": 5}), + }, + ProtocolCmd::Exec { + path: "f/test/mixed".to_string(), + args: serde_json::json!({"x": 7}), + }, + ], + ); + // preprocess: preprocessor(5) => {"x":10}, main(10) => 110 + // exec: main(7) => 107 + assert_eq!(results.len(), 3); + assert_eq!( + results[0], + DedicatedWorkerResult::PreprocessedArgs(serde_json::json!({"x": 10})) + ); + assert_eq!( + results[1], + DedicatedWorkerResult::Success(serde_json::json!(110)) + ); + assert_eq!( + results[2], + DedicatedWorkerResult::Success(serde_json::json!(107)) + ); + } + + // ==================== Argument Transformation Tests ==================== + + #[test] + fn test_bun_date_arg_transformation() { + let script = r#" +export function main(d: Date): string { + return d instanceof Date ? d.toISOString() : typeof d; +} +"#; + let results = run_worker_test( + "bun", + script, + vec![serde_json::json!({"d": "2024-01-15T10:30:00.000Z"})], + ); + assert_eq!(results.len(), 1); + assert_eq!( + results[0], + Ok(serde_json::json!("2024-01-15T10:30:00.000Z")) + ); + } + + #[test] + fn test_bun_null_and_undefined_args() { + let script = r#" +export function main(x?: number): string { + return x === null ? "null" : x === undefined ? "undefined" : String(x); +} +"#; + let results = run_worker_test( + "bun", + script, + vec![ + serde_json::json!({"x": null}), + serde_json::json!({"x": 42}), + serde_json::json!({}), + ], + ); + assert_eq!(results.len(), 3); + assert_eq!(results[0], Ok(serde_json::json!("null"))); + assert_eq!(results[1], Ok(serde_json::json!("42"))); + // Missing arg should be undefined + assert_eq!(results[2], Ok(serde_json::json!("undefined"))); + } +} + +// ============================================================================ +// Deno Dedicated Worker Protocol Tests +// ============================================================================ + +mod dedicated_worker_protocol_deno { + use std::io::{BufRead, BufReader, Write}; + use std::process::{Command, Stdio}; + use windmill_test_utils::{parse_dedicated_worker_line, DedicatedWorkerResult}; + use windmill_worker::{generate_deno_dedicated_worker_wrapper, DENO_PATH}; + + const TEST_SCRIPT_PATH: &str = "f/test/script"; + + fn run_deno_worker_test( + script: &str, + jobs: Vec, + ) -> Vec> { + let temp_dir = tempfile::tempdir().unwrap(); + std::fs::write(temp_dir.path().join("main.ts"), script).unwrap(); + + let wrapper = generate_deno_dedicated_worker_wrapper(script).unwrap(); + std::fs::write(temp_dir.path().join("wrapper.ts"), &wrapper).unwrap(); + + let mut child = Command::new(DENO_PATH.as_str()) + .args([ + "run", + "--no-check", + "--unstable-unsafe-proto", + "--unstable-bare-node-builtins", + "-A", + "wrapper.ts", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .current_dir(temp_dir.path()) + .spawn() + .expect("Failed to spawn deno process"); + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + + // Wait for "start" — deno outputs 'start\n' via console.log which adds + // its own newline, producing double newlines. Skip empty lines. + loop { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + if line.trim().is_empty() { + continue; + } + assert_eq!( + parse_dedicated_worker_line(line.trim()), + DedicatedWorkerResult::Start, + "Expected 'start', got: {}", + line.trim() + ); + break; + } + + let mut results = Vec::new(); + for job_args in jobs { + writeln!(stdin, "exec:{}:{}", TEST_SCRIPT_PATH, job_args.to_string()).unwrap(); + stdin.flush().unwrap(); + + loop { + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + let trimmed = response.trim(); + if trimmed.is_empty() { + continue; + } + match parse_dedicated_worker_line(trimmed) { + DedicatedWorkerResult::Success(value) => results.push(Ok(value)), + DedicatedWorkerResult::Error(err) => { + let msg = err["message"] + .as_str() + .unwrap_or("Unknown error") + .to_string(); + results.push(Err(msg)); + } + other => panic!("Unexpected response: {:?}", other), + } + break; + } + } + + writeln!(stdin, "end").unwrap(); + stdin.flush().unwrap(); + let _ = child.wait().expect("Worker process failed to exit"); + results + } + + #[test] + fn test_deno_dedicated_worker_simple() { + let script = r#" +export function main(x: number, y: number): number { + return x + y; +} +"#; + let results = run_deno_worker_test(script, vec![serde_json::json!({"x": 5, "y": 3})]); + assert_eq!(results.len(), 1); + assert_eq!(results[0], Ok(serde_json::json!(8))); + } + + #[test] + fn test_deno_dedicated_worker_multiple_jobs() { + let script = r#" +export function main(n: number): number { + return n * 2; +} +"#; + let jobs: Vec = (1..=5).map(|i| serde_json::json!({"n": i})).collect(); + let results = run_deno_worker_test(script, jobs); + assert_eq!(results.len(), 5); + for (i, result) in results.iter().enumerate() { + assert_eq!(*result, Ok(serde_json::json!(((i + 1) * 2) as i64))); + } + } + + #[test] + fn test_deno_dedicated_worker_error() { + let script = r#" +export function main(msg: string): never { + throw new Error(msg); +} +"#; + let results = run_deno_worker_test(script, vec![serde_json::json!({"msg": "test error"})]); + assert_eq!(results.len(), 1); + assert!(results[0].is_err()); + assert_eq!(results[0], Err("test error".to_string())); + } + + // ==================== exec_preprocess Tests ==================== + + /// Run a raw deno protocol test, reading all output lines per command + fn run_deno_raw_protocol_test( + script: &str, + commands: Vec<(&str, serde_json::Value)>, // ("exec" or "exec_preprocess", args) + ) -> Vec { + let temp_dir = tempfile::tempdir().unwrap(); + std::fs::write(temp_dir.path().join("main.ts"), script).unwrap(); + + let wrapper = generate_deno_dedicated_worker_wrapper(script).unwrap(); + std::fs::write(temp_dir.path().join("wrapper.ts"), &wrapper).unwrap(); + + let mut child = Command::new(DENO_PATH.as_str()) + .args([ + "run", + "--no-check", + "--unstable-unsafe-proto", + "--unstable-bare-node-builtins", + "-A", + "wrapper.ts", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .current_dir(temp_dir.path()) + .spawn() + .expect("Failed to spawn deno process"); + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + + // Wait for start, skip empty lines + loop { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + if line.trim().is_empty() { + continue; + } + assert_eq!( + parse_dedicated_worker_line(line.trim()), + DedicatedWorkerResult::Start, + ); + break; + } + + let mut results = Vec::new(); + + for (cmd, args) in &commands { + writeln!(stdin, "{}:{}:{}", cmd, TEST_SCRIPT_PATH, args).unwrap(); + stdin.flush().unwrap(); + + let expected_lines = if *cmd == "exec_preprocess" { 2 } else { 1 }; + + for _ in 0..expected_lines { + loop { + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + if response.trim().is_empty() { + continue; + } + let parsed = parse_dedicated_worker_line(response.trim()); + if matches!(parsed, DedicatedWorkerResult::Error(_)) { + results.push(parsed); + break; + } + results.push(parsed); + break; + } + // If last result was an error, don't read more lines for this command + if matches!(results.last(), Some(DedicatedWorkerResult::Error(_))) { + break; + } + } + } + + writeln!(stdin, "end").unwrap(); + stdin.flush().unwrap(); + let _ = child.wait().expect("Worker process failed to exit"); + + results + } + + #[test] + fn test_deno_exec_preprocess() { + let script = r#" +export function preprocessor(x: number) { + return { x: x * 10 }; +} +export function main(x: number): number { + return x + 1; +} +"#; + let results = run_deno_raw_protocol_test( + script, + vec![("exec_preprocess", serde_json::json!({"x": 5}))], + ); + assert_eq!(results.len(), 2); + assert_eq!( + results[0], + DedicatedWorkerResult::PreprocessedArgs(serde_json::json!({"x": 50})) + ); + assert_eq!( + results[1], + DedicatedWorkerResult::Success(serde_json::json!(51)) + ); + } + + // Note: no "missing preprocessor" test for Deno because the wrapper only generates + // the exec_preprocess handler when the script actually has a preprocessor function. + // Without one, exec_preprocess messages are unrecognized (by design — Rust never sends them). + + // ==================== Argument Transformation Tests ==================== + + #[test] + fn test_deno_date_arg_transformation() { + let script = r#" +export function main(d: Date): string { + return d instanceof Date ? d.toISOString() : typeof d; +} +"#; + let results = run_deno_worker_test( + script, + vec![serde_json::json!({"d": "2024-01-15T10:30:00.000Z"})], + ); + assert_eq!(results.len(), 1); + assert_eq!( + results[0], + Ok(serde_json::json!("2024-01-15T10:30:00.000Z")) + ); + } } // ============================================================================ diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index 2ffb85a418..f04e189315 100644 --- a/backend/tests/python_jobs.rs +++ b/backend/tests/python_jobs.rs @@ -1,9 +1,470 @@ use serde_json::json; +#[cfg(feature = "python")] use sqlx::postgres::Postgres; +#[cfg(feature = "python")] use sqlx::Pool; +#[cfg(feature = "python")] use windmill_common::scripts::ScriptLang; use windmill_test_utils::*; +// ============================================================================ +// Dedicated Worker Protocol Tests (Python) +// ============================================================================ + +#[cfg(feature = "python")] +mod dedicated_worker_protocol_python { + use std::io::{BufRead, BufReader, Write}; + use std::process::{Command, Stdio}; + use windmill_test_utils::{parse_dedicated_worker_line, DedicatedWorkerResult}; + use windmill_worker::{compute_py_codegen, generate_py_multi_script_wrapper, PyScriptEntry}; + + struct MultiScriptJob { + script_path: String, + args: serde_json::Value, + } + + /// Creates a multi-script Python wrapper, writes scripts to proper module paths + fn create_py_worker_files( + dir: &std::path::Path, + scripts: &[(&str, &str)], // (original_path, content) + ) -> std::path::PathBuf { + let mut codegens = Vec::new(); + for (path, content) in scripts { + let cg = compute_py_codegen(content, path); + let module_dir = dir.join(&cg.dirs); + std::fs::create_dir_all(&module_dir).unwrap(); + std::fs::write(module_dir.join(format!("{}.py", cg.module_name)), content).unwrap(); + codegens.push((path.to_string(), cg)); + } + + let entries: Vec> = codegens + .iter() + .map(|(path, cg)| PyScriptEntry { original_path: path.as_str(), codegen: cg }) + .collect(); + + let wrapper = generate_py_multi_script_wrapper(&entries, false, false); + let wrapper_path = dir.join("wrapper.py"); + std::fs::write(&wrapper_path, &wrapper).unwrap(); + wrapper_path + } + + fn run_py_multi_script_test( + scripts: &[(&str, &str)], + jobs: Vec, + ) -> Vec> { + let temp_dir = tempfile::tempdir().unwrap(); + create_py_worker_files(temp_dir.path(), scripts); + + let mut child = Command::new("python3") + .args(["-u", "-m", "wrapper"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .current_dir(temp_dir.path()) + .spawn() + .expect("Failed to spawn python3 process"); + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + + let mut start_line = String::new(); + reader.read_line(&mut start_line).unwrap(); + assert_eq!( + parse_dedicated_worker_line(start_line.trim()), + DedicatedWorkerResult::Start, + "Expected 'start', got: {}", + start_line.trim() + ); + + let mut results = Vec::new(); + for job in &jobs { + writeln!(stdin, "exec:{}:{}", job.script_path, job.args.to_string()).unwrap(); + stdin.flush().unwrap(); + + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + + match parse_dedicated_worker_line(response.trim()) { + DedicatedWorkerResult::Success(value) => results.push(Ok(value)), + DedicatedWorkerResult::Error(err) => { + let msg = err["message"] + .as_str() + .unwrap_or("Unknown error") + .to_string(); + results.push(Err(msg)); + } + other => panic!("Unexpected response: {:?}", other), + } + } + + writeln!(stdin, "end").unwrap(); + stdin.flush().unwrap(); + let _ = child.wait().expect("Worker process failed to exit"); + results + } + + fn run_py_single_script_test( + script_path: &str, + content: &str, + jobs: Vec, + ) -> Vec> { + run_py_multi_script_test( + &[(script_path, content)], + jobs.into_iter() + .map(|args| MultiScriptJob { script_path: script_path.to_string(), args }) + .collect(), + ) + } + + #[test] + fn test_python_dedicated_worker_simple() { + let results = run_py_single_script_test( + "f/test/add", + "def main(a: int, b: int):\n return a + b\n", + vec![serde_json::json!({"a": 3, "b": 4})], + ); + assert_eq!(results.len(), 1); + assert_eq!(results[0], Ok(serde_json::json!(7))); + } + + #[test] + fn test_python_dedicated_worker_multiple_jobs() { + let results = run_py_single_script_test( + "f/test/double", + "def main(n: int):\n return n * 2\n", + (1..=5).map(|i| serde_json::json!({"n": i})).collect(), + ); + assert_eq!(results.len(), 5); + for (i, result) in results.iter().enumerate() { + assert_eq!(*result, Ok(serde_json::json!(((i + 1) * 2) as i64))); + } + } + + #[test] + fn test_python_multi_script_routing() { + let results = run_py_multi_script_test( + &[ + ( + "f/math/add", + "def main(a: int, b: int):\n return a + b\n", + ), + ( + "f/math/mul", + "def main(x: int, y: int):\n return x * y\n", + ), + ], + vec![ + MultiScriptJob { + script_path: "f/math/add".to_string(), + args: serde_json::json!({"a": 3, "b": 4}), + }, + MultiScriptJob { + script_path: "f/math/mul".to_string(), + args: serde_json::json!({"x": 5, "y": 6}), + }, + MultiScriptJob { + script_path: "f/math/add".to_string(), + args: serde_json::json!({"a": 10, "b": 20}), + }, + ], + ); + assert_eq!(results.len(), 3); + assert_eq!(results[0], Ok(serde_json::json!(7))); + assert_eq!(results[1], Ok(serde_json::json!(30))); + assert_eq!(results[2], Ok(serde_json::json!(30))); + } + + #[test] + fn test_python_multi_script_error_isolation() { + let results = run_py_multi_script_test( + &[ + ("f/ok", "def main(x: int):\n return x * 2\n"), + ("f/err", "def main(msg: str):\n raise Exception(msg)\n"), + ], + vec![ + MultiScriptJob { + script_path: "f/ok".to_string(), + args: serde_json::json!({"x": 5}), + }, + MultiScriptJob { + script_path: "f/err".to_string(), + args: serde_json::json!({"msg": "boom"}), + }, + MultiScriptJob { + script_path: "f/ok".to_string(), + args: serde_json::json!({"x": 10}), + }, + ], + ); + assert_eq!(results.len(), 3); + assert_eq!(results[0], Ok(serde_json::json!(10))); + assert!(results[1].is_err()); + assert_eq!(results[1], Err("boom".to_string())); + assert_eq!(results[2], Ok(serde_json::json!(20))); + } + + #[test] + fn test_python_multi_script_unknown_path() { + let results = run_py_multi_script_test( + &[("f/known", "def main(x: int):\n return x\n")], + vec![MultiScriptJob { + script_path: "f/unknown".to_string(), + args: serde_json::json!({"x": 1}), + }], + ); + assert_eq!(results.len(), 1); + assert!(results[0].is_err()); + assert!(results[0] + .as_ref() + .unwrap_err() + .contains("Script not found")); + } + + // ==================== exec_preprocess Tests ==================== + + /// Raw protocol command for Python + enum ProtocolCmd { + Exec { path: String, args: serde_json::Value }, + ExecPreprocess { path: String, args: serde_json::Value }, + } + + /// Run a Python worker test with raw protocol commands + fn run_py_raw_protocol_test( + scripts: &[(&str, &str)], + commands: Vec, + ) -> Vec { + let temp_dir = tempfile::tempdir().unwrap(); + create_py_worker_files(temp_dir.path(), scripts); + + let mut child = Command::new("python3") + .args(["-u", "-m", "wrapper"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .current_dir(temp_dir.path()) + .spawn() + .expect("Failed to spawn python3 process"); + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + + let mut start_line = String::new(); + reader.read_line(&mut start_line).unwrap(); + assert_eq!( + parse_dedicated_worker_line(start_line.trim()), + DedicatedWorkerResult::Start, + ); + + let mut results = Vec::new(); + + for cmd in &commands { + let line = match cmd { + ProtocolCmd::Exec { path, args } => format!("exec:{}:{}", path, args), + ProtocolCmd::ExecPreprocess { path, args } => { + format!("exec_preprocess:{}:{}", path, args) + } + }; + writeln!(stdin, "{}", line).unwrap(); + stdin.flush().unwrap(); + + let expected_lines = match cmd { + ProtocolCmd::ExecPreprocess { .. } => 2, + ProtocolCmd::Exec { .. } => 1, + }; + + for _ in 0..expected_lines { + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + let parsed = parse_dedicated_worker_line(response.trim()); + if matches!(parsed, DedicatedWorkerResult::Error(_)) { + results.push(parsed); + break; + } + results.push(parsed); + } + } + + writeln!(stdin, "end").unwrap(); + stdin.flush().unwrap(); + let _ = child.wait().expect("Worker process failed to exit"); + + results + } + + #[test] + fn test_python_exec_preprocess() { + let script = r#" +def preprocessor(x: int): + return {"x": x * 10} + +def main(x: int): + return x + 1 +"#; + let results = run_py_raw_protocol_test( + &[("f/test/pre", script)], + vec![ProtocolCmd::ExecPreprocess { + path: "f/test/pre".to_string(), + args: serde_json::json!({"x": 5}), + }], + ); + assert_eq!(results.len(), 2); + assert_eq!( + results[0], + DedicatedWorkerResult::PreprocessedArgs(serde_json::json!({"x": 50})) + ); + // main(50) => 51 + assert_eq!( + results[1], + DedicatedWorkerResult::Success(serde_json::json!(51)) + ); + } + + #[test] + fn test_python_exec_preprocess_missing_preprocessor() { + let script = "def main(x: int):\n return x\n"; + let results = run_py_raw_protocol_test( + &[("f/test/nopre", script)], + vec![ProtocolCmd::ExecPreprocess { + path: "f/test/nopre".to_string(), + args: serde_json::json!({"x": 5}), + }], + ); + assert_eq!(results.len(), 1); + assert!(matches!(results[0], DedicatedWorkerResult::Error(_))); + } + + #[test] + fn test_python_exec_preprocess_then_exec() { + let script = r#" +def preprocessor(x: int): + return {"x": x * 2} + +def main(x: int): + return x + 100 +"#; + let results = run_py_raw_protocol_test( + &[("f/test/mixed", script)], + vec![ + ProtocolCmd::ExecPreprocess { + path: "f/test/mixed".to_string(), + args: serde_json::json!({"x": 5}), + }, + ProtocolCmd::Exec { + path: "f/test/mixed".to_string(), + args: serde_json::json!({"x": 7}), + }, + ], + ); + // preprocess: preprocessor(5) => {"x":10}, main(10) => 110 + // exec: main(7) => 107 + assert_eq!(results.len(), 3); + assert_eq!( + results[0], + DedicatedWorkerResult::PreprocessedArgs(serde_json::json!({"x": 10})) + ); + assert_eq!( + results[1], + DedicatedWorkerResult::Success(serde_json::json!(110)) + ); + assert_eq!( + results[2], + DedicatedWorkerResult::Success(serde_json::json!(107)) + ); + } + + // ==================== Argument Transformation Tests ==================== + + #[test] + fn test_python_datetime_arg_transformation() { + let script = r#" +from datetime import datetime + +def main(d: datetime): + return d.isoformat() +"#; + let results = run_py_single_script_test( + "f/test/dt", + script, + vec![serde_json::json!({"d": "2024-01-15T10:30:00+00:00"})], + ); + assert_eq!(results.len(), 1); + assert_eq!( + results[0], + Ok(serde_json::json!("2024-01-15T10:30:00+00:00")) + ); + } + + #[test] + fn test_python_bytes_arg_transformation() { + let script = r#" +def main(data: bytes): + return len(data) +"#; + // base64 of "hello" is "aGVsbG8=" + let results = run_py_single_script_test( + "f/test/bytes", + script, + vec![serde_json::json!({"data": "aGVsbG8="})], + ); + assert_eq!(results.len(), 1); + assert_eq!(results[0], Ok(serde_json::json!(5))); + } + + #[test] + fn test_python_kwargs_filtering() { + // Test that extra kwargs are filtered out and only declared args are passed + let script = "def main(a: int, b: int):\n return a + b\n"; + let results = run_py_single_script_test( + "f/test/kwargs", + script, + vec![serde_json::json!({"a": 1, "b": 2, "extra": 99})], + ); + assert_eq!(results.len(), 1); + assert_eq!(results[0], Ok(serde_json::json!(3))); + } + + #[test] + fn test_python_function_call_sentinel_removal() { + // Test that '' sentinel values are removed from args + let script = "def main(a: int, b: int = 10):\n return a + b\n"; + let results = run_py_single_script_test( + "f/test/sentinel", + script, + vec![serde_json::json!({"a": 5, "b": ""})], + ); + assert_eq!(results.len(), 1); + // b should be removed (sentinel), default 10 used + assert_eq!(results[0], Ok(serde_json::json!(15))); + } + + // ==================== Relative Import Tests ==================== + + #[test] + fn test_python_dedicated_worker_with_relative_import_detection() { + // Test that the wrapper includes 'import loader' when scripts have relative imports + let script_with_relative = "from f.helper import util\ndef main(x: int):\n return x\n"; + let cg = compute_py_codegen(script_with_relative, "f/test/rel"); + let entries = [PyScriptEntry { original_path: "f/test/rel", codegen: &cg }]; + let wrapper = generate_py_multi_script_wrapper(&entries, false, true); + assert!( + wrapper.contains("import loader"), + "wrapper should contain 'import loader' when any_relative_imports=true" + ); + + // Without relative imports + let script_no_relative = "def main(x: int):\n return x\n"; + let cg2 = compute_py_codegen(script_no_relative, "f/test/norel"); + let entries2 = [PyScriptEntry { original_path: "f/test/norel", codegen: &cg2 }]; + let wrapper2 = generate_py_multi_script_wrapper(&entries2, false, false); + assert!( + !wrapper2.contains("import loader"), + "wrapper should NOT contain 'import loader' when any_relative_imports=false" + ); + } +} + #[cfg(feature = "python")] #[sqlx::test(fixtures("base", "lockfile_python"))] async fn test_requirements_python(db: Pool) -> anyhow::Result<()> { diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 56de4d9892..54ac78fade 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -239,6 +239,7 @@ pub fn workspaced_service() -> Router { "/history_update/h/:hash/p/*path", post(update_script_history), ) + .route("/list_dedicated_with_deps", get(list_dedicated_with_deps)) // Temporary raw script storage for CLI lock generation .route("/raw_temp/store", post(store_raw_script_temp)) .route("/raw_temp/diff", post(diff_raw_scripts_with_deployed)) @@ -2480,6 +2481,62 @@ async fn guard_script_from_debounce_data(ns: &NewScript) -> Result<()> { } } +#[derive(Serialize)] +struct DedicatedScriptDeps { + path: String, + language: ScriptLang, + workspace_dep_names: Vec, +} + +async fn list_dedicated_with_deps( + authed: ApiAuthed, + Extension(user_db): Extension, + Path(w_id): Path, +) -> JsonResult> { + let mut tx = user_db.begin(&authed).await?; + + let rows = sqlx::query!( + "SELECT DISTINCT ON (path) path, language AS \"language: ScriptLang\", content FROM script + WHERE workspace_id = $1 + AND archived = false + AND dedicated_worker = true + AND language = ANY($2::SCRIPT_LANG[]) + ORDER BY path, created_at DESC", + &w_id, + &[ + ScriptLang::Python3, + ScriptLang::Bun, + ScriptLang::Bunnative, + ScriptLang::Deno, + ] as &[ScriptLang], + ) + .fetch_all(&mut *tx) + .await?; + + tx.commit().await?; + + let result = rows + .into_iter() + .map(|row| { + let dep_names = + windmill_common::scripts::extract_workspace_dependencies_annotated_refs( + &row.language, + &row.content, + &row.path, + ) + .map(|refs| refs.external) + .unwrap_or_default(); + DedicatedScriptDeps { + path: row.path, + language: row.language, + workspace_dep_names: dep_names, + } + }) + .collect(); + + Ok(Json(result)) +} + // ============================================================================ // Temporary Raw Script Storage for CLI Lock Generation // ============================================================================ @@ -2508,11 +2565,9 @@ async fn store_raw_script_temp( .await?; // Clean up old entries (1 week TTL) - sqlx::query!( - "DELETE FROM raw_script_temp WHERE created_at < NOW() - INTERVAL '1 week'" - ) - .execute(&db) - .await?; + sqlx::query!("DELETE FROM raw_script_temp WHERE created_at < NOW() - INTERVAL '1 week'") + .execute(&db) + .await?; Ok(Json(hash)) } @@ -2560,7 +2615,7 @@ async fn diff_raw_scripts_with_deployed( FROM script s \ WHERE s.path = local.path AND s.workspace_id = $3 AND s.archived = false \ ORDER BY s.created_at DESC LIMIT 1 \ - ) deployed ON deployed.deployed_hash = local.hash" + ) deployed ON deployed.deployed_hash = local.hash", ) .bind(&paths) .bind(&hashes) @@ -2582,7 +2637,7 @@ async fn diff_raw_scripts_with_deployed( AND wd.language = $3::SCRIPT_LANG \ AND wd.name IS NOT DISTINCT FROM $4 \ AND encode(sha256(convert_to(wd.content, 'UTF8')), 'hex') = $5 \ - )" + )", ) .bind(&dep.path) .bind(&w_id) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 825842f057..cb86c2a8b9 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -6940,6 +6940,60 @@ paths: schema: type: string + /w/{workspace}/scripts/list_dedicated_with_deps: + get: + summary: list dedicated worker scripts with workspace dependency annotations + operationId: listDedicatedWithDeps + tags: + - script + parameters: + - $ref: "#/components/parameters/WorkspaceId" + responses: + "200": + description: list of dedicated scripts with their workspace dependency names + content: + application/json: + schema: + type: array + items: + type: object + properties: + path: + type: string + language: + type: string + enum: + - python3 + - deno + - go + - bash + - powershell + - postgresql + - mysql + - bigquery + - snowflake + - mssql + - graphql + - nativets + - bun + - bunnative + - php + - rust + - ansible + - csharp + - oracledb + - duckdb + - java + - ruby + workspace_dep_names: + type: array + items: + type: string + required: + - path + - language + - workspace_dep_names + /w/{workspace}/scripts/raw/p/{path}: get: summary: raw script by path diff --git a/backend/windmill-common/src/ai_google.rs b/backend/windmill-common/src/ai_google.rs index ccf34685e5..5f8255c10a 100644 --- a/backend/windmill-common/src/ai_google.rs +++ b/backend/windmill-common/src/ai_google.rs @@ -9,7 +9,10 @@ use serde::{Deserialize, Serialize}; -use crate::ai_types::{ContentPart, ExtraContent, GoogleExtraContent, OpenAIContent, OpenAIMessage, ToolDef, UrlCitation}; +use crate::ai_types::{ + ContentPart, ExtraContent, GoogleExtraContent, OpenAIContent, OpenAIMessage, ToolDef, + UrlCitation, +}; use crate::error::Error; // ============================================================================ @@ -87,7 +90,10 @@ pub struct GeminiTextRequest { /// Tool definition — function declarations and/or Google Search grounding. #[derive(Serialize)] pub struct GeminiTool { - #[serde(rename = "functionDeclarations", skip_serializing_if = "Option::is_none")] + #[serde( + rename = "functionDeclarations", + skip_serializing_if = "Option::is_none" + )] pub function_declarations: Option>, #[serde(rename = "googleSearch", skip_serializing_if = "Option::is_none")] pub google_search: Option, @@ -115,7 +121,10 @@ pub struct GeminiToolConfig { #[derive(Serialize)] pub struct GeminiFunctionCallingConfig { pub mode: String, - #[serde(rename = "allowedFunctionNames", skip_serializing_if = "Option::is_none")] + #[serde( + rename = "allowedFunctionNames", + skip_serializing_if = "Option::is_none" + )] pub allowed_function_names: Option>, } @@ -341,10 +350,8 @@ pub fn convert_content_to_gemini_parts(content: &OpenAIContent) -> Vec { - parse_data_url(&image_url.url).map(|(mime_type, data)| { - GeminiPart::InlineData { - inline_data: GeminiInlineData { mime_type, data }, - } + parse_data_url(&image_url.url).map(|(mime_type, data)| GeminiPart::InlineData { + inline_data: GeminiInlineData { mime_type, data }, }) } // S3Objects are handled by the worker @@ -372,15 +379,12 @@ pub fn openai_messages_to_gemini( if let Some(content) = &msg.content { let parts = convert_content_to_gemini_parts(content); if !parts.is_empty() { - system_instruction = - Some(GeminiContentMessage { role: None, parts }); + system_instruction = Some(GeminiContentMessage { role: None, parts }); } } } "tool" => { - if let (Some(tool_call_id), Some(content)) = - (&msg.tool_call_id, &msg.content) - { + if let (Some(tool_call_id), Some(content)) = (&msg.tool_call_id, &msg.content) { let func_name = find_gemini_function_name(messages, tool_call_id); let response_text = match content { OpenAIContent::Text(text) => text.clone(), @@ -435,10 +439,8 @@ pub fn openai_messages_to_gemini( } if !parts.is_empty() { - contents.push(GeminiContentMessage { - role: Some(gemini_role.to_string()), - parts, - }); + contents + .push(GeminiContentMessage { role: Some(gemini_role.to_string()), parts }); } } } @@ -469,10 +471,8 @@ pub fn openai_tools_to_gemini( .collect(); if !declarations.is_empty() { - gemini_tools.push(GeminiTool { - function_declarations: Some(declarations), - google_search: None, - }); + gemini_tools + .push(GeminiTool { function_declarations: Some(declarations), google_search: None }); } if has_websearch { diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 8b20d331c0..b1ba371ea2 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -1531,6 +1531,16 @@ pub fn dedicated_worker_tag(workspace_id: &str, path: &str) -> String { ) } +/// Configuration for a runner group — a single long-lived subprocess +/// that can execute multiple scripts sharing the same workspace dependency. +/// Auto-detected from script content annotations at worker startup. +#[derive(Clone, PartialEq, Debug)] +pub struct RunnerGroupConfig { + pub workspace_id: String, + pub dep_name: String, + pub language: String, +} + pub async fn load_worker_config( db: &DB, killpill_tx: KillpillSender, @@ -1638,7 +1648,7 @@ pub async fn load_worker_config( let worker_tags = config .worker_tags .or_else(|| { - // Check for multiple dedicated workers first + // Check for multiple dedicated workers if let Some(ref dws) = dedicated_workers.as_ref() { let mut dedi_tags: Vec = dws .iter() diff --git a/backend/windmill-test-utils/src/lib.rs b/backend/windmill-test-utils/src/lib.rs index 8e83eae1da..ef2bf66513 100644 --- a/backend/windmill-test-utils/src/lib.rs +++ b/backend/windmill-test-utils/src/lib.rs @@ -899,6 +899,8 @@ pub enum DedicatedWorkerResult { Start, /// Worker returned a successful result Success(serde_json::Value), + /// Worker returned preprocessed args (from exec_preprocess) + PreprocessedArgs(serde_json::Value), /// Worker returned an error result Error(serde_json::Value), /// Line is not a protocol message (e.g., logs) @@ -908,6 +910,7 @@ pub enum DedicatedWorkerResult { /// Parse a line from dedicated worker stdout according to the protocol: /// - "start" -> Ready signal /// - "wm_res[success]:JSON" -> Success with result +/// - "wm_res[preprocessed_args]:JSON" -> Preprocessed args /// - "wm_res[error]:JSON" -> Error with details /// - anything else -> Other (logs) pub fn parse_dedicated_worker_line(line: &str) -> DedicatedWorkerResult { @@ -922,6 +925,13 @@ pub fn parse_dedicated_worker_line(line: &str) -> DedicatedWorkerResult { } } + if let Some(json_str) = line.strip_prefix("wm_res[preprocessed_args]:") { + match serde_json::from_str(json_str) { + Ok(value) => return DedicatedWorkerResult::PreprocessedArgs(value), + Err(_) => return DedicatedWorkerResult::Other(line.to_string()), + } + } + if let Some(json_str) = line.strip_prefix("wm_res[error]:") { match serde_json::from_str(json_str) { Ok(value) => return DedicatedWorkerResult::Error(value), diff --git a/backend/windmill-worker/loader.bun.js b/backend/windmill-worker/loader.bun.js index f2a00de0cf..d03f19b6ba 100644 --- a/backend/windmill-worker/loader.bun.js +++ b/backend/windmill-worker/loader.bun.js @@ -21,7 +21,7 @@ const p = { // On Windows, normalize path to POSIX format to match args.path from Bun's resolver const cdirPosix = cdir.replace(/\\/g, "/").replace(/^[a-zA-Z]:/, ""); const filterResolve = new RegExp( - `^(?!\\.\/main\\.ts)(?!${cdir}\/main\\.ts)(?!${cdirPosix}\/main\\.ts)(?!(?:/private)?${cdirNoPrivate}\/wrapper\\.mjs).*\\.ts$` + `^(?!\\.\/main\\.ts)(?!\\.\/_wm_)(?!${cdir}\/main\\.ts)(?!${cdir}\/_wm_)(?!${cdirPosix}\/main\\.ts)(?!${cdirPosix}\/_wm_)(?!(?:/private)?${cdirNoPrivate}\/wrapper\\.mjs).*\\.ts$` ); let cdirNodeModules = `${cdir}/node_modules/`; diff --git a/backend/windmill-worker/loader.bun.windows.js b/backend/windmill-worker/loader.bun.windows.js index 7d14ffcbc3..e89b70c26c 100644 --- a/backend/windmill-worker/loader.bun.windows.js +++ b/backend/windmill-worker/loader.bun.windows.js @@ -27,7 +27,7 @@ const p = { const cdirFwd = cdir.replace(/\\/g, "/"); const cdirPosix = cdirFwd.replace(/^[a-zA-Z]:/, ""); const filterResolve = new RegExp( - `^(?!\\.\/main\\.ts)(?!${cdirFwd}\/main\\.ts)(?!${cdirPosix}\/main\\.ts)(?!(?:/private)?${cdirNoPrivate}\/wrapper\\.mjs).*\\.ts$` + `^(?!\\.\/main\\.ts)(?!\\.\/_wm_)(?!${cdirFwd}\/main\\.ts)(?!${cdirFwd}\/_wm_)(?!${cdirPosix}\/main\\.ts)(?!${cdirPosix}\/_wm_)(?!(?:/private)?${cdirNoPrivate}\/wrapper\\.mjs).*\\.ts$` ); let cdirNodeModules = `${cdirFwd}/node_modules/`; diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 7dbaa6723d..6bc6cacc32 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -77,95 +77,236 @@ 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"]; -/// Generate the dedicated worker wrapper content. -/// - `arg_names`: The argument names for the main function (e.g., ["x", "y"]) -/// - `main_import`: The import path for the main module (e.g., "./main.ts") -/// - `date_conversions`: Optional date conversion statements for Datetime args -/// - `preprocessor_spread`: If the script has a preprocessor function, the comma-separated arg names for it -pub fn generate_dedicated_worker_wrapper( - arg_names: &[&str], - main_import: &str, - date_conversions: Option<&str>, - preprocessor_spread: Option<&str>, -) -> String { - let spread = arg_names.join(","); - let dates = date_conversions.unwrap_or(""); +/// 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: +/// exec:: -> wm_res[success]: | wm_res[error]: +/// exec_preprocess:: -> wm_res[preprocessed_args]: then wm_res[success]: | wm_res[error]: +/// 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(line);"# + r#"console.log("[debug] " + line);"# } else { "" }; - let preprocessor_logic = if let Some(pre_spread) = preprocessor_spread { - format!( + 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#" - if (rawLine.startsWith("preprocess:")) {{ - const preInput = rawLine.slice("preprocess:".length); - const parsedArgs = JSON.parse(preInput); - if (Main.preprocessor === undefined || typeof Main.preprocessor !== 'function') {{ - console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }})); - continue; - }} - try {{ - function preArgsObjToArr({{ {pre_spread} }}) {{ - return [ {pre_spread} ]; - }} - const preprocessedArgs = await Main.preprocessor(...preArgsObjToArr(parsedArgs)); - console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value)); - // Now call main with preprocessed args - const mainArgs = getArgs(JSON.stringify(preprocessedArgs ?? {{}})); - const res = await Main.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: rawLine }})); - }} - continue; - }}"# - ) - } else { - String::new() - }; +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#" -import * as Main from "{main_import}"; +{imports} import * as Readline from "node:readline" BigInt.prototype.toJSON = function () {{ return this.toString(); }}; -console.log('start'); +const scripts = new Map(); +{functions} +{registrations} -function getArgs(line) {{ - let {{ {spread} }} = JSON.parse(line) - {dates} - return [ {spread} ]; -}} +console.log('start'); for await (const line of Readline.createInterface({{ input: process.stdin }})) {{ {print_lines} - const rawLine = line; - if (rawLine === "end") {{ + if (line === "end") {{ process.exit(0); }} - {preprocessor_logic} - try {{ - const args = getArgs(rawLine); - const res = await Main.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: rawLine }})); + + 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) -fn split_lockfile(lockfile: &str) -> (&str, Option<&str>, bool, bool) { +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); @@ -3587,53 +3728,18 @@ pub async fn start_worker( 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 mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true, false, None)?.args; - let dates = args - .iter() - .filter_map(|x| { - if matches!(x.typ, Typ::Datetime) { - Some(x.name.clone()) - } else { - None - } - }) - .map(|x| return format!("{x} = {x} ? new Date({x}) : undefined")) - .join("\n"); - - let arg_names: Vec<&str> = args.iter().map(|x| x.name.as_str()).collect(); - - // Parse preprocessor signature if it exists - let pre_spread = 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).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_import = if codebase.is_some() { - "./main.js" - } else { - "./main.ts" - }; - let dates_opt = if dates.is_empty() { - None - } else { - Some(dates.as_str()) - }; - let wrapper_content = generate_dedicated_worker_wrapper( - &arg_names, - main_import, - dates_opt, - pre_spread.as_deref(), - ); + 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)?; } @@ -3832,4 +3938,48 @@ 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()); + } } diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 46f42af3bf..a0fd9fdde8 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -27,6 +27,16 @@ use windmill_common::{ }; use windmill_parser::Typ; +pub const DENO_UNSTABLE_ARGS: &[&str] = &[ + "--unstable-unsafe-proto", + "--unstable-bare-node-builtins", + "--unstable-webgpu", + "--unstable-ffi", + "--unstable-fs", + "--unstable-worker-options", + "--unstable-http", +]; + lazy_static::lazy_static! { static ref DENO_FLAGS: Option> = std::env::var("DENO_FLAGS") @@ -172,22 +182,14 @@ pub async fn generate_deno_lock( let mut child_cmd = Command::new(DENO_PATH.as_str()); child_cmd .current_dir(job_dir) - .args(vec![ - "cache", - "--unstable-unsafe-proto", - "--unstable-bare-node-builtins", - "--unstable-webgpu", - "--unstable-ffi", - "--unstable-fs", - "--unstable-worker-options", - "--unstable-http", + .args(["cache"].iter().chain(DENO_UNSTABLE_ARGS).chain(&[ "--lock=lock.json", "--frozen=false", "--allow-import", "--import-map", - &import_map_path, + import_map_path.as_str(), "main.ts", - ]) + ])) .envs(deno_envs) .stdout(Stdio::piped()) .stderr(Stdio::piped()); @@ -442,13 +444,7 @@ try {{ args.push("--import-map"); args.push(&import_map_path); args.push(&reload); - args.push("--unstable-unsafe-proto"); - args.push("--unstable-bare-node-builtins"); - args.push("--unstable-webgpu"); - args.push("--unstable-ffi"); - args.push("--unstable-fs"); - args.push("--unstable-worker-options"); - args.push("--unstable-http"); + args.extend_from_slice(DENO_UNSTABLE_ARGS); if !*DISABLE_DENO_LOCK { if let Some(reqs) = requirements_o { @@ -547,7 +543,7 @@ try {{ read_result(job_dir, handle_result.result_stream).await } -async fn build_import_map( +pub(crate) async fn build_import_map( w_id: &str, script_path: &str, base_internal_url: &str, @@ -592,6 +588,127 @@ async fn build_import_map( #[cfg(feature = "private")] use crate::{dedicated_worker_oss::handle_dedicated_process, JobCompletedSender}; +/// Generate the dedicated worker wrapper for Deno. +/// Parses the script signature and bakes in arg destructuring, date conversions, +/// and preprocessor logic. Uses the `exec::` protocol. +#[cfg(any(feature = "private", test))] +pub fn generate_dedicated_worker_wrapper(inner_content: &str) -> Result { + let args = windmill_parser_ts::parse_deno_signature(inner_content, true, false, None)?.args; + let dates = args + .iter() + .filter_map(|x| { + if matches!(x.typ, Typ::Datetime) { + Some(x.name.clone()) + } else { + None + } + }) + .map(|x| format!("{x} = {x} ? new Date({x}) : undefined")) + .join("\n"); + + let spread = args.into_iter().map(|x| x.name).join(","); + + let pre_spread = 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).join(",")); + + let preprocessor_import = if pre_spread.is_some() { + r#"import { preprocessor } from "./main.ts";"# + } else { + "" + }; + + let preprocessor_logic = if let Some(ref pre_spread) = pre_spread { + format!( + r#" + 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" }}) + '\n'); + continue; + }} + const argsJson = rest.slice(colonIdx + 1); + const parsedArgs = JSON.parse(argsJson); + if (typeof preprocessor !== 'function') {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }}) + '\n'); + continue; + }} + try {{ + function preArgsObjToArr({{ {pre_spread} }}: any) {{ + return [ {pre_spread} ]; + }} + const preprocessedArgs: any = await preprocessor(...preArgsObjToArr(parsedArgs)); + console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + let {{ {spread} }} = preprocessedArgs ?? {{}}; + {dates} + let res: any = await main(...[ {spread} ]); + console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + }} catch (e) {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }}) + '\n'); + }} + continue; + }}"# + ) + } else { + String::new() + }; + + Ok(format!( + r#" +import {{ main }} from "./main.ts"; +{preprocessor_import} + +BigInt.prototype.toJSON = function () {{ + return this.toString(); +}}; + +console.log('start\n'); + +const decoder = new TextDecoder(); +for await (const chunk of Deno.stdin.readable) {{ + const lines = decoder.decode(chunk); + let exit = false; + for (const line of lines.trim().split("\n")) {{ + if (line === "end") {{ + exit = true; + break; + }} + {preprocessor_logic} + 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" }}) + '\n'); + continue; + }} + const argsJson = rest.slice(colonIdx + 1); + try {{ + let {{ {spread} }} = JSON.parse(argsJson) + {dates} + let res: any = await main(...[ {spread} ]); + console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + }} catch (e) {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }}) + '\n'); + }} + continue; + }} + console.error("Unknown command:", line); + }} + if (exit) {{ + break; + }} +}} +"#, + )) +} + #[cfg(feature = "private")] use tokio::sync::mpsc::Receiver; #[cfg(feature = "private")] @@ -650,115 +767,19 @@ pub async fn start_worker( let context_envs = build_envs_map(context.to_vec()).await; { - // let mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true, false, None)?.args; - let dates = args - .iter() - .filter_map(|x| { - if matches!(x.typ, Typ::Datetime) { - Some(x.name.clone()) - } else { - None - } - }) - .map(|x| return format!("{x} = {x} ? new Date({x}) : undefined")) - .join("\n"); - - let spread = args.into_iter().map(|x| x.name).join(","); - - // Parse preprocessor signature if it exists - let pre_spread = 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).join(",")); - - let preprocessor_import = if pre_spread.is_some() { - r#"import { preprocessor } from "./main.ts";"# - } else { - "" - }; - - let preprocessor_logic = if let Some(ref pre_spread) = pre_spread { - format!( - r#" - if (line.startsWith("preprocess:")) {{ - const preInput = line.slice("preprocess:".length); - const parsedArgs = JSON.parse(preInput); - if (typeof preprocessor !== 'function') {{ - console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }}) + '\n'); - continue; - }} - try {{ - function preArgsObjToArr({{ {pre_spread} }}: any) {{ - return [ {pre_spread} ]; - }} - const preprocessedArgs: any = await preprocessor(...preArgsObjToArr(parsedArgs)); - console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); - // Now call main with preprocessed args - let {{ {spread} }} = preprocessedArgs ?? {{}}; - {dates} - let res: any = await main(...[ {spread} ]); - console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); - }} catch (e) {{ - console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }}) + '\n'); - }} - continue; - }}"# - ) - } else { - String::new() - }; - - // 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 wrapper_content: String = format!( - r#" -import {{ main }} from "./main.ts"; -{preprocessor_import} - -BigInt.prototype.toJSON = function () {{ - return this.toString(); -}}; - -{dates} - -console.log('start\n'); - -const decoder = new TextDecoder(); -for await (const chunk of Deno.stdin.readable) {{ - const lines = decoder.decode(chunk); - let exit = false; - for (const line of lines.trim().split("\n")) {{ - if (line === "end") {{ - exit = true; - break; - }} - {preprocessor_logic} - try {{ - let {{ {spread} }} = JSON.parse(line) - {dates} - let res: any = await main(...[ {spread} ]); - console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); - }} catch (e) {{ - console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }}) + '\n'); - }} - }} - if (exit) {{ - break; - }} -}} -"#, - ); + let wrapper_content = generate_dedicated_worker_wrapper(inner_content)?; write_file(job_dir, "wrapper.ts", &wrapper_content)?; } build_import_map(w_id, script_path, base_internal_url, job_dir).await?; + let import_map = format!("{job_dir}/import_map.json"); + let reload = format!("--reload={base_internal_url}"); + let wrapper = format!("{job_dir}/wrapper.ts"); + let mut deno_args = vec!["run", "--no-check", "--import-map", &import_map, &reload]; + deno_args.extend_from_slice(DENO_UNSTABLE_ARGS); + deno_args.extend_from_slice(&["-A", &wrapper]); + handle_dedicated_process( &*DENO_PATH, job_dir, @@ -766,22 +787,7 @@ for await (const chunk of Deno.stdin.readable) {{ envs, context, common_deno_proc_envs, - vec![ - "run", - "--no-check", - "--import-map", - &format!("{job_dir}/import_map.json"), - &format!("--reload={base_internal_url}"), - "--unstable-unsafe-proto", - "--unstable-bare-node-builtins", - "--unstable-webgpu", - "--unstable-ffi", - "--unstable-fs", - "--unstable-worker-options", - "--unstable-http", - "-A", - &format!("{job_dir}/wrapper.ts"), - ], + deno_args, killpill_rx, job_completed_tx, token, diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 2fe56f50b9..e97ec8ff9a 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -86,12 +86,23 @@ pub use otel_tracing_proxy_ee::{load_internal_otel_exporter, DENO_OTEL_INITIALIZ pub use worker::*; pub use bun_executor::{ - build_loader, compute_bundle_local_and_remote_path, generate_dedicated_worker_wrapper, - get_common_bun_proc_envs, install_bun_lockfile, prebundle_bun_script, prepare_job_dir, - LoaderMode, BUN_DEDICATED_WORKER_ARGS, RELATIVE_BUN_BUILDER, RELATIVE_BUN_LOADER, + build_loader, compute_bundle_local_and_remote_path, get_common_bun_proc_envs, + install_bun_lockfile, prebundle_bun_script, prepare_job_dir, LoaderMode, + BUN_DEDICATED_WORKER_ARGS, RELATIVE_BUN_BUILDER, RELATIVE_BUN_LOADER, }; -pub use deno_executor::generate_deno_lock; +#[cfg(any(feature = "private", test))] +pub use bun_executor::{ + compute_ts_codegen, generate_multi_script_wrapper, TsScriptCodegen, TsScriptEntry, +}; +#[cfg(any(feature = "private", test))] +pub use deno_executor::generate_dedicated_worker_wrapper as generate_deno_dedicated_worker_wrapper; +pub use deno_executor::{generate_deno_lock, DENO_UNSTABLE_ARGS}; pub use prepare_deps::run_prepare_deps_cli; +#[cfg(all(feature = "python", any(feature = "private", test)))] +pub use python_executor::{ + compute_py_codegen, generate_multi_script_wrapper as generate_py_multi_script_wrapper, + PyScriptCodegen, PyScriptEntry, +}; #[cfg(feature = "python")] pub use python_versions::PyV; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 3b6507a7e0..78d3d54e15 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -117,7 +117,12 @@ async fn handle_piptar_uploads(mut rx: tokio::sync::mpsc::UnboundedReceiver bool { + RELATIVE_IMPORT_REGEX.is_match(content) +} #[cfg(all(feature = "enterprise", feature = "parquet", unix))] use crate::global_cache::pull_from_tar; @@ -1084,6 +1089,320 @@ fn python_preprocessor_spread(sig: windmill_parser::MainArgSignature, indent: &s } } +/// Pre-computed codegen data for a Python script. +/// Computed in Rust from the parsed signature, then baked into the wrapper template. +#[cfg(any(feature = "private", test))] +pub struct PyScriptCodegen { + /// Python module directory dot notation (e.g., "f.my") + pub module_dir_dot: String, + /// Python module name / last path component (e.g., "script") + pub module_name: String, + /// Directory path for the module (e.g., "f/my") + pub dirs: String, + /// Inline Python code for type transforms (dates, bytes, etc.) + pub transforms: String, + /// Inline Python code for arg spread / filtering + pub spread: String, + /// Inline Python code for preprocessor arg spread (if applicable) + pub pre_spread: Option, +} + +/// Parse a Python script and compute the codegen data. +/// This reuses the same logic that was used on main in `prepare_wrapper`. +#[cfg(any(feature = "private", test))] +pub fn compute_py_codegen(content: &str, script_path: &str) -> PyScriptCodegen { + let dirs = compute_python_module_dir(script_path); + let last = script_path + .split("/") + .map(|x| { + if x.starts_with(|x: char| x.is_ascii_digit()) { + format!("_{}", x) + } else { + x.to_string() + } + }) + .last() + .unwrap() + .replace("-", "_") + .replace(" ", "_") + .to_lowercase(); + + let sig = windmill_parser_py::parse_python_signature(content, None, false).unwrap_or_default(); + let pre_sig = windmill_parser_py::parse_python_signature( + content, + Some("preprocessor".to_string()), + false, + ) + .ok() + .filter(|s| !s.args.is_empty()); + + let init_sig = pre_sig.as_ref().unwrap_or(&sig); + + let transforms = init_sig + .args + .iter() + .map(|x| match x.typ { + windmill_parser::Typ::Bytes => { + let name = &x.name; + format!( + "if \"{name}\" in kwargs and kwargs[\"{name}\"] is not None:\n \ + kwargs[\"{name}\"] = base64.b64decode(kwargs[\"{name}\"])\n", + ) + } + windmill_parser::Typ::Datetime => { + let name = &x.name; + format!( + "if \"{name}\" in kwargs and kwargs[\"{name}\"] is not None:\n \ + kwargs[\"{name}\"] = datetime.fromisoformat(kwargs[\"{name}\"])\n", + ) + } + windmill_parser::Typ::Date => { + let name = &x.name; + format!( + "if \"{name}\" in kwargs and kwargs[\"{name}\"] is not None:\n \ + try:\n \ + kwargs[\"{name}\"] = date.fromisoformat(kwargs[\"{name}\"])\n \ + except ValueError:\n \ + for _fmt in (\"%d-%m-%Y\", \"%m/%d/%Y\", \"%d/%m/%Y\", \"%Y/%m/%d\"):\n \ + try:\n \ + kwargs[\"{name}\"] = datetime.strptime(kwargs[\"{name}\"], _fmt).date()\n \ + break\n \ + except ValueError:\n \ + continue\n", + ) + } + _ => "".to_string(), + }) + .collect::>() + .join(""); + + let spread = if sig.star_kwargs { + "args = kwargs".to_string() + } else { + sig.args + .into_iter() + .map(|x| { + let name = &x.name; + if x.default.is_none() { + format!("args[\"{name}\"] = kwargs.get(\"{name}\")") + } else { + format!( + r#"args["{name}"] = kwargs.get("{name}") + if args["{name}"] is None: + del args["{name}"]"# + ) + } + }) + .join("\n ") + }; + + let pre_spread = pre_sig.map(|sig| python_preprocessor_spread(sig, " ")); + + let module_dir_dot = dirs.replace("/", ".").replace("-", "_"); + + PyScriptCodegen { module_dir_dot, module_name: last, dirs, transforms, spread, pre_spread } +} + +/// Script entry for the Python unified wrapper generator. +#[cfg(any(feature = "private", test))] +pub struct PyScriptEntry<'a> { + pub original_path: &'a str, + pub codegen: &'a PyScriptCodegen, +} + +/// Generate a wrapper for Python dedicated workers and runner groups. +/// All scripts are baked in at codegen time with proper Python imports and inline arg handling. +/// Protocol: +/// exec:: -> wm_res[success]: | wm_res[error]: +/// exec_preprocess:: -> wm_res[preprocessed_args]: then wm_res[success]: | wm_res[error]: +/// end -> exit +#[cfg(any(feature = "private", test))] +pub fn generate_multi_script_wrapper( + scripts: &[PyScriptEntry<'_>], + skip_result_postprocessing: bool, + any_relative_imports: bool, +) -> String { + let postprocessor = get_result_postprocessor(skip_result_postprocessing); + let res_to_json_body = python_res_to_json_body(postprocessor); + + let imports: String = scripts + .iter() + .enumerate() + .map(|(i, e)| { + format!( + "from {module_dir_dot} import {module_name} as _s{i}", + module_dir_dot = e.codegen.module_dir_dot, + module_name = e.codegen.module_name, + ) + }) + .collect::>() + .join("\n"); + + let mut functions = String::new(); + let mut registrations = String::new(); + + for (i, entry) in scripts.iter().enumerate() { + let cg = entry.codegen; + let indented_transforms = cg + .transforms + .split('\n') + .map(|line| { + if line.is_empty() { + String::new() + } else { + format!(" {}", line) + } + }) + .collect::>() + .join("\n"); + + functions.push_str(&format!( + r#" +def transform_{i}(kwargs): +{indented_transforms} + args = dict() + {spread} + for k, v in list(args.items()): + if v == '': + del args[k] + return args +"#, + spread = cg.spread + )); + + let pre_fn = if let Some(ref pre_spread) = cg.pre_spread { + functions.push_str(&format!( + r#" +def pre_transform_{i}(kwargs): + pre_args = dict() + {pre_spread} + for k, v in list(pre_args.items()): + if v == '': + del pre_args[k] + return pre_args +"#, + )); + format!("pre_transform_{i}") + } else { + "None".to_string() + }; + + registrations.push_str(&format!( + "scripts[\"{path}\"] = {{ 'mod': _s{i}, 'transform': transform_{i}, 'pre_transform': {pre_fn} }}\n", + path = entry.original_path, + )); + } + + let import_loader = if any_relative_imports { + "import loader" + } else { + "" + }; + + format!( + r#" +import json +import sys +import traceback +import re +import base64 +from datetime import datetime, date +{import_loader} + +{imports} + +scripts = {{}} + +def to_b_64(v: bytes): + b64 = base64.b64encode(v) + return b64.decode('ascii') + +replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\u0000|Infinity|\-Infinity)') + +def res_to_json(res, typ): +{res_to_json_body} +{functions} +{registrations} + +sys.stdout.write('start\n') +sys.stdout.flush() + +for line in sys.stdin: + line = line.strip() + if line == 'end': + break + + if line.startswith('exec_preprocess:'): + try: + rest = line[len('exec_preprocess:'):] + colon_idx = rest.index(':') + script_path = rest[:colon_idx] + args_json = rest[colon_idx + 1:] + + entry = scripts.get(script_path) + if not entry: + err_json = json.dumps({{ "message": "Script not found: " + script_path, "name": "Error" }}, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[error]:" + err_json + "\n") + sys.stdout.flush() + continue + + mod = entry['mod'] + if not hasattr(mod, 'preprocessor') or not callable(mod.preprocessor): + err_json = json.dumps({{"message": "preprocessor function is missing", "name": "Error"}}, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[error]:" + err_json + "\n") + sys.stdout.flush() + continue + kwargs = json.loads(args_json, strict=False) + pre_args = entry['pre_transform'](kwargs) + preprocessed = mod.preprocessor(**pre_args) + preprocessed_json = json.dumps(preprocessed, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[preprocessed_args]:" + preprocessed_json + "\n") + main_args = entry['transform'](preprocessed if preprocessed else {{}}) + res = mod.main(**main_args) + typ = type(res) + res_json = res_to_json(res, typ) + sys.stdout.write("wm_res[success]:" + res_json + "\n") + except BaseException as e: + exc_type, exc_value, exc_traceback = sys.exc_info() + tb = traceback.format_tb(exc_traceback) + err_json = json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[error]:" + err_json + "\n") + sys.stdout.flush() + continue + + if line.startswith('exec:'): + try: + rest = line[len('exec:'):] + colon_idx = rest.index(':') + script_path = rest[:colon_idx] + args_json = rest[colon_idx + 1:] + + entry = scripts.get(script_path) + if not entry: + err_json = json.dumps({{ "message": "Script not found: " + script_path, "name": "Error" }}, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[error]:" + err_json + "\n") + sys.stdout.flush() + continue + + kwargs = json.loads(args_json, strict=False) + args = entry['transform'](kwargs) + res = entry['mod'].main(**args) + typ = type(res) + res_json = res_to_json(res, typ) + sys.stdout.write("wm_res[success]:" + res_json + "\n") + except BaseException as e: + exc_type, exc_value, exc_traceback = sys.exc_info() + tb = traceback.format_tb(exc_traceback) + err_json = json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[error]:" + err_json + "\n") + sys.stdout.flush() + continue + + sys.stderr.write("Unknown command: " + line + "\n") +"# + ) +} + async fn prepare_wrapper( job_dir: &str, job_flow_step_id: Option<&str>, @@ -1303,7 +1622,7 @@ async fn replace_pip_secret( } } -async fn handle_python_deps( +pub(crate) async fn handle_python_deps( job_dir: &str, requirements_o: Option<&String>, inner_content: &str, @@ -2419,6 +2738,22 @@ pub async fn start_worker( use crate::PyV; tracing::info!("script path: {}", script_path); + let codegen = compute_py_codegen(inner_content, script_path); + + // Write script to proper module path (e.g., f/my/script.py) + let module_dir = format!("{}/{}", job_dir, codegen.dirs); + tokio::fs::create_dir_all(&module_dir).await?; + write_file( + &module_dir, + &format!("{}.py", codegen.module_name), + inner_content, + )?; + + let any_relative_imports = RELATIVE_IMPORT_REGEX.is_match(inner_content); + if any_relative_imports { + let _ = write_file(job_dir, "loader.py", RELATIVE_PYTHON_LOADER)?; + } + let mut mem_peak: i32 = 0; let mut canceled_by: Option = None; let context = variables::get_reserved_variables( @@ -2463,122 +2798,12 @@ pub async fn start_worker( ) .await?; - let ( - import_loader, - import_base64, - import_datetime, - module_dir_dot, - _dirs, - last, - transforms, - spread, - _, - _, - ) = prepare_wrapper(job_dir, None, None, None, inner_content, script_path).await?; - - // Parse preprocessor signature if the script has one - let pre_spread = windmill_parser_py::parse_python_signature( - inner_content, - Some("preprocessor".to_string()), - false, - ) - .ok() - .filter(|sig| !sig.args.is_empty()) - .map(|sig| python_preprocessor_spread(sig, " ")); - { - let postprocessor = get_result_postprocessor(annotations.skip_result_postprocessing); - let indented_transforms = transforms - .lines() - .map(|x| format!(" {}", x)) - .collect::>() - .join("\n"); - - let preprocessor_logic = if let Some(ref pre_spread) = pre_spread { - format!( - r#" - if line.startswith('preprocess:'): - pre_input = line[len('preprocess:'):] - kwargs = json.loads(pre_input, strict=False) - if not hasattr(inner_script, 'preprocessor') or not callable(inner_script.preprocessor): - err_json = json.dumps({{"message": "preprocessor function is missing", "name": "Error"}}, separators=(',', ':'), default=str).replace('\n', '') - sys.stdout.write("wm_res[error]:" + err_json + "\n") - sys.stdout.flush() - continue - try: - pre_args = {{}} - {pre_spread} - for k, v in list(pre_args.items()): - if v == '': - del pre_args[k] - preprocessed_kwargs = inner_script.preprocessor(**pre_args) - preprocessed_json = json.dumps(preprocessed_kwargs, separators=(',', ':'), default=str).replace('\n', '') - sys.stdout.write("wm_res[preprocessed_args]:" + preprocessed_json + "\n") - transform_and_run(preprocessed_kwargs) - except BaseException as e: - exc_type, exc_value, exc_traceback = sys.exc_info() - tb = traceback.format_tb(exc_traceback) - err_json = json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', '') - sys.stdout.write("wm_res[error]:" + err_json + "\n") - sys.stdout.flush() - continue -"# - ) - } else { - String::new() - }; - - let res_to_json_body = python_res_to_json_body(postprocessor); - let wrapper_content: String = format!( - r#" -import json -{import_loader} -{import_base64} -{import_datetime} -import traceback -import sys -from {module_dir_dot} import {last} as inner_script -import re - - -def to_b_64(v: bytes): - import base64 - b64 = base64.b64encode(v) - return b64.decode('ascii') - -def res_to_json(res, typ): -{res_to_json_body} - -def transform_and_run(kwargs): - args = {{}} -{indented_transforms} - {spread} - for k, v in list(args.items()): - if v == '': - del args[k] - res = inner_script.main(**args) - typ = type(res) - res_json = res_to_json(res, typ) - sys.stdout.write("wm_res[success]:" + res_json + "\n") - -replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\u0000|Infinity|\-Infinity)') -sys.stdout.write('start\n') - -for line in sys.stdin: - if line == 'end\n': - break - line = line.strip() - {preprocessor_logic} - kwargs = json.loads(line, strict=False) - try: - transform_and_run(kwargs) - except BaseException as e: - exc_type, exc_value, exc_traceback = sys.exc_info() - tb = traceback.format_tb(exc_traceback) - err_json = json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', '') - sys.stdout.write("wm_res[error]:" + err_json + "\n") - sys.stdout.flush() -"#, + let scripts = [PyScriptEntry { original_path: script_path, codegen: &codegen }]; + let wrapper_content = generate_multi_script_wrapper( + &scripts, + annotations.skip_result_postprocessing, + any_relative_imports, ); write_file(job_dir, "wrapper.py", &wrapper_content)?; } @@ -2632,6 +2857,7 @@ for line in sys.stdin: &mut None, ) .await?; + handle_dedicated_process( &python_path, job_dir, @@ -2704,4 +2930,43 @@ mod tests { // @ is replaced with . assert_eq!(compute_python_module_dir("u/@admin/script"), "u/.admin"); } + + #[test] + fn test_compute_py_codegen_basic_args() { + let code = "def main(x: str, y: int):\n return x\n"; + let cg = compute_py_codegen(code, "f/test/script"); + assert!(cg.spread.contains("args[\"x\"]")); + assert!(cg.spread.contains("args[\"y\"]")); + assert!(cg.transforms.is_empty()); + assert!(cg.pre_spread.is_none()); + assert_eq!(cg.module_name, "script"); + } + + #[test] + fn test_compute_py_codegen_with_datetime_and_bytes() { + let code = "import datetime\n\ndef main(name: str, created_at: datetime.datetime, file: bytes):\n return name\n"; + let cg = compute_py_codegen(code, "f/my/handler"); + assert!(cg.transforms.contains("datetime.fromisoformat")); + assert!(cg.transforms.contains("base64.b64decode")); + assert!(cg.spread.contains("args[\"name\"]")); + assert_eq!(cg.module_dir_dot, "f.my"); + assert_eq!(cg.module_name, "handler"); + } + + #[test] + fn test_compute_py_codegen_star_kwargs() { + let code = "def main(**kwargs):\n return kwargs\n"; + let cg = compute_py_codegen(code, "f/test/star"); + assert_eq!(cg.spread, "args = kwargs"); + } + + #[test] + fn test_compute_py_codegen_with_preprocessor() { + let code = "import datetime\n\ndef main(x: str, ts: datetime.datetime):\n return x\n\ndef preprocessor(input: str, when: datetime.datetime):\n return {\"x\": input, \"ts\": when}\n"; + let cg = compute_py_codegen(code, "f/test/pre"); + assert!(cg.spread.contains("args[\"x\"]")); + assert!(cg.pre_spread.is_some()); + let pre = cg.pre_spread.as_ref().unwrap(); + assert!(pre.contains("pre_args[\"input\"]")); + } } diff --git a/frontend/src/lib/components/DedicatedWorkersSelector.svelte b/frontend/src/lib/components/DedicatedWorkersSelector.svelte index 1ed9d36a7c..94086982fb 100644 --- a/frontend/src/lib/components/DedicatedWorkersSelector.svelte +++ b/frontend/src/lib/components/DedicatedWorkersSelector.svelte @@ -1,6 +1,22 @@ +{#snippet depBadge(dep: string)} + {#if existingDeps.has(dep)} + + {dep} + + + {:else} + + Workspace dependency '{dep}' not found. Create it in workspace settings to enable shared + runners. + + + + {dep} + + {/if} +{/snippet} + +{#snippet tagRow(tag: string, info: SelectedTagInfo | undefined)} +
+
+ {#if info?.type === 'flow' && info.runners && info.runners.length > 0} + + {:else} +
+ {/if} +
+ {#if info} + {#if info.type === 'flow'} + + {:else} + + {/if} + {info.path} + ({info.workspace}) + + {#if !tagRunnerGroup.has(tag)} + {#if info.workspaceDeps} + {#each info.workspaceDeps as dep} + {@render depBadge(dep)} + {/each} + {/if} + {#if info.type === 'flow' && info.runners} + + {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} + + {:else if info.language} + {info.language} + {/if} + {/if} + {:else} + {tag} + {/if} +
+ {#if !disabled} + + {/if} +
+ + {#if info?.type === 'flow' && info.expanded && info.runners} +
+ {#each info.runners as runner (runner.stepId)} +
+ {runner.stepId} + {#if runner.stepSummary} + {runner.stepSummary} + {/if} + + {runner.isInline ? runner.language : runner.scriptPath} + +
+ {/each} +
+ {/if} +
+{/snippet} +
- {#if selectedTags.length > 0}
-
- {#each selectedTags as tag (tag)} - {@const info = selectedTagsInfo.get(tag)} +
+ + {#each runnerGroups as group (`${group.depName}:${group.language}`)}
-
- {#if info?.type === 'flow' && info.runners && info.runners.length > 0} - - {:else} -
- {/if} -
-
- {#if info} - {#if info.type === 'flow'} - - {:else} - - {/if} - {info.path} - ({info.workspace}) - {#if info.type === 'flow' && info.runners} - - {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} - - {:else if info.type === 'script'} - 1 runner - {/if} - {:else} - {tag} - {/if} -
-
- {#if !disabled} - - {/if} +
+ + Shared runner + + {@render depBadge(group.depName)} + {group.language}
+
+ {#each group.tags as tag (tag)} + {@const info = selectedTagsInfo.get(tag)} + {@render tagRow(tag, info)} + {/each} +
+
+ {/each} - {#if info?.type === 'flow' && info.expanded && info.runners} -
- {#each info.runners as runner (runner.stepId)} -
- {runner.stepId} - {#if runner.stepSummary} - {runner.stepSummary} - {/if} - - {runner.isInline ? runner.language : runner.scriptPath} - -
- {/each} -
+ + {#each standaloneTags as tag (tag)} + {@const info = selectedTagsInfo.get(tag)} +
+ {#if info?.type === 'flow'} + + {:else} + + {/if} + {info?.path ?? tag} + ({info?.workspace ?? ''}) + + {#if info?.workspaceDeps} + {#each info.workspaceDeps as dep} + {@render depBadge(dep)} + {/each} + {/if} + {#if info?.type === 'flow' && info.runners} + + {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} + + {:else if info?.language} + {info.language} + {/if} + {#if !disabled} + {/if}
{/each} @@ -585,15 +824,18 @@ {/if}
- {runnable.displayName} + {runnable.displayName} {#if runnable.type === 'flow' && runnable.runners} {runnable.runners.length} {/if} - + {#if runnable.workspaceDeps} + {#each runnable.workspaceDeps as dep} + {@render depBadge(dep)} + {/each} + {/if} + {runnable.type === 'flow' ? 'flow' : runnable.language}