diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index 8ef5eecea9..be537abb11 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -1,6 +1,7 @@ name: Backend only integration tests on: + workflow_dispatch: push: branches: - "main" diff --git a/backend/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index 1617babb81..c69300028d 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -888,7 +888,7 @@ mod dedicated_worker_protocol { if bundle_for_node { // For Node.js: bundle to JavaScript first (like production's build_loader with LoaderMode::Node) - let wrapper = generate_dedicated_worker_wrapper(arg_names, "./main.js", None); + let wrapper = generate_dedicated_worker_wrapper(arg_names, "./main.js", None, None); std::fs::write(dir.join("wrapper.mjs"), wrapper).unwrap(); // Use the exact same build_loader function as production @@ -925,7 +925,7 @@ mod dedicated_worker_protocol { output_path } else { // For Bun: use TypeScript directly (like production) - let wrapper = generate_dedicated_worker_wrapper(arg_names, "./main.ts", None); + let wrapper = generate_dedicated_worker_wrapper(arg_names, "./main.ts", None, None); let wrapper_path = dir.join("wrapper.mjs"); std::fs::write(&wrapper_path, wrapper).unwrap(); wrapper_path diff --git a/backend/tests/fixtures/hello.sql b/backend/tests/fixtures/hello.sql index f273cdfeaa..962f6441d9 100644 --- a/backend/tests/fixtures/hello.sql +++ b/backend/tests/fixtures/hello.sql @@ -30,6 +30,89 @@ export async function main(foo: string, bar: string) { '', 'f/system/hello_with_preprocessor', 123413, 'deno', ''); +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +' +export async function preprocessor(foo: string, bar: string) { + return { foo: foo + "_preprocessed", bar: bar + "_preprocessed" }; +} + +export async function main(foo: string, bar: string) { + return "Hello " + foo + " " + bar; +} +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"default":null,"description":"","originalType":"string","type":"string"},"bar":{"default":null,"description":"","originalType":"string","type":"string"}},"required":["foo","bar"],"type":"object"}', +'', +'', +'f/system/hello_preprocessor_dedicated_bun', 123414, 'bun', E'{}\n//bun.lock\n{}'); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +' +def preprocessor(foo: str, bar: str): + return {"foo": foo + "_preprocessed", "bar": bar + "_preprocessed"} + +def main(foo: str, bar: str): + return "Hello " + foo + " " + bar +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"default":null,"description":"","originalType":"string","type":"string"},"bar":{"default":null,"description":"","originalType":"string","type":"string"}},"required":["foo","bar"],"type":"object"}', +'', +'', +'f/system/hello_preprocessor_dedicated_python', 123415, 'python3', ''); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +' +export async function preprocessor(foo: string, bar: string) { + return { foo: foo + "_preprocessed", bar: bar + "_preprocessed" }; +} + +export async function main(foo: string, bar: string) { + return "Hello " + foo + " " + bar; +} +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"default":null,"description":"","originalType":"string","type":"string"},"bar":{"default":null,"description":"","originalType":"string","type":"string"}},"required":["foo","bar"],"type":"object"}', +'', +'', +'f/system/hello_preprocessor_dedicated_deno', 123416, 'deno', ''); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +'//native +export async function preprocessor(foo: string, bar: string) { + return { foo: foo + "_preprocessed", bar: bar + "_preprocessed" }; +} + +export async function main(foo: string, bar: string) { + return "Hello " + foo + " " + bar; +} +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"default":null,"description":"","originalType":"string","type":"string"},"bar":{"default":null,"description":"","originalType":"string","type":"string"}},"required":["foo","bar"],"type":"object"}', +'', +'', +'f/system/hello_preprocessor_bunnative', 123417, 'bunnative', E'{}\n//bun.lock\n{}'); + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'system', +'//native +export async function preprocessor(foo: string, bar: string) { + return { foo: foo + "_preprocessed", bar: bar + "_preprocessed" }; +} + +export async function main(foo: string, bar: string) { + return "Hello " + foo + " " + bar; +} +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"default":null,"description":"","originalType":"string","type":"string"},"bar":{"default":null,"description":"","originalType":"string","type":"string"}},"required":["foo","bar"],"type":"object"}', +'', +'', +'f/system/hello_preprocessor_dedicated_bunnative', 123418, 'bunnative', E'{}\n//bun.lock\n{}'); + INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( 'test-workspace', '', diff --git a/backend/tests/job_payload.rs b/backend/tests/job_payload.rs index 87253807be..4a8355b78d 100644 --- a/backend/tests/job_payload.rs +++ b/backend/tests/job_payload.rs @@ -1,16 +1,16 @@ mod job_payload { use serde_json::json; use sqlx::{Pool, Postgres}; + use windmill_common::flow_status::RestartedFrom; use windmill_common::flows::{FlowModule, FlowModuleValue, FlowValue}; use windmill_common::jobs::JobPayload; use windmill_common::scripts::{ScriptHash, ScriptLang}; - use windmill_common::flow_status::RestartedFrom; - use windmill_test_utils::*; use windmill_common::min_version::{ MIN_VERSION, MIN_VERSION_IS_AT_LEAST_1_427, MIN_VERSION_IS_AT_LEAST_1_432, MIN_VERSION_IS_AT_LEAST_1_440, }; + use windmill_test_utils::*; pub async fn initialize_tracing() { use std::sync::Once; @@ -305,7 +305,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow".to_string(), dedicated_worker: None, version: 1443253234253454, - debouncing_settings: Default::default(), + debouncing_settings: Default::default(), }) .run_until_complete(&db, false, port) .await @@ -768,4 +768,256 @@ mod job_payload { .await; Ok(()) } + + #[sqlx::test(fixtures("base", "hello"))] + async fn test_dedicated_worker_preprocessor_bun(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let test = || async { + let db = &db; + let job = RunJob::from(JobPayload::ScriptHash { + hash: ScriptHash(123414), + path: "f/system/hello_preprocessor_dedicated_bun".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Bun, + priority: None, + apply_preprocessor: true, + concurrency_settings: + windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: + windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("foo", json!("hello")) + .arg("bar", json!("world")) + .run_until_complete_with(db, false, port, |id| async move { + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(false)); + }) + .await; + + let args = job.args.as_ref().unwrap(); + assert_eq!(args.get("foo"), Some(&json!("hello_preprocessed"))); + assert_eq!(args.get("bar"), Some(&json!("world_preprocessed"))); + assert_eq!( + job.json_result().unwrap(), + json!("Hello hello_preprocessed world_preprocessed") + ); + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(true)); + }; + test_for_versions(VERSION_FLAGS.iter().copied(), test).await; + Ok(()) + } + + #[sqlx::test(fixtures("base", "hello"))] + async fn test_dedicated_worker_preprocessor_python(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let test = || async { + let db = &db; + let job = RunJob::from(JobPayload::ScriptHash { + hash: ScriptHash(123415), + path: "f/system/hello_preprocessor_dedicated_python".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Python3, + priority: None, + apply_preprocessor: true, + concurrency_settings: + windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: + windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("foo", json!("hello")) + .arg("bar", json!("world")) + .run_until_complete_with(db, false, port, |id| async move { + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(false)); + }) + .await; + + let args = job.args.as_ref().unwrap(); + assert_eq!(args.get("foo"), Some(&json!("hello_preprocessed"))); + assert_eq!(args.get("bar"), Some(&json!("world_preprocessed"))); + assert_eq!( + job.json_result().unwrap(), + json!("Hello hello_preprocessed world_preprocessed") + ); + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(true)); + }; + test_for_versions(VERSION_FLAGS.iter().copied(), test).await; + Ok(()) + } + + #[sqlx::test(fixtures("base", "hello"))] + async fn test_dedicated_worker_preprocessor_deno(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let test = || async { + let db = &db; + let job = RunJob::from(JobPayload::ScriptHash { + hash: ScriptHash(123416), + path: "f/system/hello_preprocessor_dedicated_deno".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Deno, + priority: None, + apply_preprocessor: true, + concurrency_settings: + windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: + windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("foo", json!("hello")) + .arg("bar", json!("world")) + .run_until_complete_with(db, false, port, |id| async move { + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(false)); + }) + .await; + + let args = job.args.as_ref().unwrap(); + assert_eq!(args.get("foo"), Some(&json!("hello_preprocessed"))); + assert_eq!(args.get("bar"), Some(&json!("world_preprocessed"))); + assert_eq!( + job.json_result().unwrap(), + json!("Hello hello_preprocessed world_preprocessed") + ); + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(true)); + }; + test_for_versions(VERSION_FLAGS.iter().copied(), test).await; + Ok(()) + } + + #[sqlx::test(fixtures("base", "hello"))] + async fn test_bunnative_preprocessor(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let test = || async { + let db = &db; + let job = RunJob::from(JobPayload::ScriptHash { + hash: ScriptHash(123417), + path: "f/system/hello_preprocessor_bunnative".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Bunnative, + priority: None, + apply_preprocessor: true, + concurrency_settings: + windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: + windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("foo", json!("hello")) + .arg("bar", json!("world")) + .run_until_complete_with(db, false, port, |id| async move { + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(false)); + }) + .await; + + let args = job.args.as_ref().unwrap(); + assert_eq!(args.get("foo"), Some(&json!("hello_preprocessed"))); + assert_eq!(args.get("bar"), Some(&json!("world_preprocessed"))); + assert_eq!( + job.json_result().unwrap(), + json!("Hello hello_preprocessed world_preprocessed") + ); + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(true)); + }; + test_for_versions(VERSION_FLAGS.iter().copied(), test).await; + Ok(()) + } + + #[sqlx::test(fixtures("base", "hello"))] + async fn test_dedicated_worker_preprocessor_bunnative( + db: Pool, + ) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let test = || async { + let db = &db; + let job = RunJob::from(JobPayload::ScriptHash { + hash: ScriptHash(123418), + path: "f/system/hello_preprocessor_dedicated_bunnative".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Bunnative, + priority: None, + apply_preprocessor: true, + concurrency_settings: + windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: + windmill_common::runnable_settings::DebouncingSettings::default(), + }) + .arg("foo", json!("hello")) + .arg("bar", json!("world")) + .run_until_complete_with(db, false, port, |id| async move { + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(false)); + }) + .await; + + let args = job.args.as_ref().unwrap(); + assert_eq!(args.get("foo"), Some(&json!("hello_preprocessed"))); + assert_eq!(args.get("bar"), Some(&json!("world_preprocessed"))); + assert_eq!( + job.json_result().unwrap(), + json!("Hello hello_preprocessed world_preprocessed") + ); + let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id) + .fetch_one(db) + .await + .unwrap(); + assert_eq!(job.preprocessed, Some(true)); + }; + test_for_versions(VERSION_FLAGS.iter().copied(), test).await; + Ok(()) + } } diff --git a/backend/tests/nativets_dedicated.rs b/backend/tests/nativets_dedicated.rs index 4b730b32c7..f13759e5bc 100644 --- a/backend/tests/nativets_dedicated.rs +++ b/backend/tests/nativets_dedicated.rs @@ -80,8 +80,13 @@ mod prewarmed_isolate_tests { let mut results = Vec::new(); for job_args in &jobs { - let mut isolate = - PrewarmedIsolate::spawn("".to_string(), js.clone(), ann.clone(), arg_names.clone()); + let mut isolate = PrewarmedIsolate::spawn( + "".to_string(), + js.clone(), + ann.clone(), + arg_names.clone(), + None, + ); isolate.wait_ready().await.expect("isolate failed to warm"); let args = serde_json::to_string(job_args).unwrap(); @@ -180,8 +185,13 @@ export function main(n: number): number { let ann = default_annotation(); // Pre-warm first isolate - let mut warm = - PrewarmedIsolate::spawn("".to_string(), js.clone(), ann.clone(), arg_names.clone()); + let mut warm = PrewarmedIsolate::spawn( + "".to_string(), + js.clone(), + ann.clone(), + arg_names.clone(), + None, + ); warm.wait_ready() .await .expect("first isolate failed to warm"); @@ -193,8 +203,13 @@ export function main(n: number): number { let executing = warm.start_execution(args); // Pipeline: start pre-warming next isolate while current one runs - warm = - PrewarmedIsolate::spawn("".to_string(), js.clone(), ann.clone(), arg_names.clone()); + warm = PrewarmedIsolate::spawn( + "".to_string(), + js.clone(), + ann.clone(), + arg_names.clone(), + None, + ); let prewarmed_result = executing.wait().await.expect("isolate execution failed"); match prewarmed_result.result { diff --git a/backend/windmill-runtime-nativets/src/dedicated.rs b/backend/windmill-runtime-nativets/src/dedicated.rs index 5533c36262..5838456de9 100644 --- a/backend/windmill-runtime-nativets/src/dedicated.rs +++ b/backend/windmill-runtime-nativets/src/dedicated.rs @@ -56,6 +56,7 @@ impl PrewarmedIsolate { js_code: String, ann: NativeAnnotation, arg_names: Vec, + entrypoint: Option, ) -> Self { let (args_tx, args_rx) = tokio::sync::oneshot::channel::(); let (result_tx, result_rx) = tokio::sync::oneshot::channel::(); @@ -98,7 +99,7 @@ impl PrewarmedIsolate { }); let exec_result = tokio::select! { - r = execute_main(&mut js_runtime, None, false, None) => r, + r = execute_main(&mut js_runtime, entrypoint.as_deref(), false, None) => r, _ = memory_limit_rx.recv() => { Err(ExecuteError::Script("Memory limit reached, killing isolate".to_string())) } diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 16900dad8b..2fd194229a 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -80,10 +80,12 @@ pub const BUN_DEDICATED_WORKER_ARGS: &[&str] = &["run", "-i", "--prefer-offline" /// - `arg_names`: The argument names for the main function (e.g., ["x", "y"]) /// - `main_import`: The import path for the main module (e.g., "./main.ts") /// - `date_conversions`: Optional date conversion statements for Datetime args +/// - `preprocessor_spread`: If the script has a preprocessor function, the comma-separated arg names for it pub fn generate_dedicated_worker_wrapper( arg_names: &[&str], main_import: &str, date_conversions: Option<&str>, + preprocessor_spread: Option<&str>, ) -> String { let spread = arg_names.join(","); let dates = date_conversions.unwrap_or(""); @@ -94,6 +96,36 @@ pub fn generate_dedicated_worker_wrapper( "" }; + let preprocessor_logic = if let Some(pre_spread) = preprocessor_spread { + format!( + r#" + if (rawLine.startsWith("preprocess:")) {{ + const preInput = rawLine.slice("preprocess:".length); + const parsedArgs = JSON.parse(preInput); + if (Main.preprocessor === undefined || typeof Main.preprocessor !== 'function') {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }})); + continue; + }} + try {{ + function preArgsObjToArr({{ {pre_spread} }}) {{ + return [ {pre_spread} ]; + }} + const preprocessedArgs = await Main.preprocessor(...preArgsObjToArr(parsedArgs)); + console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value)); + // Now call main with preprocessed args + const mainArgs = getArgs(JSON.stringify(preprocessedArgs ?? {{}})); + const res = await Main.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: rawLine }})); + }} + continue; + }}"# + ) + } else { + String::new() + }; + format!( r#" import * as Main from "{main_import}"; @@ -114,15 +146,17 @@ function getArgs(line) {{ for await (const line of Readline.createInterface({{ input: process.stdin }})) {{ {print_lines} - if (line === "end") {{ + const rawLine = line; + if (rawLine === "end") {{ process.exit(0); }} + {preprocessor_logic} try {{ - const args = getArgs(line); + const args = getArgs(rawLine); const res = await Main.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: line }})); + console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: rawLine }})); }} }} "# @@ -1648,6 +1682,66 @@ try {{ append_logs(&job.id, &job.workspace_id, format!("{init_logs}\n"), conn).await; + if apply_preprocessor { + // First pass: run preprocessor function + let pre_result = crate::js_eval::eval_fetch_timeout( + env_code.clone(), + inner_content.to_string(), + js_code.clone(), + job_args, + Some("preprocessor".to_string()), + job.id, + job.timeout, + conn, + mem_peak, + canceled_by, + worker_name, + &job.workspace_id, + false, + occupancy_metrics, + None, + has_stream, + ) + .await?; + + let preprocessed: HashMap> = + serde_json::from_str(pre_result.get()).map_err(|e| { + error::Error::internal_err(format!( + "error deserializing preprocessed args: {e:#}" + )) + })?; + *new_args = Some(preprocessed.clone()); + + // Second pass: run main with preprocessed args + let preprocessed_json = sqlx::types::Json(preprocessed); + let stream_notifier = StreamNotifier::new(conn, job); + + let result = crate::js_eval::eval_fetch_timeout( + env_code, + inner_content.to_string(), + js_code, + Some(&preprocessed_json), + job.script_entrypoint_override.clone(), + job.id, + job.timeout, + conn, + mem_peak, + canceled_by, + worker_name, + &job.workspace_id, + false, + occupancy_metrics, + stream_notifier, + has_stream, + ) + .await?; + tracing::info!( + "Executed native code (with preprocessor) in {}ms", + started_at.elapsed().as_millis() + ); + return Ok(result); + } + let stream_notifier = StreamNotifier::new(conn, job); let result = crate::js_eval::eval_fetch_timeout( @@ -2735,13 +2829,35 @@ async fn handle_dedicated_bunnative( let env_code = env_code.to_string(); let js_code = js_code.to_string(); + let pre_arg_names: Option> = windmill_parser_ts::parse_deno_signature( + inner_content, + true, + false, + Some("preprocessor".to_string()), + ) + .ok() + .filter(|sig| !sig.args.is_empty()) + .map(|sig| sig.args.into_iter().map(|x| x.name).collect()); + let mut warm = PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), arg_names.clone(), + None, ); + // Pre-warm preprocessor isolate if the script has a preprocessor + let mut pre_warm = pre_arg_names.as_ref().map(|pre_names| { + PrewarmedIsolate::spawn( + env_code.clone(), + js_code.clone(), + ann.clone(), + pre_names.clone(), + Some("preprocessor".to_string()), + ) + }); + let init_log = format!("dedicated worker nativets: {worker_name}\n\n"); let alive = true; let mut killpill_rx = killpill_rx; @@ -2774,108 +2890,138 @@ async fn handle_dedicated_bunnative( "{}".to_string() }; - if let Err(e) = warm.wait_ready().await { - tracing::error!("pre-warmed isolate failed during init: {e}"); - let result = Arc::new(to_raw_value(&serde_json::json!({ - "message": format!("isolate init failed: {e}"), - "name": "Error", - }))); - append_logs(&id, &job.workspace_id, init_log.clone(), &db.into()).await; - job_completed_tx.send_job(JobCompleted { - job: MiniCompletedJob::from(job), - result, - result_columns: None, - mem_peak: 0, - canceled_by: None, - success: false, - cached_res_path: None, - token: token.to_string(), - duration: None, - preprocessed_args: None, - has_stream: Some(false), - from_cache: None, - flow_runners, - done_tx, - }, true).await?; + // Run the job: preprocess if needed, then execute main. + // Uses a labeled block to unify error handling with a single JobCompleted send. + let (result, success, preprocessed_args, logs) = 'job: { + if let Err(e) = warm.wait_ready().await { + break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": format!("isolate init failed: {e}"), "name": "Error"}))), + false, None, init_log.clone(), + ); + } + + let needs_preprocessing = job.preprocessed == Some(false); + + let (main_args, preprocessed) = if needs_preprocessing { + let Some(ref pre_names) = pre_arg_names else { + break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": "preprocessor function is missing", "name": "Error"}))), + false, None, init_log.clone(), + ); + }; + + let mut pre_isolate = pre_warm.take().unwrap_or_else(|| { + PrewarmedIsolate::spawn( + env_code.clone(), + js_code.clone(), + ann.clone(), + pre_names.clone(), + Some("preprocessor".to_string()), + ) + }); + if let Err(e) = pre_isolate.wait_ready().await { + break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": format!("preprocessor isolate init failed: {e}"), "name": "Error"}))), + false, None, init_log.clone(), + ); + } + + let pre_executing = pre_isolate.start_execution(args.clone()); + // Pipeline: start pre-warming the next preprocessor isolate + pre_warm = Some(PrewarmedIsolate::spawn( + env_code.clone(), + js_code.clone(), + ann.clone(), + pre_names.clone(), + Some("preprocessor".to_string()), + )); + + let pre_result = match pre_executing.wait().await { + Ok(r) => r, + Err(e) => break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": format!("preprocessor failed: {e}"), "name": "Error"}))), + false, None, init_log.clone(), + ), + }; + if !pre_result.logs.is_empty() { + append_logs(&id, &job.workspace_id, pre_result.logs, &db.into()).await; + } + let raw = match pre_result.result { + Ok(r) => r, + Err(e) => break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": format!("preprocessor failed: {e}"), "name": "Error"}))), + false, None, init_log.clone(), + ), + }; + + let preprocessed: HashMap> = match serde_json::from_str(raw.get()) { + Ok(v) => v, + Err(e) => break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": format!("error deserializing preprocessed args: {e:#}"), "name": "Error"}))), + false, None, init_log.clone(), + ), + }; + let main_args = serde_json::to_string(&preprocessed) + .unwrap_or_else(|_| "{}".to_string()); + (main_args, Some(preprocessed)) + } else { + (args, None) + }; + + let executing = warm.start_execution(main_args); + // Pipeline: start pre-warming the next main isolate warm = PrewarmedIsolate::spawn( env_code.clone(), js_code.clone(), ann.clone(), arg_names.clone(), + None, ); - continue; - } - let executing = warm.start_execution(args); + let main_result = match executing.wait().await { + Ok(r) => r, + Err(e) => break 'job ( + Arc::new(to_raw_value(&serde_json::json!({"message": format!("{e}"), "name": "Error"}))), + false, preprocessed, init_log.clone(), + ), + }; - warm = PrewarmedIsolate::spawn( - env_code.clone(), - js_code.clone(), - ann.clone(), - arg_names.clone(), - ); - - match executing.wait().await { - Ok(prewarmed_result) => { - let mut logs = init_log.clone(); - if !prewarmed_result.logs.is_empty() { - logs.push_str(&prewarmed_result.logs); - } - append_logs(&id, &job.workspace_id, logs, &db.into()).await; - - let (result, success) = match prewarmed_result.result { - Ok(raw) => (Arc::new(raw), true), - Err(e) => ( - Arc::new(to_raw_value(&serde_json::json!({ - "message": e, - "name": "Error", - }))), - false, - ), - }; - - job_completed_tx.send_job(JobCompleted { - job: MiniCompletedJob::from(job), - result, - result_columns: None, - mem_peak: 0, - canceled_by: None, - success, - cached_res_path: None, - token: token.to_string(), - duration: None, - preprocessed_args: None, - has_stream: Some(false), - from_cache: None, - flow_runners, - done_tx, - }, true).await?; + let mut logs = init_log.clone(); + if !main_result.logs.is_empty() { + logs.push_str(&main_result.logs); } - Err(e) => { - tracing::error!("isolate execution failed: {e}"); - append_logs(&id, &job.workspace_id, init_log.clone(), &db.into()).await; - let result = Arc::new(to_raw_value(&serde_json::json!({ - "message": format!("{e}"), - "name": "Error", - }))); - job_completed_tx.send_job(JobCompleted { - job: MiniCompletedJob::from(job), - result, - result_columns: None, - mem_peak: 0, - canceled_by: None, - success: false, - cached_res_path: None, - token: token.to_string(), - duration: None, - preprocessed_args: None, - has_stream: Some(false), - from_cache: None, - flow_runners: None, - done_tx: None, - }, true).await?; - } - } + + let (result, success) = match main_result.result { + Ok(raw) => (Arc::new(raw), true), + Err(e) => ( + Arc::new(to_raw_value(&serde_json::json!({ + "message": e, + "name": "Error", + }))), + false, + ), + }; + + (result, success, preprocessed, logs) + }; + + append_logs(&id, &job.workspace_id, logs, &db.into()).await; + job_completed_tx.send_job(JobCompleted { + job: MiniCompletedJob::from(job), + result, + result_columns: None, + mem_peak: 0, + canceled_by: None, + success, + cached_res_path: None, + token: token.to_string(), + duration: None, + preprocessed_args, + has_stream: Some(false), + from_cache: None, + flow_runners, + done_tx, + }, true).await?; } else { tracing::debug!("job channel closed for nativets dedicated worker"); break; @@ -3115,6 +3261,18 @@ pub async fn start_worker( .join("\n"); let arg_names: Vec<&str> = args.iter().map(|x| x.name.as_str()).collect(); + + // Parse preprocessor signature if it exists + let pre_spread = windmill_parser_ts::parse_deno_signature( + inner_content, + true, + false, + Some("preprocessor".to_string()), + ) + .ok() + .filter(|sig| !sig.args.is_empty()) + .map(|sig| sig.args.into_iter().map(|x| x.name).join(",")); + // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); // we cannot use Bun.read and Bun.write because it results in an EBADF error on cloud @@ -3128,7 +3286,12 @@ pub async fn start_worker( } else { Some(dates.as_str()) }; - let wrapper_content = generate_dedicated_worker_wrapper(&arg_names, main_import, dates_opt); + let wrapper_content = generate_dedicated_worker_wrapper( + &arg_names, + main_import, + dates_opt, + pre_spread.as_deref(), + ); write_file(job_dir, "wrapper.mjs", &wrapper_content)?; } diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index d9f6785eb6..d512184e39 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -655,11 +655,61 @@ pub async fn start_worker( .join("\n"); let spread = args.into_iter().map(|x| x.name).join(","); + + // Parse preprocessor signature if it exists + let pre_spread = windmill_parser_ts::parse_deno_signature( + inner_content, + true, + false, + Some("preprocessor".to_string()), + ) + .ok() + .filter(|sig| !sig.args.is_empty()) + .map(|sig| sig.args.into_iter().map(|x| x.name).join(",")); + + let preprocessor_import = if pre_spread.is_some() { + r#"import { preprocessor } from "./main.ts";"# + } else { + "" + }; + + let preprocessor_logic = if let Some(ref pre_spread) = pre_spread { + format!( + r#" + if (line.startsWith("preprocess:")) {{ + const preInput = line.slice("preprocess:".length); + const parsedArgs = JSON.parse(preInput); + if (typeof preprocessor !== 'function') {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: "preprocessor function is missing", name: "Error" }}) + '\n'); + continue; + }} + try {{ + function preArgsObjToArr({{ {pre_spread} }}: any) {{ + return [ {pre_spread} ]; + }} + const preprocessedArgs: any = await preprocessor(...preArgsObjToArr(parsedArgs)); + console.log("wm_res[preprocessed_args]:" + JSON.stringify(preprocessedArgs ?? {{}}, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + // Now call main with preprocessed args + let {{ {spread} }} = preprocessedArgs ?? {{}}; + {dates} + let res: any = await main(...[ {spread} ]); + console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + }} catch (e) {{ + console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }}) + '\n'); + }} + continue; + }}"# + ) + } else { + String::new() + }; + // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); // we cannot use Bun.read and Bun.write because it results in an EBADF error on cloud let wrapper_content: String = format!( r#" import {{ main }} from "./main.ts"; +{preprocessor_import} BigInt.prototype.toJSON = function () {{ return this.toString(); @@ -678,6 +728,7 @@ for await (const chunk of Deno.stdin.readable) {{ exit = true; break; }} + {preprocessor_logic} try {{ let {{ {spread} }} = JSON.parse(line) {dates} diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index c74feb0a0a..3a040234b6 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -680,6 +680,7 @@ pub async fn handle_python_job( String::new() }; let main_override = main_name.unwrap_or_else(|| "main".to_string()); + let res_to_json_body = python_res_to_json_body(postprocessor); let wrapper_content: String = if is_wac_v2 { format!( r#" @@ -768,19 +769,7 @@ replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\*\\u0000|Infinity|\-Infinity) result_json = os.path.join(os.path.abspath(os.path.dirname(__file__)), "result.json") def res_to_json(res, typ): - if typ.__name__ == 'DataFrame': - if typ.__module__ == 'pandas.core.frame': - res = res.values.tolist() - elif typ.__module__ == 'polars.dataframe.frame': - res = res.rows() - elif typ.__name__ == 'bytes': - res = to_b_64(res) - elif typ.__name__ == 'dict': - for k, v in res.items(): - if type(v).__name__ == 'bytes': - res[k] = to_b_64(v) - unprocessed = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '') - return {postprocessor} +{res_to_json_body} try: {preprocessor} @@ -1027,6 +1016,30 @@ mount {{ Ok(result) } +/// Generate Python code to spread preprocessor args from `kwargs` into `pre_args`. +/// The `indent` parameter controls the join separator for multi-arg spreads. +fn python_preprocessor_spread(sig: windmill_parser::MainArgSignature, indent: &str) -> String { + if sig.star_kwargs { + "pre_args = kwargs".to_string() + } else { + sig.args + .into_iter() + .map(|x| { + let name = &x.name; + if x.default.is_none() { + format!("pre_args[\"{name}\"] = kwargs.get(\"{name}\")") + } else { + format!( + r#"pre_args["{name}"] = kwargs.get("{name}") +{indent}if pre_args["{name}"] is None: +{indent} del pre_args["{name}"]"# + ) + } + }) + .join(&format!("\n{indent}")) + } +} + async fn prepare_wrapper( job_dir: &str, job_flow_step_id: Option<&str>, @@ -1193,31 +1206,7 @@ async fn prepare_wrapper( .join("\n ") }; - let pre_spread = if let Some(pre_sig) = pre_sig { - let spread = if pre_sig.star_kwargs { - "pre_args = kwargs".to_string() - } else { - pre_sig - .args - .into_iter() - .map(|x| { - let name = &x.name; - if x.default.is_none() { - format!("pre_args[\"{name}\"] = kwargs.get(\"{name}\")") - } else { - format!( - r#"pre_args["{name}"] = kwargs.get("{name}") - if pre_args["{name}"] is None: - del pre_args["{name}"]"# - ) - } - }) - .join("\n ") - }; - Some(spread) - } else { - None - }; + let pre_spread = pre_sig.map(|sig| python_preprocessor_spread(sig, " ")); let module_dir_dot = dirs.replace("/", ".").replace("-", "_"); @@ -2338,6 +2327,26 @@ This is not normal behavior, please make sure all workers have enough memory.\n }; } +/// Python function body for `res_to_json(res, typ)`. +/// Handles DataFrame, bytes, dict coercion + JSON serialization. +fn python_res_to_json_body(postprocessor: &str) -> String { + format!( + r#" if typ.__name__ == 'DataFrame': + if typ.__module__ == 'pandas.core.frame': + res = res.values.tolist() + elif typ.__module__ == 'polars.dataframe.frame': + res = res.rows() + elif typ.__name__ == 'bytes': + res = to_b_64(res) + elif typ.__name__ == 'dict': + for k, v in res.items(): + if type(v).__name__ == 'bytes': + res[k] = to_b_64(v) + unprocessed = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '') + return {postprocessor}"# + ) +} + // Returns code snippet that needs to be injected into wrapper to post-process results or leave unprocessed fn get_result_postprocessor<'a>(skip: bool) -> &'a str { if skip { @@ -2433,6 +2442,16 @@ pub async fn start_worker( _, ) = prepare_wrapper(job_dir, None, None, None, inner_content, script_path).await?; + // Parse preprocessor signature if the script has one + let pre_spread = windmill_parser_py::parse_python_signature( + inner_content, + Some("preprocessor".to_string()), + false, + ) + .ok() + .filter(|sig| !sig.args.is_empty()) + .map(|sig| python_preprocessor_spread(sig, " ")); + { let postprocessor = get_result_postprocessor(annotations.skip_result_postprocessing); let indented_transforms = transforms @@ -2441,6 +2460,41 @@ pub async fn start_worker( .collect::>() .join("\n"); + let preprocessor_logic = if let Some(ref pre_spread) = pre_spread { + format!( + r#" + if line.startswith('preprocess:'): + pre_input = line[len('preprocess:'):] + kwargs = json.loads(pre_input, strict=False) + if not hasattr(inner_script, 'preprocessor') or not callable(inner_script.preprocessor): + err_json = json.dumps({{"message": "preprocessor function is missing", "name": "Error"}}, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[error]:" + err_json + "\n") + sys.stdout.flush() + continue + try: + pre_args = {{}} + {pre_spread} + for k, v in list(pre_args.items()): + if v == '': + del pre_args[k] + preprocessed_kwargs = inner_script.preprocessor(**pre_args) + preprocessed_json = json.dumps(preprocessed_kwargs, separators=(',', ':'), default=str).replace('\n', '') + sys.stdout.write("wm_res[preprocessed_args]:" + preprocessed_json + "\n") + transform_and_run(preprocessed_kwargs) + 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 +"# + ) + } else { + String::new() + }; + + let res_to_json_body = python_res_to_json_body(postprocessor); let wrapper_content: String = format!( r#" import json @@ -2458,37 +2512,32 @@ def to_b_64(v: bytes): b64 = base64.b64encode(v) return b64.decode('ascii') -replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\u0000|Infinity|\-Infinity)') -sys.stdout.write('start\n') +def res_to_json(res, typ): +{res_to_json_body} -for line in sys.stdin: - if line == 'end\n': - break - kwargs = json.loads(line, strict=False) +def transform_and_run(kwargs): args = {{}} {indented_transforms} {spread} for k, v in list(args.items()): if v == '': del args[k] + res = inner_script.main(**args) + typ = type(res) + res_json = res_to_json(res, typ) + sys.stdout.write("wm_res[success]:" + res_json + "\n") +replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\u0000|Infinity|\-Infinity)') +sys.stdout.write('start\n') + +for line in sys.stdin: + if line == 'end\n': + break + line = line.strip() + {preprocessor_logic} + kwargs = json.loads(line, strict=False) try: - res = inner_script.main(**args) - typ = type(res) - if typ.__name__ == 'DataFrame': - if typ.__module__ == 'pandas.core.frame': - res = res.values.tolist() - elif typ.__module__ == 'polars.dataframe.frame': - res = res.rows() - elif typ.__name__ == 'bytes': - res = to_b_64(res) - elif typ.__name__ == 'dict': - for k, v in res.items(): - if type(v).__name__ == 'bytes': - res[k] = to_b_64(v) - unprocessed = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '') - res_json = {postprocessor} - sys.stdout.write("wm_res[success]:" + res_json + "\n") + transform_and_run(kwargs) except BaseException as e: exc_type, exc_value, exc_traceback = sys.exc_info() tb = traceback.format_tb(exc_traceback)