From bffa61e33f2305bbeb79a2c91989a47baa7dff31 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 3 Apr 2026 19:50:07 +0200 Subject: [PATCH] fix: dedicated worker dispatch, cross-workspace deps, UI improvements (#8689) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: restore bun as default runtime for dedicated workers Co-Authored-By: Claude Opus 4.6 (1M context) * chore: add context comment for bun dedicated worker nodejs migration Co-Authored-By: Claude Opus 4.6 (1M context) * fix: dedicated worker dispatch for flows + add E2E tests - Add workspace_id prefix to dedicated worker map lookup keys - Update ee-repo-ref for dedicated worker path handling fix - Add spawn_test_worker_dedicated/in_test_worker_dedicated test helpers - Add 6 E2E tests for dedicated workers: - test_dedicated_flow_rawscript (regression for "Script not found" bug) - test_dedicated_flow_workspace_script - test_dedicated_flow_multiple_steps - test_dedicated_standalone_script - test_dedicated_runner_group - test_dedicated_flow_runners - Add dedicated_flows.sql fixture with scripts, flows, and worker config Co-Authored-By: Claude Opus 4.6 (1M context) * fix: always run dependency job for dedicated worker scripts When a script with dedicated_worker=true is deployed with a pre-computed lock (e.g. via wmill sync push), no dependency job was created, so the dedicated worker never detected the update and kept running the old version. Now dedicated worker scripts always generate a dependency job regardless of whether a lock is provided. The dependency job runs on the dedicated worker and triggers a restart so it picks up the new script version. Fixes #8638 Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use serial_test for dedicated worker tests to avoid WORKER_CONFIG races Dedicated worker tests need non-default worker tags in the global WORKER_CONFIG. When run in parallel (CI uses --test-threads=10), multiple tests clobber each other's config. Use #[serial] to ensure dedicated worker tests run sequentially. Also load worker config from DB via load_worker_config() instead of manually setting WORKER_CONFIG fields, ensuring consistency with the monitor's reload path. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: nodejs dedicated worker script_path shadowing + add multi-language E2E tests Fix script_path shadowing in bun_executor nodejs branch where the wrapper file path was passed to handle_dedicated_process instead of the logical path, causing "Script not found" for all //nodejs dedicated workers. Add E2E tests for dedicated flows in all supported languages: - test_dedicated_flow_deno - test_dedicated_flow_python - test_dedicated_flow_bunnative (V8 PrewarmedIsolate path) - test_dedicated_flow_bun_nodejs (//nodejs annotation) Co-Authored-By: Claude Opus 4.6 (1M context) * fix: simplify dedicated worker dispatch + add serialization and E2E tests - Unified lookup: always use {workspace}:{runnable_path} for dedicated worker dispatch, replacing the flow_step_id iteration approach - Added serialization_semaphore parameter to executor start_worker fns - Added E2E tests: cross-workspace isolation, conflicting flow step IDs, preprocessor on dedicated worker - Added workspace field to RunJob for cross-workspace test support Co-Authored-By: Claude Opus 4.6 (1M context) * feat: cross-workspace workspace dependencies on workers page Add two new instance-level endpoints to the configs router: - GET /configs/list_all_workspace_dependencies - GET /configs/list_all_dedicated_with_deps Both require devops role and return data across all workspaces, enabling the workers page to show a consistent view of which workspace dependencies exist regardless of which workspace the user is browsing. Update DedicatedWorkersSelector to use the new cross-workspace endpoints with fallback to per-workspace calls for non-devops users. Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref to include dedicated worker lookup simplification Co-Authored-By: Claude Opus 4.6 (1M context) * chore: use branch name for ee-repo-ref (CI can't fetch by SHA from non-default branch) Co-Authored-By: Claude Opus 4.6 (1M context) * Update ee-repo-ref.txt with new reference * sqlx * fix: revert serialization semaphore, multi-workspace picker, dep conflict warnings - Remove serialization_semaphore from executor start_worker signatures - Remove serialization test and fixtures - Fix DedicatedWorkersSelector to preserve tags from other workspaces when toggling in the picker - Track workspace deps per-workspace for conflict detection - Show warning when dep exists in another workspace but not the script's - Group runner groups per-workspace to prevent cross-workspace merging - Add workspace to dep badge link URL Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * fix: simplify exec protocol — execd: for single-script, exec: for runner groups Add execd:/execd_preprocess: commands to bun/deno/python wrappers for single-script dedicated workers (no path needed). Runner groups keep exec:/exec_preprocess: with path for multi-script disambiguation. Co-Authored-By: Claude Opus 4.6 (1M context) * test: add unit tests for execd:/exec: wrapper protocol Verify generate_multi_script_wrapper produces both execd: (single-script) and exec: (runner group) protocol handlers, including preprocessor variants. Co-Authored-By: Claude Opus 4.6 (1M context) * Update commit reference in ee-repo-ref.txt * fix: remove beta badge from squash loop, keep tooltip Co-Authored-By: Claude Opus 4.6 (1M context) * fix: update protocol tests to use execd: for single-script wrappers Deno and bun single-script protocol tests now send execd:{args} instead of exec:{path}:{args}, matching the updated wrapper protocol. Multi-script (runner group) tests continue to use exec:{path}:{args}. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: remove unused TEST_SCRIPT_PATH in deno protocol tests Co-Authored-By: Claude Opus 4.6 (1M context) * fix: review feedback — down migration, push_as workspace, UI improvements - Use regexp_replace in down migration for positional accuracy - Fix push_as() to use self.workspace_id instead of hardcoded value - Remove per-workspace API fallbacks, use cross-workspace endpoints only - Skip devops-only API calls when user is not devops (disabled prop) - Fix duplicate key error for cross-workspace runner groups - Add workspace to RunnerGroup for unique keying - Reuse tagRow snippet for standalone items with expand/collapse - Fix picker alignment: remove empty column for non-expandable items Co-Authored-By: Claude Opus 4.6 (1M context) * test: comprehensive dedicated worker test coverage, fix Python execd_preprocess - Add Python execd_preprocess: handler (was missing for single-script dedicated workers) - Add 10 E2E tests: flow+standalone conflict, mixed lang fallback, unsupported lang flow runners, python runner group, bun/python/deno/bunnative preprocessors, runner group preprocessors, branchone flow - Add 4 Python unit tests for execd:/execd_preprocess: protocol - Update EE ref Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * fix: review feedback — migration escaping, deno try/catch, loadRunnables guard - Down migration: use E'...' so \n matches actual newlines - Up migration: anchor regex with ^ to avoid mid-content matches - Deno execd_preprocess: move JSON.parse inside try/catch - DedicatedWorkersSelector: skip devops-only API calls when disabled Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref Co-Authored-By: Claude Opus 4.6 (1M context) * test: add dedicated worker relative import tests for bun and python Verifies that build_loader's CURRENT_PATH correctly resolves workspace- relative imports when running on a dedicated worker subprocess. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: dedicated worker dispatch for nested flow structures (branches/loops) - Add extract_flow_root() to strip nesting segments from runnable_path - Dispatch uses flow_root/flow_step_id for nested paths, runnable_path for flat paths — deterministic, O(1) - Fix assert_ran_on_dedicated_worker to BFS all descendants - Fix python mode labels (python vs python3 for runner groups) - Add tests: simple forloop, multi-step forloop, whileloop, branchall, nested branch-in-loop, mixed lang fallback, unsupported lang runners Co-Authored-By: Claude Opus 4.6 (1M context) * chore: fix ee-repo-ref SHA Co-Authored-By: Claude Opus 4.6 (1M context) * fix: hide picker and skip API calls for read-only users, hide empty runner badge - Hide "Add more scripts/flows" section when disabled (read-only) - Skip per-runnable API calls (getScriptByPath, getFlowByPath) for disabled users — just show path info - Hide "0 runners" badge on flows with no eligible steps Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update ee-repo-ref to 9422b189762ae27edfc346541ae668a4ad728325 This commit updates the EE repository reference after PR #503 was merged in windmill-ee-private. Previous ee-repo-ref: 4c6ba214bfc23fff05d1dc3200ac59e650af3f4f New ee-repo-ref: 9422b189762ae27edfc346541ae668a4ad728325 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 4.6 (1M context) Co-authored-by: windmill-internal-app[bot] --- CLAUDE.md | 1 + ...153c43903f929ae5d62fbba12610f89c36d55.json | 2 +- ...20eb439a80d52ae75ebd25128ffce759cd313.json | 64 + ...887c5424562acd9769c80c5899193ef16b56b.json | 72 + backend/Cargo.toml | 1 + backend/ee-repo-ref.txt | 2 +- ...7_bun_dedicated_nodejs_annotation.down.sql | 6 + ...637_bun_dedicated_nodejs_annotation.up.sql | 10 + backend/tests/bun_jobs.rs | 16 +- backend/tests/dedicated_workers.rs | 1513 +++++++++++++++++ backend/tests/fixtures/dedicated_flows.sql | 599 +++++++ backend/tests/python_jobs.rs | 95 +- backend/windmill-api-configs/src/lib.rs | 89 + backend/windmill-api-scripts/src/scripts.rs | 6 +- backend/windmill-api/openapi.yaml | 58 + backend/windmill-test-utils/src/lib.rs | 136 +- backend/windmill-worker/src/bun_executor.rs | 85 +- backend/windmill-worker/src/common.rs | 8 + backend/windmill-worker/src/deno_executor.rs | 25 +- .../windmill-worker/src/python_executor.rs | 51 +- backend/windmill-worker/src/worker.rs | 37 +- .../DedicatedWorkersSelector.svelte | 522 +++--- .../components/flows/content/FlowLoop.svelte | 17 +- .../flows/content/FlowWhileLoop.svelte | 17 +- 24 files changed, 3091 insertions(+), 341 deletions(-) create mode 100644 backend/.sqlx/query-edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313.json create mode 100644 backend/.sqlx/query-f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b.json create mode 100644 backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql create mode 100644 backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.up.sql create mode 100644 backend/tests/dedicated_workers.rs create mode 100644 backend/tests/fixtures/dedicated_flows.sql diff --git a/CLAUDE.md b/CLAUDE.md index 0f01e67c57..825a033f94 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -26,6 +26,7 @@ Open-source platform for internal tools, workflows, API integrations, background - **DB**: `psql postgres://postgres:changeme@localhost:5432/windmill` - **Login**: `admin@windmill.dev` / `changeme` - **Instance settings**: navigate to `/#superadmin-settings` +- **Migrations**: use `cargo sqlx migrate add -r ` from `backend/` to create new migrations (never generate timestamps manually) ## Banned Patterns diff --git a/backend/.sqlx/query-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). +