starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-08 00:03:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

fmt: re-format Rust leftovers (#4897)

Browse Source
This commit is contained in:
Lucas Abel
2024-12-11 08:57:00 +01:00
committed by GitHub
parent 4ebddbd005
commit ed71efa6da
6 changed files with 326 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/apps.rs
+9 -17
View File
@@ -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<DB>,
Path((w_id, custom_path)): Path<(String, String)>,
) -> JsonResult<bool> {
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<DB>,
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.
_ => {
backend/windmill-queue/src/jobs.rs
+1 -1
View File
@@ -3221,7 +3221,7 @@ pub async fn push<'c, 'd>(
None,
None,
)
},
}
JobPayload::AppScript {
id, // app_script(id).
path,
backend/windmill-worker/src/dedicated_worker.rs
+7 -7
View File
@@ -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 => (),
}
backend/windmill-worker/src/python_executor.rs
+60 -51
View File
@@ -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<Vec<String>> {
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<tokio::sync::broadcast::Receiver<()>> =
(0..req_with_penv.len()).map(|_| kill_tx.subscribe()).collect();
let kill_rxs: Vec<tokio::sync::broadcast::Receiver<()>> = (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 {
backend/windmill-worker/src/worker.rs
+27 -20
View File
@@ -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<Postgres>,
same_worker_tx: SameWorkerSender,
@@ -1874,7 +1884,6 @@ async fn handle_queued_job(
) -> windmill_common::error::Result<bool> {
// 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<Vec<String>>,
) -> windmill_common::error::Result<HashMap<String, String>> {
@@ -2326,7 +2334,6 @@ async fn handle_code_execution_job(
new_args: &mut Option<HashMap<String, Box<RawValue>>>,
occupancy_metrics: &mut OccupancyMetrics,
killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
) -> error::Result<Box<RawValue>> {
let ContentReqLangEnvs {
content: inner_content,
backend/windmill-worker/src/worker_flow.rs
+222 -174
View File
@@ -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::<GetType>(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::<GetType>(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::<Error>::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::<Error>::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<Box<sqlx::types::JsonRawValue>>\"
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<SendResult>,
) -> 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::<Vec<_>>();
@@ -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::<Vec<_>>();
@@ -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<FlowModule>, failure_module: Option<&Box<FlowModule>>) -> bool {
pub(super) fn is_simple_modules(
modules: &Vec<FlowModule>,
failure_module: Option<&Box<FlowModule>>,
) -> 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 }
}