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