diff --git a/backend/parsers/windmill-parser-ts/src/lib.rs b/backend/parsers/windmill-parser-ts/src/lib.rs index e0b1640ffb..e95a7848b7 100644 --- a/backend/parsers/windmill-parser-ts/src/lib.rs +++ b/backend/parsers/windmill-parser-ts/src/lib.rs @@ -128,7 +128,11 @@ pub fn parse_expr_for_ids(code: &str) -> anyhow::Result> { Ok(visitor.idents.into_iter().collect()) } -pub fn parse_deno_signature(code: &str, skip_dflt: bool) -> anyhow::Result { +pub fn parse_deno_signature( + code: &str, + skip_dflt: bool, + main_override: Option, +) -> anyhow::Result { let cm: Lrc = Default::default(); let fm = cm.new_source_file(FileName::Custom("main.ts".into()), code.into()); let lexer = Lexer::new( @@ -154,11 +158,12 @@ pub fn parse_deno_signature(code: &str, skip_dflt: bool) -> anyhow::Result Some(function.params), + })) if &sym.to_string() == &main_name => Some(function.params), _ => None, }); diff --git a/backend/parsers/windmill-parser-wasm/src/lib.rs b/backend/parsers/windmill-parser-wasm/src/lib.rs index 0ecfc32727..e9c491f850 100644 --- a/backend/parsers/windmill-parser-wasm/src/lib.rs +++ b/backend/parsers/windmill-parser-wasm/src/lib.rs @@ -13,7 +13,7 @@ fn wrap_sig(r: anyhow::Result) -> String { #[wasm_bindgen] pub fn parse_deno(code: &str) -> String { - wrap_sig(windmill_parser_ts::parse_deno_signature(code, false)) + wrap_sig(windmill_parser_ts::parse_deno_signature(code, false, None)) } #[wasm_bindgen] diff --git a/backend/parsers/windmill-parser-wasm/tests/wasm.rs b/backend/parsers/windmill-parser-wasm/tests/wasm.rs index 191e64655a..84d358c636 100644 --- a/backend/parsers/windmill-parser-wasm/tests/wasm.rs +++ b/backend/parsers/windmill-parser-wasm/tests/wasm.rs @@ -1,8 +1,8 @@ use serde_json::json; use wasm_bindgen_test::wasm_bindgen_test; use windmill_parser::{Arg, MainArgSignature, ObjectProperty, Typ}; +use windmill_parser_bash::parse_powershell_sig; use windmill_parser_ts::{parse_deno_signature, parse_expr_for_ids, parse_expr_for_imports}; -use windmill_parser_bash::{parse_powershell_sig}; #[wasm_bindgen_test] fn test_parse_deno_sig() -> anyhow::Result<()> { @@ -17,7 +17,7 @@ export function main(test1?: string, test2: string = \"burkina\", } "; assert_eq!( - parse_deno_signature(code, false)?, + parse_deno_signature(code, false, None)?, MainArgSignature { star_args: false, star_kwargs: false, @@ -135,7 +135,7 @@ export function main(test2 = \"burkina\", } "; assert_eq!( - parse_deno_signature(code, false)?, + parse_deno_signature(code, false, None)?, MainArgSignature { star_args: false, star_kwargs: false, @@ -204,7 +204,7 @@ export function main(foo: FooBar, {a, b}: FooBar, {c, d}: FooBar = {a: \"foo\", } "; assert_eq!( - parse_deno_signature(code, false)?, + parse_deno_signature(code, false, None)?, MainArgSignature { star_args: false, star_kwargs: false, @@ -245,7 +245,7 @@ export function main(foo: (\"foo\" | \"bar\")[]) { } "; assert_eq!( - parse_deno_signature(code, false)?, + parse_deno_signature(code, false, None)?, MainArgSignature { star_args: false, star_kwargs: false, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 6b3c83b38b..de238e1d71 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -565,7 +565,7 @@ pub async fn add_completed_job< // tracing::error!("2 {:?}", start.elapsed()); if !queued_job.is_flow_step { - if _duration > 1000 { + if _duration > 500 { if let Err(e) = sqlx::query!( "UPDATE completed_job SET flow_status = q.flow_status FROM queue q WHERE completed_job.id = $1 AND q.id = $1 AND q.workspace_id = $2 AND completed_job.workspace_id = $2", &queued_job.id, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 3729151497..3c11018c9e 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -13,8 +13,9 @@ use crate::common::build_envs_map; use crate::{ common::{ - create_args_and_out_file, get_reserved_variables, handle_child, parse_npm_config, - read_result, set_logs, start_child_process, write_file, write_file_binary, + create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, + parse_npm_config, read_result, set_logs, start_child_process, write_file, + write_file_binary, }, AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, PATH_ENV, @@ -356,6 +357,8 @@ pub async fn handle_bun_job( let annotation = get_annotation(inner_content); + let main_override = get_main_override(job.args.as_ref()); + #[cfg(not(feature = "enterprise"))] if annotation.nodejs_mode || annotation.npm_mode { return Err(error::Error::ExecutionErr( @@ -446,7 +449,9 @@ pub async fn handle_bun_job( let write_wrapper_f = async { // let mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; + let args = + windmill_parser_ts::parse_deno_signature(inner_content, true, main_override.clone())? + .args; let dates = args .iter() .enumerate() @@ -463,9 +468,11 @@ pub async fn handle_bun_job( let spread = 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 + let main_name = main_override.unwrap_or("main".to_string()); + let wrapper_content: String = format!( r#" -import {{ main }} from "./main.ts"; +import {{ {main_name} }} from "./main.ts"; const fs = require('fs/promises'); @@ -478,7 +485,7 @@ BigInt.prototype.toJSON = function () {{ {dates} async function run() {{ - let res: any = await main(...args); + let res: any = await {main_name}(...args); const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); await fs.writeFile("result.json", res_json); process.exit(0); @@ -786,6 +793,7 @@ pub async fn start_worker( let _ = write_file(job_dir, "main.ts", inner_content).await?; let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(&base_internal_url).await; + let context = variables::get_reserved_variables( w_id, &token, @@ -872,7 +880,7 @@ pub async fn start_worker( { // let mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; + let args = windmill_parser_ts::parse_deno_signature(inner_content, true, None)?.args; let dates = args .iter() .enumerate() diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 42230b34a6..22c2809145 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -11,10 +11,12 @@ use regex::Regex; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use serde_json::{json, Value}; +use sqlx::types::Json; use sqlx::{Pool, Postgres}; use tokio::process::Command; use tokio::{fs::File, io::AsyncReadExt}; use windmill_common::error::to_anyhow; +use windmill_common::jobs::ENTRYPOINT_OVERRIDE; #[cfg(feature = "parquet")] use windmill_common::s3_helpers::{ get_etag_or_empty, AzureBlobResource, LargeFileStorage, ObjectStoreResource, S3Object, @@ -419,6 +421,16 @@ pub async fn build_envs_map(context: Vec) -> HashMap>>>) -> Option { + return args + .map(|x| { + x.0.get(ENTRYPOINT_OVERRIDE) + .map(|x| x.get().to_string().replace("\"", "")) + }) + .flatten(); +} + async fn get_mem_peak(pid: Option, nsjail: bool) -> i32 { if pid.is_none() { return -1; @@ -1001,8 +1013,6 @@ fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize) { } } - - pub async fn hash_args( _db: &DB, _client: &AuthedClient, @@ -1161,16 +1171,18 @@ pub async fn get_cached_resource_value_if_valid( } else { get_workspace_s3_resource_path(_db, &client, _workspace_id, _job_id).await }; - + if !s3_etags.is_empty() && object_store_resource_opt.is_none() { tracing::warn!("Cached result references s3 files that are not retrievable anymore because the workspace S3 resource can't be fetched. Cache will be invalidated"); return None; } for (s3_file_key, s3_file_etag) in s3_etags { if let Some(object_store_resource) = object_store_resource_opt.clone() { - let etag = - get_etag_or_empty(&object_store_resource, S3Object { s3: s3_file_key.clone() }) - .await; + let etag = get_etag_or_empty( + &object_store_resource, + S3Object { s3: s3_file_key.clone() }, + ) + .await; if etag.is_none() || etag.clone().unwrap() != s3_file_etag { tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag); return None; @@ -1196,7 +1208,7 @@ pub async fn save_in_cache( let s3_etags = arg_value_hash_additions(db, _client, job.workspace_id.as_str(), &job.id, r).await; - #[cfg(feature = "parquet")] + #[cfg(feature = "parquet")] let s3_etags = if s3_etags.is_empty() { None } else { @@ -1206,12 +1218,7 @@ pub async fn save_in_cache( #[cfg(not(feature = "parquet"))] let s3_etags = None; - - let store_cache_resource = CachedResource { - expire, - s3_etags, - value: r.clone(), - }; + let store_cache_resource = CachedResource { expire, s3_etags, value: r.clone() }; let raw_json = sqlx::types::Json(store_cache_resource); if let Err(e) = sqlx::query!( diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 8ce54d047a..ef45b777ab 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -7,8 +7,8 @@ use windmill_queue::CanceledBy; use crate::{ common::{ - create_args_and_out_file, get_reserved_variables, handle_child, parse_npm_config, - read_result, set_logs, start_child_process, write_file, + create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, + parse_npm_config, read_result, set_logs, start_child_process, write_file, }, AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV, NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV, @@ -185,11 +185,15 @@ pub async fn handle_deno_job( Ok(()) as error::Result<()> }; + let main_override = get_main_override(job.args.as_ref()); + let write_main_f = write_file(job_dir, "main.ts", inner_content); let write_wrapper_f = async { // let mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; + let args = + windmill_parser_ts::parse_deno_signature(inner_content, true, main_override.clone())? + .args; let dates = args .iter() .enumerate() @@ -204,10 +208,11 @@ pub async fn handle_deno_job( .join("\n"); let spread = args.into_iter().map(|x| x.name).join(","); + let main_name = main_override.unwrap_or("main".to_string()); // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); let wrapper_content: String = format!( r#" -import {{ main }} from "./main.ts"; +import {{ {main_name} }} from "./main.ts"; const args = await Deno.readTextFile("args.json") .then(JSON.parse) @@ -219,7 +224,7 @@ BigInt.prototype.toJSON = function () {{ {dates} async function run() {{ - let res: any = await main(...args); + let res: any = await {main_name}(...args); const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); await Deno.writeTextFile("result.json", res_json); Deno.exit(0); @@ -447,7 +452,7 @@ pub async fn start_worker( { // let mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; + let args = windmill_parser_ts::parse_deno_signature(inner_content, true, None)?.args; let dates = args .iter() .enumerate() diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 886e52baf8..377ad29658 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -599,7 +599,7 @@ pub async fn eval_fetch_timeout( ) -> anyhow::Result<(Box, String)> { let (sender, mut receiver) = oneshot::channel::(); - let parsed_args = windmill_parser_ts::parse_deno_signature(&ts_expr, true)?.args; + let parsed_args = windmill_parser_ts::parse_deno_signature(&ts_expr, true, None)?.args; let spread = parsed_args .into_iter() .map(|x| { diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 818f7ee8ff..e66ef04536 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -14,7 +14,7 @@ use uuid::Uuid; use windmill_common::ee::{get_license_plan, LicensePlan}; use windmill_common::{ error::{self, Error}, - jobs::{QueuedJob, ENTRYPOINT_OVERRIDE}, + jobs::QueuedJob, utils::calculate_hash, worker::WORKER_CONFIG, DB, @@ -55,8 +55,8 @@ use crate::S3_CACHE_BUCKET; use crate::{ common::{ - create_args_and_out_file, get_reserved_variables, handle_child, read_result, set_logs, - start_child_process, write_file, + create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, + read_result, set_logs, start_child_process, write_file, }, AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, HTTPS_PROXY, HTTP_PROXY, LOCK_CACHE_DIR, NO_PROXY, NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL, @@ -463,12 +463,7 @@ async fn prepare_wrapper( String, Option, )> { - let main_override = args - .map(|x| { - x.0.get(ENTRYPOINT_OVERRIDE) - .map(|x| x.get().to_string().replace("\"", "")) - }) - .flatten(); + let main_override = get_main_override(args); let relative_imports = RELATIVE_IMPORT_REGEX.is_match(&inner_content); diff --git a/typescript-client/client.ts b/typescript-client/client.ts index f5d37a7940..cfe035c7d1 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -110,6 +110,155 @@ export async function getRootJobId(jobId?: string): Promise { return await JobService.getRootJobId({ workspace, id: jobId }); } +export async function runScript( + path: string | null = null, + hash_: string | null = null, + args: Record | null = null, + verbose: boolean = false +): Promise { + args = args || {}; + + if (verbose) { + console.info(`running \`${path}\` synchronously with args:`, args); + } + + const jobId = await runScriptAsync(path, hash_, args); + return await waitJob(jobId, verbose); +} + +export async function waitJob( + jobId: string, + verbose: boolean = false, + assertResultIsNotNull: boolean = false +): Promise { + while (true) { + // Implement your HTTP request logic here to get job result + const resultRes = await getResult(jobId); + + const started = resultRes.started; + const completed = resultRes.completed; + const success = resultRes.success; + + if (!started && verbose) { + console.info(`job ${jobId} has not started yet`); + } + + if (completed) { + const result = resultRes.result; + if (success) { + if (result === null && assertResultIsNotNull) { + throw new Error("Result was null"); + } + return result; + } else { + const error = result.error; + throw new Error(`Job ${jobId} was not successful: ${error}`); + } + } + + if (verbose) { + console.info(`sleeping 0.5 seconds for jobId: ${jobId}`); + } + + await new Promise((resolve) => setTimeout(resolve, 500)); + } +} + +export async function getResult(jobId: string): Promise { + !clientSet && setClient(); + const workspace = getWorkspace(); + return await JobService.getCompletedJobResult({ workspace, id: jobId }); +} + +export async function getResultMaybe(jobId: string): Promise { + !clientSet && setClient(); + const workspace = getWorkspace(); + return await JobService.getCompletedJobResultMaybe({ workspace, id: jobId }); +} +const STRIP_COMMENTS = + /(\/\/.*$)|(\/\*[\s\S]*?\*\/)|(\s*=[^,\)]*(('(?:\\'|[^'\r\n])*')|("(?:\\"|[^"\r\n])*"))|(\s*=[^,\)]*))/gm; +const ARGUMENT_NAMES = /([^\s,]+)/g; +function getParamNames(func: Function): string[] { + const fnStr = func.toString().replace(STRIP_COMMENTS, ""); + let result: string[] | null = fnStr + .slice(fnStr.indexOf("(") + 1, fnStr.indexOf(")")) + .match(ARGUMENT_NAMES); + if (result === null) result = []; + return result; +} + +export function task(f: (_: P) => T): (_: P) => Promise { + return async (...y) => { + const args: Record = {}; + const paramNames = getParamNames(f); + y.forEach((x, i) => (args[paramNames[i]] = x)); + let req = await fetch( + `http://0.0.0.0:8000/api/w/foo/jobs/run/workflow_as_code/${getEnv( + "WM_JOB_ID" + )}/${f.name}`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${getEnv("WM_TOKEN")}`, + }, + body: JSON.stringify({ args }), + } + ); + let jobId = await req.text(); + return await waitJob(jobId); + }; +} + +export async function runScriptAsync( + path: string | null, + hash_: string | null, + args: Record | null, + scheduledInSeconds: number | null = null +): Promise { + // Create a script job and return its job id. + if (path && hash_) { + throw new Error("path and hash_ are mutually exclusive"); + } + args = args || {}; + const params: Record = {}; + + if (scheduledInSeconds) { + params["scheduled_in_secs"] = scheduledInSeconds; + } + + let parentJobId = getEnv("WM_JOB_ID"); + if (parentJobId !== undefined) { + params["parent_job"] = parentJobId; + } + + let rootJobId = getEnv("WM_ROOT_FLOW_JOB_ID"); + if (rootJobId != undefined) { + params["root_job"] = rootJobId; + } + + let endpoint: string; + if (path) { + endpoint = `/w/${getWorkspace()}/jobs/run/p/${path}`; + } else if (hash_) { + endpoint = `/w/${getWorkspace()}/jobs/run/h/${hash_}`; + } else { + throw new Error("path or hash_ must be provided"); + } + let url = new URL(OpenAPI.BASE + endpoint); + url.search = new URLSearchParams(params).toString(); + + return fetch(url, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${OpenAPI.TOKEN}`, + }, + body: JSON.stringify(args), + }) + .then((res) => res.json()) + .then((data) => data.id); +} /** * Resolve a resource value in case the default value was picked because the input payload was undefined * @param obj resource value or path of the resource under the format `$res:path`