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-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 36ddb8ab9f..713ccb9dd3 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - true + null ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" 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 4b6444b413..571b689977 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 9ba2542cc7..0ff64f2b50 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -b30d25cf5f7ca7f94509fe9a732e22ab2186f929 \ No newline at end of file +9422b189762ae27edfc346541ae668a4ad728325 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..569308d97a --- /dev/null +++ b/backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql @@ -0,0 +1,6 @@ +-- 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 = regexp_replace(content, E'^//nodejs\n// dedicated workers were previously running in nodejs mode by default, remove this annotation to use bun\n', '') +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..5dec8b8d77 --- /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..c5f5ac7617 --- /dev/null +++ b/backend/tests/dedicated_workers.rs @@ -0,0 +1,1513 @@ +#[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 any descendant job) was executed + /// by a dedicated worker process with the expected mode by checking job logs + /// for the "dedicated worker {mode}:" prefix. + /// Recurses through all descendants (children, grandchildren, etc.) to handle + /// nested flow structures like branches and loops. + async fn assert_ran_on_dedicated_worker( + db: &Pool, + job_id: uuid::Uuid, + expected_mode: &str, + ) { + let marker = format!("dedicated worker {expected_mode}:"); + + // BFS through job tree to find any descendant with the marker + let mut queue = vec![job_id]; + let mut all_logs = Vec::new(); + + while let Some(id) = queue.pop() { + let logs: Option = + sqlx::query_scalar("SELECT logs FROM job_logs WHERE job_id = $1") + .bind(id) + .fetch_optional(db) + .await + .unwrap() + .flatten(); + + if let Some(ref l) = logs { + if l.contains(&marker) { + return; + } + } + all_logs.push((id, logs)); + + let child_ids: Vec = + sqlx::query_scalar("SELECT id FROM v2_job WHERE parent_job = $1") + .bind(id) + .fetch_all(db) + .await + .unwrap(); + queue.extend(child_ids); + } + + panic!( + "No job in tree had '{marker}' in logs. All logs: {:?}", + all_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 a flow's Script step referencing a standalone dedicated script + /// correctly routes through the standalone worker (first-registered-wins collision). + /// Both the flow and the script are configured as dedicated workers. + /// The standalone dedicated_double worker handles the job from both contexts. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_script_standalone_conflict( + db: Pool, + ) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // Run as standalone script first + let uuid_standalone = 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; + + // Run via flow that references the same script + let uuid_flow = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_conflict_standalone_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000011, + }) + .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_standalone).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_flow).await; + }, + port, + vec![ + wp("test-workspace", "f/system/dedicated_double"), + wp( + "test-workspace", + "flow/f/system/dedicated_conflict_standalone_flow", + ), + ], + ) + .await; + + let job_standalone = completed_job(uuid_standalone, &db).await; + let job_flow = completed_job(uuid_flow, &db).await; + // Both should compute x * 2 = 14 + assert_eq!(job_standalone.json_result().unwrap(), json!(14)); + assert_eq!(job_flow.json_result().unwrap(), json!(14)); + assert_ran_on_dedicated_worker(&db, uuid_standalone, "bun").await; + assert_ran_on_dedicated_worker(&db, uuid_flow, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with mixed languages: bun step (dedicated) + bash step (normal). + /// The bash step should fall back to normal execution while the bun step uses dedicated worker. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_mixed_lang_fallback(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_mixed_lang_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000012, + }) + .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_mixed_lang_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + // bun: 5 + 10 = 15, bash: echo done = "done" + assert_eq!(job.json_result().unwrap(), json!("done")); + // The bun step ran on dedicated worker + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a squashed for-loop with a bash step (unsupported for flow runners). + /// The bash step can't spawn a flow runner, so each iteration falls back to normal execution. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_runners_unsupported_lang( + 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_bash".to_string(), + dedicated_worker: None, + apply_preprocessor: false, + version: 3000000000000013, + }) + .run_until_complete(&db, false, port) + .await + }, + port, + ) + .await; + + // for-loop over [1, 2, 3], bash echo done = ["done", "done", "done"] + assert_eq!(job.json_result().unwrap(), json!(["done", "done", "done"])); + Ok(()) + } + + /// Test Python runner group: two Python scripts sharing a workspace dep annotation. + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_python_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(300050), + path: "f/system/py_rg_script_a".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Python3, + 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(300051), + path: "f/system/py_rg_script_b".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Python3, + 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 { + 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/py_rg_script_a"), + wp("test-workspace", "f/system/py_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, "python3").await; + assert_ran_on_dedicated_worker(&db, uuid_b, "python3").await; + Ok(()) + } + + /// Test preprocessor in a runner group. Script with preprocessor shares a workspace dep + /// with another script, forming a runner group. Tests exec_preprocess:{path}:{args} protocol. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_runner_group_preprocessor(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // Run the preprocessor script with apply_preprocessor=true + let uuid_pre = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300052), + path: "f/system/rg_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; + + // Also run the other script in the group (no preprocessor) + let uuid_other = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300053), + path: "f/system/rg_preprocess_other".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 { + listener.find(&uuid_pre).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_other).await; + }, + port, + vec![ + wp("test-workspace", "f/system/rg_preprocess_script"), + wp("test-workspace", "f/system/rg_preprocess_other"), + ], + ) + .await; + + let job_pre = completed_job(uuid_pre, &db).await; + let job_other = completed_job(uuid_other, &db).await; + // preprocessor(5) → {x: 10}, main(10) → 110 + assert_eq!(job_pre.json_result().unwrap(), json!(110)); + // main(5) → 305 + assert_eq!(job_other.json_result().unwrap(), json!(305)); + assert_ran_on_dedicated_worker(&db, uuid_pre, "bun").await; + assert_ran_on_dedicated_worker(&db, uuid_other, "bun").await; + + // Verify preprocessed flag was set on the preprocessor job + let preprocessed: Option = + sqlx::query_scalar("SELECT preprocessed FROM v2_job WHERE id = $1") + .bind(uuid_pre) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(preprocessed, Some(true)); + + Ok(()) + } + + /// Test a dedicated flow with a BranchOne module. + /// When x > 10, branch path (x + 100) is taken; otherwise default (x + 200). + /// Tests that dedicated workers are spawned for modules inside branches. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_branchone(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // Test with x=20 (branch taken: x + 100 = 120) + let uuid_branch = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_branch_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000014, + }) + .arg("x", json!(20)) + .push(&db) + .await; + + // Test with x=5 (default taken: x + 200 = 205) + let uuid_default = RunJob::from(JobPayload::Flow { + path: "f/system/dedicated_branch_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000014, + }) + .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_branch).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_default).await; + }, + port, + vec![wp("test-workspace", "flow/f/system/dedicated_branch_flow")], + ) + .await; + + let job_branch = completed_job(uuid_branch, &db).await; + let job_default = completed_job(uuid_default, &db).await; + assert_eq!(job_branch.json_result().unwrap(), json!(120)); + assert_eq!(job_default.json_result().unwrap(), json!(205)); + assert_ran_on_dedicated_worker(&db, uuid_branch, "bun").await; + assert_ran_on_dedicated_worker(&db, uuid_default, "bun").await; + Ok(()) + } + + /// Test Python standalone preprocessor (execd_preprocess: protocol). + /// preprocessor doubles x, main adds 100: preprocessor(5) → {x:10}, main(10) → 110. + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_python_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(300054), + path: "f/system/py_preprocess_script".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Python3, + 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/py_preprocess_script")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(110)); + assert_ran_on_dedicated_worker(&db, uuid, "python").await; + + 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 Deno standalone preprocessor (execd_preprocess: protocol). + /// preprocessor doubles x, main adds 100: preprocessor(5) → {x:10}, main(10) → 110. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_deno_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(300055), + path: "f/system/deno_preprocess_script".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Deno, + 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/deno_preprocess_script")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(110)); + assert_ran_on_dedicated_worker(&db, uuid, "deno").await; + + 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 Bunnative standalone preprocessor (V8 isolate path). + /// preprocessor doubles x, main adds 100: preprocessor(5) → {x:10}, main(10) → 110. + #[cfg(feature = "deno_core")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_bunnative_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(300058), + path: "f/system/bunnative_preprocess_script".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Bunnative, + 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/bunnative_preprocess_script")], + ) + .await; + + let job = completed_job(uuid, &db).await; + assert_eq!(job.json_result().unwrap(), json!(110)); + assert_ran_on_dedicated_worker(&db, uuid, "nativets").await; + + 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 Python runner group preprocessor (exec_preprocess:{path}:{args} protocol). + /// Two Python scripts sharing a workspace dep, one has a preprocessor. + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_python_runner_group_preprocessor( + db: Pool, + ) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let uuid_pre = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300056), + path: "f/system/py_rg_preprocess_a".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Python3, + 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 uuid_other = RunJob::from(JobPayload::ScriptHash { + hash: windmill_common::scripts::ScriptHash(300057), + path: "f/system/py_rg_preprocess_b".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Python3, + 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 { + listener.find(&uuid_pre).await; + let listener2 = listen_for_completed_jobs(&db).await; + listener2.find(&uuid_other).await; + }, + port, + vec![ + wp("test-workspace", "f/system/py_rg_preprocess_a"), + wp("test-workspace", "f/system/py_rg_preprocess_b"), + ], + ) + .await; + + let job_pre = completed_job(uuid_pre, &db).await; + let job_other = completed_job(uuid_other, &db).await; + // preprocessor(5) → {x: 10}, main(10) → 110 + assert_eq!(job_pre.json_result().unwrap(), json!(110)); + // main(5) → 305 + assert_eq!(job_other.json_result().unwrap(), json!(305)); + assert_ran_on_dedicated_worker(&db, uuid_pre, "python3").await; + assert_ran_on_dedicated_worker(&db, uuid_other, "python3").await; + + let preprocessed: Option = + sqlx::query_scalar("SELECT preprocessed FROM v2_job WHERE id = $1") + .bind(uuid_pre) + .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(()) + } + + /// Test a dedicated worker script that uses a relative import to another workspace script. + /// Verifies that build_loader's CURRENT_PATH resolves imports correctly. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_bun_relative_import(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(300061), + path: "f/system/dedicated_with_import".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(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "f/system/dedicated_with_import")], + ) + .await; + + let job = completed_job(uuid, &db).await; + // helper(5) = 50, main = 50 + 1 = 51 + assert_eq!(job.json_result().unwrap(), json!(51)); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated Python worker script that uses a relative import. + /// Verifies that the Python loader resolves relative imports correctly. + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_python_relative_import(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(300063), + path: "f/system/py_dedicated_with_import".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: windmill_common::scripts::ScriptLang::Python3, + 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(), + listener.find(&uuid), + port, + vec![wp("test-workspace", "f/system/py_dedicated_with_import")], + ) + .await; + + let job = completed_job(uuid, &db).await; + // helper(5) = 50, main = 50 + 1 = 51 + assert_eq!(job.json_result().unwrap(), json!(51)); + assert_ran_on_dedicated_worker(&db, uuid, "python").await; + Ok(()) + } + + /// Test a dedicated flow with a simple non-squashed for-loop (single step). + /// is_simple_modules optimization inlines the step into the sub-flow job. + /// The dedicated worker map uses node_id (parent loop id) as key. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_forloop_simple(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_forloop_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000015, + }) + .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_forloop_flow")], + ) + .await; + + let job = completed_job(uuid, &db).await; + // for-loop over [1, 2, 3], each * 10 = [10, 20, 30] + assert_eq!(job.json_result().unwrap(), json!([10, 20, 30])); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with a non-simple for-loop (multi-step). + /// Two inner steps make is_simple_modules false, so each step runs as a + /// separate job with runnable_path including /forloop-N/ nesting segments. + /// Tests the extract_flow_root dispatch path. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_forloop_multi_step(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_forloop_multi_step_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000019, + }) + .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_forloop_multi_step_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + // for-loop over [1, 2, 3]: step b = value+1, step c = b*10 + // iter 1: 1+1=2, 2*10=20; iter 2: 2+1=3, 3*10=30; iter 3: 3+1=4, 4*10=40 + assert_eq!(job.json_result().unwrap(), json!([20, 30, 40])); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with a while-loop. + /// Inner step (x + 1) runs until result >= 3, with early stop. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_whileloop(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_whileloop_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000016, + }) + .arg("x", json!(1)) + .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_whileloop_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + // x=1: n=1→2 (continue), n=2→3 (stop). While-loop returns array of results. + assert_eq!(job.json_result().unwrap(), json!([2, 3])); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with BranchAll (parallel branches). + /// Two branches: x+100 and x+200, both should dispatch to dedicated workers. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_branchall(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_branchall_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000017, + }) + .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_branchall_flow", + )], + ) + .await; + + let job = completed_job(uuid, &db).await; + // BranchAll returns array of branch results: [105, 205] + assert_eq!(job.json_result().unwrap(), json!([105, 205])); + assert_ran_on_dedicated_worker(&db, uuid, "bun").await; + Ok(()) + } + + /// Test a dedicated flow with deeply nested structure: for-loop containing a branchone. + /// Tests dispatch through forloop-N/branchone-0/step_id path segments. + #[sqlx::test(fixtures("base", "dedicated_flows"))] + #[serial] + async fn test_dedicated_flow_nested_branch_in_loop(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_nested_flow".to_string(), + dedicated_worker: Some(true), + apply_preprocessor: false, + version: 3000000000000018, + }) + .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_nested_flow")], + ) + .await; + + let job = completed_job(uuid, &db).await; + // for-loop [1, 2]: + // iter 1 (value=1): branchone condition 1>1 false → default: 1+200=201 + // iter 2 (value=2): branchone condition 2>1 true → branch: 2+100=102 + assert_eq!(job.json_result().unwrap(), json!([201, 102])); + assert_ran_on_dedicated_worker(&db, uuid, "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..b3b63685df --- /dev/null +++ b/backend/tests/fixtures/dedicated_flows.sql @@ -0,0 +1,599 @@ +-- 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); + +-- ============================================================ +-- Test: flow Script step + standalone conflict +-- Flow references dedicated_double (also configured as standalone). +-- Standalone worker should handle both its own jobs and the flow's Script step jobs. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_conflict_standalone_flow', +'{3000000000000011}', +'{"$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 ( +3000000000000011, +'test-workspace', +'f/system/dedicated_conflict_standalone_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' +); + +-- ============================================================ +-- Test: non-dedicated step inside dedicated flow +-- Flow with bun step (x + 10) then bash step (echo result * 2). +-- The bun step runs on dedicated worker; bash falls back to normal execution. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_mixed_lang_flow', +'{3000000000000012}', +'{"$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"}},{"id":"b","value":{"type":"rawscript","content":"echo done","language":"bash","input_transforms":{"results_a":{"expr":"results.a","type":"javascript"}}}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000012, +'test-workspace', +'f/system/dedicated_mixed_lang_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"}},{"id":"b","value":{"type":"rawscript","content":"echo done","language":"bash","input_transforms":{"results_a":{"expr":"results.a","type":"javascript"}}}}]}', +'system' +); + +-- ============================================================ +-- Test: non-dedicated step inside flow runners (squashed loop) +-- Squashed for-loop with a bash step. Bash can't spawn a flow runner, +-- so each iteration falls back to normal execution. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_flow_runners_bash', +'{3000000000000013}', +'{"$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":"echo done","language":"bash","input_transforms":{"iter_value":{"expr":"flow_input.iter.value","type":"javascript"}}}}]}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000013, +'test-workspace', +'f/system/dedicated_flow_runners_bash', +'{"$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":"echo done","language":"bash","input_transforms":{"iter_value":{"expr":"flow_input.iter.value","type":"javascript"}}}}]}}]}', +'system' +); + +-- ============================================================ +-- Test: Python runner group +-- Two Python scripts sharing a workspace dependency 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_requirements: f/system/dedicated_double\ndef main(x: int):\n return x + 100', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/py_rg_script_a', 300050, 'python3', '', 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_requirements: f/system/dedicated_double\ndef main(x: int):\n return x + 200', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/py_rg_script_b', 300051, 'python3', '', true); + +-- ============================================================ +-- Test: preprocessor in runner group +-- Two bun scripts sharing a workspace dep, one has a preprocessor. +-- Tests exec_preprocess:{path}:{args} protocol in runner groups. +-- ============================================================ +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 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/rg_preprocess_script', 300052, '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 + 300; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/rg_preprocess_other', 300053, 'bun', E'{}\n//bun.lock\n', true); + +-- ============================================================ +-- Test: flow with BranchOne +-- Flow with branchone: condition true → branch (x + 100), default → (x + 200). +-- Tests recursive module traversal for spawning dedicated workers. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_branch_flow', +'{3000000000000014}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"branchone","branches":[{"expr":"flow_input.x > 10","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(x: number) { return x + 100; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}],"default":[{"id":"c","value":{"type":"rawscript","content":"export function main(x: number) { return x + 200; }","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 ( +3000000000000014, +'test-workspace', +'f/system/dedicated_branch_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":"branchone","branches":[{"expr":"flow_input.x > 10","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(x: number) { return x + 100; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}],"default":[{"id":"c","value":{"type":"rawscript","content":"export function main(x: number) { return x + 200; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}}]}', +'system' +); + +-- ============================================================ +-- Test: Python standalone preprocessor +-- Python script with preprocessor function. Tests execd_preprocess: protocol. +-- ============================================================ +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'def preprocessor(x: int):\n return {"x": x * 2}\n\ndef main(x: int):\n return x + 100', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/py_preprocess_script', 300054, 'python3', '', true); + +-- ============================================================ +-- Test: Deno standalone preprocessor +-- Deno script with preprocessor function. Tests execd_preprocess: protocol. +-- ============================================================ +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/deno_preprocess_script', 300055, 'deno', '', true); + +-- ============================================================ +-- Test: Python runner group preprocessor +-- Two Python scripts sharing a workspace dep, one has a preprocessor. +-- Tests exec_preprocess:{path}:{args} protocol for Python runner groups. +-- ============================================================ +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'# extra_requirements: f/system/py_preprocess_script\ndef preprocessor(x: int):\n return {"x": x * 2}\n\ndef main(x: int):\n return x + 100', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/py_rg_preprocess_a', 300056, 'python3', '', 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_requirements: f/system/py_preprocess_script\ndef main(x: int):\n return x + 300', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/py_rg_preprocess_b', 300057, 'python3', '', true); + +-- ============================================================ +-- Test: Bunnative standalone preprocessor +-- Bunnative script with preprocessor function. Tests V8 isolate preprocessing. +-- ============================================================ +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'//native\nexport 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/bunnative_preprocess_script', 300058, 'bunnative', E'{}\n//bun.lock\n', true); + +-- ============================================================ +-- Test: Bun dedicated worker with relative imports +-- Helper script at f/system/dedicated_helper (returns x * 10). +-- Dedicated script imports from it using a relative path. +-- ============================================================ +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +E'export function helper(x: number) { return x * 10; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', '', +'f/system/dedicated_helper', 300060, 'bun', E'{}\n//bun.lock\n'); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'import { helper } from "./dedicated_helper.ts";\n\nexport function main(x: number) { return helper(x) + 1; }', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/dedicated_with_import', 300061, 'bun', E'{}\n//bun.lock\n', true); + +-- ============================================================ +-- Test: Python dedicated worker with relative imports +-- Helper script at f/system/py_dedicated_helper (returns x * 10). +-- Dedicated script imports from it using a relative path. +-- ============================================================ +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +E'def helper(x: int):\n return x * 10', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', '', +'f/system/py_dedicated_helper', 300062, 'python3', ''); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, dedicated_worker) VALUES ( +'test-workspace', +'system', +E'from .py_dedicated_helper import helper\n\ndef main(x: int):\n return helper(x) + 1', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +'', '', +'f/system/py_dedicated_with_import', 300063, 'python3', '', true); + +-- ============================================================ +-- Test: non-squashed for-loop in dedicated flow +-- For-loop iterates over [1, 2, 3], inner bun step returns iter.value * 10. +-- NOT squashed, so inner steps use the dedicated worker map (not flow runners). +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_forloop_flow', +'{3000000000000015}', +'{"$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":false,"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 ( +3000000000000015, +'test-workspace', +'f/system/dedicated_forloop_flow', +'{"$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":false,"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' +); + +-- ============================================================ +-- Test: non-simple for-loop in dedicated flow (multi-step) +-- For-loop iterates over [1, 2, 3], two steps: step b (iter.value + 1), step c (results.b * 10). +-- Two steps makes is_simple_modules false, so inner steps run as separate jobs. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_forloop_multi_step_flow', +'{3000000000000019}', +'{"$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":false,"modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(iter: {value: number, index: number}) { return iter.value + 1; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}},{"id":"c","value":{"type":"rawscript","content":"export function main(prev: number) { return prev * 10; }","language":"bun","input_transforms":{"prev":{"expr":"results.b","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000019, +'test-workspace', +'f/system/dedicated_forloop_multi_step_flow', +'{"$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":false,"modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(iter: {value: number, index: number}) { return iter.value + 1; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}},{"id":"c","value":{"type":"rawscript","content":"export function main(prev: number) { return prev * 10; }","language":"bun","input_transforms":{"prev":{"expr":"results.b","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}}]}', +'system' +); + +-- ============================================================ +-- Test: while-loop in dedicated flow +-- While-loop with an inner bun step that returns x + 1. +-- Early stop breaks when result >= 3. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_whileloop_flow', +'{3000000000000016}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"whileloopflow","skip_failures":false,"squash":false,"modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(n: number) { return n + 1; }","language":"bun","input_transforms":{"n":{"expr":"results.b ?? flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"},"stop_after_if":{"expr":"result >= 3","skip_if_stopped":false}}]}}]}', +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +3000000000000016, +'test-workspace', +'f/system/dedicated_whileloop_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":"whileloopflow","skip_failures":false,"squash":false,"modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(n: number) { return n + 1; }","language":"bun","input_transforms":{"n":{"expr":"results.b ?? flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"},"stop_after_if":{"expr":"result >= 3","skip_if_stopped":false}}]}}]}', +'system' +); + +-- ============================================================ +-- Test: branchall in dedicated flow +-- BranchAll with 2 parallel branches, each with a bun step. +-- Branch 0: x + 100, Branch 1: x + 200. +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_branchall_flow', +'{3000000000000017}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"x":{"type":"number","description":""}},"required":["x"],"type":"object"}', +E'{"modules":[{"id":"a","value":{"type":"branchall","parallel":true,"branches":[{"expr":"true","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(x: number) { return x + 100; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]},{"expr":"true","modules":[{"id":"c","value":{"type":"rawscript","content":"export function main(x: number) { return x + 200; }","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 ( +3000000000000017, +'test-workspace', +'f/system/dedicated_branchall_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":"branchall","parallel":true,"branches":[{"expr":"true","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(x: number) { return x + 100; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]},{"expr":"true","modules":[{"id":"c","value":{"type":"rawscript","content":"export function main(x: number) { return x + 200; }","language":"bun","input_transforms":{"x":{"expr":"flow_input.x","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}]}}]}', +'system' +); + +-- ============================================================ +-- Test: nested branch inside loop in dedicated flow +-- For-loop [1,2] with inner branchone: if iter.value > 1 → x + 100, else → x + 200. +-- Tests deeply nested dispatch (forloop-N/branchone-0/step_id). +-- ============================================================ +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/dedicated_nested_flow', +'{3000000000000018}', +'{"$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]"},"skip_failures":false,"parallel":false,"squash":false,"modules":[{"id":"b","value":{"type":"branchone","branches":[{"expr":"flow_input.iter.value > 1","modules":[{"id":"c","value":{"type":"rawscript","content":"export function main(iter: {value: number}) { return iter.value + 100; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}],"default":[{"id":"d","value":{"type":"rawscript","content":"export function main(iter: {value: number}) { return iter.value + 200; }","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 ( +3000000000000018, +'test-workspace', +'f/system/dedicated_nested_flow', +'{"$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]"},"skip_failures":false,"parallel":false,"squash":false,"modules":[{"id":"b","value":{"type":"branchone","branches":[{"expr":"flow_input.iter.value > 1","modules":[{"id":"c","value":{"type":"rawscript","content":"export function main(iter: {value: number}) { return iter.value + 100; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}],"default":[{"id":"d","value":{"type":"rawscript","content":"export function main(iter: {value: number}) { return iter.value + 200; }","language":"bun","input_transforms":{"iter":{"expr":"flow_input.iter","type":"javascript"}},"lock":"{}\\n//bun.lock\\n"}}]}}]}}]}', +'system' +); + diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index f04e189315..4819a1da60 100644 --- a/backend/tests/python_jobs.rs +++ b/backend/tests/python_jobs.rs @@ -227,6 +227,8 @@ mod dedicated_worker_protocol_python { enum ProtocolCmd { Exec { path: String, args: serde_json::Value }, ExecPreprocess { path: String, args: serde_json::Value }, + Execd { args: serde_json::Value }, + ExecdPreprocess { args: serde_json::Value }, } /// Run a Python worker test with raw protocol commands @@ -265,13 +267,15 @@ mod dedicated_worker_protocol_python { ProtocolCmd::ExecPreprocess { path, args } => { format!("exec_preprocess:{}:{}", path, args) } + ProtocolCmd::Execd { args } => format!("execd:{}", args), + ProtocolCmd::ExecdPreprocess { args } => format!("execd_preprocess:{}", args), }; writeln!(stdin, "{}", line).unwrap(); stdin.flush().unwrap(); let expected_lines = match cmd { - ProtocolCmd::ExecPreprocess { .. } => 2, - ProtocolCmd::Exec { .. } => 1, + ProtocolCmd::ExecPreprocess { .. } | ProtocolCmd::ExecdPreprocess { .. } => 2, + ProtocolCmd::Exec { .. } | ProtocolCmd::Execd { .. } => 1, }; for _ in 0..expected_lines { @@ -374,6 +378,93 @@ def main(x: int): ); } + // ==================== execd / execd_preprocess Tests ==================== + + #[test] + fn test_python_execd_single_script() { + let results = run_py_raw_protocol_test( + &[( + "f/test/add", + "def main(a: int, b: int):\n return a + b\n", + )], + vec![ProtocolCmd::Execd { args: serde_json::json!({"a": 3, "b": 4}) }], + ); + assert_eq!(results.len(), 1); + assert_eq!( + results[0], + DedicatedWorkerResult::Success(serde_json::json!(7)) + ); + } + + #[test] + fn test_python_execd_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::ExecdPreprocess { 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_execd_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::ExecdPreprocess { args: serde_json::json!({"x": 5}) }], + ); + assert_eq!(results.len(), 1); + assert!(matches!(results[0], DedicatedWorkerResult::Error(_))); + } + + #[test] + fn test_python_execd_preprocess_then_execd() { + 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::ExecdPreprocess { args: serde_json::json!({"x": 5}) }, + ProtocolCmd::Execd { args: serde_json::json!({"x": 7}) }, + ], + ); + // preprocess: preprocessor(5) => {"x":10}, main(10) => 110 + // execd: 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] 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 cbe2450c29..cf386a7ebf 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -843,7 +843,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 789e0a410d..b97aabe606 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -15815,6 +15815,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 dca6ab1657..4e1534c777 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", @@ -245,7 +252,7 @@ impl RunJob { /// Push the job as a specific user (for testing permissions) pub async fn push_as(self, db: &Pool, username: &str, email: &str) -> Uuid { - let RunJob { payload, args, scheduled_for_o, job_id, .. } = self; + let RunJob { payload, args, scheduled_for_o, 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)); @@ -255,7 +262,7 @@ impl RunJob { let (uuid, tx) = windmill_queue::push( db, tx, - "test-workspace", + &workspace_id, payload, windmill_queue::PushArgs::from(&hm_args), username, @@ -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 30a6ec3e2c..b3c74c5d63 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(":"); @@ -3549,11 +3588,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, @@ -3787,7 +3821,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, @@ -3796,16 +3830,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 { @@ -3833,6 +3868,7 @@ pub async fn start_worker( script_path, "bun", client, + false, ) .await } @@ -3988,4 +4024,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/common.rs b/backend/windmill-worker/src/common.rs index 48efb3ee37..30616228aa 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1297,6 +1297,14 @@ pub fn use_flow_root_path(flow_path: &str) -> String { } } +/// Extract the flow root path by stripping /branchone-N/, /branchall-N/, /forloop-N/, /loop-N/ +/// and everything after. Returns None if no nesting segments are found. +pub fn extract_flow_root(runnable_path: &str) -> Option<&str> { + RE_FLOW_ROOT + .captures(runnable_path) + .and_then(|c| c.get(1).map(|m| m.as_str())) +} + pub fn build_http_client(timeout_duration: std::time::Duration) -> error::Result { configure_client( reqwest::ClientBuilder::new() diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 3013d10689..e48a6028ef 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,20 +627,14 @@ 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); - const parsedArgs = JSON.parse(argsJson); + if (line.startsWith("execd_preprocess:")) {{ + const argsJson = line.slice("execd_preprocess:".length); if (typeof preprocessor !== 'function') {{ console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }}) + '\n'); continue; }} try {{ + const parsedArgs = JSON.parse(argsJson); function preArgsObjToArr({{ {pre_spread} }}: any) {{ return [ {pre_spread} ]; }} @@ -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..453b4e986c 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,52 @@ for line in sys.stdin: sys.stdout.flush() continue + if line.startswith('execd_preprocess:'): + try: + args_json = line[len('execd_preprocess:'):] + entry = next(iter(scripts.values())) + 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('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 +2921,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..bea06588ed 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,21 @@ 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| { + // For flow steps inside branches/loops, runnable_path includes + // nesting segments (e.g. f/flow/branchone-0/a) but the dedicated + // worker map is keyed by flow_root/step_id (e.g. f/flow/a). + // When nesting segments are present, use flow_root + flow_step_id + // to construct the correct key. + let key = + if let Some(flow_root) = crate::common::extract_flow_root(path) { + let step_id = job.flow_step_id.as_deref().unwrap_or(""); + format!("{}:{}/{}", job.workspace_id, flow_root, step_id) + } else { + 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 +5457,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..670e0384bd 100644 --- a/frontend/src/lib/components/DedicatedWorkersSelector.svelte +++ b/frontend/src/lib/components/DedicatedWorkersSelector.svelte @@ -3,7 +3,7 @@ ScriptService, FlowService, WorkspaceService, - WorkspaceDependenciesService, + ConfigService, type FlowModule } from '$lib/gen' import { @@ -66,8 +66,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 +129,53 @@ 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) + // Fetch dep info only when not disabled (requires devops role) + let needsFetch = !disabled && currentExistingDeps.size === 0 + if (!needsFetch && !disabled) { + 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 (requires devops role) 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 - }) - } - } - depsPerWorkspace.set(ws, depsMap) - // Merge into existingDeps - for (const d of wsDeps) { - if (!d.archived && d.name) { - currentExistingDeps.add(d.name) - } - } - } catch { - // ignore + if (needsFetch) { + 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) } - }) - ) - existingDeps = new Set(currentExistingDeps) + wsMap.set(d.path, { + deps: d.workspace_dep_names, + language: d.language + }) + } + } + 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) + } + } + existingDeps = new Map(currentExistingDeps) } try { @@ -219,6 +218,17 @@ const parsed = parseTag(tag) if (!parsed) return + if (disabled) { + // Read-only: just show path info, skip API calls + newInfo.set(tag, { + tag, + workspace: parsed.workspace, + type: parsed.type, + path: parsed.path + }) + return + } + if (parsed.type === 'script') { const depInfo = depsPerWorkspace.get(parsed.workspace)?.get(parsed.path) let scriptLang = depInfo?.language @@ -394,7 +404,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 @@ -403,21 +413,29 @@ workspace: workspaceId, dedicatedWorker: true }), - ScriptService.listDedicatedWithDeps({ - workspace: workspaceId - }).catch(() => []), - WorkspaceDependenciesService.listWorkspaceDependencies({ - workspace: workspaceId - }).catch(() => []) + disabled + ? Promise.resolve([]) + : ScriptService.listDedicatedWithDeps({ + workspace: workspaceId + }).catch(() => []), + disabled + ? Promise.resolve([]) + : ConfigService.listAllWorkspaceDependencies().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 +515,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) } @@ -519,26 +541,32 @@ let selectedCount = $derived(runnables.filter((r) => r.selected).length) - // Compute shared runner groups: scripts sharing a (dep_name, language) pair + // Compute shared runner groups: scripts sharing a (workspace, dep_name, language) tuple interface RunnerGroup { + workspace: string depName: string language: string tags: string[] } let runnerGroups: RunnerGroup[] = $derived.by(() => { - const groupMap = new Map() + 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) } else { - groupMap.set(key, { depName: dep, language: lang, tags: [tag] }) + groupMap.set(key, { + workspace: info.workspace, + depName: dep, + language: lang, + tags: [tag] + }) } } } @@ -562,16 +590,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. @@ -580,10 +620,15 @@ {/if} {/snippet} -{#snippet tagRow(tag: string, info: SelectedTagInfo | undefined)} +{#snippet tagRow(tag: string, info: SelectedTagInfo | undefined, standalone: boolean)} + {@const hasChevron = info?.type === 'flow' && info.runners && info.runners.length > 0}
-
- {#if info?.type === 'flow' && info.runners && info.runners.length > 0} +
+ {#if hasChevron} - {:else} + {:else if !standalone}
{/if} -
+
{#if info} {#if info.type === 'flow'} @@ -613,10 +658,10 @@ {#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} + {#if info.type === 'flow' && info.runners && info.runners.length > 0} {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} @@ -664,19 +709,19 @@
- {#each runnerGroups as group (`${group.depName}:${group.language}`)} + {#each runnerGroups as group (`${group.workspace}:${group.depName}:${group.language}`)}
Shared runner - {@render depBadge(group.depName)} + {@render depBadge(group.depName, group.workspace)} {group.language}
{#each group.tags as tag (tag)} {@const info = selectedTagsInfo.get(tag)} - {@render tagRow(tag, info)} + {@render tagRow(tag, info, false)} {/each}
@@ -685,199 +730,168 @@ {#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} -
+ {@render tagRow(tag, info, true)} {/each}
{/if} - -
- + {#if selectorExpanded} - - {:else} - - {/if} - - {selectedTags.length > 0 ? 'Add more scripts/flows' : 'Select scripts/flows'} - - +
+ +
+ Workspace + ({ value: w.id, label: `${w.name} (${w.id})` }))} - placeholder="Select workspace..." - disabled={disabled || workspacesLoading} - /> -
+ + {#if selectedWorkspace} +
+
+ Scripts/flows with dedicated worker enabled + {#if !loading && runnables.length > 0} +
+ + +
+ {/if} +
- - {#if selectedWorkspace} -
-
- Scripts/flows with dedicated worker enabled - {#if !loading && runnables.length > 0} -
- - -
- - {#if loading} -
- - Loading... -
- {:else if runnables.length === 0} -
- No scripts or flows with dedicated worker enabled found. -
- {:else} -
- {#each runnables as runnable (runnable.tag)} -
-
- {#if runnable.type === 'flow' && runnable.runners && runnable.runners.length > 0} + {:else if runnables.length === 0} +
+ No scripts or flows with dedicated worker enabled found. +
+ {:else} +
+ {#each runnables as runnable (runnable.tag)} +
+
+ {#if runnable.type === 'flow' && runnable.runners && runnable.runners.length > 0} + + {/if} - {:else} -
- {/if} - + {/if}
- - {#if runnable.type === 'flow' && runnable.expanded && runnable.runners} -
- {#if runnable.runners.length === 0} -
- No eligible steps (python3/bun/bunnative/deno) -
- {:else} - {#each runnable.runners as runner (runner.stepId)} -
- {runner.stepId} - {#if runner.stepSummary} - - {runner.stepSummary} - - {/if} - - {runner.isInline ? runner.language : runner.scriptPath} - -
- {/each} - {/if} -
- {/if} -
- {/each} -
-
- {selectedCount} selected -
- {/if} -
- {/if} -
- {/if} -
+ {/each} +
+
+ {selectedCount} selected +
+ {/if} +
+ {/if} +
+ {/if} +
+ {/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). +