kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "feat: restore bun for dedicated workers, fix dispatch & serialization, cross-workspace deps (#8645)"

Browse Source 
This reverts commit 619ebb65ce.
This commit is contained in:
hugocasa
2026-04-03 01:08:10 +02:00
parent 8581a3300d
commit ee5420e401
21 changed files with 146 additions and 1618 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CLAUDE.md
View File

@@ -26,7 +26,6 @@ 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 <name>` from `backend/` to create new migrations (never generate timestamps manually)
## Banned Patterns

64
backend/.sqlx/query-edd6c09b7f012588788fd3c572d20eb439a80d52ae75ebd25128ffce759cd313.json generated
View File

@@ -1,64 +0,0 @@
{
"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"
}

72
backend/.sqlx/query-f0858450021df721d8a48b1b5dc887c5424562acd9769c80c5899193ef16b56b.json generated
View File

@@ -1,72 +0,0 @@
{
"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"
}

1
backend/Cargo.toml
View File

@@ -267,7 +267,6 @@ 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

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
91cd9ca431ec5b80b038573b5a5efa9773dafaa1
57dd88faa3b0b354f813385cf3f6a34eca54a4a1

8
backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.down.sql
View File

@@ -1,8 +0,0 @@
-- 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;

10
backend/migrations/20260401101637_bun_dedicated_nodejs_annotation.up.sql
View File

@@ -1,10 +0,0 @@
-- 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';

16
backend/tests/bun_jobs.rs
View File

@@ -1066,8 +1066,8 @@ mod dedicated_worker_protocol {
let mut results = Vec::new();
for job_args in jobs {
// Protocol: execd:<json_args> (single-script, no path needed)
writeln!(stdin, "execd:{}", job_args.to_string()).unwrap();
// Protocol: exec:<script_path>:<json_args>
writeln!(stdin, "exec:{}:{}", TEST_SCRIPT_PATH, job_args.to_string()).unwrap();
stdin.flush().unwrap();
let mut response = String::new();
@@ -1653,6 +1653,8 @@ 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<serde_json::Value>,
@@ -1702,7 +1704,7 @@ mod dedicated_worker_protocol_deno {
let mut results = Vec::new();
for job_args in jobs {
writeln!(stdin, "execd:{}", job_args.to_string()).unwrap();
writeln!(stdin, "exec:{}:{}", TEST_SCRIPT_PATH, job_args.to_string()).unwrap();
stdin.flush().unwrap();
loop {
@@ -1823,13 +1825,7 @@ export function main(msg: string): never {
let mut results = Vec::new();
for (cmd, args) in &commands {
// 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();
writeln!(stdin, "{}:{}:{}", cmd, TEST_SCRIPT_PATH, args).unwrap();
stdin.flush().unwrap();
let expected_lines = if *cmd == "exec_preprocess" { 2 } else { 1 };

660
backend/tests/dedicated_workers.rs
View File

@@ -1,660 +0,0 @@
#[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<Postgres>,
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<String> =
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<uuid::Uuid> =
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<String> =
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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<bool> =
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<Postgres>) -> 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(())
}
}

252
backend/tests/fixtures/dedicated_flows.sql vendored
View File

@@ -1,252 +0,0 @@
-- 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<empty>', 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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>"}},{"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<empty>"}}]}',
'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<empty>"}},{"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<empty>"}}]}',
'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<empty>', 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<empty>', 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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>"}}]}}]}',
'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<empty>"}}]}}]}',
'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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>"}}]}',
'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<empty>', 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<empty>', true);

89
backend/windmill-api-configs/src/lib.rs
View File

@@ -43,14 +43,6 @@ 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)]
@@ -327,84 +319,3 @@ async fn list_configs() -> error::JsonResult<String> {
"Config listing available only in the enterprise version".to_string(),
))
}
#[derive(Serialize)]
struct WorkspaceDependencySummary {
workspace_id: String,
name: Option<String>,
language: windmill_common::scripts::ScriptLang,
}
async fn list_all_workspace_dependencies(
authed: ApiAuthed,
Extension(db): Extension<DB>,
) -> error::JsonResult<Vec<WorkspaceDependencySummary>> {
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<String>,
}
async fn list_all_dedicated_with_deps(
authed: ApiAuthed,
Extension(db): Extension<DB>,
) -> error::JsonResult<Vec<DedicatedScriptDepsWithWorkspace>> {
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))
}

6
backend/windmill-api-scripts/src/scripts.rs
View File

@@ -834,11 +834,7 @@ async fn create_script_internal<'c>(
})
};
// 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 needs_lock_gen = lock.is_none() && codebase.is_none();
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

58
backend/windmill-api/openapi.yaml
View File

@@ -15697,64 +15697,6 @@ 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

132
backend/windmill-test-utils/src/lib.rs
View File

@@ -160,7 +160,6 @@ pub struct RunJob {
pub scheduled_for_o: Option<chrono::DateTime<chrono::Utc>>,
pub email: String,
pub job_id: Option<Uuid>,
pub workspace_id: String,
}
impl From<JobPayload> for RunJob {
@@ -171,7 +170,6 @@ impl From<JobPayload> for RunJob {
scheduled_for_o: None,
email: "test@windmill.dev".to_string(),
job_id: None,
workspace_id: "test-workspace".to_string(),
}
}
}
@@ -200,13 +198,8 @@ impl RunJob {
self
}
pub fn workspace(mut self, workspace_id: impl Into<String>) -> Self {
self.workspace_id = workspace_id.into();
self
}
pub async fn push(self, db: &Pool<Postgres>) -> Uuid {
let RunJob { payload, args, scheduled_for_o, email, job_id, workspace_id } = self;
let RunJob { payload, args, scheduled_for_o, email, job_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));
@@ -216,7 +209,7 @@ impl RunJob {
let (uuid, tx) = windmill_queue::push(
db,
tx,
&workspace_id,
"test-workspace",
payload,
windmill_queue::PushArgs::from(&hm_args),
/* user */ "test-user",
@@ -443,127 +436,6 @@ 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<windmill_common::worker::WorkspacedPath>,
) -> (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<String> = 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<Fut: std::future::Future>(
conn: impl Into<Connection>,
inner: Fut,
port: u16,
dedicated_workers: Vec<windmill_common::worker::WorkspacedPath>,
) -> <Fut as std::future::Future>::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<Postgres>) -> impl Stream<Item = Uuid> + Unpin {
listen_for_uuid_on(db, "completed").await
}

85
backend/windmill-worker/src/bun_executor.rs
View File

@@ -153,10 +153,8 @@ 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:
/// execd:<json_args> -> execute the single registered script (non-runner-group)
/// execd_preprocess:<json_args> -> preprocess + execute the single registered script
/// exec:<path>:<json_args> -> execute script by path (runner groups with multiple scripts)
/// exec_preprocess:<path>:<json> -> preprocess + execute script by path
/// exec:<path>:<json_args> -> wm_res[success]:<result> | wm_res[error]:<err>
/// exec_preprocess:<path>:<json> -> wm_res[preprocessed_args]:<result> then wm_res[success]:<result> | wm_res[error]:<err>
/// end -> exit
#[cfg(any(feature = "private", test))]
pub fn generate_multi_script_wrapper(scripts: &[TsScriptEntry<'_>], ext: &str) -> String {
@@ -242,43 +240,6 @@ 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(":");
@@ -3586,6 +3547,11 @@ 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,
@@ -3819,7 +3785,7 @@ pub async fn start_worker(
}
if annotation.nodejs {
let wrapper_path = format!("{job_dir}/wrapper.mjs");
let script_path = format!("{job_dir}/wrapper.mjs");
handle_dedicated_process(
&*NODE_BIN_PATH,
@@ -3828,17 +3794,16 @@ pub async fn start_worker(
envs,
context,
common_bun_proc_envs,
vec![&wrapper_path],
vec![&script_path],
killpill_rx,
job_completed_tx,
token,
jobs_rx,
worker_name,
db,
script_path,
&script_path,
"nodejs",
client,
false,
)
.await
} else {
@@ -3866,7 +3831,6 @@ pub async fn start_worker(
script_path,
"bun",
client,
false,
)
.await
}
@@ -4022,33 +3986,4 @@ 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:")"#));
}
}

23
backend/windmill-worker/src/deno_executor.rs
View File

@@ -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 `execd:<args>` protocol (single-script, no path needed).
/// and preprocessor logic. Uses the `exec:<path>:<args>` protocol.
#[cfg(any(feature = "private", test))]
pub fn generate_dedicated_worker_wrapper(inner_content: &str) -> Result<String> {
let args = windmill_parser_ts::parse_deno_signature(inner_content, true, false, None)?.args;
@@ -627,8 +627,14 @@ pub fn generate_dedicated_worker_wrapper(inner_content: &str) -> Result<String>
let preprocessor_logic = if let Some(ref pre_spread) = pre_spread {
format!(
r#"
if (line.startsWith("execd_preprocess:")) {{
const argsJson = line.slice("execd_preprocess:".length);
if (line.startsWith("exec_preprocess:")) {{
const rest = line.slice("exec_preprocess:".length);
const colonIdx = rest.indexOf(":");
if (colonIdx === -1) {{
console.log("wm_res[error]:" + JSON.stringify({{ message: "Malformed exec_preprocess command: missing colon separator", name: "Error" }}) + '\n');
continue;
}}
const argsJson = rest.slice(colonIdx + 1);
const parsedArgs = JSON.parse(argsJson);
if (typeof preprocessor !== 'function') {{
console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }}) + '\n');
@@ -680,8 +686,14 @@ for await (const chunk of Deno.stdin.readable) {{
break;
}}
{preprocessor_logic}
if (line.startsWith("execd:")) {{
const argsJson = line.slice("execd:".length);
if (line.startsWith("exec:")) {{
const rest = line.slice("exec:".length);
const colonIdx = rest.indexOf(":");
if (colonIdx === -1) {{
console.log("wm_res[error]:" + JSON.stringify({{ message: "Malformed exec command: missing colon separator", name: "Error" }}) + '\n');
continue;
}}
const argsJson = rest.slice(colonIdx + 1);
try {{
let {{ {spread} }} = JSON.parse(argsJson)
{dates}
@@ -790,7 +802,6 @@ pub async fn start_worker(
script_path,
"deno",
client,
false,
)
.await
}

23
backend/windmill-worker/src/python_executor.rs
View File

@@ -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:
/// execd:<json_args> -> execute the single registered script (non-runner-group)
/// exec:<path>:<json_args> -> execute script by path (runner groups)
/// exec:<path>:<json_args> -> wm_res[success]:<result> | wm_res[error]:<err>
/// exec_preprocess:<path>:<json> -> wm_res[preprocessed_args]:<result> then wm_res[success]:<result> | wm_res[error]:<err>
/// end -> exit
#[cfg(any(feature = "private", test))]
pub fn generate_multi_script_wrapper(
@@ -1370,24 +1370,6 @@ 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:'):]
@@ -2893,7 +2875,6 @@ pub async fn start_worker(
script_path,
"python",
client,
false,
)
.await
}

26
backend/windmill-worker/src/worker.rs
View File

@@ -1994,8 +1994,9 @@ pub async fn run_worker(
// Option<JoinHandle<()>>,
#[cfg(all(feature = "private", feature = "enterprise"))]
let (dedicated_workers, dedicated_handles): (
let (dedicated_workers, dedicated_flow_paths, dedicated_handles): (
HashMap<String, Sender<DedicatedWorkerJob>>,
HashSet<String>,
Vec<JoinHandle<()>>,
) = match conn {
Connection::Sql(pool) => {
@@ -2010,14 +2011,15 @@ pub async fn run_worker(
)
.await
}
Connection::Http(_) => (HashMap::new(), vec![]),
Connection::Http(_) => (HashMap::new(), HashSet::new(), vec![]),
};
#[cfg(any(not(feature = "private"), not(feature = "enterprise")))]
let (dedicated_workers, dedicated_handles): (
let (dedicated_workers, dedicated_flow_paths, dedicated_handles): (
HashMap<String, Sender<DedicatedWorkerJob>>,
HashSet<String>,
Vec<JoinHandle<()>>,
) = (HashMap::new(), vec![]);
) = (HashMap::new(), HashSet::new(), vec![]);
if i_worker == 1 {
// Initialize runtime asset inserter for batched database inserts
@@ -2478,10 +2480,17 @@ pub async fn run_worker(
JobKind::Script | JobKind::Preview | JobKind::FlowScript
) {
if !dedicated_workers.is_empty() {
let dedicated_worker_tx = job.runnable_path.as_ref().and_then(|path| {
let key = format!("{}:{}", job.workspace_id, path);
dedicated_workers.get(&key)
});
// 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))
};
if let Some(dedicated_worker_tx) = dedicated_worker_tx {
let dedicated_job = DedicatedWorkerJob {
job: Arc::new(job.job()),
@@ -5446,3 +5455,4 @@ pub fn get_worker_internal_server_inline_utils(
)),
}
}

202
frontend/src/lib/components/DedicatedWorkersSelector.svelte
View File

@@ -4,7 +4,6 @@
FlowService,
WorkspaceService,
WorkspaceDependenciesService,
ConfigService,
type FlowModule
} from '$lib/gen'
import {
@@ -67,8 +66,8 @@
let workspaces: { id: string; name: string }[] = $state([])
let workspacesLoading = $state(true)
let selectorExpanded = $state(false)
// Map of dep name → set of workspaces that have it (for cross-workspace validation)
let existingDeps: Map<string, Set<string>> = $state(new Map())
// Set of existing workspace dependency names for the current workspace (for validation)
let existingDeps: Set<string> = $state(new Set())
// Track detailed info for each selected tag (for displaying in summary)
interface SelectedTagInfo {
@@ -130,98 +129,54 @@
const newInfo = new SvelteMap<string, SelectedTagInfo>()
// 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
}
// Collect workspaces that need dep info fetched
const workspacesNeedingDeps = new Set<string>()
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 across all workspaces in one call
// Fetch workspace dep info for workspaces that need it
const depsPerWorkspace = new Map<string, Map<string, { deps: string[]; language: string }>>()
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)
}
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)
}
}
} catch {
// Fallback to per-workspace calls if cross-workspace endpoint unavailable
const workspacesNeedingDeps = new Set<string>()
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<string, { deps: string[]; language: string }>()
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 (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<string, { deps: string[]; language: string }>()
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)
depsPerWorkspace.set(ws, depsMap)
// Merge into existingDeps
for (const d of wsDeps) {
if (!d.archived && d.name) {
currentExistingDeps.add(d.name)
}
}
} catch {
// ignore
}
})
)
existingDeps = new Set(currentExistingDeps)
}
try {
@@ -439,7 +394,7 @@
loading = true
runnables = []
const [scripts, flows, dedicatedDeps, allWsDeps] = await Promise.all([
const [scripts, flows, dedicatedDeps, wsDeps] = await Promise.all([
ScriptService.listScripts({
workspace: workspaceId,
dedicatedWorker: true
@@ -451,28 +406,18 @@
ScriptService.listDedicatedWithDeps({
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(() => [])
)
WorkspaceDependenciesService.listWorkspaceDependencies({
workspace: workspaceId
}).catch(() => [])
])
// Track existing workspace dep names with their workspaces for validation
const newDeps = new Map<string, Set<string>>()
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
// 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)
)
// Build a map from path -> workspace dep names
const depsMap = new Map<string, string[]>()
@@ -552,11 +497,7 @@
}
function updateSelectedTags() {
// 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]
selectedTags = runnables.filter((r) => r.selected).map((r) => r.tag)
onchange?.(selectedTags)
}
@@ -586,14 +527,13 @@
}
let runnerGroups: RunnerGroup[] = $derived.by(() => {
// Group by (workspace, dep_name, language) — runner groups are per-workspace
const groupMap = new Map<string, { depName: string; language: string; tags: string[] }>()
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 = `${info.workspace}:${dep}:${lang}`
const key = `${dep}:${lang}`
const existing = groupMap.get(key)
if (existing) {
existing.tags.push(tag)
@@ -622,28 +562,16 @@
let standaloneTags: string[] = $derived(selectedTags.filter((tag) => !tagRunnerGroup.has(tag)))
</script>
{#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}
<Badge color="indigo" small href="/workspace_settings?tab=dependencies&workspace={workspace}">
{#snippet depBadge(dep: string)}
{#if existingDeps.has(dep)}
<Badge color="indigo" small href="/workspace_settings?tab=dependencies">
{dep}
<ExternalLink class="h-2.5 w-2.5" />
</Badge>
{:else if existsElsewhere}
<Tooltip small>
Workspace dependency '{dep}' exists in {[...(depWorkspaces ?? [])].join(', ')} but not in '{workspace}'.
Each workspace needs its own dependency.
</Tooltip>
<Badge color="yellow" small>
<TriangleAlert class="h-2.5 w-2.5" />
{dep} (not in {workspace})
</Badge>
{:else}
<Tooltip small>
Workspace dependency '{dep}' not found in any workspace. Create it in workspace settings to
enable shared runners.
Workspace dependency '{dep}' not found. Create it in workspace settings to enable shared
runners.
</Tooltip>
<Badge color="yellow" small>
<TriangleAlert class="h-2.5 w-2.5" />
@@ -685,7 +613,7 @@
{#if !tagRunnerGroup.has(tag)}
{#if info.workspaceDeps}
{#each info.workspaceDeps as dep}
{@render depBadge(dep, info?.workspace)}
{@render depBadge(dep)}
{/each}
{/if}
{#if info.type === 'flow' && info.runners}
@@ -742,7 +670,7 @@
<Layers size={12} class="flex-shrink-0 text-secondary" />
<span class="text-xs font-medium text-emphasis">Shared runner</span>
<span class="flex-1"></span>
{@render depBadge(group.depName, selectedTagsInfo.get(group.tags[0])?.workspace)}
{@render depBadge(group.depName)}
<Badge color="gray" small>{group.language}</Badge>
</div>
<div class="divide-y">
@@ -768,7 +696,7 @@
<span class="flex-1"></span>
{#if info?.workspaceDeps}
{#each info.workspaceDeps as dep}
{@render depBadge(dep, info?.workspace)}
{@render depBadge(dep)}
{/each}
{/if}
{#if info?.type === 'flow' && info.runners}
@@ -904,7 +832,7 @@
{/if}
{#if runnable.workspaceDeps}
{#each runnable.workspaceDeps as dep}
{@render depBadge(dep, selectedWorkspace)}
{@render depBadge(dep)}
{/each}
{/if}
<Badge color={runnable.type === 'flow' ? 'indigo' : 'gray'} small>

17
frontend/src/lib/components/flows/content/FlowLoop.svelte
View File

@@ -32,6 +32,7 @@
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>('FlowEditorContext')
@@ -190,11 +191,17 @@
<div class="flex-shrink-0">
<div class="mb-2 text-sm font-bold"
>Squash
<Tooltip documentationLink="https://www.windmill.dev/docs/flows/flow_loops">
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).
</Tooltip>
<Badge
>Beta <Tooltip documentationLink="https://www.windmill.dev/docs/flows/flow_loops">
<span class="font-semibold"
>This can result in unexpected behavior, use at your own risk for now.</span
><br />
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).
</Tooltip>
</Badge>
</div>
<Toggle
bind:checked={mod.value.squash}

17
frontend/src/lib/components/flows/content/FlowWhileLoop.svelte
View File

@@ -20,6 +20,7 @@
import FlowModuleSkip from './FlowModuleSkip.svelte'
import TabsV2 from '$lib/components/common/tabs/TabsV2.svelte'
import { useUiIntent } from '$lib/components/copilot/chat/flow/useUiIntent'
import Badge from '$lib/components/common/badge/Badge.svelte'
const { flowStateStore } = getContext<FlowEditorContext>('FlowEditorContext')
@@ -106,11 +107,17 @@
<div class="flex-shrink-0">
<div class="mb-2 text-sm font-bold"
>Squash
<Tooltip documentationLink="https://www.windmill.dev/docs/flows/while_loops">
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).
</Tooltip>
<Badge
>Beta <Tooltip documentationLink="https://www.windmill.dev/docs/flows/while_loops">
<span class="font-semibold"
>This can result in unexpected behavior, use at your own risk for now.</span
><br />
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).
</Tooltip>
</Badge>
</div>
<Toggle
bind:checked={mod.value.squash}