diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 352a6e7fb2..069502d549 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -4873,6 +4873,12 @@ dependencies = [ "malachite-nz", ] +[[package]] +name = "mappable-rc" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "204651f31b0a6a7b2128d2b92c372cd94607b210c3a6b6e542c57a8cfd4db996" + [[package]] name = "match_cfg" version = "0.1.0" @@ -10468,6 +10474,7 @@ dependencies = [ "itertools 0.13.0", "jsonwebtoken", "lazy_static", + "mappable-rc", "mysql_async", "native-tls", "nix", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 32c9bd493c..98a7214d7e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -214,6 +214,7 @@ convert_case = "0.6.0" getrandom = "0.2" tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1", "with-bit-vec-0_6"]} bit-vec = "=0.6.3" +mappable-rc = "^0" mysql_async = { version = "*", default-features = false, features = ["minimal", "default", "native-tls-tls"]} postgres-native-tls = "^0" native-tls = "^0" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index a6ea4d7c72..1fd91cf9b4 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -c7ef7778d68c86977a5d803fe98c418ef6a485c7 +ffc6e57d99f19adda60fa4baa8623413c44bb4dd \ No newline at end of file diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 4a3069be8e..68b3c3f061 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -899,7 +899,7 @@ impl RunJob { tx, "test-workspace", payload, - hm_args.into(), + windmill_queue::PushArgs::from(&hm_args), /* user */ "test-user", /* email */ "test@windmill.dev", /* permissioned_as */ "u/test-user".to_string(), diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index bf7ad1a730..fad23ac8c4 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -46,7 +46,7 @@ use windmill_common::{ }; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; -use windmill_queue::{push, PushArgs, PushIsolationLevel, QueueTransaction}; +use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel, QueueTransaction}; pub fn workspaced_service() -> Router { Router::new() @@ -653,7 +653,7 @@ async fn create_app( tx, &w_id, JobPayload::AppDependencies { path: app.path.clone(), version: v_id }, - PushArgs { args, extra: HashMap::new() }, + PushArgs { args: &args, extra: None }, &authed.username, &authed.email, windmill_common::users::username_to_permissioned_as(&authed.username), @@ -944,7 +944,7 @@ async fn update_app( tx, &w_id, JobPayload::AppDependencies { path: npath.clone(), version: v_id }, - PushArgs { args, extra: HashMap::new() }, + PushArgs { args: &args, extra: None }, &authed.username, &authed.email, windmill_common::users::username_to_permissioned_as(&authed.username), @@ -1157,7 +1157,7 @@ async fn execute_component( tx, &w_id, job_payload, - args, + PushArgs { args: &args.args, extra: args.extra }, &username, &email, permissioned_as, @@ -1239,12 +1239,12 @@ async fn build_args( policy: Policy, component: &str, path: String, - args: HashMap>, + mut args: HashMap>, authed: Option<&ApiAuthed>, user_db: &UserDB, db: &DB, w_id: &str, -) -> Result<(PushArgs, Option)> { +) -> Result<(PushArgsOwned, Option)> { let mut job_id: Option = None; let key = format!("{}:{}", component, &path); let (static_inputs, one_of_inputs, allow_user_resources) = match policy { @@ -1294,7 +1294,6 @@ async fn build_args( )))?, }; - let mut args = args.clone(); let mut safe_args = HashMap::>::new(); // tracing::error!("{:?}", allow_user_resources); @@ -1413,5 +1412,8 @@ async fn build_args( for (k, v) in static_inputs { extra.insert(k.to_string(), v.to_owned()); } - Ok((PushArgs { extra, args: safe_args }, job_id)) + Ok(( + PushArgsOwned { extra: Some(extra), args: safe_args }, + job_id, + )) } diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index 99f155b0f8..241469d05a 100644 --- a/backend/windmill-api/src/capture.rs +++ b/backend/windmill-api/src/capture.rs @@ -18,7 +18,7 @@ use windmill_common::{ error::{JsonResult, Result}, utils::{not_found_if_none, StripPath}, }; -use windmill_queue::PushArgs; +use windmill_queue::{PushArgs, PushArgsOwned}; use crate::db::{ApiAuthed, DB}; @@ -86,7 +86,7 @@ pub async fn new_payload( pub async fn update_payload( Extension(db): Extension, Path((w_id, path)): Path<(String, StripPath)>, - args: PushArgs, + args: PushArgsOwned, ) -> Result { let mut tx = db.begin().await?; @@ -99,7 +99,7 @@ pub async fn update_payload( ", &w_id, &path.to_path(), - Json(args) as Json, + Json(PushArgs { args: &args.args, extra: args.extra }) as Json, ) .execute(&mut *tx) .await?; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index e5a27972f6..42962619ed 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -425,7 +425,7 @@ async fn create_flow( dedicated_worker: nf.dedicated_worker, version: version, }, - args.into(), + windmill_queue::PushArgs { args: &args, extra: None }, &authed.username, &authed.email, windmill_common::users::username_to_permissioned_as(&authed.username), @@ -820,7 +820,7 @@ async fn update_flow( dedicated_worker: nf.dedicated_worker, version: version, }, - windmill_queue::PushArgs { args, extra: HashMap::new() }, + windmill_queue::PushArgs { args: &args, extra: None }, &authed.username, &authed.email, windmill_common::users::username_to_permissioned_as(&authed.username), diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index a3298b8e2d..8cbf551961 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -77,7 +77,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED}; use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL}; use windmill_queue::{ cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push, - DecodeQueries, PushArgs, PushIsolationLevel, QueueTransaction, + DecodeQueries, PushArgs, PushArgsOwned, PushIsolationLevel, QueueTransaction, }; #[cfg(feature = "prometheus")] @@ -588,14 +588,20 @@ async fn get_flow_job_debug_info( "This endpoint is only for root flow jobs".to_string(), )); } + let leaf_jobs: HashMap = job + .leaf_jobs + .clone() + .and_then(|x| serde_json::from_value(x).ok()) + .unwrap_or_else(HashMap::new); let mut jobs = HashMap::new(); - jobs.insert("root_job".to_string(), Job::QueuedJob(job.clone())); + let id = job.id.clone(); + jobs.insert("root_job".to_string(), Job::QueuedJob(job)); let mut job_ids = vec![]; let jobs_with_root = sqlx::query_scalar!( "SELECT id FROM queue WHERE workspace_id = $1 and root_job = $2", &w_id, - &job.id, + &id, ) .fetch_all(&db) .await?; @@ -604,10 +610,6 @@ async fn get_flow_job_debug_info( job_ids.push(job); } - let leaf_jobs: HashMap = job - .leaf_jobs - .and_then(|x| serde_json::from_value(x).ok()) - .unwrap_or_else(HashMap::new); for job in leaf_jobs.iter() { match job.1 { JobResult::ListJob(jobs) => job_ids.extend(jobs.to_owned()), @@ -920,7 +922,7 @@ async fn get_args( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::JsonResult> { - let record = sqlx::query( + let record = sqlx::query_as::<_, RawArgs>( "SELECT created_by, args FROM completed_job WHERE completed_job.id = $1 AND completed_job.workspace_id = $2", @@ -931,8 +933,6 @@ async fn get_args( .await?; if let Some(record) = record { - let record = RawArgs::from_row(&record) - .map_err(|e| Error::InternalErr(format!("error parsing args: {e:#}")))?; if opt_authed.is_none() && record.created_by != "anonymous" { return Err(Error::BadRequest( "As a non logged in user, you can only see jobs ran by anonymous users".to_string(), @@ -940,7 +940,7 @@ async fn get_args( } Ok(Json(record.args.map(|x| x.0).unwrap_or_default())) } else { - let record = sqlx::query( + let record = sqlx::query_as::<_, RawArgs>( "SELECT created_by, args FROM queue WHERE queue.id = $1 AND queue.workspace_id = $2", @@ -950,8 +950,6 @@ async fn get_args( .fetch_optional(&db) .await?; let record = not_found_if_none(record, "Job Args", id.to_string())?; - let record = RawArgs::from_row(&record) - .map_err(|e| Error::InternalErr(format!("error parsing args: {e:#}")))?; if opt_authed.is_none() && record.created_by != "anonymous" { return Err(Error::BadRequest( "As a non logged in user, you can only see jobs ran by anonymous users".to_string(), @@ -2636,7 +2634,7 @@ pub async fn run_flow_by_path( Extension(rsmq): Extension>, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result<(StatusCode, String)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -2669,7 +2667,7 @@ pub async fn run_flow_by_path( tx, &w_id, JobPayload::Flow { path: flow_path.to_string(), dedicated_worker }, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -2744,10 +2742,12 @@ pub async fn restart_flow( .with_context(|| "No flow path set for completed flow job")?; check_scopes(&authed, || format!("run:flow/{flow_path}"))?; + let ehm = HashMap::new(); let push_args = completed_job .args - .map(|json| PushArgs { args: json.0, extra: HashMap::new() }) - .unwrap_or_else(PushArgs::empty); + .as_ref() + .map(|json| PushArgs { args: &json.0, extra: None }) + .unwrap_or_else(|| PushArgs::from(&ehm)); let scheduled_for = run_query.get_scheduled_for(&db).await?; @@ -2793,7 +2793,7 @@ pub async fn run_script_by_path( Extension(rsmq): Extension>, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result<(StatusCode, String)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -2818,7 +2818,7 @@ pub async fn run_script_by_path( tx, &w_id, job_payload, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -2886,7 +2886,7 @@ pub async fn run_workflow_as_code( let mut extra = HashMap::new(); extra.insert(ENTRYPOINT_OVERRIDE.to_string(), to_raw_value(&entrypoint)); - let args = PushArgs { args: task.args.unwrap_or_else(HashMap::new), extra }; + let args = PushArgs { args: &task.args.unwrap_or_else(HashMap::new), extra: Some(extra) }; let scheduled_for = run_query.get_scheduled_for(&db).await?; let tag = run_query.tag.clone().or(tag).or(Some(job.tag)); @@ -2898,7 +2898,7 @@ pub async fn run_workflow_as_code( tx, &w_id, job_payload, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3021,15 +3021,14 @@ async fn run_wait_result( } if result.is_none() { - let row = sqlx::query( + let row = sqlx::query_as::<_, RawResult>( "SELECT null as created_by, result, language, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(uuid) .bind(&w_id) .fetch_optional(db) .await?; - if let Some(row) = row { - let raw_result = RawResult::from_row(&row)?; + if let Some(raw_result) = row { result = match format_result( raw_result.language.as_ref(), raw_result.flow_status.map(|x| x.0), @@ -3203,7 +3202,7 @@ pub async fn run_wait_result_job_by_path_get( }); let inner_args: HashMap> = HashMap::new(); - let args = PushArgs { extra: payload_args, args: inner_args }; + let args = PushArgs { extra: Some(payload_args), args: &inner_args }; check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?; let script_path = script_path.to_path(); @@ -3224,7 +3223,7 @@ pub async fn run_wait_result_job_by_path_get( tx, &w_id, job_payload, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3285,7 +3284,7 @@ pub async fn run_wait_result_flow_by_path_get( payload_args.insert(k.to_string(), v.clone()); }); - let args = PushArgs { extra: payload_args, args: HashMap::new() }; + let args = PushArgsOwned { extra: Some(payload_args), args: HashMap::new() }; run_wait_result_flow_by_path_internal( db, run_query, flow_path, authed, rsmq, user_db, args, w_id, @@ -3300,7 +3299,7 @@ pub async fn run_wait_result_script_by_path( Extension(db): Extension, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -3326,7 +3325,7 @@ async fn run_wait_result_script_by_path_internal( rsmq: Option, user_db: UserDB, w_id: String, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result { check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?; let script_path = script_path.to_path(); @@ -3347,7 +3346,7 @@ async fn run_wait_result_script_by_path_internal( tx, &w_id, job_payload, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3383,7 +3382,7 @@ pub async fn run_wait_result_script_by_hash( Extension(db): Extension, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -3431,7 +3430,7 @@ pub async fn run_wait_result_script_by_hash( dedicated_worker, priority, }, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3467,7 +3466,7 @@ pub async fn run_wait_result_flow_by_path( Extension(db): Extension, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -3485,7 +3484,7 @@ async fn run_wait_result_flow_by_path_internal( authed: ApiAuthed, rsmq: Option, user_db: UserDB, - args: PushArgs, + args: PushArgsOwned, w_id: String, ) -> error::Result { check_queue_too_long(&db, run_query.queue_limit).await?; @@ -3509,11 +3508,11 @@ async fn run_wait_result_flow_by_path_internal( .fetch_optional(&mut tx) .await? .map(|x| (x.tag, x.dedicated_worker, x.early_return)) - .ok_or_else( - || Error::NotFound( - format!("flow not found at path {flow_path} in workspace {w_id}") - ) - )?; + .ok_or_else(|| { + Error::NotFound(format!( + "flow not found at path {flow_path} in workspace {w_id}" + )) + })?; let tag = run_query.tag.clone().or(tag); check_tag_available_for_workspace(&w_id, &tag).await?; @@ -3525,7 +3524,7 @@ async fn run_wait_result_flow_by_path_internal( tx, &w_id, JobPayload::Flow { path: flow_path.to_string(), dedicated_worker }, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3593,7 +3592,7 @@ async fn run_preview_script( dedicated_worker: preview.dedicated_worker, }), }, - preview.args.unwrap_or_default().into(), + PushArgs::from(&preview.args.unwrap_or_default()), authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3675,7 +3674,7 @@ async fn run_bundle_preview_script( custom_concurrency_key: None, }), }, - args.into(), + PushArgs::from(&args), authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3789,16 +3788,17 @@ async fn run_dependencies_job( } let raw_script = req.raw_scripts[0].clone(); let script_path = raw_script.script_path; + let ehm = HashMap::new(); let (args, raw_code) = if let Some(deps) = req.raw_deps { let mut hm = HashMap::new(); hm.insert( "raw_deps".to_string(), JsonRawValue::from_string("true".to_string()).unwrap(), ); - (PushArgs { extra: hm, args: HashMap::new() }, deps) + (PushArgs { extra: Some(hm), args: &ehm }, deps) } else { ( - PushArgs::empty(), + PushArgs::from(&ehm), raw_script.raw_code.unwrap_or_else(|| "".to_string()), ) }; @@ -3869,7 +3869,10 @@ async fn run_flow_dependencies_job( PushIsolationLevel::IsolatedRoot(db.clone(), rsmq), &w_id, JobPayload::RawFlowDependencies { path: req.path, flow_value: req.flow_value }, - HashMap::from([("skip_flow_update".to_string(), to_raw_value(&true))]).into(), + PushArgs::from(&HashMap::from([( + "skip_flow_update".to_string(), + to_raw_value(&true), + )])), authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -3971,12 +3974,13 @@ async fn add_batch_jobs( } }; for _ in 0..n { + let ehm = HashMap::new(); let (uuid, ntx) = push( &db, tx, &w_id, payload.clone(), - PushArgs::empty(), + PushArgs::from(&ehm), authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -4106,7 +4110,7 @@ async fn run_preview_flow_job( path: raw_flow.path, restarted_from: raw_flow.restarted_from, }, - raw_flow.args.unwrap_or_default().into(), + PushArgs::from(&raw_flow.args.unwrap_or_default()), authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -4139,7 +4143,7 @@ pub async fn run_job_by_hash( Extension(rsmq): Extension>, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, - args: PushArgs, + args: PushArgsOwned, ) -> error::Result<(StatusCode, String)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -4185,7 +4189,7 @@ pub async fn run_job_by_hash( dedicated_worker, priority, }, - args, + PushArgs { args: &args.args, extra: args.extra }, authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -4593,7 +4597,7 @@ async fn get_completed_job<'a>( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::Result { - let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, + let job_o = sqlx::query_as::<_, CompletedJob>("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped, raw_lock, email, visible_to_owner, mem_peak, tag, priority, result->'wm_labels' as labels FROM completed_job WHERE id = $1 AND workspace_id = $2") @@ -4602,9 +4606,7 @@ async fn get_completed_job<'a>( .fetch_optional(&db) .await?; - let job = not_found_if_none(job_o, "Completed Job", id.to_string())?; - - let cj = CompletedJob::from_row(&job)?; + let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?; if opt_authed.is_none() && cj.created_by != "anonymous" { return Err(Error::BadRequest( @@ -4649,7 +4651,7 @@ async fn get_completed_job_result( Query(JsonPath { json_path, suspended_job, approver, resume_id, secret }): Query, ) -> error::Result { let result_o = if let Some(json_path) = json_path { - sqlx::query( + sqlx::query_as::<_, RawResult>( "SELECT result #> $3 as result, flow_status, language, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(id) @@ -4663,16 +4665,14 @@ async fn get_completed_job_result( .fetch_optional(&db) .await? } else { - sqlx::query("SELECT result, flow_status, language, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2") + sqlx::query_as::<_, RawResult>("SELECT result, flow_status, language, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2") .bind(id) .bind(&w_id) .fetch_optional(&db) .await? }; - let result = not_found_if_none(result_o, "Completed Job", id.to_string())?; - - let raw_result = RawResult::from_row(&result)?; + let raw_result = not_found_if_none(result_o, "Completed Job", id.to_string())?; if opt_authed.is_none() && raw_result.created_by.unwrap_or_default() != "anonymous" { match (suspended_job, resume_id, approver, secret) { @@ -4740,7 +4740,7 @@ async fn get_completed_job_result_maybe( Path((w_id, id)): Path<(String, Uuid)>, Query(GetCompletedJobQuery { get_started }): Query, ) -> error::Result { - let result_o = sqlx::query( + let result_o = sqlx::query_as::<_, RawResultWithSuccess>( "SELECT result, success, language, flow_status, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(id) @@ -4748,8 +4748,7 @@ async fn get_completed_job_result_maybe( .fetch_optional(&db) .await?; - if let Some(result) = result_o { - let res = RawResultWithSuccess::from_row(&result)?; + if let Some(res) = result_o { let result = format_result( res.language.as_ref(), res.flow_status.map(|x| x.0), @@ -4804,7 +4803,7 @@ async fn delete_completed_job<'a>( let mut tx = user_db.begin(&authed).await?; require_admin(authed.is_admin, &authed.username)?; - let job_o = sqlx::query( + let job_o = sqlx::query_as::<_, CompletedJob>( "UPDATE completed_job SET args = null, logs = '', result = null, deleted = true WHERE id = $1 AND workspace_id = $2 \ RETURNING *", ) @@ -4817,7 +4816,7 @@ async fn delete_completed_job<'a>( .execute(&mut *tx) .await?; - let job = not_found_if_none(job_o, "Completed Job", id.to_string())?; + let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?; audit_log( &mut *tx, @@ -4831,7 +4830,6 @@ async fn delete_completed_job<'a>( .await?; tx.commit().await?; - let cj = CompletedJob::from_row(&job)?; let cj = format_completed_job_result(cj); diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 967df3f201..a274fcb3c1 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -745,7 +745,7 @@ async fn create_script_internal<'c>( path: ns.path, dedicated_worker: ns.dedicated_worker, }, - args.into(), + windmill_queue::PushArgs::from(&args), &authed.username, &authed.email, permissioned_as, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 2590087496..ab9a52a602 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -113,14 +113,6 @@ pub struct QueuedJob { } impl QueuedJob { - pub fn get_args(&self) -> HashMap> { - if let Some(args) = self.args.as_ref() { - args.0.clone() - } else { - HashMap::new() - } - } - pub fn script_path(&self) -> &str { self.script_path .as_ref() @@ -287,8 +279,8 @@ impl CompletedJob { } #[derive(sqlx::FromRow)] -pub struct BranchResults<'a> { - pub result: &'a RawValue, +pub struct BranchResults { + pub result: sqlx::types::Json>, pub id: Uuid, } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 323d6d37ad..860a00b10d 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -31,7 +31,7 @@ use reqwest::{ Client, StatusCode, }; use rsmq_async::RsmqConnection; -use serde::{Deserialize, Serialize}; +use serde::{ser::SerializeMap, Deserialize, Serialize}; use serde_json::{json, value::RawValue}; use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction}; #[cfg(feature = "benchmark")] @@ -137,7 +137,7 @@ pub struct CanceledBy { pub async fn cancel_single_job<'c>( username: &str, reason: Option, - job_running: &QueuedJob, + job_running: Arc, w_id: &str, mut tx: Transaction<'c, Postgres>, db: &Pool, @@ -146,10 +146,10 @@ pub async fn cancel_single_job<'c>( ) -> error::Result<(Transaction<'c, Postgres>, Option)> { if force_cancel || (job_running.parent_job.is_none() && !job_running.running) { let username = username.to_string(); - let job_running = job_running.clone(); let w_id = w_id.to_string(); let db = db.clone(); let rsmq = rsmq.clone(); + let job_running = job_running.clone(); tokio::task::spawn(async move { let reason: String = reason .clone() @@ -223,8 +223,8 @@ pub async fn cancel_job<'c>( "You are not logged in and this job was not created by an anonymous user like you so you cannot cancel it".to_string(), )); } - let mut job = job.unwrap(); + let mut job = job.unwrap(); if force_cancel { // if force canceling a flow step, make sure we force cancel from the highest parent loop { @@ -240,6 +240,8 @@ pub async fn cancel_job<'c>( } } + let job = Arc::new(job); + // get all children let mut jobs = vec![job.id]; let mut jobs_to_cancel = vec![]; @@ -260,7 +262,7 @@ pub async fn cancel_job<'c>( let (ntx, _) = cancel_single_job( username, reason.clone(), - &job, + job.clone(), w_id, tx, db, @@ -278,7 +280,7 @@ pub async fn cancel_job<'c>( let (ntx, _) = cancel_single_job( username, reason.clone(), - &job, + Arc::new(job), w_id, tx, db, @@ -417,6 +419,12 @@ impl ValidableJson for Box { } } +impl ValidableJson for Arc> { + fn is_valid_json(&self) -> bool { + !self.get().is_empty() + } +} + impl ValidableJson for serde_json::Value { fn is_valid_json(&self) -> bool { true @@ -984,6 +992,7 @@ pub async fn add_completed_job< None }; + let ehm = HashMap::new(); let (_uuid, tx) = push( db, tx, @@ -1005,8 +1014,8 @@ pub async fn add_completed_job< queued_job .args .as_ref() - .map(|x| PushArgs { args: x.0.clone(), extra: HashMap::new() }) - .unwrap_or_else(PushArgs::empty), + .map(|x| PushArgs::from(&x.0)) + .unwrap_or_else(|| PushArgs::from(&ehm)), &queued_job.created_by, &queued_job.email, queued_job.permissioned_as.clone(), @@ -1357,7 +1366,7 @@ async fn apply_schedule_handlers< } else { #[cfg(feature = "enterprise")] if let Some(on_recovery_path) = schedule.on_recovery.clone() { - let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); + let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); let times = schedule.on_recovery_times.unwrap_or(1).max(1); let past_jobs = sqlx::query_as::<_, CompletedJobSubset>( "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5", @@ -1382,7 +1391,7 @@ async fn apply_schedule_handlers< let failed_job = past_jobs[past_jobs.len() - 1].clone(); if !failed_job.success { - tx = handle_recovered_schedule( + handle_recovered_schedule( db, tx, job_id, @@ -1398,9 +1407,9 @@ async fn apply_schedule_handlers< schedule.on_recovery_extra_args.clone(), ) .await?; + } else { + tx.commit().await?; } - - tx.commit().await?; } } @@ -1480,7 +1489,7 @@ pub async fn push_error_handler< tx, handler_w_id, payload, - PushArgs { extra, args: result }, + PushArgs { extra: Some(extra), args: &result }, if is_global_error_handler { "global" } else if is_schedule_error_handler { @@ -1553,7 +1562,7 @@ async fn handle_recovered_schedule< successful_times: i32, successful_job_started_at: DateTime, extra_args: Option>>, -) -> windmill_common::error::Result> { +) -> windmill_common::error::Result<()> { let (payload, tag) = get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?; let mut extra = HashMap::new(); @@ -1606,7 +1615,7 @@ async fn handle_recovered_schedule< tx, w_id, payload, - PushArgs { extra: extra, args: args }, + PushArgs { extra: Some(extra), args: &args }, SCHEDULE_RECOVERY_HANDLER_USERNAME, SCHEDULE_RECOVERY_HANDLER_USER_EMAIL, ERROR_HANDLER_USER_GROUP.to_string(), @@ -1631,7 +1640,8 @@ async fn handle_recovered_schedule< uuid, schedule_path ); - return Ok(tx); + tx.commit().await?; + Ok(()) } pub async fn pull( @@ -1930,7 +1940,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< .map_err(|_| anyhow::anyhow!("Failed to parsed Redis message"))?, ); - let m2r = sqlx::query( + let m2r = sqlx::query_as::<_, QueuedJob>( "UPDATE queue SET running = true , started_at = coalesce(started_at, now()) @@ -1948,11 +1958,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< .bind(uuid) .fetch_optional(db) .await?; - let m2 = if let Some(row) = m2r { - Some(QueuedJob::from_row(&row)?) - } else { - None - }; rsmq.delete_message(&tag.unwrap(), &msg.id) .await @@ -1961,7 +1966,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< #[cfg(feature = "benchmark")] println!("rsmq 2: {:?}", instant.elapsed()); - m2 + m2r } else { None } @@ -1981,7 +1986,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< let priority_tags_sorted = config.priority_tags_sorted.clone(); drop(config); let r = if suspend_first { - sqlx::query("UPDATE queue + sqlx::query_as::<_, QueuedJob>("UPDATE queue SET running = true , started_at = coalesce(started_at, now()) , last_ping = now() @@ -2007,18 +2012,13 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< } else { None }; - let r = if let Some(row) = r { - Some(QueuedJob::from_row(&row)?) - } else { - None - }; if r.is_none() { // #[cfg(feature = "benchmark")] // let instant = Instant::now(); let mut highest_priority_job: Option = None; for priority_tags in priority_tags_sorted { - let r = sqlx::query( + let r = sqlx::query_as::<_, QueuedJob>( "UPDATE queue SET running = true , started_at = coalesce(started_at, now()) @@ -2044,12 +2044,11 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< .fetch_optional(db) .await?; - if let Some(pulled_row) = r { - let pulled_job = QueuedJob::from_row(&pulled_row)?; - highest_priority_job = Some(pulled_job.clone()); + if let Some(pulled_job) = r { + let id = pulled_job.id; + highest_priority_job = Some(pulled_job); tracing::debug!( - "Pulling for job {} with tags {:?} with priority {}", - pulled_job.id, + "Pulling for job {id} with tags {:?} with priority {}", priority_tags.tags, priority_tags.priority ); @@ -2103,18 +2102,13 @@ async fn legacy_concurrency_key(db: &Pool, queued_job: &QueuedJob) -> .flatten() .flatten(); - r.map(|x| { - interpolate_args( - x, - &queued_job - .args - .clone() - .map(|x| x.0) - .unwrap_or_default() - .into(), - &queued_job.workspace_id, - ) - }) + let ehm = HashMap::new(); + let push_args = queued_job + .args + .as_ref() + .map(|x| PushArgs::from(&x.0)) + .unwrap_or_else(|| PushArgs::from(&ehm)); + r.map(|x| interpolate_args(x, &push_args, &queued_job.workspace_id)) } async fn concurrency_key( @@ -2138,7 +2132,7 @@ fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String { let arg_value = args .args .get(arg_name) - .or(args.extra.get(arg_name)) + .or(args.extra.as_ref().and_then(|x| x.get(arg_name))) .map(|x| x.get()) .unwrap_or_default() .trim_matches('"'); @@ -2427,7 +2421,7 @@ async fn extract_result_from_job_result( ) -> error::Result> { match job_result { JobResult::ListJob(job_ids) => { - let rows = sqlx::query( + let rows = sqlx::query_as::<_, ResultWithId>( "SELECT id, result FROM completed_job WHERE id = ANY($1) AND workspace_id = $2", ) .bind(job_ids.as_slice()) @@ -2435,11 +2429,7 @@ async fn extract_result_from_job_result( .fetch_all(db) .await? .into_iter() - .filter_map(|x| { - ResultWithId::from_row(&x) - .ok() - .and_then(|x| x.result.map(|y| (x.id, y))) - }) + .filter_map(|x| x.result.map(|y| (x.id, y))) .collect::>>>(); let result = job_ids .into_iter() @@ -2451,7 +2441,7 @@ async fn extract_result_from_job_result( .collect::>(); Ok(to_raw_value(&result)) } - JobResult::SingleJob(x) => Ok(sqlx::query( + JobResult::SingleJob(x) => Ok(sqlx::query_as::<_, ResultR>( "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(x) @@ -2463,11 +2453,7 @@ async fn extract_result_from_job_result( ) .fetch_optional(db) .await? - .map(|r| { - ResultR::from_row(&r) - .ok() - .and_then(|x| x.result.map(|x| x.0)) - }) + .map(|r| r.result.map(|x| x.0)) .flatten() .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))), } @@ -2523,35 +2509,27 @@ pub async fn get_queued_job_tx<'c>( w_id: &str, tx: &mut Transaction<'c, Postgres>, ) -> error::Result> { - let r = sqlx::query( + sqlx::query_as::<_, QueuedJob>( "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2", ) .bind(id) .bind(w_id) .fetch_optional(&mut **tx) - .await?; - if let Some(row) = r { - Ok(Some(QueuedJob::from_row(&row)?.to_owned())) - } else { - Ok(None) - } + .await + .map_err(Into::into) } pub async fn get_queued_job(id: &Uuid, w_id: &str, db: &DB) -> error::Result> { - let r = sqlx::query( + sqlx::query_as::<_, QueuedJob>( "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2", ) .bind(id) .bind(w_id) .fetch_optional(db) - .await?; - if let Some(row) = r { - Ok(Some(QueuedJob::from_row(&row)?.to_owned())) - } else { - Ok(None) - } + .await + .map_err(Into::into) } pub enum PushIsolationLevel<'c, R: rsmq_async::RsmqConnection + Send + 'c> { @@ -2586,14 +2564,48 @@ macro_rules! fetch_scalar_isolated { use sqlx::types::JsonRawValue; -#[derive(Serialize, Debug)] -pub struct PushArgs { - #[serde(flatten)] - pub extra: HashMap>, - #[serde(flatten)] +#[derive(Debug)] +pub struct PushArgsOwned { + pub extra: Option>>, pub args: HashMap>, } +#[derive(Debug)] +pub struct PushArgs<'c> { + pub extra: Option>>, + pub args: &'c HashMap>, +} + +impl<'c> From<&'c HashMap>> for PushArgs<'c> { + fn from(args: &'c HashMap>) -> Self { + PushArgs { extra: None, args } + } +} + +impl<'c> Serialize for PushArgs<'c> { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + let mut map = serializer.serialize_map(Some( + self.args.len() + self.extra.as_ref().map(|x| x.len()).unwrap_or_default(), + ))?; + let mut in_extra = vec![]; + if let Some(extra) = &self.extra { + for (k, v) in extra { + map.serialize_entry(k, v)?; + in_extra.push(k); + } + } + for (k, v) in self.args { + if !in_extra.contains(&k) { + map.serialize_entry(k, v)?; + } + } + map.end() + } +} + #[derive(Deserialize)] pub struct DecodeQuery { pub include_query: Option, @@ -2648,11 +2660,11 @@ impl DecodeQueries { } } -impl PushArgs { - pub fn insert, V: Into>>(&mut self, k: K, v: V) { - self.extra.insert(k.into(), v.into()); - } -} +// impl<'c> PushArgs<'c> { +// pub fn insert, V: Into>>(&mut self, k: K, v: V) { +// self.extra.insert(k.into(), v.into()); +// } +// } #[derive(Deserialize)] pub struct RequestQuery { @@ -2687,7 +2699,7 @@ fn restructure_cloudevents_metadata( } } -impl PushArgs { +impl PushArgsOwned { async fn from_json( mut extra: HashMap>, use_raw: bool, @@ -2705,12 +2717,12 @@ impl PushArgs { .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); let mut hm = HashMap::new(); hm.insert("body".to_string(), args); - Ok(PushArgs { extra, args: hm }) + Ok(PushArgsOwned { extra: Some(extra), args: hm }) } else { let hm = serde_json::from_str::>>>(&str) .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())? .unwrap_or_else(HashMap::new); - Ok(PushArgs { extra, args: hm }) + Ok(PushArgsOwned { extra: Some(extra), args: hm }) } } @@ -2727,7 +2739,7 @@ impl PushArgs { Error::BadRequest(format!("invalid cloudevents+json: {}", e)).into_response() })?; let hm = restructure_cloudevents_metadata(hm).map_err(|e| e.into_response())?; - Ok(PushArgs { extra, args: hm }) + Ok(PushArgsOwned { extra: Some(extra), args: hm }) } } @@ -2771,10 +2783,10 @@ mod tests { "#; let extra = HashMap::new(); - let a1 = PushArgs::from_ce_json(extra.clone(), false, r1.to_string()) + let a1 = PushArgsOwned::from_ce_json(extra.clone(), false, r1.to_string()) .await .expect("Failed to parse the cloudevent"); - let a2 = PushArgs::from_ce_json(extra.clone(), false, r2.to_string()) + let a2 = PushArgsOwned::from_ce_json(extra.clone(), false, r2.to_string()) .await .expect("Failed to parse the cloudevent"); @@ -2792,7 +2804,7 @@ mod tests { } #[axum::async_trait] -impl FromRequest for PushArgs +impl FromRequest for PushArgsOwned where S: Send + Sync, { @@ -2824,7 +2836,7 @@ where let str = String::from_utf8(bytes.to_vec()) .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; - PushArgs::from_json(extra, use_raw, str).await + PushArgsOwned::from_json(extra, use_raw, str).await } else if content_type .unwrap() .starts_with("application/cloudevents+json") @@ -2835,7 +2847,7 @@ where let str = String::from_utf8(bytes.to_vec()) .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; - PushArgs::from_ce_json(extra, use_raw, str).await + PushArgsOwned::from_ce_json(extra, use_raw, str).await } else if content_type .unwrap() .starts_with("application/cloudevents-batch+json") @@ -2868,7 +2880,7 @@ where .map(|(k, v)| (k, to_raw_value(&v))) .collect::>(); - return Ok(PushArgs { extra, args: payload }); + return Ok(PushArgsOwned { extra: Some(extra), args: payload }); } else { Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response()) } @@ -2906,9 +2918,9 @@ pub fn build_extra( args } -impl PushArgs { +impl PushArgsOwned { pub fn empty() -> Self { - PushArgs { extra: HashMap::new(), args: HashMap::new() } + PushArgsOwned { extra: None, args: HashMap::new() } } } @@ -2916,12 +2928,6 @@ pub fn empty_result() -> Box { return JsonRawValue::from_string("{}".to_string()).unwrap(); } -impl From>> for PushArgs { - fn from(value: HashMap>) -> Self { - PushArgs { extra: HashMap::new(), args: value } - } -} - // impl From> for PushArgs { // fn from(value: PushArgsInner) -> Self { // PushArgs::Unwrapped(value) @@ -2938,12 +2944,12 @@ struct FlowRawValue { } // #[instrument(level = "trace", skip_all)] -pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( +pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>( _db: &Pool, mut tx: PushIsolationLevel<'c, R>, workspace_id: &str, job_payload: JobPayload, - args: PushArgs, + args: PushArgs<'d>, user: &str, mut email: &str, mut permissioned_as: String, @@ -3731,6 +3737,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( .map_err(|e| Error::InternalErr(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?; } + tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}"); let uuid = sqlx::query_scalar!( "INSERT INTO queue (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, @@ -3775,6 +3782,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( .await .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?; + tracing::debug!("Pushed {job_id}"); // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction. #[cfg(feature = "prometheus")] if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 03605105bb..ede0433a2d 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -208,7 +208,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( tx, &schedule.workspace_id, payload, - crate::PushArgs { args, extra: HashMap::new() }, + crate::PushArgs { args: &args, extra: None }, &schedule_to_user(&schedule.path), &schedule.email, username_to_permissioned_as(&schedule.edited_by), @@ -228,6 +228,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( authed, ) .await?; + Ok(tx) // TODO: Bubble up pushed UUID from here } diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index f4fda2a201..36e8bfcedd 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -49,6 +49,7 @@ chrono.workspace = true dotenv.workspace = true rand.workspace = true # TODO: Remove. only used by token creation hack. const_format.workspace = true +mappable-rc.workspace = true git-version.workspace = true dyn-iter.workspace = true once_cell.workspace = true diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 44085ecfbb..1944b66ad8 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -894,8 +894,6 @@ pub async fn get_common_bun_proc_envs(base_internal_url: &str) -> HashMap>, + jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> Result<()> { let mut logs = "".to_string(); diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 1dc06a156d..d4ed0c1db3 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -70,13 +70,14 @@ pub async fn handle_dedicated_process( mut killpill_rx: tokio::sync::broadcast::Receiver<()>, job_completed_tx: JobCompletedSender, token: &str, - mut jobs_rx: Receiver>, + mut jobs_rx: Receiver>, worker_name: &str, db: &DB, script_path: &str, mode: &str, ) -> std::result::Result<(), error::Error> { //do not cache local dependencies + let mut child = { let mut cmd = Command::new(command_path); cmd.current_dir(job_dir) @@ -130,7 +131,7 @@ pub async fn handle_dedicated_process( } }); - let mut jobs = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS); + let mut jobs: VecDeque> = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS); // let mut i = 0; // let mut j = 0; let mut alive = true; @@ -173,6 +174,7 @@ pub async fn handle_dedicated_process( tracing::info!("job completed on dedicated worker {script_path}: {}", job.id); match serde_json::from_str::>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) { Ok(result) => { + let result = Arc::new(result); append_logs(&job.id, &job.workspace_id, logs.clone(), db).await; if line.starts_with("wm_res[success]:") { job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap() @@ -182,7 +184,7 @@ pub async fn handle_dedicated_process( }, Err(e) => { tracing::error!("Could not deserialize job result `{line}`: {e:?}"); - job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap(); + job_completed_tx.send(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap(); }, }; logs = init_log.clone(); @@ -222,7 +224,11 @@ pub async fn handle_dedicated_process( Ok(()) } -type DedicatedWorker = (String, Sender>, Option>); +type DedicatedWorker = ( + String, + Sender>, + Option>, +); // spawn one dedicated worker per compatible steps of the flow, associating the node id to the dedicated worker channel send #[async_recursion] @@ -240,7 +246,8 @@ async fn spawn_dedicated_workers_for_flow( job_completed_tx: &JobCompletedSender, ) -> Vec { let mut workers = vec![]; - let mut script_path_to_worker: HashMap>> = HashMap::new(); + let mut script_path_to_worker: HashMap>> = + HashMap::new(); for module in modules.iter() { let value = module.get_value(); if let Ok(value) = value { @@ -389,7 +396,7 @@ pub async fn create_dedicated_worker_map( worker_name: &str, job_completed_tx: &JobCompletedSender, ) -> ( - HashMap>>, + HashMap>>, bool, Vec>, ) { @@ -519,7 +526,7 @@ async fn spawn_dedicated_worker( #[cfg(feature = "enterprise")] { let (dedicated_worker_tx, dedicated_worker_rx) = - tokio::sync::mpsc::channel::>(MAX_BUFFERED_DEDICATED_JOBS); + tokio::sync::mpsc::channel::>(MAX_BUFFERED_DEDICATED_JOBS); let killpill_rx = killpill_rx.resubscribe(); let db = db.clone(); let base_internal_url = base_internal_url.to_string(); diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index ac1b4a6edd..cc3d71b1e1 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -394,8 +394,7 @@ async fn build_import_map( #[cfg(feature = "enterprise")] use crate::{dedicated_worker::handle_dedicated_process, JobCompletedSender}; -#[cfg(feature = "enterprise")] -use std::sync::Arc; + #[cfg(feature = "enterprise")] use tokio::sync::mpsc::Receiver; @@ -410,7 +409,7 @@ pub async fn start_worker( script_path: &str, token: &str, job_completed_tx: JobCompletedSender, - jobs_rx: Receiver>, + jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, db: &sqlx::Pool, ) -> Result<()> { diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index f0a3414d91..0fc75d0209 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -138,7 +138,7 @@ pub struct OptAuthedClient(Option); pub async fn eval_timeout( expr: String, transform_context: HashMap>>, - flow_input: Option>>>, + flow_input: Option>>>, authed_client: Option<&AuthedClient>, by_id: Option, ) -> anyhow::Result> { @@ -563,7 +563,7 @@ async fn op_resource( pub struct TransformContext { pub envs: HashMap>>, - pub flow_input: Option>>>, + pub flow_input: Option>>>, } #[op2] @@ -575,7 +575,7 @@ fn op_get_context(op_state: Rc>, #[string] id: &str) -> String client .flow_input .as_ref() - .and_then(|x| serde_json::to_string(&x).ok()) + .and_then(|x| serde_json::to_string(x.as_ref()).ok()) .unwrap_or_else(|| "null".to_string()) } else { client diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 84bbd8c31c..72a482da3d 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -1007,9 +1007,6 @@ pub async fn handle_python_reqs( Ok(req_paths) } -#[cfg(feature = "enterprise")] -use std::sync::Arc; - #[cfg(feature = "enterprise")] use crate::JobCompletedSender; #[cfg(feature = "enterprise")] @@ -1032,7 +1029,7 @@ pub async fn start_worker( script_path: &str, token: &str, job_completed_tx: JobCompletedSender, - jobs_rx: Receiver>, + jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> error::Result<()> { let mut mem_peak: i32 = 0; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 1129f35c14..11805dd730 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1344,7 +1344,7 @@ pub async fn run_worker( ) -> error::Result<()> { if let Some(content) = WORKER_CONFIG.read().await.init_bash.clone() { let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq); + let ehm = HashMap::new(); let (uuid, inner_tx) = push( &db, tx, @@ -1952,7 +1953,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>( cache_ttl: None, dedicated_worker: None, }), - PushArgs::empty(), + PushArgs::from(&ehm), worker_name, "worker@windmill.dev", SUPERADMIN_SECRET_EMAIL.to_string(), @@ -2017,6 +2018,10 @@ pub async fn process_completed_job, - pub result: Box, + pub result: Arc>, pub mem_peak: i32, pub success: bool, pub cached_res_path: Option, @@ -2458,7 +2464,7 @@ async fn handle_queued_job( job_completed_tx .send(JobCompleted { job: job, - result: cached_resource_value, + result: Arc::new(cached_resource_value), mem_peak: 0, canceled_by: None, success: true, @@ -2475,7 +2481,7 @@ async fn handle_queued_job( #[cfg(feature = "prometheus")] let timer = _worker_flow_initial_transition_duration.map(|x| x.start_timer()); handle_flow( - &job, + job, db, &client.get_authed().await, None, @@ -2615,7 +2621,7 @@ async fn handle_queued_job( process_result( job, - result, + result.map(|x| Arc::new(x)), job_dir, job_completed_tx, mem_peak, @@ -2632,7 +2638,7 @@ async fn handle_queued_job( async fn process_result( job: Arc, - result: error::Result>, + result: error::Result>>, job_dir: &str, job_completed_tx: JobCompletedSender, mem_peak: i32, @@ -2707,7 +2713,7 @@ async fn process_result( job_completed_tx .send(JobCompleted { job: job, - result: to_raw_value(&error_value), + result: Arc::new(to_raw_value(&error_value)), mem_peak, canceled_by, success: false, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 8b558bd5bb..2bed31c9e8 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -17,6 +17,7 @@ use crate::common::{hash_args, save_in_cache}; use crate::js_eval::{eval_timeout, IdContext}; use crate::{AuthedClient, PreviousResult, SameWorkerPayload, SendResult, JOB_TOKEN, KEEP_JOB_DIR}; use anyhow::Context; +use mappable_rc::Marc; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use serde_json::{json, Value}; @@ -47,7 +48,7 @@ use windmill_common::{ use windmill_queue::schedule::get_schedule_opt; use windmill_queue::{ add_completed_job, add_completed_job_error, append_logs, get_queued_job, - handle_maybe_scheduled_job, CanceledBy, PushIsolationLevel, WrappedError, + handle_maybe_scheduled_job, CanceledBy, PushArgs, PushIsolationLevel, WrappedError, }; type DB = sqlx::Pool; @@ -56,7 +57,6 @@ use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTrans // #[instrument(level = "trace", skip_all)] pub async fn update_flow_status_after_job_completion< - 'a, R: rsmq_async::RsmqConnection + Send + Sync + Clone, >( db: &DB, @@ -65,7 +65,7 @@ pub async fn update_flow_status_after_job_completion< job_id_for_status: &Uuid, w_id: &str, success: bool, - result: &'a RawValue, + result: Arc>, unrecoverable: bool, same_worker_tx: Sender, worker_dir: &str, @@ -105,7 +105,7 @@ pub async fn update_flow_status_after_job_completion< &nrec.job_id_for_status, w_id, nrec.success, - nrec.result.as_ref(), + nrec.result, false, same_worker_tx.clone(), worker_dir, @@ -127,7 +127,9 @@ pub async fn update_flow_status_after_job_completion< &nrec.job_id_for_status, w_id, false, - &to_raw_value(&Json(&WrappedError { error: json!(e.to_string()) })), + Arc::new(to_raw_value(&Json(&WrappedError { + error: json!(e.to_string()), + }))), true, same_worker_tx.clone(), worker_dir, @@ -147,7 +149,7 @@ pub struct RecUpdateFlowStatusAfterJobCompletion { flow: uuid::Uuid, job_id_for_status: Uuid, success: bool, - result: Box, + result: Arc>, stop_early_override: Option, skip_error_handler: bool, } @@ -166,7 +168,6 @@ pub struct RowFlowStatus { } // #[instrument(level = "trace", skip_all)] pub async fn update_flow_status_after_job_completion_internal< - 'a, R: rsmq_async::RsmqConnection + Send + Sync + Clone, >( db: &DB, @@ -175,7 +176,7 @@ pub async fn update_flow_status_after_job_completion_internal< job_id_for_status: &Uuid, w_id: &str, mut success: bool, - result: &'a RawValue, + result: Arc>, unrecoverable: bool, same_worker_tx: Sender, worker_dir: &str, @@ -258,7 +259,7 @@ pub async fn update_flow_status_after_job_completion_internal< } else if is_failure_step { (false, false, false) } else { - let row = sqlx::query( + let r = sqlx::query_as::<_, SkipIfStopped>( "SELECT raw_flow->'modules'->$1::int->'stop_after_if'->>'expr' as stop_early_expr, (raw_flow->'modules'->$1::int->'stop_after_if'->>'skip_if_stopped')::bool as skip_if_stopped, @@ -272,19 +273,18 @@ pub async fn update_flow_status_after_job_completion_internal< .fetch_one(db) .await .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e:#}")))?; - let r = SkipIfStopped::from_row(&row)?; let stop_early = success && if let Some(expr) = r.stop_early_expr.clone() { compute_bool_from_expr( expr, - Arc::new( + Marc::new( r.args .map(|x| x.0) .unwrap_or_else(|| serde_json::from_str("{}").unwrap()) .to_owned(), ), - Arc::new(result.to_owned()), + result.clone(), None, Some(client), None, @@ -654,9 +654,9 @@ pub async fn update_flow_status_after_job_completion_internal< let nresult = match &new_status { Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. }) | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => { - retrieve_flow_jobs_results(db, w_id, jobs).await? + Arc::new(retrieve_flow_jobs_results(db, w_id, jobs).await?) } - _ => result.to_owned(), + _ => result.clone(), }; if old_status.retry.fail_count > 0 @@ -730,6 +730,8 @@ pub async fn update_flow_status_after_job_completion_internal< ) }; + let flow_job = Arc::new(flow_job); + let done = if !should_continue_flow { { let logs = if flow_job.canceled { @@ -837,10 +839,10 @@ pub async fn update_flow_status_after_job_completion_internal< } else { tracing::debug!(id = %flow_job.id, "start handle flow"); match handle_flow( - &flow_job, + flow_job.clone(), db, client, - Some(nresult.to_owned()), + Some(nresult.clone()), same_worker_tx.clone(), worker_dir, rsmq.clone(), @@ -886,7 +888,7 @@ pub async fn update_flow_status_after_job_completion_internal< flow: parent_job, job_id_for_status: flow, success: success && !is_failure_step, - result: nresult, + result: nresult.clone(), stop_early_override: if stop_early { Some(skip_if_stop_early) } else { @@ -906,7 +908,7 @@ async fn retrieve_flow_jobs_results( w_id: &str, job_uuids: &Vec, ) -> error::Result> { - let results = sqlx::query( + let results = sqlx::query_as::<_, BranchResults>( "SELECT result, id FROM completed_job WHERE id = ANY($1) AND workspace_id = $2", @@ -916,10 +918,7 @@ async fn retrieve_flow_jobs_results( .fetch_all(db) .await? .into_iter() - .map(|r| { - let br = BranchResults::from_row(&r).unwrap(); - (br.id, br.result.to_owned()) - }) + .map(|br| (br.id, br.result)) .collect::>(); let results = job_uuids @@ -1068,7 +1067,7 @@ fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> { async fn compute_bool_from_expr( expr: String, - flow_args: Arc>>, + flow_args: Marc>>, result: Arc>, by_id: Option, client: Option<&AuthedClient>, @@ -1163,7 +1162,7 @@ pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result { /// resumes should be in order of timestamp ascending, so that more recent are at the end #[instrument(level = "trace", skip_all)] async fn transform_input( - flow_args: Arc>>, + flow_args: Marc>>, last_result: Arc>, input_transforms: &HashMap, resumes: Arc>, @@ -1216,10 +1215,10 @@ async fn transform_input( #[instrument(level = "trace", skip_all)] pub async fn handle_flow( - flow_job: &QueuedJob, + flow_job: Arc, db: &sqlx::Pool, client: &AuthedClient, - last_result: Option>, + last_result: Option>>, same_worker_tx: Sender, worker_dir: &str, rsmq: Option, @@ -1251,7 +1250,7 @@ pub async fn handle_flow( if let Err(err) = handle_maybe_scheduled_job( rsmq.clone(), db, - flow_job, + &flow_job, &schedule, flow_job.script_path.as_ref().unwrap(), &flow_job.workspace_id, @@ -1278,7 +1277,7 @@ pub async fn handle_flow( flow, db, client, - last_result.to_owned(), + last_result, same_worker_tx, worker_dir, rsmq, @@ -1328,15 +1327,19 @@ fn potentially_crash_for_testing() { } } +// static +lazy_static::lazy_static! { + pub static ref EHM: HashMap> = HashMap::new(); +} // #[async_recursion] // #[instrument(level = "trace", skip_all)] async fn push_next_flow_job( - flow_job: &QueuedJob, + flow_job: Arc, mut status: FlowStatus, flow: FlowValue, db: &sqlx::Pool, client: &AuthedClient, - last_job_result: Option>, + last_job_result: Option>>, same_worker_tx: Sender, worker_dir: &str, rsmq: Option, @@ -1357,7 +1360,14 @@ async fn push_next_flow_job .cloned() .unwrap_or_else(|| status.failure_module.module_status.clone()); - let flow_job_args = flow_job.get_args(); + let fj: mappable_rc::Marc = flow_job.clone().into(); + let arc_flow_job_args: Marc>> = Marc::map(fj, |x| { + if let Some(args) = &x.args { + &args.0 + } else { + &EHM + } + }); // if this is an empty module of if the module has already been completed, successfully, update the parent flow if flow.modules.is_empty() || matches!(status_module, FlowStatusModule::Success { .. }) { @@ -1366,7 +1376,7 @@ async fn push_next_flow_job flow: flow_job.id, success: true, result: if flow.modules.is_empty() { - to_raw_value(&flow_job_args) + to_raw_value(arc_flow_job_args.as_ref()) } else { // it has to be an empty for loop event serde_json::from_str("[]").unwrap() @@ -1386,8 +1396,6 @@ async fn push_next_flow_job return Ok(()); } - let arc_flow_job_args = Arc::new(flow_job_args.clone()); - if i == 0 { if !flow_job.is_flow_step && flow_job.schedule_path.is_some() { let no_flow_overlap = sqlx::query_scalar!( @@ -1470,7 +1478,7 @@ async fn push_next_flow_job // Compute and initialize last_job_result let arc_last_job_result = if status_module.is_failure() { // if job is being retried, pass the result of its previous failure - Arc::new(last_job_result.unwrap_or(to_raw_value(&json!("{}")))) + last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}")))) } else if i == 0 { // if it's the first job executed in the flow, pass the flow args Arc::new(to_raw_value(&flow_job.args)) @@ -1479,7 +1487,7 @@ async fn push_next_flow_job // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a // flow restart from a specific step if last_job_result.is_some() { - Arc::new(last_job_result.unwrap()) + last_job_result.unwrap() } else { match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status).await? { None => Arc::new(to_raw_value(&json!("{}"))), @@ -1514,25 +1522,19 @@ async fn push_next_flow_job .await .context("lock flow in queue")?; - let resumes = sqlx::query( + let resumes = sqlx::query_as::<_, ResumeRow>( "SELECT value, approver, resume_id, approved FROM resume_job WHERE job = $1 ORDER BY created_at ASC", ) .bind(last) .fetch_all(&mut *tx) .await? .into_iter() - .map(|x| ResumeRow::from_row(&x)) .collect::>(); - resume_messages.extend( - resumes - .iter() - .map(|r| to_raw_value(&r.as_ref().map(|x| x.value.clone()).ok())), - ); + resume_messages.extend(resumes.iter().map(|r| to_raw_value(&r.value))); approvers.extend(resumes.iter().map(|r| { - r.as_ref() - .ok() - .and_then(|x| x.approver.clone()) + r.approver + .clone() .as_deref() .unwrap_or_else(|| "anonymous") .to_string() @@ -1599,9 +1601,7 @@ async fn push_next_flow_job .await?; } - let is_disapproved = resumes - .iter() - .find(|x| x.as_ref().is_ok_and(|x| !x.approved)); + let is_disapproved = resumes.iter().find(|x| !x.approved); if is_disapproved.is_none() && resume_messages.len() >= required_events as usize { sqlx::query( "UPDATE queue @@ -1612,11 +1612,9 @@ async fn push_next_flow_job .bind(json!(resumes .into_iter() .map(|r| Approval { - resume_id: r.as_ref().map(|x| x.resume_id).unwrap_or_default() as u16, + resume_id: r.resume_id as u16, approver: r - .as_ref() - .ok() - .and_then(|x| x.approver.clone()) + .approver.clone() .unwrap_or_else(|| "unknown".to_string()) }) .collect::>())) @@ -1676,10 +1674,7 @@ async fn push_next_flow_job let (logs, error_name) = if let Some(disapprover) = is_disapproved { ( - format!( - "Disapproved by {:?}", - disapprover.as_ref().unwrap().approver - ), + format!("Disapproved by {:?}", disapprover.approver), "SuspendedDisapproved", ) } else { @@ -1875,80 +1870,83 @@ async fn push_next_flow_job drop(resume_messages); - let args: windmill_common::error::Result<_> = if module.mock.is_some() - && module.mock.as_ref().unwrap().enabled - { - let mut hm = HashMap::new(); - hm.insert( - "previous_result".to_string(), - to_raw_value( - &module - .mock - .as_ref() - .unwrap() - .return_value - .clone() - .unwrap_or_else(|| serde_json::from_str("null").unwrap()), - ), - ); - Ok(hm) - } else if let Some(id) = get_args_from_id { - let row = sqlx::query("SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2") + let args: windmill_common::error::Result<_> = + if module.mock.is_some() && module.mock.as_ref().unwrap().enabled { + let mut hm = HashMap::new(); + hm.insert( + "previous_result".to_string(), + to_raw_value( + &module + .mock + .as_ref() + .unwrap() + .return_value + .clone() + .unwrap_or_else(|| serde_json::from_str("null").unwrap()), + ), + ); + Ok(Marc::new(hm)) + } else if let Some(id) = get_args_from_id { + let row = sqlx::query_as::<_, RawArgs>( + "SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2", + ) .bind(id) .bind(&flow_job.workspace_id) .fetch_optional(db) .await?; - if let Some(row) = row { - RawArgs::from_row(&row) - .map(|x| x.args.map(|x| x.0).unwrap_or_else(HashMap::new)) - .map_err(|e| error::Error::InternalErr(format!("Impossible to build args: {e:#}"))) + if let Some(raw_args) = row { + Ok(Marc::new( + raw_args.args.map(|x| x.0).unwrap_or_else(HashMap::new), + )) + } else { + Ok(Marc::new(HashMap::new())) + } } else { - Ok(HashMap::new()) - } - } else { - match &module.get_value() { - Ok( - FlowModuleValue::Script { input_transforms, .. } - | FlowModuleValue::RawScript { input_transforms, .. } - | FlowModuleValue::Flow { input_transforms, .. }, - ) => { - let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; - transform_context = Some(ctx); - let by_id = transform_context.as_ref().unwrap(); - transform_input( - arc_flow_job_args.clone(), - arc_last_job_result.clone(), - input_transforms, - resumes.clone(), - resume.clone(), - approvers.clone(), - by_id, - client, + match &module.get_value() { + Ok( + FlowModuleValue::Script { input_transforms, .. } + | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::Flow { input_transforms, .. }, + ) => { + let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; + transform_context = Some(ctx); + let by_id = transform_context.as_ref().unwrap(); + transform_input( + arc_flow_job_args.clone(), + arc_last_job_result.clone(), + input_transforms, + resumes.clone(), + resume.clone(), + approvers.clone(), + by_id, + client, + ) + .await + .map(Marc::new) + } + Ok(FlowModuleValue::Identity) => serde_json::from_str( + &serde_json::to_string(&PreviousResult { + previous_result: Some(&arc_last_job_result), + }) + .unwrap(), ) - .await - } - Ok(FlowModuleValue::Identity) => serde_json::from_str( - &serde_json::to_string(&PreviousResult { - previous_result: Some(&arc_last_job_result), - }) - .unwrap(), - ) - .map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))), + .map(Marc::new) + .map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))), - Ok(_) => Ok(flow_job_args), - Err(e) => { - return Err(error::Error::InternalErr(format!( - "module was not convertible to acceptable value {e:?}" - ))) + Ok(_) => Ok(arc_flow_job_args.clone()), + Err(e) => { + return Err(error::Error::InternalErr(format!( + "module was not convertible to acceptable value {e:?}" + ))) + } } - } - }; + }; tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed"); let next_flow_transform = compute_next_flow_transform( arc_flow_job_args.clone(), arc_last_job_result.clone(), - flow_job, + &flow_job, &flow, transform_context, db, @@ -2007,7 +2005,7 @@ async fn push_next_flow_job }; let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); - + let nargs = args.as_ref(); for i in (0..len).into_iter() { if i % 100 == 0 && i != 0 { tracing::info!(id = %flow_job.id, root_id = %job_root, "pushed (non-commited yet) first {i} subflows of {len}"); @@ -2039,19 +2037,20 @@ async fn push_next_flow_job None }; - let transform_inp; + let marc; + let me; let args = match &next_status { NextStatus::AllFlowJobs { branchall: Some(BranchAllStatus { .. }), iterator: None, .. - } => args.as_ref().map(|args| args.clone()), + } => nargs, NextStatus::NextLoopIteration { next: ForloopNextIteration { new_args, .. }, simple_input_transforms, } => { - let mut args = if let Ok(args) = args.as_ref() { - args.clone() + let mut args = if let Ok(args) = nargs { + args.as_ref().clone() } else { HashMap::new() }; @@ -2061,8 +2060,8 @@ async fn push_next_flow_job if let Some(input_transforms) = simple_input_transforms { //previous id is none because we do not want to use previous id if we are in a for loop let ctx = get_transform_context(&flow_job, "", &status).await?; - transform_inp = transform_input( - Arc::new(args), + let ti = transform_input( + Marc::new(args), arc_last_job_result.clone(), input_transforms, resumes.clone(), @@ -2071,10 +2070,23 @@ async fn push_next_flow_job &ctx, client, ) - .await; - transform_inp.as_ref().map(|args| args.clone()) + .await + .map_err(|e| { + Error::ExecutionErr( + format!("could not transform input using an expr: {e}",), + ) + }) + .map(Marc::new); + if let Ok(ti) = ti { + marc = ti; + Ok(&marc) + } else { + me = ti.unwrap_err(); + Err(&me) + } } else { - Ok(args) + marc = Marc::new(args); + Ok(&marc) } } NextStatus::AllFlowJobs { @@ -2084,7 +2096,7 @@ async fn push_next_flow_job } => { if let Ok(args) = args.as_ref() { let mut hm = HashMap::new(); - for (k, v) in args { + for (k, v) in args.iter() { hm.insert(k.to_string(), v.to_owned()); } insert_iter_arg( @@ -2094,8 +2106,8 @@ async fn push_next_flow_job ); if let Some(input_transforms) = simple_input_transforms { let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; - transform_inp = transform_input( - Arc::new(hm), + let ti = transform_input( + Marc::new(hm), arc_last_job_result.clone(), input_transforms, resumes.clone(), @@ -2104,20 +2116,45 @@ async fn push_next_flow_job &ctx, client, ) - .await; - transform_inp.as_ref().map(|args| args.clone()) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "could not transform input using an expr: {e}" + )) + }) + .map(Marc::new); + if let Ok(ti) = ti { + marc = ti; + Ok(&marc) + } else { + me = ti.unwrap_err(); + Err(&me) + } } else { - Ok(hm) + marc = Marc::new(hm); + Ok(&marc) } } else { - args.as_ref().map(|args| args.clone()) + nargs } } - _ => args.as_ref().map(|args| args.clone()), + _ => nargs, }; - let (ok, err) = match args { - Ok(v) => (Some(v), None), - Err(e) => (None, Some(e)), + + let push_args; + let err; + let ov; + + match args { + Ok(v) => { + ov = v; + push_args = PushArgs::from(ov.as_ref()); + err = None; + } + Err(e) => { + push_args = PushArgs::from(&*EHM); + err = Some(e); + } }; tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed args for job {i} of {len}"); @@ -2161,10 +2198,7 @@ async fn push_next_flow_job tx2, &flow_job.workspace_id, payload_tag.payload, - windmill_queue::PushArgs { - args: ok.unwrap_or_else(|| serde_json::from_str("{}").unwrap()), - extra: HashMap::new(), - }, + push_args, &flow_job.created_by, &flow_job.email, flow_job.permissioned_as.to_owned(), @@ -2514,7 +2548,7 @@ fn insert_iter_arg( } async fn compute_next_flow_transform( - arc_flow_job_args: Arc>>, + arc_flow_job_args: Marc>>, arc_last_job_result: Arc>, flow_job: &QueuedJob, flow: &FlowValue, @@ -3041,7 +3075,7 @@ async fn next_forloop_status( resumes: Arc>, resume: Arc>, approvers: Arc>, - arc_flow_job_args: Arc>>, + arc_flow_job_args: Marc>>, client: &AuthedClient, parallel: &bool, ) -> Result { diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index c0ea748464..5f50648863 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -451,7 +451,7 @@ async fn trigger_dependents_to_recompute_dependencies< tx, &w_id, job_payload, - windmill_queue::PushArgs { args, extra: HashMap::new() }, + windmill_queue::PushArgs { args: &args, extra: None }, &created_by, email, permissioned_as.to_string(), diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index 658268dd30..a1b7edc7a1 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -714,7 +714,11 @@ multiple={false} /> {#if value?.length} -
File length: {value.length} base64 chars
+
File length: {value.length} base64 chars ({(value.length / 1024 / 1024).toFixed( + 2 + )}MB)
{/if} {:else if inputCat == 'resource-string'} diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 327335f41a..d693bfd45b 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -586,7 +586,7 @@ $localDurationStatuses[flowJobIds?.moduleId ?? '']?.iteration_from ?? 0 )}

- Embedded flows: ({flowJobIds?.flowJobs.length} items) + Subflows: ({flowJobIds?.flowJobs.length} items)

{#if (flowJobIds?.flowJobs.length ?? 0) > 20 && lenToAdd > 0} {@const allToAdd = (flowJobIds?.length ?? 0) - (slicedListJobIds.length ?? 0)} diff --git a/frontend/src/lib/components/JobArgs.svelte b/frontend/src/lib/components/JobArgs.svelte index 6942545e80..5880a71c3b 100644 --- a/frontend/src/lib/components/JobArgs.svelte +++ b/frontend/src/lib/components/JobArgs.svelte @@ -53,7 +53,7 @@ ${Object.entries(args) {#if id && workspace && args && typeof args === 'object' && deepEqual( Object.keys(args), ['reason'] ) && args['reason'] == 'WINDMILL_TOO_BIG'} - The args are too big in size to be able to fetch s3. Please download the JSON file to view them. diff --git a/frontend/src/lib/components/LightweightArgInput.svelte b/frontend/src/lib/components/LightweightArgInput.svelte index 872233291a..b46ac00d24 100644 --- a/frontend/src/lib/components/LightweightArgInput.svelte +++ b/frontend/src/lib/components/LightweightArgInput.svelte @@ -480,7 +480,10 @@ multiple={false} /> {#if value?.length} -
File length: {value.length} base64 chars
File length: {value.length} base64 chars ({(value.length / 1024 / 1024).toFixed( + 2 + )}MB) {/if} diff --git a/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte b/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte index ae859591c8..e52e71b5d7 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte @@ -8,12 +8,23 @@ import { getContext } from 'svelte' import { NEVER_TESTED_THIS_FAR } from '../models' import Section from '$lib/components/Section.svelte' + import { getStepPropPicker } from '../previousResults' - const { flowStateStore } = getContext('FlowEditorContext') + const { flowStateStore, flowStore, previewArgs } = + getContext('FlowEditorContext') export let flowModule: FlowModule let editor: SimpleEditor | undefined = undefined + $: stepPropPicker = getStepPropPicker( + $flowStateStore, + undefined, + undefined, + flowModule.id, + $flowStore, + $previewArgs, + false + ) $: isStopAfterIfEnabled = Boolean(flowModule.stop_after_if) $: result = $flowStateStore[flowModule.id]?.previewResult ?? NEVER_TESTED_THIS_FAR @@ -52,16 +63,19 @@ > {#if flowModule.stop_after_if} - Stop condition expression + Stop condition expression
{ editor?.insertAtCursor(detail) editor?.focus() @@ -79,10 +93,11 @@ {:else} Stop condition expression + /> Stop condition expression