diff --git a/backend/.sqlx/query-0722f461da9edb169c3c73423440eaa3e1b9f2ca3b7033bd7ed978a3b4ddf3fc.json b/backend/.sqlx/query-0722f461da9edb169c3c73423440eaa3e1b9f2ca3b7033bd7ed978a3b4ddf3fc.json new file mode 100644 index 0000000000..86b1a7790a --- /dev/null +++ b/backend/.sqlx/query-0722f461da9edb169c3c73423440eaa3e1b9f2ca3b7033bd7ed978a3b4ddf3fc.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_flow AS \"raw_flow!: Json>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "raw_flow!: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "0722f461da9edb169c3c73423440eaa3e1b9f2ca3b7033bd7ed978a3b4ddf3fc" +} diff --git a/backend/.sqlx/query-217a5291438d23597b2c7f05d2c481f406364d56a129089a268f6423c548bca6.json b/backend/.sqlx/query-217a5291438d23597b2c7f05d2c481f406364d56a129089a268f6423c548bca6.json new file mode 100644 index 0000000000..9dac2a2451 --- /dev/null +++ b/backend/.sqlx/query-217a5291438d23597b2c7f05d2c481f406364d56a129089a268f6423c548bca6.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\"\n FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "raw_code", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "raw_lock", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "raw_flow: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + true, + true + ] + }, + "hash": "217a5291438d23597b2c7f05d2c481f406364d56a129089a268f6423c548bca6" +} diff --git a/backend/.sqlx/query-45950064cce9f53f73a01ddcd6911ec677297009b71041d39019c4700a571c0f.json b/backend/.sqlx/query-45950064cce9f53f73a01ddcd6911ec677297009b71041d39019c4700a571c0f.json new file mode 100644 index 0000000000..ecba5ab5c0 --- /dev/null +++ b/backend/.sqlx/query-45950064cce9f53f73a01ddcd6911ec677297009b71041d39019c4700a571c0f.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM completed_job WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "45950064cce9f53f73a01ddcd6911ec677297009b71041d39019c4700a571c0f" +} diff --git a/backend/.sqlx/query-b3a9ab85a78edf292ddb0af33f7435bbcc84f63c9079e29d84043f258eba208e.json b/backend/.sqlx/query-b3a9ab85a78edf292ddb0af33f7435bbcc84f63c9079e29d84043f258eba208e.json new file mode 100644 index 0000000000..59bcc04054 --- /dev/null +++ b/backend/.sqlx/query-b3a9ab85a78edf292ddb0af33f7435bbcc84f63c9079e29d84043f258eba208e.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n flow_status AS \"flow_status!: Json>\",\n coalesce(job.raw_flow, queue.raw_flow)->'modules'->(flow_status->'step')::int AS \"module: Json>\"\n FROM queue LEFT JOIN job USING(id, workspace_id) WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow_status!: Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "module: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + null + ] + }, + "hash": "b3a9ab85a78edf292ddb0af33f7435bbcc84f63c9079e29d84043f258eba208e" +} diff --git a/backend/.sqlx/query-da861c06f6448f68069ba3be9b677641c5c6b2c1e4960c74db27dbaffcbd4c40.json b/backend/.sqlx/query-da861c06f6448f68069ba3be9b677641c5c6b2c1e4960c74db27dbaffcbd4c40.json new file mode 100644 index 0000000000..95a5b6fbf6 --- /dev/null +++ b/backend/.sqlx/query-da861c06f6448f68069ba3be9b677641c5c6b2c1e4960c74db27dbaffcbd4c40.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "raw_code", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "raw_lock", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "raw_flow: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + true, + true + ] + }, + "hash": "da861c06f6448f68069ba3be9b677641c5c6b2c1e4960c74db27dbaffcbd4c40" +} diff --git a/backend/.sqlx/query-e1923bc755bd6b8cc871ae9381a97d3c63a0fce3dd57c93f48a2136f7395fa3a.json b/backend/.sqlx/query-e1923bc755bd6b8cc871ae9381a97d3c63a0fce3dd57c93f48a2136f7395fa3a.json new file mode 100644 index 0000000000..a477ebf141 --- /dev/null +++ b/backend/.sqlx/query-e1923bc755bd6b8cc871ae9381a97d3c63a0fce3dd57c93f48a2136f7395fa3a.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM job WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "e1923bc755bd6b8cc871ae9381a97d3c63a0fce3dd57c93f48a2136f7395fa3a" +} diff --git a/backend/.sqlx/query-f24303a2386575a750a48a21d1b6cdecd6e64a0415c60ad1495f100c5d85bbe1.json b/backend/.sqlx/query-f24303a2386575a750a48a21d1b6cdecd6e64a0415c60ad1495f100c5d85bbe1.json new file mode 100644 index 0000000000..74200d658d --- /dev/null +++ b/backend/.sqlx/query-f24303a2386575a750a48a21d1b6cdecd6e64a0415c60ad1495f100c5d85bbe1.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_flow AS \"raw_flow!: Json>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "raw_flow!: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "f24303a2386575a750a48a21d1b6cdecd6e64a0415c60ad1495f100c5d85bbe1" +} diff --git a/backend/.sqlx/query-fb1a32318b35ec5c8129eb3660b79eb5e6d1e01fcf01cc05d3c7ddf47295c2f5.json b/backend/.sqlx/query-fb1a32318b35ec5c8129eb3660b79eb5e6d1e01fcf01cc05d3c7ddf47295c2f5.json new file mode 100644 index 0000000000..6997a5e2f4 --- /dev/null +++ b/backend/.sqlx/query-fb1a32318b35ec5c8129eb3660b79eb5e6d1e01fcf01cc05d3c7ddf47295c2f5.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "raw_code", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "raw_lock", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "raw_flow: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + true, + true + ] + }, + "hash": "fb1a32318b35ec5c8129eb3660b79eb5e6d1e01fcf01cc05d3c7ddf47295c2f5" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 1faa25d419..d6e5b41e81 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -11110,6 +11110,7 @@ dependencies = [ "chrono", "chrono-tz 0.10.0", "cron", + "futures", "futures-core", "hex", "hmac", diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 4931ba9857..d357a5ec72 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -16,7 +16,6 @@ use rand::Rng; use serde::{Deserialize, Serialize, Serializer}; use crate::{ - error::Error, more_serde::{default_empty_string, default_id, default_null, default_true, is_default}, scripts::{Schema, ScriptHash, ScriptLang}, }; @@ -652,29 +651,3 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec) { }); } } - -pub async fn has_failure_module<'c>(flow: sqlx::types::Uuid, db: &sqlx::Pool, completed: bool) -> Result { - if completed { - sqlx::query_scalar!( - "SELECT raw_flow->'failure_module' != 'null'::jsonb - FROM completed_job_view - WHERE id = $1", - flow - ) - } else { - sqlx::query_scalar!( - "SELECT raw_flow->'failure_module' != 'null'::jsonb - FROM queue_view - WHERE id = $1", - flow - ) - } - .fetch_one(db) - .await - .map_err(|e| { - Error::InternalErr(format!( - "error during retrieval of has_failure_module: {e:#}" - )) - }) - .map(|v| v.unwrap_or(false)) -} diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index ee6b95a19b..d94151f75b 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -119,8 +119,9 @@ pub async fn make_suspended_pull_query(wc: &WorkerConfig) { canceled_reason, last_ping, job_kind, schedule_path, permissioned_as, flow_status, is_flow_step, language, suspend, suspend_until, same_worker, pre_run_error, email, visible_to_owner, mem_peak, - root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, - timeout, flow_step_id, cache_ttl, priority", wc.worker_tags.iter().map(|x| format!("'{x}'")).join(", ")); + root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, + timeout, flow_step_id, cache_ttl, priority, + raw_code, raw_lock, raw_flow", wc.worker_tags.iter().map(|x| format!("'{x}'")).join(", ")); let mut l = WORKER_SUSPENDED_PULL_QUERY.write().await; *l = query; } @@ -150,8 +151,9 @@ pub async fn make_pull_query(wc: &WorkerConfig) { canceled_reason, last_ping, job_kind, schedule_path, permissioned_as, flow_status, is_flow_step, language, suspend, suspend_until, same_worker, pre_run_error, email, visible_to_owner, mem_peak, - root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, - timeout, flow_step_id, cache_ttl, priority", tags.tags.iter().map(|x| format!("'{x}'")).join(", ")); + root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, + timeout, flow_step_id, cache_ttl, priority, + raw_code, raw_lock, raw_flow", tags.tags.iter().map(|x| format!("'{x}'")).join(", ")); queries.push(query); } diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index 435b7c71f7..1c5fb38320 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -37,6 +37,7 @@ cron.workspace = true rsmq_async.workspace = true tokio.workspace = true futures-core.workspace = true +futures.workspace = true itertools.workspace = true async-recursion.workspace = true bigdecimal.workspace = true diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 8ee176ed76..2173351fba 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -17,6 +17,7 @@ use axum::{ response::{IntoResponse, Response}, }; use chrono::{DateTime, Duration, Utc}; +use futures::future::TryFutureExt; use itertools::Itertools; #[cfg(feature = "prometheus")] use prometheus::IntCounter; @@ -71,9 +72,6 @@ use windmill_common::BASE_URL; #[cfg(feature = "cloud")] use windmill_common::users::SUPERADMIN_SYNC_EMAIL; -#[cfg(feature = "enterprise")] -use windmill_common::flows::has_failure_module; - #[cfg(feature = "enterprise")] use windmill_common::worker::CLOUD_HOSTED; @@ -584,12 +582,21 @@ pub async fn add_completed_job< let (raw_code, raw_lock, raw_flow) = if !*MIN_VERSION_IS_AT_LEAST_1_427.read().await { sqlx::query!( "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\" - FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1", + FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", &job_id, &queued_job.workspace_id ) .fetch_one(db) + .map_ok(|record| (record.raw_code, record.raw_lock, record.raw_flow)) + .or_else(|_| { + sqlx::query!( + "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\" + FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1", + &job_id, &queued_job.workspace_id + ) + .fetch_one(db) + .map_ok(|record| (record.raw_code, record.raw_lock, record.raw_flow)) + }) .await - .map(|record| (record.raw_code, record.raw_lock, record.raw_flow)) .unwrap_or_default() } else { (None, None, None) @@ -923,6 +930,18 @@ pub async fn add_completed_job< #[cfg(feature = "enterprise")] if !success { + async fn has_failure_module(db: &Pool, job_id: Uuid) -> bool { + sqlx::query_scalar!("SELECT raw_flow->'failure_module' != 'null'::jsonb FROM job WHERE id = $1", job_id) + .fetch_one(db) + .or_else(|_| + sqlx::query_scalar!("SELECT raw_flow->'failure_module' != 'null'::jsonb FROM completed_job WHERE id = $1", job_id) + .fetch_one(db) + ) + .await + .unwrap_or(Some(false)) + .unwrap_or(false) + } + if queued_job.email == ERROR_HANDLER_USER_EMAIL { let base_url = BASE_URL.read().await; let w_id = &queued_job.workspace_id; @@ -964,7 +983,7 @@ pub async fn add_completed_job< } else if !skip_downstream_error_handlers && (matches!(queued_job.job_kind, JobKind::Script) || matches!(queued_job.job_kind, JobKind::Flow) - && !has_failure_module(job_id, db, true).await.unwrap_or(false)) + && !has_failure_module(db, job_id).await) && queued_job.parent_job.is_none() { let result = serde_json::from_str( @@ -1779,11 +1798,27 @@ async fn handle_successful_schedule< Ok(()) } +#[derive(sqlx::FromRow)] +pub struct PulledJob { + #[sqlx(flatten)] + pub job: QueuedJob, + pub raw_code: Option, + pub raw_lock: Option, + pub raw_flow: Option>>, +} + +impl std::ops::Deref for PulledJob { + type Target = QueuedJob; + fn deref(&self) -> &Self::Target { + &self.job + } +} + pub async fn pull( db: &Pool, rsmq: Option, suspend_first: bool, -) -> windmill_common::error::Result<(Option, bool)> { +) -> windmill_common::error::Result<(Option, bool)> { loop { let (job, suspended) = pull_single_job_and_mark_as_running_no_concurrency_limit( db, @@ -1792,11 +1827,11 @@ pub async fn pull( ) .await?; - if job.is_none() { + let Some(job) = job else { return Ok((None, suspended)); - } + }; - let has_concurent_limit = job.as_ref().unwrap().concurrent_limit.is_some(); + let has_concurent_limit = job.concurrent_limit.is_some(); #[cfg(not(feature = "enterprise"))] if has_concurent_limit { @@ -1807,7 +1842,7 @@ pub async fn pull( let has_concurent_limit = false; // concurrency check. If more than X jobs for this path are already running, we re-queue and pull another job from the queue - let pulled_job = job.unwrap(); + let pulled_job = job; if pulled_job.script_path.is_none() || !has_concurent_limit || pulled_job.canceled { #[cfg(feature = "prometheus")] if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { @@ -1987,7 +2022,7 @@ pub async fn pull( let job_log_event = format!( "\nRe-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits with key {job_concurrency_key} and limit {job_custom_concurrent_limit} in the last {job_custom_concurrency_time_window_s} seconds", ); - let _ = append_logs(&job_uuid, pulled_job.workspace_id, job_log_event, db).await; + let _ = append_logs(&job_uuid, &pulled_job.workspace_id, job_log_event, db).await; if rsmq.is_some() { // if let Some(ref mut rsmq) = tx.rsmq { // if using redis, only one message at a time can be poped from the queue. Process only this message and move to the next elligible job @@ -2041,8 +2076,8 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< db: &Pool, rsmq: Option, suspend_first: bool, -) -> windmill_common::error::Result<(Option, bool)> { - let job_and_suspended: (Option, bool) = if let Some(mut rsmq) = rsmq { +) -> windmill_common::error::Result<(Option, bool)> { + let job_and_suspended: (Option, bool) = if let Some(mut rsmq) = rsmq { #[cfg(feature = "benchmark")] let instant = Instant::now(); @@ -2074,7 +2109,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_as::<_, QueuedJob>( + let m2r = sqlx::query_as::<_, PulledJob>( "UPDATE queue SET running = true , started_at = coalesce(started_at, now()) @@ -2087,7 +2122,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< flow_status, is_flow_step, language, suspend, suspend_until, same_worker, pre_run_error, email, visible_to_owner, mem_peak, root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, - timeout, flow_step_id, cache_ttl, priority", + timeout, flow_step_id, cache_ttl, priority, raw_code, raw_lock, raw_flow", ) .bind(uuid) .fetch_optional(db) @@ -2121,7 +2156,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< let r = if suspend_first { // tracing::info!("Pulling job with query: {}", query); - sqlx::query_as::<_, QueuedJob>(&query) + sqlx::query_as::<_, PulledJob>(&query) .fetch_optional(db) .await? } else { @@ -2130,7 +2165,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< if r.is_none() { // #[cfg(feature = "benchmark")] // let instant = Instant::now(); - let mut highest_priority_job: Option = None; + let mut highest_priority_job: Option = None; let queries = WORKER_PULL_QUERIES.read().await; @@ -2141,7 +2176,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< for query in queries.iter() { // tracing::info!("Pulling job with query: {}", query); - let r = sqlx::query_as::<_, QueuedJob>(query) + let r = sqlx::query_as::<_, PulledJob>(query) .fetch_optional(db) .await?; @@ -2581,7 +2616,7 @@ pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result( +async fn get_queued_job_tx<'c>( id: Uuid, w_id: &str, tx: &mut Transaction<'c, Postgres>, @@ -4112,7 +4147,14 @@ async fn restarted_flows_resolution( ), Error, > { - let completed_job = sqlx::query_as::<_, CompletedJob>( + #[derive(sqlx::FromRow)] + struct CompletedJobWithRawFlow { + #[sqlx(flatten)] + completed_job: CompletedJob, + raw_flow: Option>>, + } + + let CompletedJobWithRawFlow { completed_job, raw_flow } = sqlx::query_as::<_, CompletedJobWithRawFlow>( "SELECT *, null as labels FROM completed_job WHERE id = $1 and workspace_id = $2", ) .bind(completed_flow_id) @@ -4127,22 +4169,23 @@ async fn restarted_flows_resolution( })?; let flow_value = if let Some(flow_value) = flow_value_if_any { - flow_value + Some(flow_value) + } else if let Some(raw_flow) = raw_flow.as_ref() { + serde_json::from_str::(raw_flow.get()).ok() } else { sqlx::query_scalar!( "SELECT raw_flow AS \"raw_flow!: Json>\" - FROM completed_job_view WHERE id = $1 AND workspace_id = $2 LIMIT 1", + FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", &completed_flow_id, workspace_id ) .fetch_one(db) .await .ok() .and_then(|raw_flow| serde_json::from_str::(raw_flow.get()).ok()) - .ok_or(Error::InternalErr(format!( - "Unable to parse raw definition for job {} in workspace {}", - completed_flow_id, workspace_id, - )))? - }; + }.ok_or(Error::InternalErr(format!( + "Unable to parse raw definition for job {} in workspace {}", + completed_flow_id, workspace_id, + )))?; let flow_status = completed_job .parse_flow_status() .ok_or(Error::InternalErr(format!( diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index cd3bca8b61..256ff7cc9e 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -54,8 +54,8 @@ use windmill_common::{ }; use windmill_queue::{ - append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PushArgs, - PushIsolationLevel, HTTP_CLIENT, + append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PulledJob, + PushArgs, PushIsolationLevel, HTTP_CLIENT, }; #[cfg(feature = "prometheus")] @@ -1190,7 +1190,7 @@ pub async fn run_worker( + let r = sqlx::query_as::<_, PulledJob>( "UPDATE queue SET last_ping = now() WHERE id = $1 RETURNING *", ) .bind(same_worker_job.job_id) @@ -1327,7 +1327,7 @@ pub async fn run_worker { #[tracing::instrument(name = "job", level = "info", skip_all, fields(job_id = %job.id))] async fn handle_queued_job( job: Arc, + raw_code: Option, + raw_lock: Option, + raw_flow: Option>>, db: &DB, client: &AuthedClientBackgroundTask, hostname: &str, @@ -1829,16 +1836,18 @@ async fn handle_queued_job( None }; - let (raw_code, raw_lock, raw_flow) = - sqlx::query!( + let (raw_code, raw_lock, raw_flow) = match (raw_code, raw_lock, raw_flow) { + (None, None, None) => sqlx::query!( "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\" - FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1", + FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1", &job.id, job.workspace_id ) .fetch_one(db) .await .map(|record| (record.raw_code, record.raw_lock, record.raw_flow)) - .unwrap_or_default(); + .unwrap_or_default(), + (raw_code, raw_lock, raw_flow) => (raw_code, raw_lock, raw_flow), + }; let cached_res_path = if job.cache_ttl.is_some() { let version_hash = if let Some(h) = job.script_hash { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 43ba839725..3d041c60f5 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -48,17 +48,17 @@ use windmill_common::{ Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, }, - flows::{has_failure_module, FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend}, + flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend}, }; use windmill_queue::schedule::get_schedule_opt; use windmill_queue::{ add_completed_job, add_completed_job_error, append_logs, handle_maybe_scheduled_job, - CanceledBy, PushArgs, PushIsolationLevel, WrappedError, + CanceledBy, PulledJob, PushArgs, PushIsolationLevel, WrappedError, }; type DB = sqlx::Pool; -use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction}; +use windmill_queue::{canceled_job_to_result, push, QueueTransaction}; // #[instrument(level = "trace", skip_all)] pub async fn update_flow_status_after_job_completion< @@ -214,8 +214,8 @@ pub async fn update_flow_status_after_job_completion_internal< let (old_status, current_module) = sqlx::query!( "SELECT flow_status AS \"flow_status!: Json>\", - raw_flow->'modules'->(flow_status->'step')::int AS \"module: Json>\" - FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1", + coalesce(job.raw_flow, queue.raw_flow)->'modules'->(flow_status->'step')::int AS \"module: Json>\" + FROM queue LEFT JOIN job USING(id, workspace_id) WHERE id = $1 AND workspace_id = $2 LIMIT 1", flow, w_id ) .fetch_one(db) @@ -290,17 +290,15 @@ pub async fn update_flow_status_after_job_completion_internal< Step::Step(i) => Some(i), }; - let is_flow = if let Some(step) = step { - sqlx::query_scalar!( - "SELECT raw_flow->'modules'->($1)::text->'value'->>'type' = 'flow' FROM queue_view WHERE id = $2 LIMIT 1", - step as i32, flow - ) - .fetch_one(db) - .await - .map_err(|e| { - Error::InternalErr(format!("error during retrieval of step's type: {e:#}")) - })? - .unwrap_or(false) + 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 }; @@ -915,8 +913,12 @@ pub async fn update_flow_status_after_job_completion_internal< .context("remove flow status retry")?; } - let flow_job = get_queued_job_tx(flow, w_id, tx.transaction_mut()) - .await? + 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?; @@ -926,14 +928,19 @@ pub async fn update_flow_status_after_job_completion_internal< .unwrap_or_else(|| "none".to_string()); tracing::info!(id = %flow_job.id, root_id = %job_root, "update flow status"); - let raw_flow = sqlx::query_scalar!( - "SELECT raw_flow AS \"raw_flow!: Json>\" - FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1", - &flow_job.id, w_id - ) - .fetch_one(db) - .await - .ok(); + let PulledJob { job: flow_job, raw_flow, .. } = flow_job; + let raw_flow = if let Some(raw_flow) = raw_flow { + Some(raw_flow) + } else { + 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 + ) + .fetch_one(db) + .await + .ok() + }; let flow_value = raw_flow .as_ref() .and_then(|raw_flow| serde_json::from_str::(raw_flow.get()).ok()); @@ -965,7 +972,7 @@ pub async fn update_flow_status_after_job_completion_internal< false if !is_failure_step && !skip_error_handler - && has_failure_module(flow, db, false).await? => + && flow_value.as_ref().map(|v| v.failure_module.is_some()).unwrap_or(false) => { true }