From 84992cd8ffa94e009aeb6b25db224a3001581e15 Mon Sep 17 00:00:00 2001 From: Pyra <92104930+pyranota@users.noreply.github.com> Date: Mon, 10 Nov 2025 14:43:50 +0100 Subject: [PATCH] fix: proper error handling in pulled job preprocessor (#7098) * fix: proper error handling in pulled job preprocessor Signed-off-by: pyranota * follow up for merge Signed-off-by: pyranota * make it safe Signed-off-by: pyranota * clippy Signed-off-by: pyranota * remove unused import Signed-off-by: pyranota * use String instead of Value Signed-off-by: pyranota * update ee ref Signed-off-by: pyranota * implement Error for PulledJobResultToErr Signed-off-by: pyranota * updatesqlx Signed-off-by: pyranota --------- Signed-off-by: pyranota --- ...153c43903f929ae5d62fbba12610f89c36d55.json | 2 +- ...420ef920d472286cf4c1f78452446a4bcb604.json | 22 + ...39ae72faf0c8fe097e6ad6d309aee9a8aede2.json | 24 + backend/Cargo.lock | 1 + backend/ee-repo-ref.txt | 2 +- backend/windmill-common/src/worker.rs | 9 +- backend/windmill-queue/Cargo.toml | 1 + backend/windmill-queue/src/jobs.rs | 534 ++++++++++-------- backend/windmill-worker/src/ai/tools.rs | 4 +- backend/windmill-worker/src/common.rs | 7 - .../windmill-worker/src/result_processor.rs | 7 +- backend/windmill-worker/src/worker.rs | 73 ++- 12 files changed, 394 insertions(+), 292 deletions(-) create mode 100644 backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json create mode 100644 backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json b/backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json new file mode 100644 index 0000000000..27a5df6de9 --- /dev/null +++ b/backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n SELECT app_id, value, created_by, raw_app\n FROM app_version WHERE id = $1\n RETURNING id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604" +} diff --git a/backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json b/backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json new file mode 100644 index 0000000000..32616f298f --- /dev/null +++ b/backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO flow_version\n (workspace_id, path, value, schema, created_by)\n\n SELECT workspace_id, path, value, schema, created_by\n FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3\n\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index cb19092a20..c0129f7f57 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15725,6 +15725,7 @@ dependencies = [ "serde_urlencoded", "sql-builder", "sqlx", + "thiserror 2.0.17", "tokio", "tracing", "ulid", diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index b9762186b4..3564b14a2a 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -5b7afe50da442441747e7a8f6ef461c96faa9dc2 +1cc227920a3799fe24a4135f1c6ae7026c0abcb5 diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 29d5488be8..5235e96de2 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -7,7 +7,7 @@ use regex::Regex; use reqwest_middleware::ClientWithMiddleware; use semver::Version; use serde::{de::DeserializeOwned, Deserialize, Serialize}; -use serde_json::value::RawValue; +use serde_json::{json, value::RawValue}; use sqlx::{types::Json, Pool, Postgres}; use std::{ cmp::Reverse, @@ -1701,6 +1701,13 @@ pub fn load_env_vars( .collect() } +pub fn error_to_value(err: &error::Error) -> serde_json::Value { + match err { + error::Error::JsonErr(err) => err.clone(), + _ => json!({"message": err.to_string(), "name": err.name()}), + } +} + #[derive(Clone, PartialEq, Debug)] pub struct WorkspacedPath { pub workspace_id: String, diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index 62b8b5962b..6349aee84e 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -46,3 +46,4 @@ serde_urlencoded.workspace = true regex.workspace = true backon.workspace = true quick_cache.workspace = true +thiserror.workspace = true diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index a6370163d6..f7e61de65c 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2456,12 +2456,18 @@ pub struct PulledJobResult { pub job: Option, pub suspended: bool, pub missing_concurrency_key: bool, + pub error_while_preprocessing: Option } +#[derive(thiserror::Error, Debug)] pub enum PulledJobResultToJobErr { + #[error("missing concurrency key")] MissingConcurrencyKey(JobCompleted), + #[error("pulled job preprocessor error: {}", .0.result)] + ErrorWhilePreprocessing(JobCompleted), } + impl PulledJobResult { pub fn to_pulled_job(self) -> Result, PulledJobResultToJobErr> { match self { @@ -2484,9 +2490,278 @@ impl PulledJobResult { from_cache: None, }), ), + PulledJobResult { job: Some(job), error_while_preprocessing: Some(e), .. } => Err( + PulledJobResultToJobErr::ErrorWhilePreprocessing(JobCompleted { + preprocessed_args: None, + job: MiniCompletedJob::from(job.job), + success: false, + result: Arc::new(windmill_common::worker::to_raw_value(&json!({ + "name": "Pulled job preprocessing error", + "message": e + }))), + result_columns: None, + mem_peak: 0, + cached_res_path: None, + token: "".to_string(), + canceled_by: None, + duration: None, + has_stream: Some(false), + from_cache: None, + }), + ), PulledJobResult { job, .. } => Ok(job), } } + + /// Generic preprocess function + /// Can be used for any kind of preprocessing + pub async fn preprocess(&mut self, db: &DB) -> error::Result<()> { + let PulledJobResult { job: Some(ref mut pulled_job), .. } = self else { + return Ok(()); + }; + + let kind = pulled_job.kind; + // Handle dependency job debouncing cleanup when a job is pulled for execution + if kind.is_dependency() + && pulled_job + .args + .as_ref() + .map(|x| x.get("triggered_by_relative_import").is_some()) + .unwrap_or_default() + && !*WMDEBUG_NO_DJOB_DEBOUNCING + { + return Box::pin(async move { + // Only used for testing in tests/relative_imports.rs + // Give us some space to work with. + #[cfg(debug_assertions)] + if let Some(duration) = pulled_job + .args + .as_ref() + .map(|x| { + x.get("dbg_sleep_between_pull_and_debounce_key_removal") + .map(|v| serde_json::from_str::(v.get()).ok()) + .flatten() + }) + .flatten() + { + tracing::debug!("going to sleep",); + sleep(std::time::Duration::from_secs(duration as u64)).await; + } + + tracing::debug!( + "Processing debounce cleanup for dependency job {} at path {:?}", + &pulled_job.id, + &pulled_job.runnable_path + ); + + let key = format!("{}:{}:dependency", &pulled_job.workspace_id, pulled_job.runnable_path()); + let mut tx = db.begin().await?; + + // === DEBOUNCE CLEANUP === + // + // Clean up the debounce_key entry for this job (if it exists). + // + // IMPORTANT: We delete by key (not job_id) to avoid race conditions: + // If pusher has locked this row then this call will be blocked until all txs are commited. + // + // The idea is that the worker_lockfiles::trigger_dependents_to_recompute_locks will fetch the latest version of the obj. + // This object needs to be created before the djob is executed and it happens right here. + // + // This way the next pusher can fetch the latest version of object and base their djob payload on newest version. + // The concurrency limit on djobs will make sure that by the time next djob is started executing the base version it is referencing + // has already calculated all locks. This way even next djob will always use the fully finalized version of object. + // + // + // + // Note: We don't use a transaction here for performance (it's called during job pull). + // This means there's a tiny window where the job is running but key isn't deleted yet, + // which is acceptable because new requests will just accumulate data to this job. + tracing::debug!( + job_id = %pulled_job.id, + "Cleaning up debounce_key entry for completed/pulled job" + ); + + // This will either: + // 1. Block until pusher pushed. Which gives us: + // - If there was any stale data in pusher, then we will read it here (couple of lines below) + // 2. Block pusher until we are done here. This gives us: + // - We will clone objects and retrieve the latest version. So when we are done the pusher can read latest version. + sqlx::query!("DELETE FROM debounce_key WHERE key = $1", &key) + .execute(&mut *tx) + .await + .map_err(|e| { + tracing::error!( + error = %e, + job_id = %pulled_job.id, + "Failed to delete debounce_key" + ); + e + })?; + + let Some(base_hash) = pulled_job.runnable_id else { + return Err(Error::InternalErr( + "Missing runnable_id for dependency job triggered by relative import" + .to_string(), + )); + }; + + tracing::debug!( + job_id = %pulled_job.id, + base_hash = %base_hash, + job_kind = ?kind, + "Creating new version for dependency job triggered by relative import" + ); + + let new_id = match kind { + JobKind::Dependencies => { + let deployment_message = pulled_job + .args + .clone() + .map(|hashmap| { + hashmap + .get("deployment_message") + .map(|map_value| { + serde_json::from_str::(map_value.get()).ok() + }) + .flatten() + }) + .flatten(); + + // This way we tell downstream which script we should archive when the resolution is finished. + // (not used at the moment) + pulled_job.args.as_mut().map(|args| { + args.insert("base_hash".to_owned(), to_raw_value(&*base_hash)) + }); + + let cloned_script = windmill_common::scripts::clone_script( + base_hash, + &pulled_job.workspace_id, + deployment_message, + &mut tx, + ) + .await?; + + if is_generated_from_raw_requirements(&Some(cloned_script.old_script.language), &cloned_script.old_script.lock.map(|v| v.to_string())) { + return Err(Error::BadRequest(format!( + "Script at path {} is generated from raw requirements, not overriding", + pulled_job.runnable_path() + ))); + } + + cloned_script.new_hash + } + JobKind::FlowDependencies => { + sqlx::query_scalar!( + "INSERT INTO flow_version + (workspace_id, path, value, schema, created_by) + + SELECT workspace_id, path, value, schema, created_by + FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3 + + RETURNING id + ", + pulled_job.runnable_path(), + pulled_job.workspace_id, + *base_hash, + ) + .fetch_one(&mut *tx) + .await? + } + JobKind::AppDependencies => { + sqlx::query_scalar!( + "INSERT INTO app_version + (app_id, value, created_by, raw_app) + SELECT app_id, value, created_by, raw_app + FROM app_version WHERE id = $1 + RETURNING id", + *base_hash + ) + .fetch_one(&mut *tx) + .await? + } + _ => { + return Err(Error::InternalErr(format!( + "Matched unexpected JobKind ({:?}). This is a bug!", + kind + ))) + } + }; + + pulled_job.runnable_id.replace(new_id.into()); + + if *windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await { + // === RETRIEVE ACCUMULATED DEBOUNCE DATA === + // + // For flows and apps, retrieve all nodes/components that were accumulated + // during the debounce window. This data comes from requests that were merged + // into this job instead of creating their own jobs. + // + // Scripts don't need this because they don't have nodes/components to relock. + if let Some(to_relock_field) = match &pulled_job.kind { + JobKind::FlowDependencies => Some("nodes_to_relock"), + JobKind::AppDependencies => Some("components_to_relock"), + _ => None, // Scripts don't use accumulated stale data + } { + tracing::debug!( + job_id = %pulled_job.id, + job_kind = ?pulled_job.kind, + field = %to_relock_field, + "Retrieving accumulated stale data from debounced requests" + ); + + if let Some(stale_data) = sqlx::query_scalar!( + "DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock", + &pulled_job.id + ) + .fetch_optional(&mut *tx) + .await + .map_err(|e| { + tracing::error!( + error = %e, + job_id = %pulled_job.id, + "Failed to retrieve debounce_stale_data" + ); + e + })? + .flatten() + { + tracing::debug!( + job_id = %pulled_job.id, + node_count = stale_data.len(), + nodes = ?stale_data, + "Retrieved accumulated nodes/components from {} debounced requests", + stale_data.len() + ); + + // Replace the job's relock list with the accumulated data + // This ensures all nodes from all debounced requests are processed + if let Some(args) = pulled_job.args.as_mut() { + args.insert(to_relock_field.to_owned(), to_raw_value(&stale_data)); + tracing::debug!( + field = %to_relock_field, + "Updated job args with accumulated debounce data" + ); + } + } else { + tracing::trace!( + job_id = %pulled_job.id, + "No accumulated stale data found (no debounced requests or already cleaned up)" + ); + } + } + } else { + tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support."); + + } + // This will unblock pusher. + tx.commit().await?; + + Ok(()) + }).await; + } + + Ok(()) + } } /// Pull the job from queue @@ -2512,6 +2787,7 @@ pub async fn pull( job: None, suspended: false, missing_concurrency_key: false, + error_while_preprocessing: None, }); } @@ -2567,20 +2843,21 @@ pub async fn pull( // Concurrency limit is available for either enterprise job or dependency job && (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)) => { - let job = crate::jobs_ee::apply_concurrency_limit( + crate::jobs_ee::apply_concurrency_limit( db, pull_loop_count, suspended, job, ) - .await?; - job.unwrap_or(PulledJobResult { + .await? + .unwrap_or(PulledJobResult { job: None, suspended, missing_concurrency_key: false, + error_while_preprocessing: None, }) } - _ => PulledJobResult { job, suspended, missing_concurrency_key: false }, + _ => PulledJobResult { job, suspended, missing_concurrency_key: false, error_while_preprocessing: None }, }; Ok::<_, Error>(pulled_job_result) @@ -2598,7 +2875,7 @@ pub async fn pull( ) .await?; let Some(job) = job else { - return Ok(PulledJobResult { job: None, suspended, missing_concurrency_key: false }); + return Ok(PulledJobResult { job: None, suspended, missing_concurrency_key: false, error_while_preprocessing: None }); }; let has_concurent_limit = job.concurrent_limit.is_some(); @@ -2626,6 +2903,7 @@ pub async fn pull( job: Some(pulled_job), suspended, missing_concurrency_key: false, + error_while_preprocessing: None, }); } @@ -2633,11 +2911,11 @@ pub async fn pull( if cfg!(feature = "enterprise") || (pulled_job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING) { - if let Some(pulled_job) = + if let Some(pulled_job_res) = crate::jobs_ee::apply_concurrency_limit(db, pull_loop_count, suspended, pulled_job) .await? { - return Ok(pulled_job); + return Ok(pulled_job_res); } } } @@ -5547,245 +5825,3 @@ pub async fn get_same_worker_job( )) }) } - -pub async fn preprocess_dependency_job(job: &mut PulledJob, db: &DB) -> error::Result<()> { - let kind = job.kind; - // Handle dependency job debouncing cleanup when a job is pulled for execution - if kind.is_dependency() - && job - .args - .as_ref() - .map(|x| x.get("triggered_by_relative_import").is_some()) - .unwrap_or_default() - && !*WMDEBUG_NO_DJOB_DEBOUNCING - { - return Box::pin(async move { - // Only used for testing in tests/relative_imports.rs - // Give us some space to work with. - #[cfg(debug_assertions)] - if let Some(duration) = job - .args - .as_ref() - .map(|x| { - x.get("dbg_sleep_between_pull_and_debounce_key_removal") - .map(|v| serde_json::from_str::(v.get()).ok()) - .flatten() - }) - .flatten() - { - tracing::debug!("going to sleep",); - sleep(std::time::Duration::from_secs(duration as u64)).await; - } - - tracing::debug!( - "Processing debounce cleanup for dependency job {} at path {:?}", - &job.id, - &job.runnable_path - ); - - let key = format!("{}:{}:dependency", &job.workspace_id, job.runnable_path()); - let mut tx = db.begin().await?; - - // === DEBOUNCE CLEANUP === - // - // Clean up the debounce_key entry for this job (if it exists). - // - // IMPORTANT: We delete by key (not job_id) to avoid race conditions: - // If pusher has locked this row then this call will be blocked until all txs are commited. - // - // The idea is that the worker_lockfiles::trigger_dependents_to_recompute_locks will fetch the latest version of the obj. - // This object needs to be created before the djob is executed and it happens right here. - // - // This way the next pusher can fetch the latest version of object and base their djob payload on newest version. - // The concurrency limit on djobs will make sure that by the time next djob is started executing the base version it is referencing - // has already calculated all locks. This way even next djob will always use the fully finalized version of object. - // - // - // - // Note: We don't use a transaction here for performance (it's called during job pull). - // This means there's a tiny window where the job is running but key isn't deleted yet, - // which is acceptable because new requests will just accumulate data to this job. - tracing::debug!( - job_id = %job.id, - "Cleaning up debounce_key entry for completed/pulled job" - ); - - // This will either: - // 1. Block until pusher pushed. Which gives us: - // - If there was any stale data in pusher, then we will read it here (couple of lines below) - // 2. Block pusher until we are done here. This gives us: - // - We will clone objects and retrieve the latest version. So when we are done the pusher can read latest version. - sqlx::query!("DELETE FROM debounce_key WHERE key = $1", &key) - .execute(&mut *tx) - .await - .map_err(|e| { - tracing::error!( - error = %e, - job_id = %job.id, - "Failed to delete debounce_key" - ); - e - })?; - - let Some(base_hash) = job.runnable_id else { - return Err(Error::InternalErr( - "Missing runnable_id for dependency job triggered by relative import" - .to_string(), - )); - }; - - tracing::debug!( - job_id = %job.id, - base_hash = %base_hash, - job_kind = ?kind, - "Creating new version for dependency job triggered by relative import" - ); - - let new_id = match kind { - JobKind::Dependencies => { - let deployment_message = job - .args - .clone() - .map(|hashmap| { - hashmap - .get("deployment_message") - .map(|map_value| { - serde_json::from_str::(map_value.get()).ok() - }) - .flatten() - }) - .flatten(); - - // This way we tell downstream which script we should archive when the resolution is finished. - // (not used at the moment) - job.args.as_mut().map(|args| { - args.insert("base_hash".to_owned(), to_raw_value(&*base_hash)) - }); - - let cloned_script = windmill_common::scripts::clone_script( - base_hash, - &job.workspace_id, - deployment_message, - &mut tx, - ) - .await?; - if is_generated_from_raw_requirements(&Some(cloned_script.old_script.language), &cloned_script.old_script.lock.map(|v| v.to_string())) { - return Err(Error::BadRequest(format!( - "Script at path {} is generated from raw requirements, not overriding", - job.runnable_path() - ))); - } - cloned_script.new_hash - } - JobKind::FlowDependencies => { - sqlx::query_scalar!( - "INSERT INTO flow_version - (workspace_id, path, value, schema, created_by) - - SELECT workspace_id, path, value, schema, created_by - FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3 - - RETURNING id - ", - job.runnable_path(), - job.workspace_id, - *base_hash, - ) - .fetch_one(&mut *tx) - .await? - } - JobKind::AppDependencies => { - sqlx::query_scalar!( - "INSERT INTO app_version - (app_id, value, created_by, raw_app) - SELECT app_id, value, created_by, raw_app - FROM app_version WHERE id = $1 - RETURNING id", - *base_hash - ) - .fetch_one(&mut *tx) - .await? - } - _ => { - return Err(Error::InternalErr(format!( - "Matched unexpected JobKind ({:?}). This is a bug!", - kind - ))) - } - }; - - job.runnable_id.replace(new_id.into()); - - if !*windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await { - tx.commit().await?; - tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support."); - return Ok(()); - } - // === RETRIEVE ACCUMULATED DEBOUNCE DATA === - // - // For flows and apps, retrieve all nodes/components that were accumulated - // during the debounce window. This data comes from requests that were merged - // into this job instead of creating their own jobs. - // - // Scripts don't need this because they don't have nodes/components to relock. - if let Some(to_relock_field) = match &job.kind { - JobKind::FlowDependencies => Some("nodes_to_relock"), - JobKind::AppDependencies => Some("components_to_relock"), - _ => None, // Scripts don't use accumulated stale data - } { - tracing::debug!( - job_id = %job.id, - job_kind = ?job.kind, - field = %to_relock_field, - "Retrieving accumulated stale data from debounced requests" - ); - - if let Some(stale_data) = sqlx::query_scalar!( - "DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock", - &job.id - ) - .fetch_optional(&mut *tx) - .await - .map_err(|e| { - tracing::error!( - error = %e, - job_id = %job.id, - "Failed to retrieve debounce_stale_data" - ); - e - })? - .flatten() - { - tracing::debug!( - job_id = %job.id, - node_count = stale_data.len(), - nodes = ?stale_data, - "Retrieved accumulated nodes/components from {} debounced requests", - stale_data.len() - ); - - // Replace the job's relock list with the accumulated data - // This ensures all nodes from all debounced requests are processed - if let Some(args) = job.args.as_mut() { - args.insert(to_relock_field.to_owned(), to_raw_value(&stale_data)); - tracing::debug!( - field = %to_relock_field, - "Updated job args with accumulated debounce data" - ); - } - } else { - tracing::trace!( - job_id = %job.id, - "No accumulated stale data found (no debounced requests or already cleaned up)" - ); - } - } - - // This will unblock pusher. - tx.commit().await?; - Ok(()) - }).await; - } - - Ok(()) -} diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 243a67fb4e..74f46c2bf4 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -6,7 +6,7 @@ use crate::ai::utils::{ update_flow_status_module_with_actions, update_flow_status_module_with_actions_success, FlowContext, }; -use crate::common::{error_to_value, OccupancyMetrics}; +use crate::common::OccupancyMetrics; use crate::result_processor::handle_non_flow_job_error; use crate::worker_flow::{ evaluate_input_transform, raw_script_to_payload, script_to_payload, JobPayloadWithTag, @@ -591,7 +591,7 @@ async fn handle_tool_execution_error( final_events_str: &mut String, ) -> Result<(), Error> { let err_string = format!("{}: {}", err.name(), err.to_string()); - let err_json = error_to_value(&err); + let err_json = windmill_common::worker::error_to_value(&err); let _ = handle_non_flow_job_error( ctx.db, tool_job, diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 259b12ec84..4a6edbd55d 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -539,13 +539,6 @@ pub async fn update_worker_ping_for_failed_init_script( } } -pub fn error_to_value(err: &Error) -> serde_json::Value { - match err { - Error::JsonErr(err) => err.clone(), - _ => json!({"message": err.to_string(), "name": err.name()}), - } -} - #[derive(Clone)] pub struct OccupancyMetrics { pub running_job_started_at: Option, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 7cfe63dc34..1e168d3a30 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -22,7 +22,7 @@ use windmill_common::{ flow_status::FlowJobDuration, jobs::JobKind, utils::WarnAfterExt, - worker::{to_raw_value, Connection, WORKER_GROUP}, + worker::{error_to_value, to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, KillpillSender, DB, }; @@ -31,7 +31,8 @@ use windmill_common::{ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ - CanceledBy, INIT_SCRIPT_TAG, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, append_logs, get_mini_completed_job + append_logs, get_mini_completed_job, CanceledBy, JobCompleted, MiniCompletedJob, MiniPulledJob, + ValidableJson, WrappedError, INIT_SCRIPT_TAG, }; use serde_json::{json, value::RawValue, Value}; @@ -42,7 +43,7 @@ use windmill_queue::{add_completed_job, add_completed_job_error}; use crate::{ bash_executor::ANSI_ESCAPE_RE, - common::{error_to_value, read_result, save_in_cache}, + common::{read_result, save_in_cache}, otel_oss::add_root_flow_job_to_otlp, worker_flow::update_flow_status_after_job_completion, JobCompletedReceiver, JobCompletedSender, SameWorkerSender, SendResult, SendResultPayload, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index c457dc5343..e58d521a5d 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -18,6 +18,7 @@ use windmill_common::scripts::hash_to_codebase_id; use windmill_common::scripts::is_special_codebase_hash; use windmill_common::utils::report_critical_error; use windmill_common::utils::retrieve_common_worker_prefix; +use windmill_common::worker::error_to_value; use windmill_common::{ agent_workers::DECODED_AGENT_TOKEN, apps::AppScriptId, @@ -58,7 +59,6 @@ use std::{ time::Duration, }; use windmill_parser::MainArgSignature; -use windmill_queue::preprocess_dependency_job; use windmill_queue::MiniCompletedJob; use windmill_queue::PulledJobResultToJobErr; @@ -112,7 +112,7 @@ use crate::{ bash_executor::handle_bash_job, bun_executor::handle_bun_job, common::{ - build_args_map, cached_result_path, error_to_value, get_cached_resource_value_if_valid, + build_args_map, cached_result_path, get_cached_resource_value_if_valid, get_reserved_variables, update_worker_ping_for_failed_init_script, OccupancyMetrics, }, csharp_executor::handle_csharp_job, @@ -841,10 +841,22 @@ pub fn start_interactive_worker_shell( match job { Ok(j) => match j.to_pulled_job() { - Ok(j) => Ok(j.map(NextJob::Sql)), - Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) => { - if let Err(err) = job_completed_tx.send_job(jc, true).await { - tracing::error!("An error occurred while sending job completed (missing concurrency key): {:#?}", err) + Ok(j) => Ok(j.clone().map(NextJob::Sql)), + ref e @ (Err(PulledJobResultToJobErr::MissingConcurrencyKey( + ref jc, + )) + | Err(PulledJobResultToJobErr::ErrorWhilePreprocessing( + ref jc, + ))) => { + if let Err(err) = + job_completed_tx.send_job(jc.clone(), true).await + { + let e_fmt = match e { + Ok(_) => "unknown error".to_owned(), + Err(e) => e.to_string(), + }; + + tracing::error!("An error occurred while sending job completed ({e_fmt}): {:#?}", err) } Ok(None) } @@ -877,7 +889,7 @@ pub fn start_interactive_worker_shell( token, precomputed_agent_info: precomputed_bundle, } = extract_job_and_perms(job, &conn).await; - + let authed_client = AuthedClient::new( base_internal_url.to_owned(), job.workspace_id.clone(), @@ -886,7 +898,7 @@ pub fn start_interactive_worker_shell( ); let arc_job = Arc::new(job); - + let _ = handle_queued_job( arc_job.clone(), raw_code, @@ -1605,30 +1617,23 @@ pub async fn run_worker( } }; - // Essential debouncing job preprocessing. - if let Ok(windmill_queue::PulledJobResult { - job: Some(ref mut pulled_job), - .. - }) = &mut job - { - match timeout( + // Preprocess pulled job result + if let Ok(ref mut pulled_job_res) = job { + if let Err(e) = timeout( + // Will fail if longer than 10 seconds core::time::Duration::from_secs(10), - preprocess_dependency_job(pulled_job, &db), + pulled_job_res.preprocess(db), ) .warn_after_seconds(2) .await + // Flatten result + .map_err(error::Error::from) + .and_then(|r| r) { - Ok(Err(e)) => { - tracing::error!(worker = %worker_name, hostname = %hostname, "critical: debouncing job preprocessor failed: {e:?}"); - job = Err(e.into()); - } - Err(e) => { - tracing::error!(worker = %worker_name, hostname = %hostname, "critical: debouncing job preprocessor has timed out: {e:?}"); - job = Err(e.into()); - } - _ => {} + pulled_job_res.error_while_preprocessing = Some(e.to_string()); } } + add_time!(bench, "job pulled from DB"); let duration_pull_s = pull_time.elapsed().as_secs_f64(); let err_pull = job.is_ok(); @@ -1688,9 +1693,21 @@ pub async fn run_worker( match job { Ok(pulled_job_result) => match pulled_job_result.to_pulled_job() { Ok(j) => Ok(j.map(NextJob::Sql)), - Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) => { - if let Err(err) = job_completed_tx.send_job(jc, true).await { - tracing::error!("An error occurred while sending job completed (missing concurrency key): {:#?}", err) + ref e @ (Err(PulledJobResultToJobErr::MissingConcurrencyKey( + ref jc, + )) + | Err(PulledJobResultToJobErr::ErrorWhilePreprocessing( + ref jc, + ))) => { + if let Err(err) = + job_completed_tx.send_job(jc.clone(), true).await + { + let e_fmt = match e { + Ok(_) => "unknown error".to_owned(), + Err(e) => e.to_string(), + }; + + tracing::error!("An error occurred while sending job completed ({e_fmt}): {:#?}", err) } Ok(None) }