From 619ebb65ce8dce8264add31c3147919802a8286a Mon Sep 17 00:00:00 2001 From: hugocasa Date: Thu, 2 Apr 2026 21:37:02 +0200 Subject: [PATCH] feat: restore bun for dedicated workers, fix dispatch & serialization, cross-workspace deps (#8645) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: restore bun as default runtime for dedicated workers Co-Authored-By: Claude Opus 4.6 (1M context) * chore: add context comment for bun dedicated worker nodejs migration Co-Authored-By: Claude Opus 4.6 (1M context) * fix: dedicated worker dispatch for flows + add E2E tests - Add workspace_id prefix to dedicated worker map lookup keys - Update ee-repo-ref for dedicated worker path handling fix - Add spawn_test_worker_dedicated/in_test_worker_dedicated test helpers - Add 6 E2E tests for dedicated workers: - test_dedicated_flow_rawscript (regression for "Script not found" bug) - test_dedicated_flow_workspace_script - test_dedicated_flow_multiple_steps - test_dedicated_standalone_script - test_dedicated_runner_group - test_dedicated_flow_runners - Add dedicated_flows.sql fixture with scripts, flows, and worker config Co-Authored-By: Claude Opus 4.6 (1M context) * fix: always run dependency job for dedicated worker scripts When a script with dedicated_worker=true is deployed with a pre-computed lock (e.g. via wmill sync push), no dependency job was created, so the dedicated worker never detected the update and kept running the old version. Now dedicated worker scripts always generate a dependency job regardless of whether a lock is provided. The dependency job runs on the dedicated worker and triggers a restart so it picks up the new script version. Fixes #8638 Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use serial_test for dedicated worker tests to avoid WORKER_CONFIG races Dedicated worker tests need non-default worker tags in the global WORKER_CONFIG. When run in parallel (CI uses --test-threads=10), multiple tests clobber each other's config. Use #[serial] to ensure dedicated worker tests run sequentially. Also load worker config from DB via load_worker_config() instead of manually setting WORKER_CONFIG fields, ensuring consistency with the monitor's reload path. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: nodejs dedicated worker script_path shadowing + add multi-language E2E tests Fix script_path shadowing in bun_executor nodejs branch where the wrapper file path was passed to handle_dedicated_process instead of the logical path, causing "Script not found" for all //nodejs dedicated workers. Add E2E tests for dedicated flows in all supported languages: - test_dedicated_flow_deno - test_dedicated_flow_python - test_dedicated_flow_bunnative (V8 PrewarmedIsolate path) - test_dedicated_flow_bun_nodejs (//nodejs annotation) Co-Authored-By: Claude Opus 4.6 (1M context) * fix: simplify dedicated worker dispatch + add serialization and E2E tests - Unified lookup: always use {workspace}:{runnable_path} for dedicated worker dispatch, replacing the flow_step_id iteration approach - Added serialization_semaphore parameter to executor start_worker fns - Added E2E tests: cross-workspace isolation, conflicting flow step IDs, preprocessor on dedicated worker - Added workspace field to RunJob for cross-workspace test support Co-Authored-By: Claude Opus 4.6 (1M context) * feat: cross-workspace workspace dependencies on workers page Add two new instance-level endpoints to the configs router: - GET /configs/list_all_workspace_dependencies - GET /configs/list_all_dedicated_with_deps Both require devops role and return data across all workspaces, enabling the workers page to show a consistent view of which workspace dependencies exist regardless of which workspace the user is browsing. Update DedicatedWorkersSelector to use the new cross-workspace endpoints with fallback to per-workspace calls for non-devops users. Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref to include dedicated worker lookup simplification Co-Authored-By: Claude Opus 4.6 (1M context) * chore: use branch name for ee-repo-ref (CI can't fetch by SHA from non-default branch) Co-Authored-By: Claude Opus 4.6 (1M context) * Update ee-repo-ref.txt with new reference * sqlx * fix: revert serialization semaphore, multi-workspace picker, dep conflict warnings - Remove serialization_semaphore from executor start_worker signatures - Remove serialization test and fixtures - Fix DedicatedWorkersSelector to preserve tags from other workspaces when toggling in the picker - Track workspace deps per-workspace for conflict detection - Show warning when dep exists in another workspace but not the script's - Group runner groups per-workspace to prevent cross-workspace merging - Add workspace to dep badge link URL Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * fix: simplify exec protocol — execd: for single-script, exec: for runner groups Add execd:/execd_preprocess: commands to bun/deno/python wrappers for single-script dedicated workers (no path needed). Runner groups keep exec:/exec_preprocess: with path for multi-script disambiguation. Co-Authored-By: Claude Opus 4.6 (1M context) * test: add unit tests for execd:/exec: wrapper protocol Verify generate_multi_script_wrapper produces both execd: (single-script) and exec: (runner group) protocol handlers, including preprocessor variants. Co-Authored-By: Claude Opus 4.6 (1M context) * Update commit reference in ee-repo-ref.txt * fix: remove beta badge from squash loop, keep tooltip Co-Authored-By: Claude Opus 4.6 (1M context) * fix: update protocol tests to use execd: for single-script wrappers Deno and bun single-script protocol tests now send execd:{args} instead of exec:{path}:{args}, matching the updated wrapper protocol. Multi-script (runner group) tests continue to use exec:{path}:{args}. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: remove unused TEST_SCRIPT_PATH in deno protocol tests Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- CLAUDE.md | 1 + ...20eb439a80d52ae75ebd25128ffce759cd313.json | 64 ++ ...887c5424562acd9769c80c5899193ef16b56b.json | 72 ++ backend/Cargo.toml | 1 + backend/ee-repo-ref.txt | 2 +- ...7_bun_dedicated_nodejs_annotation.down.sql | 8 + ...637_bun_dedicated_nodejs_annotation.up.sql | 10 + backend/tests/bun_jobs.rs | 16 +- backend/tests/dedicated_workers.rs | 660 ++++++++++++++++++ backend/tests/fixtures/dedicated_flows.sql | 252 +++++++ backend/windmill-api-configs/src/lib.rs | 89 +++ backend/windmill-api-scripts/src/scripts.rs | 6 +- backend/windmill-api/openapi.yaml | 58 ++ backend/windmill-test-utils/src/lib.rs | 132 +++- backend/windmill-worker/src/bun_executor.rs | 85 ++- backend/windmill-worker/src/deno_executor.rs | 23 +- .../windmill-worker/src/python_executor.rs | 23 +- backend/windmill-worker/src/worker.rs | 26 +- .../DedicatedWorkersSelector.svelte | 202 ++++-- .../components/flows/content/FlowLoop.svelte | 17 +- .../flows/content/FlowWhileLoop.svelte | 17 +- 21 files changed, 1618 insertions(+), 146 deletions(-) create mode 100644 backend/.sqlx/query-edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313.json create mode 100644 backend/.sqlx/query-f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b.json create mode 100644 backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql create mode 100644 backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.up.sql create mode 100644 backend/tests/dedicated_workers.rs create mode 100644 backend/tests/fixtures/dedicated_flows.sql diff --git a/CLAUDE.md b/CLAUDE.md index 0f01e67c57..825a033f94 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -26,6 +26,7 @@ Open-source platform for internal tools, workflows, API integrations, background - **DB**: `psql postgres://postgres:changeme@localhost:5432/windmill` - **Login**: `admin@windmill.dev` / `changeme` - **Instance settings**: navigate to `/#superadmin-settings` +- **Migrations**: use `cargo sqlx migrate add -r ` from `backend/` to create new migrations (never generate timestamps manually) ## Banned Patterns diff --git a/backend/.sqlx/query-edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313.json b/backend/.sqlx/query-edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313.json new file mode 100644 index 0000000000..33b0b1f144 --- /dev/null +++ b/backend/.sqlx/query-edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313.json @@ -0,0 +1,64 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_id, name, language AS \"language: windmill_common::scripts::ScriptLang\"\n FROM workspace_dependencies\n WHERE archived = false\n ORDER BY workspace_id, name", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "name", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "language: windmill_common::scripts::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", + "rlang" + ] + } + } + } + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + true, + false + ] + }, + "hash": "edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313" +} diff --git a/backend/.sqlx/query-f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b.json b/backend/.sqlx/query-f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b.json new file mode 100644 index 0000000000..ad6028c14f --- /dev/null +++ b/backend/.sqlx/query-f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b.json @@ -0,0 +1,72 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT ON (workspace_id, path)\n workspace_id, path, language AS \"language: windmill_common::scripts::ScriptLang\", content\n FROM script\n WHERE archived = false\n AND dedicated_worker = true\n AND language = ANY($1::text[]::SCRIPT_LANG[])\n ORDER BY workspace_id, path, created_at DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "language: windmill_common::scripts::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", + "rlang" + ] + } + } + } + }, + { + "ordinal": 3, + "name": "content", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "TextArray" + ] + }, + "nullable": [ + false, + false, + false, + false + ] + }, + "hash": "f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b" +} diff --git a/backend/Cargo.toml b/backend/Cargo.toml index d636d4e6f4..b9a9413cb2 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -267,6 +267,7 @@ windmill-types.workspace = true opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true } windmill-trigger.workspace = true +serial_test = "3" windmill-trigger-websocket.workspace = true windmill-trigger-postgres.workspace = true windmill-trigger-mqtt.workspace = true diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 8d70a431af..b572bde77b 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -0e001bab643e449b3310b0692dd3598ee0902ecc +9d9e0284b336c0ee4493143391b931d47872bf3e diff --git a/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql b/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql new file mode 100644 index 0000000000..f6e66116d3 --- /dev/null +++ b/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql @@ -0,0 +1,8 @@ +-- Remove the //nodejs annotation that was prepended by the up migration. +-- Only removes it if it's at the very start of the content. +UPDATE script +SET content = REPLACE(content, '//nodejs +// dedicated workers were previously running in nodejs mode by default, remove this annotation to use bun +', '') +WHERE language = 'bun' + AND dedicated_worker = true; diff --git a/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.up.sql b/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.up.sql new file mode 100644 index 0000000000..acb60af405 --- /dev/null +++ b/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.up.sql @@ -0,0 +1,10 @@ +-- Bun dedicated workers were previously forced to run in nodejs mode at runtime. +-- Now that bun is the default again, add the //nodejs annotation to existing +-- bun dedicated scripts that don't already have it, so their behavior is preserved. +UPDATE script +SET content = '//nodejs +// dedicated workers were previously running in nodejs mode by default, remove this annotation to use bun +' || content +WHERE language = 'bun' + AND dedicated_worker = true + AND content !~ '//\s*nodejs'; diff --git a/backend/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index dc36cca9f5..cf937bc931 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -1066,8 +1066,8 @@ mod dedicated_worker_protocol { let mut results = Vec::new(); for job_args in jobs { - // Protocol: exec:: - writeln!(stdin, "exec:{}:{}", TEST_SCRIPT_PATH, job_args.to_string()).unwrap(); + // Protocol: execd: (single-script, no path needed) + writeln!(stdin, "execd:{}", job_args.to_string()).unwrap(); stdin.flush().unwrap(); let mut response = String::new(); @@ -1653,8 +1653,6 @@ mod dedicated_worker_protocol_deno { 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, @@ -1704,7 +1702,7 @@ mod dedicated_worker_protocol_deno { let mut results = Vec::new(); for job_args in jobs { - writeln!(stdin, "exec:{}:{}", TEST_SCRIPT_PATH, job_args.to_string()).unwrap(); + writeln!(stdin, "execd:{}", job_args.to_string()).unwrap(); stdin.flush().unwrap(); loop { @@ -1825,7 +1823,13 @@ export function main(msg: string): never { let mut results = Vec::new(); for (cmd, args) in &commands { - writeln!(stdin, "{}:{}:{}", cmd, TEST_SCRIPT_PATH, args).unwrap(); + // Single-script Deno wrapper uses execd:/execd_preprocess: (no path) + let direct_cmd = if *cmd == "exec_preprocess" { + "execd_preprocess" + } else { + "execd" + }; + writeln!(stdin, "{}:{}", direct_cmd, args).unwrap(); stdin.flush().unwrap(); let expected_lines = if *cmd == "exec_preprocess" { 2 } else { 1 }; diff --git a/backend/tests/dedicated_workers.rs b/backend/tests/dedicated_workers.rs new file mode 100644 index 0000000000..4e593b20b4 --- /dev/null +++ b/backend/tests/dedicated_workers.rs @@ -0,0 +1,660 @@ +#[cfg(all(feature = "private", feature = "enterprise"))] +mod dedicated_worker_tests { + use serde_json::json; + use serial_test::serial; + use sqlx::{Pool, Postgres}; + use windmill_common::jobs::JobPayload; + use windmill_common::worker::WorkspacedPath; + use windmill_test_utils::*; + + pub async fn initialize_tracing() { + use std::sync::Once; + static ONCE: Once = Once::new(); + ONCE.call_once(|| { + let _ = windmill_common::tracing_init::initialize_tracing( + "test", + &windmill_common::utils::Mode::Standalone, + "test", + ); + }); + } + + fn wp(workspace: &str, path: &str) -> WorkspacedPath { + WorkspacedPath { workspace_id: workspace.to_string(), path: path.to_string() } + } + + /// Assert that a completed job (or one of its child step jobs) was executed + /// by a dedicated worker process with the expected mode by checking job logs + /// for the "dedicated worker {mode}:" prefix. + async fn assert_ran_on_dedicated_worker( + db: &Pool, + job_id: uuid::Uuid, + expected_mode: &str, + ) { + let marker = format!("dedicated worker {expected_mode}:"); + + // Check the job's logs (stored in job_logs table) + let logs: Option = + sqlx::query_scalar("SELECT logs FROM job_logs WHERE job_id = $1") + .bind(job_id) + .fetch_optional(db) + .await + .unwrap() + .flatten(); + + if let Some(ref l) = logs { + if l.contains(&marker) { + return; + } + } + + // For flow jobs, check child step jobs' logs + let child_ids: Vec = + sqlx::query_scalar("SELECT id FROM v2_job WHERE parent_job = $1") + .bind(job_id) + .fetch_all(db) + .await + .unwrap(); + + let mut child_logs = Vec::new(); + for child_id in &child_ids { + let cl: Option = + sqlx::query_scalar("SELECT logs FROM job_logs WHERE job_id = $1") + .bind(child_id) + .fetch_optional(db) + .await + .unwrap() + .flatten(); + child_logs.push(cl); + } + + assert!( + child_logs + .iter() + .any(|l| l.as_ref().is_some_and(|l| l.contains(&marker))), + "No job or child job had '{marker}' in logs. Job logs: {:?}, Child logs: {:?}", + logs, + child_logs, + ); + } + + /// Test a dedicated flow with a single inline RawScript bun step. + /// This is the regression test for the "Script not found" bug. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_rawscript(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_rawscript_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000001, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp( + "test-workspace", + "flow/f/system/dedicated_rawscript_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(15)); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with a Script step referencing an external workspace script. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_workspace_script(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_script_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000002, + }) + .arg("x", json!(7)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + // Need both the flow tag (for parent job) and the script tag (for step job) + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![ + wp("test-workspace", "flow/f/system/dedicated_script_flow"), + wp("test-workspace", "f/system/dedicated_double"), + ], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(14)); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with multiple inline RawScript steps. + /// Validates that each step gets its own wrapper key and no collisions occur. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_multiple_steps(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_multi_step_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000003, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp( + "test-workspace", + "flow/f/system/dedicated_multi_step_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(18)); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a standalone dedicated script (not in a flow). + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_standalone_script(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300001), + path: "f/system/dedicated_double".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default( + ), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("x", json!(21)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "f/system/dedicated_double")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(42)); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test runner groups: two scripts sharing a workspace dependency are auto-grouped + /// into a single runner process. Both should execute correctly. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_runner_group(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid_a = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300010), + path: "f/system/rg_script_a".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default( + ), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let uuid_b = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300011), + path: "f/system/rg_script_b".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default( + ), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + async { + // Wait for both jobs to complete + listener.find(&uuid_a).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_b).await; + }, + port, + vec![ + wp("test-workspace", "f/system/rg_script_a"), + wp("test-workspace", "f/system/rg_script_b"), + ], + ) + .await; + + let job_a = completed_job(uuid_a, &db).await; + let job_b = completed_job(uuid_b, &db).await; + assert_eq!(job_a.json_result().unwrap(), json!(105)); + assert_eq!(job_b.json_result().unwrap(), json!(205)); + assert_ran_on_dedicated_worker(&db, uuid_a, "bun").await; + assert_ran_on_dedicated_worker(&db, uuid_b, "bun").await; + Ok(()) + } + + /// Test flow runners: a squashed for-loop spawns dedicated subprocesses. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_runners(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + // Reset dedicated_workers in WORKER_CONFIG to avoid pollution from previous tests. + { + let mut wc = windmill_common::worker::WORKER_CONFIG.write().await; + wc.dedicated_worker = None; + wc.dedicated_workers = None; + wc.worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); + wc.priority_tags_sorted = vec![windmill_common::worker::PriorityTags { + priority: 0, + tags: wc.worker_tags.clone(), + }]; + windmill_common::worker::store_suspended_pull_query(&wc).await; + windmill_common::worker::store_pull_query(&wc).await; + } + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let job = in_test_worker( + db.clone(), + async { + RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_flow_runners".to_string(), + dedicated_worker: None, + apply_preprocessor: false, + version: 3000000000000004, + }) + .run_until_complete(&db, false, port) + .await + }, + port, + ) + .await; + + // for-loop over [1, 2, 3], each * 10 = [10, 20, 30] + assert_eq!(job.json_result().unwrap(), json!([10, 20, 30])); + // Verify the loop iterations ran on dedicated worker subprocesses + assert_ran_on_dedicated_worker(&db, job.id, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with a Deno inline RawScript step. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_deno(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_deno_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000005, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "flow/f/system/dedicated_deno_flow")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(105)); + assert_ran_on_dedicated_worker(&db, uuid, "deno").await; + Ok(()) + } + + /// Test a dedicated flow with a Python inline RawScript step. + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_python(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_python_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000006, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "flow/f/system/dedicated_python_flow")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(105)); + assert_ran_on_dedicated_worker(&db, uuid, "python").await; + Ok(()) + } + + /// Test a dedicated flow with a Bunnative (//native) inline RawScript step. + /// Bunnative uses the V8 PrewarmedIsolate path instead of a subprocess wrapper. + #[cfg(feature = "deno_core")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_bunnative(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_bunnative_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000007, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp( + "test-workspace", + "flow/f/system/dedicated_bunnative_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(105)); + assert_ran_on_dedicated_worker(&db, uuid, "nativets").await; + Ok(()) + } + + /// Test a dedicated flow with a Bun script using the //nodejs annotation. + /// Validates that the annotation routes execution through Node.js instead of Bun. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_bun_nodejs(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_nodejs_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000008, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "flow/f/system/dedicated_nodejs_flow")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(105)); + assert_ran_on_dedicated_worker(&db, uuid, "nodejs").await; + Ok(()) + } + + /// Test that two dedicated flows with conflicting step IDs (both "a") are correctly + /// disambiguated via runnable_path-based lookup (not step ID). + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_conflicting_step_ids(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid_a = RunJob::from(JobPayload::Flow { + path: "f/system/conflict_flow_a".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000009, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let uuid_b = RunJob::from(JobPayload::Flow { + path: "f/system/conflict_flow_b".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000010, + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + async { + listener.find(&uuid_a).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_b).await; + }, + port, + vec![ + wp("test-workspace", "flow/f/system/conflict_flow_a"), + wp("test-workspace", "flow/f/system/conflict_flow_b"), + ], + ) + .await; + + let job_a = completed_job(uuid_a, &db).await; + let job_b = completed_job(uuid_b, &db).await; + // Flow A: x + 1000 = 1005, Flow B: x + 2000 = 2005 + assert_eq!(job_a.json_result().unwrap(), json!(1005)); + assert_eq!(job_b.json_result().unwrap(), json!(2005)); + assert_ran_on_dedicated_worker(&db, uuid_a, "bun").await; + assert_ran_on_dedicated_worker(&db, uuid_b, "bun").await; + Ok(()) + } + + /// Test that a dedicated worker script with a preprocessor function works correctly. + /// The preprocessor doubles x, then main adds 100: preprocessor(5) → {x:10}, main(10) → 110. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_preprocessor(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300030), + path: "f/system/preprocess_script".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bun, + priority: None, + apply_preprocessor: true, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default( + ), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("x", json!(5)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "f/system/preprocess_script")], + ) + .await; + + let job = completed_job(uuid, &db).await; + // preprocessor(5) → {x: 10}, main(10) → 110 + assert_eq!(job.json_result().unwrap(), json!(110)); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + + // Verify preprocessed flag was set + let preprocessed: Option = + sqlx::query_scalar("SELECT preprocessed FROM v2_job WHERE id = $1") + .bind(uuid) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(preprocessed, Some(true)); + + Ok(()) + } + + /// Test that two workspaces with the same script path are correctly isolated. + /// Workspace 1: dedicated_double returns x * 2, Workspace 2: returns x * 3. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_cross_workspace_isolation(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid_ws1 = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300001), + path: "f/system/dedicated_double".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default( + ), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("x", json!(7)) + .push(&db) + .await; + + let uuid_ws2 = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300040), + path: "f/system/dedicated_double".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default( + ), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .workspace("test-workspace-2") + .arg("x", json!(7)) + .push(&db) + .await; + + let listener = listen_for_completed_jobs(&db).await; + in_test_worker_dedicated( + db.clone(), + async { + listener.find(&uuid_ws1).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_ws2).await; + }, + port, + vec![ + wp("test-workspace", "f/system/dedicated_double"), + wp("test-workspace-2", "f/system/dedicated_double"), + ], + ) + .await; + + let job_ws1 = completed_job(uuid_ws1, &db).await; + let job_ws2 = completed_job(uuid_ws2, &db).await; + // Workspace 1: x * 2 = 14, Workspace 2: x * 3 = 21 + assert_eq!(job_ws1.json_result().unwrap(), json!(14)); + assert_eq!(job_ws2.json_result().unwrap(), json!(21)); + assert_ran_on_dedicated_worker(&db, uuid_ws1, "bun").await; + assert_ran_on_dedicated_worker(&db, uuid_ws2, "bun").await; + Ok(()) + } +} diff --git a/backend/tests/fixtures/dedicated_flows.sql b/backend/tests/fixtures/dedicated_flows.sql new file mode 100644 index 0000000000..071f6d976a --- /dev/null +++ b/backend/tests/fixtures/dedicated_flows.sql @@ -0,0 +1,252 @@ +-- Fixtures for dedicated worker E2E tests + +-- A simple Bun script for testing dedicated Script steps in flows +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'export function main(x: number) { return x * 2; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/dedicated_double', 300001, 'bun', E'{}\n//bun.lock\n', true); + +-- Flow with a single RawScript inline bun step (the "Script not found" bug case) +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_rawscript_flow', +'{3000000000000001}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 10; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000001, +'test-workspace', +'f/system/dedicated_rawscript_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 10; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +-- Flow with a Script step referencing the external dedicated_double script +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_script_flow', +'{3000000000000002}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'{"modules":[{"id":"a","value":{"type":"script","path":"f/system/dedicated_double","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}}}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000002, +'test-workspace', +'f/system/dedicated_script_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'{"modules":[{"id":"a","value":{"type":"script","path":"f/system/dedicated_double","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}}}}]}', +'system' +); + +-- Flow with two inline RawScript steps (tests multi-step key uniqueness) +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_multi_step_flow', +'{3000000000000003}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 1; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}},{"id":"b","value":{"type":"rawscript","content":"export function main(x: number) { return x * 3; }","language":"bun","input_transforms":{"x":{"expr":"results.a","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000003, +'test-workspace', +'f/system/dedicated_multi_step_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 1; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}},{"id":"b","value":{"type":"rawscript","content":"export function main(x: number) { return x * 3; }","language":"bun","input_transforms":{"x":{"expr":"results.a","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +-- Two scripts sharing a workspace dependency for runner group testing. +-- Both reference the same external dep "f/system/dedicated_double" via extra_package_json annotation. +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'// extra_package_json: f/system/dedicated_double\nexport function main(x: number) { return x + 100; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/rg_script_a', 300010, 'bun', E'{}\n//bun.lock\n', true); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'// extra_package_json: f/system/dedicated_double\nexport function main(x: number) { return x + 200; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/rg_script_b', 300011, 'bun', E'{}\n//bun.lock\n', true); + +-- Flow with a Deno inline RawScript step +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_deno_flow', +'{3000000000000005}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 100; }","language":"deno","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}}}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000005, +'test-workspace', +'f/system/dedicated_deno_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 100; }","language":"deno","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}}}}]}', +'system' +); + +-- Flow with a Python inline RawScript step +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_python_flow', +'{3000000000000006}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"def main(x: int):\\n return x + 100","language":"python3","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}}}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000006, +'test-workspace', +'f/system/dedicated_python_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"def main(x: int):\\n return x + 100","language":"python3","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}}}}]}', +'system' +); + +-- Flow with a Bunnative (//native) inline RawScript step +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_bunnative_flow', +'{3000000000000007}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"//native\\nexport function main(x: number) { return x + 100; }","language":"bunnative","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000007, +'test-workspace', +'f/system/dedicated_bunnative_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"//native\\nexport function main(x: number) { return x + 100; }","language":"bunnative","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +-- Flow with a Bun + //nodejs annotation inline RawScript step +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_nodejs_flow', +'{3000000000000008}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"//nodejs\\nexport function main(x: number) { return x + 100; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000008, +'test-workspace', +'f/system/dedicated_nodejs_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"//nodejs\\nexport function main(x: number) { return x + 100; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +-- Flow with a squashed for-loop for testing flow runners. +-- The for-loop iterates over [1, 2, 3], each iteration runs a simple bun rawscript. +-- squash=true triggers spawn_flow_module_runners to create dedicated subprocesses. +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_flow_runners', +'{3000000000000004}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"forloopflow","iterator":{"type":"javascript","expr":"[1, 2, 3]"},"skip_failures":false,"parallel":false,"squash":true,"modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(iter: {value: number, index: number}) { return iter.value * 10; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000004, +'test-workspace', +'f/system/dedicated_flow_runners', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"forloopflow","iterator":{"type":"javascript","expr":"[1, 2, 3]"},"skip_failures":false,"parallel":false,"squash":true,"modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(iter: {value: number, index: number}) { return iter.value * 10; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}}]}', +'system' +); + +-- Two flows with conflicting step IDs (both have module "a") but different RawScript content. +-- Tests that runnable_path-based lookup correctly disambiguates them. +-- Flow A: x + 1000 +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/conflict_flow_a', +'{3000000000000009}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 1000; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000009, +'test-workspace', +'f/system/conflict_flow_a', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 1000; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +-- Flow B: x + 2000 +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/conflict_flow_b', +'{3000000000000010}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 2000; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000010, +'test-workspace', +'f/system/conflict_flow_b', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export function main(x: number) { return x + 2000; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}', +'system' +); + +-- A dedicated worker script with a preprocessor function. +-- preprocessor doubles x, main adds 100. +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'export function preprocessor(x: number) { return { x: x * 2 }; }\nexport function main(x: number) { return x + 100; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/preprocess_script', 300030, 'bun', E'{}\n//bun.lock\n', true); + +-- Second workspace for cross-workspace isolation testing. +INSERT INTO workspace(id, name, owner) VALUES ('test-workspace-2', 'test-workspace-2', 'test-user'); +INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES + ('test-workspace-2', 'test@windmill.dev', 'test-user', true, 'Admin'); +INSERT INTO workspace_key(workspace_id, kind, key) VALUES + ('test-workspace-2', 'cloud', 'test-key-2'); +INSERT INTO workspace_settings (workspace_id) VALUES ('test-workspace-2'); +INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES + ('test-workspace-2', 'all', 'All users', '{}'); + +-- Same path as dedicated_double but in workspace 2, returns x * 3 instead of x * 2. +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace-2', +'system', +E'export function main(x: number) { return x * 3; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/dedicated_double', 300040, 'bun', E'{}\n//bun.lock\n', true); + diff --git a/backend/windmill-api-configs/src/lib.rs b/backend/windmill-api-configs/src/lib.rs index 8485de5e95..fd77ff52bb 100644 --- a/backend/windmill-api-configs/src/lib.rs +++ b/backend/windmill-api-configs/src/lib.rs @@ -43,6 +43,14 @@ pub fn global_service() -> Router { "/list_available_python_versions", get(list_available_python_versions), ) + .route( + "/list_all_workspace_dependencies", + get(list_all_workspace_dependencies), + ) + .route( + "/list_all_dedicated_with_deps", + get(list_all_dedicated_with_deps), + ) } #[derive(Serialize, Deserialize, FromRow)] @@ -319,3 +327,84 @@ async fn list_configs() -> error::JsonResult { "Config listing available only in the enterprise version".to_string(), )) } + +#[derive(Serialize)] +struct WorkspaceDependencySummary { + workspace_id: String, + name: Option, + language: windmill_common::scripts::ScriptLang, +} + +async fn list_all_workspace_dependencies( + authed: ApiAuthed, + Extension(db): Extension, +) -> error::JsonResult> { + require_devops_role(&db, &authed.email).await?; + let deps = sqlx::query!( + r#"SELECT workspace_id, name, language AS "language: windmill_common::scripts::ScriptLang" + FROM workspace_dependencies + WHERE archived = false + ORDER BY workspace_id, name"#, + ) + .fetch_all(&db) + .await?; + Ok(Json( + deps.into_iter() + .map(|r| WorkspaceDependencySummary { + workspace_id: r.workspace_id, + name: r.name, + language: r.language, + }) + .collect(), + )) +} + +#[derive(Serialize)] +struct DedicatedScriptDepsWithWorkspace { + workspace_id: String, + path: String, + language: windmill_common::scripts::ScriptLang, + workspace_dep_names: Vec, +} + +async fn list_all_dedicated_with_deps( + authed: ApiAuthed, + Extension(db): Extension, +) -> error::JsonResult> { + require_devops_role(&db, &authed.email).await?; + + let rows = sqlx::query!( + r#"SELECT DISTINCT ON (workspace_id, path) + workspace_id, path, language AS "language: windmill_common::scripts::ScriptLang", content + FROM script + WHERE archived = false + AND dedicated_worker = true + AND language = ANY($1::text[]::SCRIPT_LANG[]) + ORDER BY workspace_id, path, created_at DESC"#, + &["python3", "bun", "bunnative", "deno"] as &[&str], + ) + .fetch_all(&db) + .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(); + DedicatedScriptDepsWithWorkspace { + workspace_id: row.workspace_id, + path: row.path, + language: row.language, + workspace_dep_names: dep_names, + } + }) + .collect(); + + Ok(Json(result)) +} diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index afa5ded9dc..84abdecddd 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -834,7 +834,11 @@ async fn create_script_internal<'c>( }) }; - let needs_lock_gen = lock.is_none() && codebase.is_none(); + // Always generate a lock for dedicated worker scripts, even if a lock was provided. + // The dependency job runs on the dedicated worker and triggers a restart so it picks + // up the new script version (see result_processor.rs: is_dependency_job && is_dedicated_worker). + let needs_lock_gen = + (lock.is_none() && codebase.is_none()) || ns.dedicated_worker.is_some_and(|x| x); let envs = ns.envs.as_ref().map(|x| x.as_slice()); let envs = if ns.envs.is_none() || ns.envs.as_ref().unwrap().is_empty() { None diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index f33ca19888..a019d887cd 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -15697,6 +15697,64 @@ paths: items: type: string + /configs/list_all_workspace_dependencies: + get: + summary: list all workspace dependencies + operationId: listAllWorkspaceDependencies + tags: + - config + responses: + "200": + description: a list of workspace dependency summaries + content: + application/json: + schema: + type: array + items: + type: object + properties: + workspace_id: + type: string + name: + type: string + language: + $ref: "#/components/schemas/ScriptLang" + required: + - workspace_id + - language + + /configs/list_all_dedicated_with_deps: + get: + summary: list all dedicated scripts with their dependencies + operationId: listAllDedicatedWithDeps + tags: + - config + responses: + "200": + description: a list of dedicated scripts with workspace dependencies + content: + application/json: + schema: + type: array + items: + type: object + properties: + workspace_id: + type: string + path: + type: string + language: + $ref: "#/components/schemas/ScriptLang" + workspace_dep_names: + type: array + items: + type: string + required: + - workspace_id + - path + - language + - workspace_dep_names + /agent_workers/create_agent_token: post: summary: create agent token diff --git a/backend/windmill-test-utils/src/lib.rs b/backend/windmill-test-utils/src/lib.rs index 153bdeeb7c..2778e7589f 100644 --- a/backend/windmill-test-utils/src/lib.rs +++ b/backend/windmill-test-utils/src/lib.rs @@ -160,6 +160,7 @@ pub struct RunJob { pub scheduled_for_o: Option>, pub email: String, pub job_id: Option, + pub workspace_id: String, } impl From for RunJob { @@ -170,6 +171,7 @@ impl From for RunJob { scheduled_for_o: None, email: "test@windmill.dev".to_string(), job_id: None, + workspace_id: "test-workspace".to_string(), } } } @@ -198,8 +200,13 @@ impl RunJob { self } + pub fn workspace(mut self, workspace_id: impl Into) -> Self { + self.workspace_id = workspace_id.into(); + self + } + pub async fn push(self, db: &Pool) -> Uuid { - let RunJob { payload, args, scheduled_for_o, email, job_id } = self; + let RunJob { payload, args, scheduled_for_o, email, job_id, workspace_id } = self; let mut hm_args = std::collections::HashMap::new(); for (k, v) in args { hm_args.insert(k, windmill_common::worker::to_raw_value(&v)); @@ -209,7 +216,7 @@ impl RunJob { let (uuid, tx) = windmill_queue::push( db, tx, - "test-workspace", + &workspace_id, payload, windmill_queue::PushArgs::from(&hm_args), /* user */ "test-user", @@ -436,6 +443,127 @@ pub fn spawn_test_worker( (tx, tokio::task::spawn(future)) } +/// Spawn a test worker configured with dedicated workers. +/// `dedicated_workers` should be in format `["workspace_id:path", ...]`. +pub fn spawn_test_worker_dedicated( + conn: &Connection, + port: u16, + dedicated_workers: Vec, +) -> (KillpillSender, tokio::task::JoinHandle<()>) { + #[cfg(feature = "deno_core")] + windmill_runtime_nativets::setup_deno_runtime().expect("V8 init failed"); + + std::fs::DirBuilder::new() + .recursive(true) + .create(&*windmill_worker::GO_BIN_CACHE_DIR) + .expect("could not create initial worker dir"); + + let (tx, rx) = KillpillSender::new(1); + let worker_instance: &str = "test worker instance"; + let worker_name: String = next_worker_name(); + let ip: &str = Default::default(); + let conn = conn.to_owned(); + + let tx2 = tx.clone(); + let future = async move { + let base_internal_url = format!("http://localhost:{}", port); + + // Insert dedicated worker config into the DB so that load_worker_config + // (called by the monitor and at worker startup) picks it up consistently. + // This avoids global WORKER_CONFIG races between parallel tests. + let dedicated_strs: Vec = dedicated_workers + .iter() + .map(|dw| format!("{}:{}", dw.workspace_id, dw.path)) + .collect(); + let mut all_tags = windmill_common::worker::DEFAULT_TAGS.clone(); + for dw in &dedicated_workers { + all_tags.push(windmill_common::worker::dedicated_worker_tag( + &dw.workspace_id, + &dw.path, + )); + } + let config_value = serde_json::json!({ + "dedicated_workers": dedicated_strs, + "worker_tags": all_tags, + }); + let db_ref = match &conn { + Connection::Sql(db) => db, + _ => panic!("dedicated worker tests require SQL connection"), + }; + sqlx::query("INSERT INTO config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = $2") + .bind(format!("worker__{}", *windmill_common::worker::WORKER_GROUP)) + .bind(&config_value) + .execute(db_ref) + .await + .expect("insert dedicated worker config"); + + // Load the config from DB (same path as the monitor) to set WORKER_CONFIG + // consistently. This ensures tags, pull queries, and dedicated_workers are + // all set from the same source of truth. + { + let killpill_tx2 = tx2.clone(); + let config = windmill_common::worker::load_worker_config(db_ref, killpill_tx2) + .await + .expect("load worker config"); + let mut wc = WORKER_CONFIG.write().await; + windmill_common::worker::store_suspended_pull_query(&config).await; + windmill_common::worker::store_pull_query(&config).await; + *wc = config; + } + windmill_worker::run_worker( + &conn, + worker_instance, + worker_name, + 1, + 1, + ip, + rx, + tx2, + &base_internal_url, + ) + .await + }; + + (tx, tokio::task::spawn(future)) +} + +/// Like `in_test_worker` but with dedicated worker configuration. +pub async fn in_test_worker_dedicated( + conn: impl Into, + inner: Fut, + port: u16, + dedicated_workers: Vec, +) -> ::Output { + set_jwt_secret().await; + // Reset WORKER_CONFIG to avoid stale state from previous tests' monitor reloads. + { + let mut wc = WORKER_CONFIG.write().await; + wc.dedicated_worker = None; + wc.dedicated_workers = None; + } + let (quit, worker) = spawn_test_worker_dedicated(&conn.into(), port, dedicated_workers); + let worker = tokio::time::timeout(std::time::Duration::from_secs(90), worker); + tokio::pin!(worker); + + let res = tokio::select! { + biased; + res = inner => res, + res = &mut worker => match + res.expect("worker timed out") + .expect("worker panicked") { + _ => panic!("worker quit early"), + }, + }; + + quit.send(); + + let _: () = worker + .await + .expect("worker timed out") + .expect("worker panicked"); + res +} + pub async fn listen_for_completed_jobs(db: &Pool) -> impl Stream + Unpin { listen_for_uuid_on(db, "completed").await } diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 433f96bcec..7592c90573 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -153,8 +153,10 @@ pub struct TsScriptEntry<'a> { /// 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]: +/// execd: -> execute the single registered script (non-runner-group) +/// execd_preprocess: -> preprocess + execute the single registered script +/// exec:: -> execute script by path (runner groups with multiple scripts) +/// exec_preprocess:: -> preprocess + execute script by path /// end -> exit #[cfg(any(feature = "private", test))] pub fn generate_multi_script_wrapper(scripts: &[TsScriptEntry<'_>], ext: &str) -> String { @@ -240,6 +242,43 @@ for await (const line of Readline.createInterface({{ input: process.stdin }})) { process.exit(0); }} + // Direct execution: single-script dedicated workers (no path needed) + if (line.startsWith("execd_preprocess:")) {{ + const argsJson = line.slice("execd_preprocess:".length); + const entry = scripts.values().next().value; + + try {{ + if (!entry.getPreArgs) {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }})); + continue; + }} + const preArgs = entry.getPreArgs(argsJson); + const preprocessedArgs = await entry.module.preprocessor(...preArgs); + console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value)); + const mainArgs = entry.getArgs(JSON.stringify(preprocessedArgs ?? {{}})); + const res = await entry.module.main(...mainArgs); + console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); + }} catch (e) {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }})); + }} + continue; + }} + + if (line.startsWith("execd:")) {{ + const argsJson = line.slice("execd:".length); + const entry = scripts.values().next().value; + + try {{ + const args = entry.getArgs(argsJson); + const res = await entry.module.main(...args); + console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); + }} catch (e) {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: argsJson }})); + }} + continue; + }} + + // Path-based execution: runner groups with multiple scripts if (line.startsWith("exec_preprocess:")) {{ const rest = line.slice("exec_preprocess:".length); const colonIdx = rest.indexOf(":"); @@ -3547,11 +3586,6 @@ pub async fn start_worker( let mut annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content); - //TODO: remove this when bun dedicated workers work without issues - if !annotation.native { - annotation.nodejs = true; - } - let context = variables::get_reserved_variables( &Connection::from(db.clone()), w_id, @@ -3785,7 +3819,7 @@ pub async fn start_worker( } if annotation.nodejs { - let script_path = format!("{job_dir}/wrapper.mjs"); + let wrapper_path = format!("{job_dir}/wrapper.mjs"); handle_dedicated_process( &*NODE_BIN_PATH, @@ -3794,16 +3828,17 @@ pub async fn start_worker( envs, context, common_bun_proc_envs, - vec![&script_path], + vec![&wrapper_path], killpill_rx, job_completed_tx, token, jobs_rx, worker_name, db, - &script_path, + script_path, "nodejs", client, + false, ) .await } else { @@ -3831,6 +3866,7 @@ pub async fn start_worker( script_path, "bun", client, + false, ) .await } @@ -3986,4 +4022,33 @@ export function preprocessor(input: string, when: Date) { return { x: input, ts: assert!(cg.date_conversions.is_empty()); assert!(cg.preprocessor_spread.is_none()); } + + #[test] + fn test_wrapper_contains_execd_protocol() { + let code = r#"export function main(x: number) { return x; }"#; + let cg = compute_ts_codegen(code); + let scripts = + vec![TsScriptEntry { import_name: "main", original_path: "test/script", codegen: &cg }]; + let wrapper = generate_multi_script_wrapper(&scripts, "ts"); + // Single-script wrapper must support execd: (direct, no path) + assert!(wrapper.contains(r#"line.startsWith("execd:")"#)); + // Must also support exec: for backward compat / runner groups + assert!(wrapper.contains(r#"line.startsWith("exec:")"#)); + // Must register the script in the map + assert!(wrapper.contains(r#"scripts.set("test/script""#)); + } + + #[test] + fn test_wrapper_contains_execd_preprocess_protocol() { + let code = r#"export function preprocessor(x: number) { return { x }; } +export function main(x: number) { return x; }"#; + let cg = compute_ts_codegen(code); + let scripts = + vec![TsScriptEntry { import_name: "main", original_path: "test/script", codegen: &cg }]; + let wrapper = generate_multi_script_wrapper(&scripts, "ts"); + assert!(wrapper.contains(r#"line.startsWith("execd_preprocess:")"#)); + assert!(wrapper.contains(r#"line.startsWith("execd:")"#)); + assert!(wrapper.contains(r#"line.startsWith("exec_preprocess:")"#)); + assert!(wrapper.contains(r#"line.startsWith("exec:")"#)); + } } diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 3013d10689..717a1fb931 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -590,7 +590,7 @@ pub(crate) async fn build_import_map( 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. +/// and preprocessor logic. Uses the `execd:` protocol (single-script, no path needed). #[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; @@ -627,14 +627,8 @@ pub fn generate_dedicated_worker_wrapper(inner_content: &str) -> Result 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); + if (line.startsWith("execd_preprocess:")) {{ + const argsJson = line.slice("execd_preprocess:".length); const parsedArgs = JSON.parse(argsJson); if (typeof preprocessor !== 'function') {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }}) + '\n'); @@ -686,14 +680,8 @@ for await (const chunk of Deno.stdin.readable) {{ 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); + if (line.startsWith("execd:")) {{ + const argsJson = line.slice("execd:".length); try {{ let {{ {spread} }} = JSON.parse(argsJson) {dates} @@ -802,6 +790,7 @@ pub async fn start_worker( script_path, "deno", client, + false, ) .await } diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 78d3d54e15..3c6ea7a474 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -1213,8 +1213,8 @@ pub struct PyScriptEntry<'a> { /// 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]: +/// execd: -> execute the single registered script (non-runner-group) +/// exec:: -> execute script by path (runner groups) /// end -> exit #[cfg(any(feature = "private", test))] pub fn generate_multi_script_wrapper( @@ -1370,6 +1370,24 @@ for line in sys.stdin: sys.stdout.flush() continue + if line.startswith('execd:'): + try: + args_json = line[len('execd:'):] + entry = next(iter(scripts.values())) + 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 + if line.startswith('exec:'): try: rest = line[len('exec:'):] @@ -2875,6 +2893,7 @@ pub async fn start_worker( script_path, "python", client, + false, ) .await } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index e40170dcd3..6b1e401b66 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1994,9 +1994,8 @@ pub async fn run_worker( // Option>, #[cfg(all(feature = "private", feature = "enterprise"))] - let (dedicated_workers, dedicated_flow_paths, dedicated_handles): ( + let (dedicated_workers, dedicated_handles): ( HashMap>, - HashSet, Vec>, ) = match conn { Connection::Sql(pool) => { @@ -2011,15 +2010,14 @@ pub async fn run_worker( ) .await } - Connection::Http(_) => (HashMap::new(), HashSet::new(), vec![]), + Connection::Http(_) => (HashMap::new(), vec![]), }; #[cfg(any(not(feature = "private"), not(feature = "enterprise")))] - let (dedicated_workers, dedicated_flow_paths, dedicated_handles): ( + let (dedicated_workers, dedicated_handles): ( HashMap>, - HashSet, Vec>, - ) = (HashMap::new(), HashSet::new(), vec![]); + ) = (HashMap::new(), vec![]); if i_worker == 1 { // Initialize runtime asset inserter for batched database inserts @@ -2480,17 +2478,10 @@ pub async fn run_worker( JobKind::Script | JobKind::Preview | JobKind::FlowScript ) { if !dedicated_workers.is_empty() { - // Try flow path + step_id combinations for flow jobs, otherwise use runnable_path - let dedicated_worker_tx = if let Some(step_id) = job.flow_step_id.as_ref() { - dedicated_flow_paths.iter().find_map(|flow_path| { - let key = format!("{}:{}", flow_path, step_id); - dedicated_workers.get(&key) - }) - } else { - job.runnable_path - .as_ref() - .and_then(|path| dedicated_workers.get(path)) - }; + let dedicated_worker_tx = job.runnable_path.as_ref().and_then(|path| { + let key = format!("{}:{}", job.workspace_id, path); + dedicated_workers.get(&key) + }); if let Some(dedicated_worker_tx) = dedicated_worker_tx { let dedicated_job = DedicatedWorkerJob { job: Arc::new(job.job()), @@ -5455,4 +5446,3 @@ pub fn get_worker_internal_server_inline_utils( )), } } - diff --git a/frontend/src/lib/components/DedicatedWorkersSelector.svelte b/frontend/src/lib/components/DedicatedWorkersSelector.svelte index 94086982fb..a54d856189 100644 --- a/frontend/src/lib/components/DedicatedWorkersSelector.svelte +++ b/frontend/src/lib/components/DedicatedWorkersSelector.svelte @@ -4,6 +4,7 @@ FlowService, WorkspaceService, WorkspaceDependenciesService, + ConfigService, type FlowModule } from '$lib/gen' import { @@ -66,8 +67,8 @@ let workspaces: { id: string; name: string }[] = $state([]) let workspacesLoading = $state(true) let selectorExpanded = $state(false) - // Set of existing workspace dependency names for the current workspace (for validation) - let existingDeps: Set = $state(new Set()) + // Map of dep name → set of workspaces that have it (for cross-workspace validation) + let existingDeps: Map> = $state(new Map()) // Track detailed info for each selected tag (for displaying in summary) interface SelectedTagInfo { @@ -129,54 +130,98 @@ const newInfo = new SvelteMap() - // Collect workspaces that need dep info fetched - const workspacesNeedingDeps = new Set() - for (const tag of tags) { - const existing = currentInfo.get(tag) - const existingRunnable = currentRunnables.find((r) => r.tag === tag) - // If we don't have workspaceDeps cached, we need to fetch for this workspace - if (!existing?.workspaceDeps && !existingRunnable?.workspaceDeps) { - const parsed = parseTag(tag) - if (parsed?.type === 'script') { - workspacesNeedingDeps.add(parsed.workspace) + // Check if any tags need dep info fetched + let needsFetch = currentExistingDeps.size === 0 + if (!needsFetch) { + for (const tag of tags) { + const existing = currentInfo.get(tag) + const existingRunnable = currentRunnables.find((r) => r.tag === tag) + if (!existing?.workspaceDeps && !existingRunnable?.workspaceDeps) { + const parsed = parseTag(tag) + if (parsed?.type === 'script') { + needsFetch = true + break + } } } } - // Fetch workspace dep info for workspaces that need it + // Fetch dep info across all workspaces in one call const depsPerWorkspace = new Map>() - if (workspacesNeedingDeps.size > 0 || currentExistingDeps.size === 0) { - await Promise.all( - Array.from(workspacesNeedingDeps).map(async (ws) => { - try { - const [dedicatedDeps, wsDeps] = await Promise.all([ - ScriptService.listDedicatedWithDeps({ workspace: ws }).catch(() => []), - WorkspaceDependenciesService.listWorkspaceDependencies({ - workspace: ws - }).catch(() => []) - ]) - const depsMap = new Map() - for (const d of dedicatedDeps) { - if (d.workspace_dep_names.length > 0) { - depsMap.set(d.path, { - deps: d.workspace_dep_names, - language: d.language - }) - } + if (needsFetch) { + try { + const [allDedicatedDeps, allWsDeps] = await Promise.all([ + ConfigService.listAllDedicatedWithDeps().catch(() => []), + ConfigService.listAllWorkspaceDependencies().catch(() => []) + ]) + for (const d of allDedicatedDeps) { + if (d.workspace_dep_names.length > 0) { + let wsMap = depsPerWorkspace.get(d.workspace_id) + if (!wsMap) { + wsMap = new Map() + depsPerWorkspace.set(d.workspace_id, wsMap) } - depsPerWorkspace.set(ws, depsMap) - // Merge into existingDeps - for (const d of wsDeps) { - if (!d.archived && d.name) { - currentExistingDeps.add(d.name) - } - } - } catch { - // ignore + wsMap.set(d.path, { + deps: d.workspace_dep_names, + language: d.language + }) } - }) - ) - existingDeps = new Set(currentExistingDeps) + } + for (const d of allWsDeps) { + if (d.name) { + let wsSet = currentExistingDeps.get(d.name) + if (!wsSet) { + wsSet = new Set() + currentExistingDeps.set(d.name, wsSet) + } + wsSet.add(d.workspace_id) + } + } + } catch { + // Fallback to per-workspace calls if cross-workspace endpoint unavailable + const workspacesNeedingDeps = new Set() + for (const tag of tags) { + const parsed = parseTag(tag) + if (parsed?.type === 'script') { + workspacesNeedingDeps.add(parsed.workspace) + } + } + await Promise.all( + Array.from(workspacesNeedingDeps).map(async (ws) => { + try { + const [dedicatedDeps, wsDeps] = await Promise.all([ + ScriptService.listDedicatedWithDeps({ workspace: ws }).catch(() => []), + WorkspaceDependenciesService.listWorkspaceDependencies({ + workspace: ws + }).catch(() => []) + ]) + const depsMap = new Map() + for (const d of dedicatedDeps) { + if (d.workspace_dep_names.length > 0) { + depsMap.set(d.path, { + deps: d.workspace_dep_names, + language: d.language + }) + } + } + depsPerWorkspace.set(ws, depsMap) + for (const d of wsDeps) { + if (!d.archived && d.name) { + let wsSet = currentExistingDeps.get(d.name) + if (!wsSet) { + wsSet = new Set() + currentExistingDeps.set(d.name, wsSet) + } + wsSet.add(ws) + } + } + } catch { + // ignore + } + }) + ) + } + existingDeps = new Map(currentExistingDeps) } try { @@ -394,7 +439,7 @@ loading = true runnables = [] - const [scripts, flows, dedicatedDeps, wsDeps] = await Promise.all([ + const [scripts, flows, dedicatedDeps, allWsDeps] = await Promise.all([ ScriptService.listScripts({ workspace: workspaceId, dedicatedWorker: true @@ -406,18 +451,28 @@ ScriptService.listDedicatedWithDeps({ workspace: workspaceId }).catch(() => []), - WorkspaceDependenciesService.listWorkspaceDependencies({ - workspace: workspaceId - }).catch(() => []) + // Load workspace deps across all workspaces for accurate validation + ConfigService.listAllWorkspaceDependencies().catch(() => + // Fallback to current workspace if cross-workspace endpoint unavailable + WorkspaceDependenciesService.listWorkspaceDependencies({ + workspace: workspaceId + }).catch(() => []) + ) ]) - // Track existing workspace dep names for validation - existingDeps = new Set( - wsDeps - .filter((d) => !d.archived) - .map((d) => d.name) - .filter((n): n is string => !!n) - ) + // Track existing workspace dep names with their workspaces for validation + const newDeps = new Map>() + for (const d of allWsDeps) { + if (d.name) { + let wsSet = newDeps.get(d.name) + if (!wsSet) { + wsSet = new Set() + newDeps.set(d.name, wsSet) + } + wsSet.add(d.workspace_id) + } + } + existingDeps = newDeps // Build a map from path -> workspace dep names const depsMap = new Map() @@ -497,7 +552,11 @@ } function updateSelectedTags() { - selectedTags = runnables.filter((r) => r.selected).map((r) => r.tag) + // Keep tags from other workspaces, update only the currently visible ones + const visibleTags = new Set(runnables.map((r) => r.tag)) + const otherTags = selectedTags.filter((t) => !visibleTags.has(t)) + const newVisibleTags = runnables.filter((r) => r.selected).map((r) => r.tag) + selectedTags = [...otherTags, ...newVisibleTags] onchange?.(selectedTags) } @@ -527,13 +586,14 @@ } let runnerGroups: RunnerGroup[] = $derived.by(() => { + // Group by (workspace, dep_name, language) — runner groups are per-workspace const groupMap = new Map() for (const tag of selectedTags) { const info = selectedTagsInfo.get(tag) if (info?.type === 'script' && info.workspaceDeps) { const lang = info.language ?? runnables.find((r) => r.tag === tag)?.language ?? 'unknown' for (const dep of info.workspaceDeps) { - const key = `${dep}:${lang}` + const key = `${info.workspace}:${dep}:${lang}` const existing = groupMap.get(key) if (existing) { existing.tags.push(tag) @@ -562,16 +622,28 @@ let standaloneTags: string[] = $derived(selectedTags.filter((tag) => !tagRunnerGroup.has(tag))) -{#snippet depBadge(dep: string)} - {#if existingDeps.has(dep)} - +{#snippet depBadge(dep: string, workspace: string | undefined)} + {@const depWorkspaces = existingDeps.get(dep)} + {@const existsInWorkspace = depWorkspaces?.has(workspace ?? '')} + {@const existsElsewhere = depWorkspaces && depWorkspaces.size > 0 && !existsInWorkspace} + {#if existsInWorkspace} + {dep} + {:else if existsElsewhere} + + Workspace dependency '{dep}' exists in {[...(depWorkspaces ?? [])].join(', ')} but not in '{workspace}'. + Each workspace needs its own dependency. + + + + {dep} (not in {workspace}) + {:else} - Workspace dependency '{dep}' not found. Create it in workspace settings to enable shared - runners. + Workspace dependency '{dep}' not found in any workspace. Create it in workspace settings to + enable shared runners. @@ -613,7 +685,7 @@ {#if !tagRunnerGroup.has(tag)} {#if info.workspaceDeps} {#each info.workspaceDeps as dep} - {@render depBadge(dep)} + {@render depBadge(dep, info?.workspace)} {/each} {/if} {#if info.type === 'flow' && info.runners} @@ -670,7 +742,7 @@ Shared runner - {@render depBadge(group.depName)} + {@render depBadge(group.depName, selectedTagsInfo.get(group.tags[0])?.workspace)} {group.language}
@@ -696,7 +768,7 @@ {#if info?.workspaceDeps} {#each info.workspaceDeps as dep} - {@render depBadge(dep)} + {@render depBadge(dep, info?.workspace)} {/each} {/if} {#if info?.type === 'flow' && info.runners} @@ -832,7 +904,7 @@ {/if} {#if runnable.workspaceDeps} {#each runnable.workspaceDeps as dep} - {@render depBadge(dep)} + {@render depBadge(dep, selectedWorkspace)} {/each} {/if} diff --git a/frontend/src/lib/components/flows/content/FlowLoop.svelte b/frontend/src/lib/components/flows/content/FlowLoop.svelte index 886f57386f..10b1bee573 100644 --- a/frontend/src/lib/components/flows/content/FlowLoop.svelte +++ b/frontend/src/lib/components/flows/content/FlowLoop.svelte @@ -32,7 +32,6 @@ import { slide } from 'svelte/transition' import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' - import Badge from '$lib/components/common/badge/Badge.svelte' const { previewArgs, flowStateStore, flowStore, currentEditor } = getContext('FlowEditorContext') @@ -191,17 +190,11 @@
Squash - - Beta - This can result in unexpected behavior, use at your own risk for now.
- Squashing a for loop runs all iterations on the same worker, using a single runner - per step for the entire loop. This eliminates cold starts between iterations for supported - languages (Bun, Deno, and Python). - - + + Squashing a for loop runs all iterations on the same worker, using a single runner + per step for the entire loop. This eliminates cold starts between iterations for + supported languages (Bun, Deno, and Python). +
('FlowEditorContext') @@ -107,17 +106,11 @@
Squash - - Beta - This can result in unexpected behavior, use at your own risk for now.
- Squashing a for loop runs all iterations on the same worker, using a single runner - per step for the entire loop. This eliminates cold starts between iterations for supported - languages (Bun, Deno, and Python). - - + + Squashing a while loop runs all iterations on the same worker, using a single runner + per step for the entire loop. This eliminates cold starts between iterations for + supported languages (Bun, Deno, and Python). +