diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 9ce7a15c02..e7b8f6067a 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -58,7 +58,8 @@ use windmill_common::{ jobs::{get_payload_tag_from_prefixed_path, JobPayload, RawCode}, users::username_to_permissioned_as, utils::{ - http_get_from_hub, not_found_if_none, paginate, query_elems_from_hub, require_admin, Pagination, StripPath + http_get_from_hub, not_found_if_none, paginate, query_elems_from_hub, require_admin, + Pagination, StripPath, }, variables::{build_crypt, build_crypt_with_key_suffix}, worker::{to_raw_value, CLOUD_HOSTED}, @@ -558,12 +559,11 @@ async fn update_app_history( return Ok(()); } - async fn custom_path_exists( Extension(db): Extension, Path((w_id, custom_path)): Path<(String, String)>, ) -> JsonResult { - let exists = + let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM app WHERE custom_path = $1 AND ($2::TEXT IS NULL OR workspace_id = $2))", custom_path, @@ -658,7 +658,6 @@ async fn get_public_app_by_secret( Ok(Json(app)) } - async fn get_public_resource( Extension(db): Extension, Path((w_id, path)): Path<(String, StripPath)>, @@ -742,7 +741,6 @@ async fn create_app( } if let Some(custom_path) = &app.custom_path { - require_admin(authed.is_admin, &authed.username)?; let exists = sqlx::query_scalar!( @@ -778,7 +776,9 @@ async fn create_app( app.summary, json!(app.policy), app.draft_only, - app.custom_path.map(|s| if s.is_empty() { None } else { Some(s) }).flatten() + app.custom_path + .map(|s| if s.is_empty() { None } else { Some(s) }) + .flatten() ) .fetch_one(&mut *tx) .await?; @@ -1019,7 +1019,6 @@ async fn update_app( } if let Some(ncustom_path) = &ns.custom_path { - require_admin(authed.is_admin, &authed.username)?; if ncustom_path.is_empty() { @@ -1034,7 +1033,7 @@ async fn update_app( ) .fetch_one(&mut *tx) .await?.unwrap_or(false); - + if exists { return Err(Error::BadRequest(format!( "App with custom path {} already exists", @@ -1306,15 +1305,8 @@ async fn execute_component( force_viewer_allow_user_resources: Some(allow_user_resources), .. } => ( - &Policy { - execution_mode: ExecutionMode::Viewer, - ..Default::default() - }, - &PolicyTriggerableInputs { - static_inputs, - one_of_inputs, - allow_user_resources, - }, + &Policy { execution_mode: ExecutionMode::Viewer, ..Default::default() }, + &PolicyTriggerableInputs { static_inputs, one_of_inputs, allow_user_resources }, ), // 2. "run" mode. _ => { diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ef37834400..4b442588ed 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3221,7 +3221,7 @@ pub async fn push<'c, 'd>( None, None, ) - }, + } JobPayload::AppScript { id, // app_script(id). path, diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index c16413c573..ad1333c28d 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -16,8 +16,7 @@ use windmill_common::error::Error; use windmill_common::flows::FlowValue; use windmill_common::worker::WORKER_CONFIG; use windmill_common::{ - cache, - error, + cache, error, flows::{FlowModule, FlowModuleValue}, jobs::QueuedJob, scripts::{ScriptHash, ScriptLang}, @@ -416,17 +415,18 @@ async fn spawn_dedicated_workers_for_flow( job_completed_tx, Some(module.id.clone()), ) - .await + .await { workers.push(dedi_w); } - }, + } Err(err) => tracing::error!( "failed to get script for module: {:?}, err: {:?}", - module, err - ) + module, + err + ), } - }, + } FlowModuleValue::Flow { .. } => (), FlowModuleValue::Identity => (), } diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 6016c9349c..90969fae04 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -3,7 +3,7 @@ use std::{ fs, path::Path, process::Stdio, - sync::Arc + sync::Arc, }; use anyhow::anyhow; @@ -351,7 +351,12 @@ pub async fn uv_pip_compile( occupancy_metrics, ) .await - .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed.\n\ncommand: {uv_cmd} {}\n\n{e:?}", args.join(" "))))?; + .map_err(|e| { + Error::ExecutionErr(format!( + "Lock file generation failed.\n\ncommand: {uv_cmd} {}\n\n{e:?}", + args.join(" ") + )) + })?; } let path_lock = format!("{job_dir}/requirements.txt"); @@ -578,7 +583,6 @@ pub async fn handle_python_job( create_args_and_out_file(&client, job, job_dir, db).await?; tracing::debug!("Finished preparing wrapper"); - let preprocessor = if let Some(pre_spread) = pre_spread { format!( r#"if inner_script.preprocessor is None or not callable(inner_script.preprocessor): @@ -763,7 +767,8 @@ mount {{ .stdout(Stdio::piped()) .stderr(Stdio::piped()); - #[cfg(windows)] { + #[cfg(windows)] + { python_cmd.env("SystemRoot", SYSTEM_ROOT.as_str()); python_cmd.env("USERPROFILE", crate::USERPROFILE_ENV.as_str()); } @@ -1253,29 +1258,28 @@ async fn spawn_uv_install( #[cfg(unix)] { if no_uv_install { - let mut flock_cmd = Command::new(FLOCK_PATH.as_str()); - flock_cmd - .env_clear() - .envs(PROXY_ENVS.clone()) - .envs(envs) - .args([ - "-x", - &format!( - "{}/{}-{}.lock", - LOCK_CACHE_DIR, - if no_uv_install { "pip" } else { "py311" }, - fssafe_req - ), - "--command", - &command_args.join(" "), - ]) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()); - start_child_process(flock_cmd, FLOCK_PATH.as_str()).await + let mut flock_cmd = Command::new(FLOCK_PATH.as_str()); + flock_cmd + .env_clear() + .envs(PROXY_ENVS.clone()) + .envs(envs) + .args([ + "-x", + &format!( + "{}/{}-{}.lock", + LOCK_CACHE_DIR, + if no_uv_install { "pip" } else { "py311" }, + fssafe_req + ), + "--command", + &command_args.join(" "), + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + start_child_process(flock_cmd, FLOCK_PATH.as_str()).await } else { let mut cmd = Command::new(command_args[0]); - cmd - .env_clear() + cmd.env_clear() .envs(PROXY_ENVS.clone()) .envs(envs) .args(&command_args[1..]) @@ -1335,7 +1339,6 @@ pub async fn handle_python_reqs( mut no_uv_install: bool, is_ansible: bool, ) -> error::Result> { - let counter_arc = Arc::new(tokio::sync::Mutex::new(0)); // Append logs with line like this: // [9/21] + requests==2.32.3 << (S3) | in 57ms @@ -1419,7 +1422,6 @@ pub async fn handle_python_reqs( .map(handle_ephemeral_token), ); - // Prepare NSJAIL if !*DISABLE_NSJAIL { let _ = write_file( @@ -1475,12 +1477,19 @@ pub async fn handle_python_reqs( } } if in_cache.len() > 0 { - append_logs(&job_id, w_id, format!("\nenv deps from local cache: {}\n", in_cache.join(", ")), db).await; + append_logs( + &job_id, + w_id, + format!("\nenv deps from local cache: {}\n", in_cache.join(", ")), + db, + ) + .await; } let (kill_tx, ..) = tokio::sync::broadcast::channel::<()>(1); - let kill_rxs: Vec> = - (0..req_with_penv.len()).map(|_| kill_tx.subscribe()).collect(); + let kill_rxs: Vec> = (0..req_with_penv.len()) + .map(|_| kill_tx.subscribe()) + .collect(); // ________ Read comments at the end of the function to get more context let (_done_tx, mut done_rx) = tokio::sync::mpsc::channel::<()>(1); @@ -1520,21 +1529,21 @@ pub async fn handle_python_reqs( if canceled { tracing::info!( - // If there is listener on other side, + // If there is listener on other side, workspace_id = %w_id_2, "cancelling installations", ); if let Err(ref e) = kill_tx.send(()){ tracing::error!( - // If there is listener on other side, + // If there is listener on other side, workspace_id = %w_id_2, "failed to send done: Probably receiving end closed too early or have not opened yet\n{}", // If there is no listener, it will be dropped safely e ); } - } + } } // Once done_tx is dropped, this will be fired _ = done_rx.recv() => break @@ -1569,9 +1578,15 @@ pub async fn handle_python_reqs( // Do we use Nsjail? if !*DISABLE_NSJAIL { - logs.push_str(&format!("\nStarting isolated installation... ({} tasks in parallel) \n", parallel_limit)); + logs.push_str(&format!( + "\nStarting isolated installation... ({} tasks in parallel) \n", + parallel_limit + )); } else { - logs.push_str(&format!("\nStarting installation... ({} tasks in parallel) \n", parallel_limit)); + logs.push_str(&format!( + "\nStarting installation... ({} tasks in parallel) \n", + parallel_limit + )); } append_logs(&job_id, w_id, logs, db).await; } @@ -1631,13 +1646,12 @@ pub async fn handle_python_reqs( tokio::select! { // Cancel was called on the job _ = kill_rx.recv() => return Err(anyhow::anyhow!("S3 pull was canceled")), - pull = pull_from_tar(os, venv_p.clone(), no_uv_install) => { if let Err(e) = pull { tracing::info!( workspace_id = %w_id, "No tarball was found on S3 or different problem occured {job_id}:\n{e}", - ); + ); } else { print_success( true, @@ -1667,12 +1681,12 @@ pub async fn handle_python_reqs( no_uv_install, ).await { Ok(r) => r, - Err(e) => { + Err(e) => { append_logs( &job_id, w_id, format!( - "\nError while spawning proccess:\n{e}", + "\nError while spawning proccess:\n{e}", ), db, ) @@ -1712,7 +1726,7 @@ pub async fn handle_python_reqs( &job_id, w_id, format!( - "\nError while installing {}:\n{buf}", + "\nError while installing {}:\n{buf}", &req ), db, @@ -1771,7 +1785,10 @@ pub async fn handle_python_reqs( let mut failed = false; for (handle, (_, venv_p)) in handles.into_iter().zip(req_with_penv.into_iter()) { - if let Err(e) = handle.await.unwrap_or(Err(anyhow!("Problem by joining handle"))) { + if let Err(e) = handle + .await + .unwrap_or(Err(anyhow!("Problem by joining handle"))) + { failed = true; tracing::warn!( workspace_id = %w_id, @@ -1792,19 +1809,11 @@ pub async fn handle_python_reqs( if has_work { let total_time = total_time.elapsed().as_millis(); - append_logs( - &job_id, - w_id, - format!( - "\nenv set in {}ms", - total_time - ), - db, - ).await; + append_logs(&job_id, w_id, format!("\nenv set in {}ms", total_time), db).await; } // Usually done_tx will drop after this return - // If there is listener on other side, + // If there is listener on other side, // it will be triggered // If there is no listener, it will be dropped safely return if failed { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 4af815cd76..1d6dcc37d4 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -9,7 +9,6 @@ // #[cfg(feature = "otel")] // use opentelemetry::{global, KeyValue}; - use windmill_common::{ apps::AppScriptId, auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET}, @@ -92,12 +91,30 @@ use tokio::{ use rand::Rng; use crate::{ - ansible_executor::handle_ansible_job, bash_executor::{handle_bash_job, handle_powershell_job}, bun_executor::handle_bun_job, common::{ + ansible_executor::handle_ansible_job, + bash_executor::{handle_bash_job, handle_powershell_job}, + bun_executor::handle_bun_job, + common::{ build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args, update_worker_ping_for_failed_init_script, OccupancyMetrics, - }, deno_executor::handle_deno_job, go_executor::handle_go_job, graphql_executor::do_graphql, handle_child::SLOW_LOGS, handle_job_error, job_logger::NO_LOGS_AT_ALL, js_eval::{eval_fetch_timeout, transpile_ts}, mysql_executor::do_mysql, pg_executor::do_postgresql, php_executor::handle_php_job, python_executor::handle_python_job, result_processor::{process_result, start_background_processor}, rust_executor::handle_rust_job, worker_flow::{handle_flow, update_flow_status_in_progress, Step}, worker_lockfiles::{ + }, + deno_executor::handle_deno_job, + go_executor::handle_go_job, + graphql_executor::do_graphql, + handle_child::SLOW_LOGS, + handle_job_error, + job_logger::NO_LOGS_AT_ALL, + js_eval::{eval_fetch_timeout, transpile_ts}, + mysql_executor::do_mysql, + pg_executor::do_postgresql, + php_executor::handle_php_job, + python_executor::handle_python_job, + result_processor::{process_result, start_background_processor}, + rust_executor::handle_rust_job, + worker_flow::{handle_flow, update_flow_status_in_progress, Step}, + worker_lockfiles::{ handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job, - } + }, }; use backon::ConstantBuilder; @@ -951,7 +968,6 @@ pub async fn run_worker( None }; - // let worker_resource = &[ // KeyValue::new("hostname", hostname.to_string()), // KeyValue::new("worker", worker_name.to_string()), @@ -960,7 +976,6 @@ pub async fn run_worker( // let meter = global::meter("windmill"); // let counter = meter.u64_counter("jobs.execution").build(); - let mut occupancy_metrics = OccupancyMetrics::new(start_time); let mut jobs_executed = 0; @@ -1098,7 +1113,7 @@ pub async fn run_worker( if !valid_key { tracing::error!( - worker = %worker_name, hostname = %hostname, + worker = %worker_name, hostname = %hostname, "Invalid license key, workers require a valid license key, sleeping for 30s waiting for valid key to be set" ); tokio::time::sleep(Duration::from_secs(10)).await; @@ -1168,19 +1183,19 @@ pub async fn run_worker( ) .notify(|err, dur| { tracing::error!( - worker = %worker_name, hostname = %hostname, + worker = %worker_name, hostname = %hostname, "retrying updating worker ping in {dur:#?}, err: {err:#?}" ); }) .sleep(tokio::time::sleep) .await { tracing::error!( - worker = %worker_name, hostname = %hostname, + worker = %worker_name, hostname = %hostname, "failed to update worker ping, exiting: {}", e); killpill_tx.send(()).unwrap_or_default(); } tracing::info!( - worker = %worker_name, hostname = %hostname, + worker = %worker_name, hostname = %hostname, "ping update, memory: container={}MB, windmill={}MB", memory_usage.unwrap_or_default() / (1024 * 1024), wm_memory_usage.unwrap_or_default() / (1024 * 1024) @@ -1239,7 +1254,7 @@ pub async fn run_worker( if let Ok(same_worker_job) = same_worker_rx.try_recv() { same_worker_queue_size.fetch_sub(1, Ordering::SeqCst); tracing::debug!( - worker = %worker_name, hostname = %hostname, + worker = %worker_name, hostname = %hostname, "received {} from same worker channel", same_worker_job.job_id ); @@ -1252,7 +1267,7 @@ pub async fn run_worker( .map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string())); if r.is_err() && !same_worker_job.recoverable { tracing::error!( - worker = %worker_name, hostname = %hostname, + worker = %worker_name, hostname = %hostname, "failed to fetch same_worker job on a non recoverable job, exiting" ); job_completed_tx @@ -1528,7 +1543,6 @@ pub async fn run_worker( let arc_job = Arc::new(job); add_time!(bench, "handle_queued_job START"); - let span = tracing::span!(tracing::Level::INFO, "job", job_id = %arc_job.id, root_job = field::Empty, workspace_id = %arc_job.workspace_id, worker = %worker_name, hostname = %hostname, tag = %arc_job.tag, language = field::Empty, @@ -1620,8 +1634,6 @@ pub async fn run_worker( _ => {} } - - #[cfg(feature = "prometheus")] if let Some(duration) = _timer.map(|x| x.stop_and_record()) { register_metric( @@ -1728,8 +1740,6 @@ pub async fn run_worker( tracing::info!(worker = %worker_name, hostname = %hostname, "number of jobs executed: {}", jobs_executed); } - - async fn queue_init_bash_maybe<'c>( db: &Pool, same_worker_tx: SameWorkerSender, @@ -1874,7 +1884,6 @@ async fn handle_queued_job( ) -> windmill_common::error::Result { // Extract the active span from the context - if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job))); } @@ -2219,7 +2228,6 @@ async fn handle_queued_job( } } - pub fn build_envs( envs: Option>, ) -> windmill_common::error::Result> { @@ -2326,7 +2334,6 @@ async fn handle_code_execution_job( new_args: &mut Option>>, occupancy_metrics: &mut OccupancyMetrics, killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, - ) -> error::Result> { let ContentReqLangEnvs { content: inner_content, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index b5770166c5..d7bdaed7fd 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -146,11 +146,11 @@ pub async fn update_flow_status_after_job_completion( } UpdateFlowStatusAfterJobCompletion::Rec(nrec) => { rec = nrec; - }, + } UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch => { add_time!(bench, "update flow status internal END"); return Ok(None); - }, + } UpdateFlowStatusAfterJobCompletion::NotDone => { add_time!(bench, "update flow status internal END"); return Ok(None); @@ -285,93 +285,96 @@ pub async fn update_flow_status_after_job_completion_internal( let is_failure_step = old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0; - let (mut stop_early, mut skip_if_stop_early, continue_on_error) = if let Some(se) = - stop_early_override - { - //do not stop early if module is a flow step - let step = match module_step { - Step::PreprocessorStep => None, - Step::FailureStep => None, - Step::Step(i) => Some(i), - }; + let (mut stop_early, mut skip_if_stop_early, continue_on_error) = + if let Some(se) = stop_early_override { + //do not stop early if module is a flow step + let step = match module_step { + Step::PreprocessorStep => None, + Step::FailureStep => None, + Step::Step(i) => Some(i), + }; - let is_flow = if let Some(_) = step { - #[derive(Deserialize)] - struct GetType<'j> { r#type: &'j str } + let is_flow = if let Some(_) = step { + #[derive(Deserialize)] + struct GetType<'j> { + r#type: &'j str, + } - current_module - .as_ref() - .map(|module| serde_json::from_str::(module.value.get()).map(|v| v.r#type == "flow")) - .unwrap_or(Ok(false)) - .unwrap_or(false) - } else { - false - }; - - if is_flow { - (false, false, false) - } else { - (true, se, false) - } - } else if is_failure_step || matches!(module_step, Step::PreprocessorStep) { - (false, false, false) - } else if let Some(current_module) = current_module.as_ref() { - let stop_early = success - && !is_branch_all - && if let Some(ref expr) = current_module - .stop_after_if - .as_ref() - .map(|x| x.expr.clone()) - { - let all_iters = match &module_status { - FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), .. } - if expr.contains("all_iters") => - { - Some(Arc::new( - retrieve_flow_jobs_results(db, w_id, flow_jobs).await?, - )) - } - _ => None, - }; - let args = sqlx::query_as::<_, RowArgs>( - "SELECT - args - FROM queue - WHERE id = $2", - ) - .bind(old_status.step) - .bind(flow) - .fetch_one(db) - .await - .map_err(|e| { - Error::InternalErr(format!("retrieval of args from state: {e:#}")) - })?; - compute_bool_from_expr( - &expr, - Marc::new(args.args.unwrap_or_default().0), - result.clone(), - all_iters, - None, - Some(client), - None, - None, - ) - .await? + current_module + .as_ref() + .map(|module| { + serde_json::from_str::(module.value.get()) + .map(|v| v.r#type == "flow") + }) + .unwrap_or(Ok(false)) + .unwrap_or(false) } else { false }; - ( - stop_early, - current_module - .stop_after_if - .as_ref() - .map(|x| x.skip_if_stopped) - .unwrap_or(false), - current_module.continue_on_error.unwrap_or(false), - ) - } else { - (false, false, false) - }; + + if is_flow { + (false, false, false) + } else { + (true, se, false) + } + } else if is_failure_step || matches!(module_step, Step::PreprocessorStep) { + (false, false, false) + } else if let Some(current_module) = current_module.as_ref() { + let stop_early = success + && !is_branch_all + && if let Some(ref expr) = current_module + .stop_after_if + .as_ref() + .map(|x| x.expr.clone()) + { + let all_iters = + match &module_status { + FlowStatusModule::InProgress { + flow_jobs: Some(flow_jobs), .. + } if expr.contains("all_iters") => Some(Arc::new( + retrieve_flow_jobs_results(db, w_id, flow_jobs).await?, + )), + _ => None, + }; + let args = sqlx::query_as::<_, RowArgs>( + "SELECT + args + FROM queue + WHERE id = $2", + ) + .bind(old_status.step) + .bind(flow) + .fetch_one(db) + .await + .map_err(|e| { + Error::InternalErr(format!("retrieval of args from state: {e:#}")) + })?; + compute_bool_from_expr( + &expr, + Marc::new(args.args.unwrap_or_default().0), + result.clone(), + all_iters, + None, + Some(client), + None, + None, + ) + .await? + } else { + false + }; + ( + stop_early, + current_module + .stop_after_if + .as_ref() + .map(|x| x.skip_if_stopped) + .unwrap_or(false), + current_module.continue_on_error.unwrap_or(false), + ) + } else { + (false, false, false) + }; let skip_branch_failure = match module_status { FlowStatusModule::InProgress { @@ -433,7 +436,7 @@ pub async fn update_flow_status_after_job_completion_internal( None }; - let nindex = if let Some(position) = position { + let nindex = if let Some(position) = position { sqlx::query_scalar!( "UPDATE queue SET flow_status = JSONB_SET( @@ -480,7 +483,7 @@ pub async fn update_flow_status_after_job_completion_internal( None }; - let nindex = if let Some(position) = position { + let nindex = if let Some(position) = position { sqlx::query_scalar!( "UPDATE queue SET flow_status = JSONB_SET( @@ -690,16 +693,15 @@ pub async fn update_flow_status_after_job_completion_internal( } } if success || (flow_jobs.is_some() && (skip_loop_failures || skip_branch_failure)) { - let is_skipped = if current_module.as_ref().is_some_and(|m| m.skip_if.is_some()) { + let is_skipped = if current_module.as_ref().is_some_and(|m| m.skip_if.is_some()) + { sqlx::query_scalar!( "SELECT job_kind = 'identity' FROM completed_job WHERE id = $1", job_id_for_status ) .fetch_one(db) .await - .map_err(|e| { - Error::InternalErr(format!("error during skip check: {e:#}")) - })? + .map_err(|e| Error::InternalErr(format!("error during skip check: {e:#}")))? .unwrap_or(false) } else { false @@ -918,13 +920,15 @@ pub async fn update_flow_status_after_job_completion_internal( .context("remove flow status retry")?; } - let flow_job = sqlx::query_as::<_, PulledJob>("SELECT * FROM queue WHERE id = $1 AND workspace_id = $2") - .bind(flow) - .bind(w_id) - .fetch_optional(&mut *tx) - .await - .map_err(Into::::into)? - .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?; + let flow_job = sqlx::query_as::<_, PulledJob>( + "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2", + ) + .bind(flow) + .bind(w_id) + .fetch_optional(&mut *tx) + .await + .map_err(Into::::into)? + .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?; tx.commit().await?; let job_root = flow_job @@ -940,7 +944,8 @@ pub async fn update_flow_status_after_job_completion_internal( sqlx::query_scalar!( "SELECT raw_flow AS \"raw_flow!: Json>\" FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", - &flow_job.id, w_id + &flow_job.id, + w_id ) .fetch_one(db) .await @@ -963,9 +968,16 @@ pub async fn update_flow_status_after_job_completion_internal( flow_value .as_ref() .and_then(|value| match module_step { - Step::PreprocessorStep => value.preprocessor_module.as_ref().and_then(|m| m.retry.as_ref()), - Step::Step(i) => value.modules.get(i).as_ref().and_then(|m| m.retry.as_ref()), - Step::FailureStep => value.failure_module.as_ref().and_then(|m| m.retry.as_ref()), + Step::PreprocessorStep => value + .preprocessor_module + .as_ref() + .and_then(|m| m.retry.as_ref()), + Step::Step(i) => { + value.modules.get(i).as_ref().and_then(|m| m.retry.as_ref()) + } + Step::FailureStep => { + value.failure_module.as_ref().and_then(|m| m.retry.as_ref()) + } }) .unwrap_or(&Retry::default()), &old_status.retry, @@ -977,7 +989,10 @@ pub async fn update_flow_status_after_job_completion_internal( false if !is_failure_step && !skip_error_handler - && flow_value.as_ref().map(|v| v.failure_module.is_some()).unwrap_or(false) => + && flow_value + .as_ref() + .map(|v| v.failure_module.is_some()) + .unwrap_or(false) => { true } @@ -1045,7 +1060,7 @@ pub async fn update_flow_status_after_job_completion_internal( reason: flow_job.canceled_reason.clone(), }), canceled_job_to_result(&flow_job), - worker_name, + worker_name, true, None, ) @@ -1121,7 +1136,7 @@ pub async fn update_flow_status_after_job_completion_internal( Some(nresult.clone()), same_worker_tx.clone(), worker_dir, - job_completed_tx, + job_completed_tx, ) .warn_after_seconds(10) .await @@ -1135,17 +1150,8 @@ pub async fn update_flow_status_after_job_completion_internal( db, ) .await; - let _ = add_completed_job_error( - db, - &flow_job, - 0, - None, - e, - worker_name, - true, - None, - ) - .await; + let _ = add_completed_job_error(db, &flow_job, 0, None, e, worker_name, true, None) + .await; true } Ok(_) => false, @@ -1161,18 +1167,20 @@ pub async fn update_flow_status_after_job_completion_internal( if let Some(parent_job) = flow_job.parent_job { tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, "subflow is finished, updating parent flow status"); - return Ok(UpdateFlowStatusAfterJobCompletion::Rec(RecUpdateFlowStatusAfterJobCompletion { - flow: parent_job, - job_id_for_status: flow, - success: success && !is_failure_step, - result: nresult.clone(), - stop_early_override: if stop_early { - Some(skip_if_stop_early) - } else { - None + return Ok(UpdateFlowStatusAfterJobCompletion::Rec( + RecUpdateFlowStatusAfterJobCompletion { + flow: parent_job, + job_id_for_status: flow, + success: success && !is_failure_step, + result: nresult.clone(), + stop_early_override: if stop_early { + Some(skip_if_stop_early) + } else { + None + }, + skip_error_handler: skip_error_handler || is_failure_step, }, - skip_error_handler: skip_error_handler || is_failure_step, - })); + )); } } Ok(UpdateFlowStatusAfterJobCompletion::Done(flow_job)) @@ -1508,8 +1516,7 @@ pub async fn handle_flow( worker_dir: &str, job_completed_tx: Sender, ) -> anyhow::Result<()> { - let flow = flow_value - .with_context(|| "Unable to parse flow definition")?; + let flow = flow_value.with_context(|| "Unable to parse flow definition")?; let status = flow_job .parse_flow_status() .with_context(|| "Unable to parse flow status")?; @@ -1524,14 +1531,15 @@ pub async fn handle_flow( let schedule_path = flow_job.schedule_path.as_ref().unwrap(); - let schedule = - get_schedule_opt(&mut tx, &flow_job.workspace_id, schedule_path).warn_after_seconds(5).await?; + let schedule = get_schedule_opt(&mut tx, &flow_job.workspace_id, schedule_path) + .warn_after_seconds(5) + .await?; tx.commit().await?; if let Some(schedule) = schedule { if let Err(err) = handle_maybe_scheduled_job( - db, + db, &flow_job, &schedule, flow_job.script_path.as_ref().unwrap(), @@ -2260,14 +2268,16 @@ async fn push_next_flow_job( } else { let value = module.get_value(); match &value { - Ok(_) if matches!(value, Ok(FlowModuleValue::Identity)) || is_skipped => serde_json::from_str( - &serde_json::to_string(&PreviousResult { - previous_result: Some(&arc_last_job_result), - }) - .unwrap(), - ) - .map(Marc::new) - .map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))), + Ok(_) if matches!(value, Ok(FlowModuleValue::Identity)) || is_skipped => { + serde_json::from_str( + &serde_json::to_string(&PreviousResult { + previous_result: Some(&arc_last_job_result), + }) + .unwrap(), + ) + .map(Marc::new) + .map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))) + } Ok( FlowModuleValue::Script { input_transforms, .. } | FlowModuleValue::RawScript { input_transforms, .. } @@ -2289,7 +2299,7 @@ async fn push_next_flow_job( ) .await .map(Marc::new) - }, + } Ok(_) => Ok(arc_flow_job_args.clone()), Err(e) => { return Err(error::Error::InternalErr(format!( @@ -2546,7 +2556,9 @@ async fn push_next_flow_job( }; tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}"); - let tag = if flow_job.tag == "flow" || flow_job.tag == format!("flow-{}", flow_job.workspace_id) { + let tag = if flow_job.tag == "flow" + || flow_job.tag == format!("flow-{}", flow_job.workspace_id) + { payload_tag.tag.clone() } else { Some(flow_job.tag.clone()) @@ -2963,7 +2975,7 @@ fn payload_from_modules<'a>( if let Some(id) = modules_node { return Some(JobPayload::FlowNode { id, path: path() }); } - + add_virtual_items_if_necessary(&mut modules); let mut failure_module = failure_module.cloned(); @@ -3107,7 +3119,7 @@ async fn compute_next_flow_transform( ContinuePayload::SingleJob(payload), NextStatus::NextStep, )) - }, + } FlowModuleValue::WhileloopFlow { modules, modules_node, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg let is_simple = is_simple_modules(&modules, flow.failure_module.as_ref()); @@ -3209,15 +3221,21 @@ async fn compute_next_flow_transform( .into_iter() .filter_map(|i| { let Some(payload) = payload_from_modules( - modules.clone(), modules_node, flow.failure_module.as_ref(), flow.same_worker, + modules.clone(), + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, || format!("{}-{i}", status.step), || format!("{}/forloop-{i}", flow_job.script_path()), - true + true, ) else { return None; }; Some(JobPayloadWithTag { - payload, tag: None, delete_after_use, timeout: None + payload, + tag: None, + delete_after_use, + timeout: None, }) }) .collect::>(); @@ -3281,28 +3299,37 @@ async fn compute_next_flow_transform( )))?, }; - let (modules, modules_node) = match branch { BranchChosen::Default => (default, default_node), BranchChosen::Branch { branch } => branches .into_iter() .nth(branch) .map(|Branch { modules, modules_node, .. }| (modules, modules_node)) - .ok_or_else(|| Error::BadRequest(format!("Unrecognized branch for BranchOne {status_module:?}")))? + .ok_or_else(|| { + Error::BadRequest(format!( + "Unrecognized branch for BranchOne {status_module:?}" + )) + })?, }; let Some(payload) = payload_from_modules( - modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + modules, + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, || status.step.to_string(), || format!("{}/branchone-{}", flow_job.script_path(), status.step), - true + true, ) else { return Ok(NextFlowTransform::EmptyInnerFlows); }; Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(JobPayloadWithTag { - payload, tag: None, delete_after_use, timeout: None + payload, + tag: None, + delete_after_use, + timeout: None, }), NextStatus::BranchChosen(branch), )) @@ -3321,15 +3348,21 @@ async fn compute_next_flow_transform( .enumerate() .filter_map(|(i, Branch { modules, modules_node, .. })| { let Some(payload) = payload_from_modules( - modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + modules, + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, || format!("{}-{i}", status.step), || format!("{}/branchall-{}", flow_job.script_path(), i), - false + false, ) else { return None; }; Some(JobPayloadWithTag { - payload, tag: None, delete_after_use, timeout: None + payload, + tag: None, + delete_after_use, + timeout: None, }) }) .collect::>(); @@ -3371,20 +3404,36 @@ async fn compute_next_flow_transform( let Branch { modules, modules_node, .. } = branches .into_iter() .nth(branch_status.branch) - .ok_or_else(|| Error::BadRequest(format!("Unrecognized branch for BranchAll {status_module:?}")))?; + .ok_or_else(|| { + Error::BadRequest(format!( + "Unrecognized branch for BranchAll {status_module:?}" + )) + })?; let Some(payload) = payload_from_modules( - modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + modules, + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, || format!("{}-{}", status.step, branch_status.branch), - || format!("{}/branchall-{}", flow_job.script_path(), branch_status.branch), - false + || { + format!( + "{}/branchall-{}", + flow_job.script_path(), + branch_status.branch + ) + }, + false, ) else { return Ok(NextFlowTransform::EmptyInnerFlows); }; Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(JobPayloadWithTag { - payload, tag: None, delete_after_use, timeout: None + payload, + tag: None, + delete_after_use, + timeout: None, }), NextStatus::NextBranchStep(NextBranch { status: branch_status, @@ -3417,25 +3466,25 @@ async fn next_loop_iteration( | FlowModuleValue::FlowScript { input_transforms, .. } | FlowModuleValue::Flow { input_transforms, .. } => { Some(std::mem::take(input_transforms)) - }, + } _ => None, }; return Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob( - payload_from_simple_module(value, db, flow_job, module, Some(inner_path())).await? + payload_from_simple_module(value, db, flow_job, module, Some(inner_path())).await?, ), - NextStatus::NextLoopIteration { - next: ns, - simple_input_transforms, - }, + NextStatus::NextLoopIteration { next: ns, simple_input_transforms }, )); } - + let Some(payload) = payload_from_modules( - modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + modules, + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, || format!("{}-{}", status.step, ns.index), inner_path, - true + true, ) else { return Ok(NextFlowTransform::EmptyInnerFlows); }; @@ -3451,7 +3500,10 @@ async fn next_loop_iteration( )) } -pub(super) fn is_simple_modules(modules: &Vec, failure_module: Option<&Box>) -> bool { +pub(super) fn is_simple_modules( + modules: &Vec, + failure_module: Option<&Box>, +) -> bool { let is_simple = modules.len() == 1 && modules[0].is_simple() && modules[0].sleep.is_none() @@ -3698,11 +3750,7 @@ fn raw_script_to_payload( } fn flow_to_payload(path: String, delete_after_use: bool) -> JobPayloadWithTag { - let payload = JobPayload::Flow { - path, - dedicated_worker: None, - apply_preprocessor: false, - }; + let payload = JobPayload::Flow { path, dedicated_worker: None, apply_preprocessor: false }; JobPayloadWithTag { payload, tag: None, delete_after_use, timeout: None } }