diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index 555d7a4d8b..e4359e19da 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -248,8 +248,11 @@ lazy_static::lazy_static! { #[derive(Deserialize)] pub struct WindmillCompositeResult { + #[serde(alias = "wm_status_code")] windmill_status_code: Option, + #[serde(alias = "wm_content_type")] windmill_content_type: Option, + #[serde(alias = "wm_headers")] windmill_headers: Option>, result: Option>, } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 897f8e2ce0..52e6b09c6f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -630,12 +630,45 @@ pub struct WrappedError { pub trait ValidableJson { fn is_valid_json(&self) -> bool; fn wm_labels(&self) -> Option>; + fn wm_failure(&self) -> Option; + fn result_metadata(&self) -> ResultMetadata; fn size(&self) -> usize; } -#[derive(serde::Deserialize)] -struct ResultLabels { - wm_labels: Vec, +/// The Windmill-specific markers we look for inside a job's result. +/// `wm_failure` retags a successful run as a failure with the +/// given message; `wm_labels` adds runtime labels to the job row. +#[derive(serde::Deserialize, Default, Debug, Clone)] +pub struct ResultMetadata { + pub wm_labels: Option>, + pub wm_failure: Option, +} + +/// Sentinel `error.name` we inject into a result when retagging a successful +/// run as a failure due to `wm_failure`. Used downstream to detect that +/// the result is already in the standard `{ error: { name, message }, ... }` +/// shape and must not be wrapped a second time by `WrappedError`. +pub const MANUAL_FAILURE_ERROR_NAME: &str = "ManualFailure"; + +/// Returns true when the result already carries our injected +/// `error: { name: "ManualFailure", ... }` marker — i.e. it was shaped by +/// `process_jc`'s wm_failure path. A real runtime failure whose raw +/// result happens to contain a `wm_failure` field but no such error +/// key returns false (and so still goes through the standard wrap path). +pub fn is_pre_shaped_wm_failure_result(result: &str) -> bool { + #[derive(serde::Deserialize)] + struct Marker { + error: Option, + } + #[derive(serde::Deserialize)] + struct NameOnly { + name: String, + } + serde_json::from_str::(result) + .ok() + .and_then(|m| m.error) + .map(|e| e.name == MANUAL_FAILURE_ERROR_NAME) + .unwrap_or(false) } impl ValidableJson for WrappedError { @@ -647,6 +680,14 @@ impl ValidableJson for WrappedError { None } + fn wm_failure(&self) -> Option { + None + } + + fn result_metadata(&self) -> ResultMetadata { + ResultMetadata::default() + } + fn size(&self) -> usize { 0 } @@ -658,9 +699,15 @@ impl ValidableJson for Box { } fn wm_labels(&self) -> Option> { - serde_json::from_str::(self.get()) - .ok() - .map(|r| r.wm_labels) + self.result_metadata().wm_labels + } + + fn wm_failure(&self) -> Option { + self.result_metadata().wm_failure + } + + fn result_metadata(&self) -> ResultMetadata { + serde_json::from_str::(self.get()).unwrap_or_default() } fn size(&self) -> usize { @@ -677,6 +724,14 @@ impl ValidableJson for Arc { T::wm_labels(&self) } + fn wm_failure(&self) -> Option { + T::wm_failure(&self) + } + + fn result_metadata(&self) -> ResultMetadata { + T::result_metadata(&self) + } + fn size(&self) -> usize { T::size(&self) } @@ -688,9 +743,15 @@ impl ValidableJson for serde_json::Value { } fn wm_labels(&self) -> Option> { - serde_json::from_value::(self.clone()) - .ok() - .map(|r| r.wm_labels) + self.result_metadata().wm_labels + } + + fn wm_failure(&self) -> Option { + self.result_metadata().wm_failure + } + + fn result_metadata(&self) -> ResultMetadata { + serde_json::from_value::(self.clone()).unwrap_or_default() } fn size(&self) -> usize { @@ -707,6 +768,14 @@ impl ValidableJson for Json { self.0.wm_labels() } + fn wm_failure(&self) -> Option { + self.0.wm_failure() + } + + fn result_metadata(&self) -> ResultMetadata { + self.0.result_metadata() + } + fn size(&self) -> usize { self.0.size() } @@ -742,16 +811,7 @@ where } } -pub async fn add_completed_job_error( - db: &Pool, - completed_job: &MiniCompletedJob, - mem_peak: i32, - canceled_by: Option, - e: serde_json::Value, - _worker_name: &str, - flow_is_done: bool, - duration: Option, -) -> Result { +async fn record_failure_metrics(completed_job: &MiniCompletedJob, _worker_name: &str) { #[cfg(feature = "prometheus")] register_metric( &WORKER_EXECUTION_FAILED, @@ -772,6 +832,64 @@ pub async fn add_completed_job_error( .await; otel_incr_worker_execution_failed(&completed_job.tag); +} + +/// Tag a completed job as a failure while storing the result as-is, without +/// the standard `WrappedError` `{ error: ... }` wrap. Use for jobs whose result +/// is already shaped (e.g. when `wm_failure` injected a top-level +/// `error` key, while preserving sibling fields like `windmill_status_code`). +/// +/// This is a worker-internal helper called by trusted result-processing code +/// after the worker has authenticated and pulled the job. Callers MUST verify +/// upstream auth (i.e. the job was legitimately pulled by this worker) — this +/// function performs no authorization check itself, mirroring the contract of +/// `add_completed_job_error`. +pub async fn add_completed_job_pre_shaped_failure( + db: &Pool, + completed_job: &MiniCompletedJob, + mem_peak: i32, + canceled_by: Option, + result: Json<&T>, + worker_name: &str, + flow_is_done: bool, + duration: Option, +) -> Result<(), Error> { + record_failure_metrics(completed_job, worker_name).await; + + tracing::error!( + "job {} in {} did not succeed (wm_failure)", + completed_job.id, + completed_job.workspace_id, + ); + let _ = add_completed_job( + db, + completed_job, + false, + false, + result, + None, + mem_peak, + canceled_by, + flow_is_done, + duration, + false, + ) + .warn_after_seconds(10) + .await?; + Ok(()) +} + +pub async fn add_completed_job_error( + db: &Pool, + completed_job: &MiniCompletedJob, + mem_peak: i32, + canceled_by: Option, + e: serde_json::Value, + worker_name: &str, + flow_is_done: bool, + duration: Option, +) -> Result { + record_failure_metrics(completed_job, worker_name).await; let result = WrappedError { error: e }; tracing::error!( diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 1f771f6879..9eac0a5dcd 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -35,8 +35,9 @@ use windmill_common::{ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ - append_logs, get_mini_completed_job, CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, - MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG, + append_logs, get_mini_completed_job, is_pre_shaped_wm_failure_result, CanceledBy, FlowRunners, + JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG, + MANUAL_FAILURE_ERROR_NAME, }; use serde_json::{json, value::RawValue, Value}; @@ -61,8 +62,37 @@ struct ErrorMessage { name: String, } +#[derive(Debug, Deserialize)] +struct NestedErrorMessage { + error: ErrorMessage, +} + +/// Extract `{ name, message }` from a result. Accepts both the standard +/// top-level shape (regular runtime errors) and the nested `{ error: { name, +/// message }, ... }` shape produced by the wm_failure injection. +/// +/// For wm_failure-injected results, we prefer the nested error: a successful +/// run may legitimately contain top-level `name`/`message` fields (user data +/// named `name`/`message`), and we want OTel to record the ManualFailure +/// rather than the user's sibling fields. +fn extract_error_message(raw: &str) -> Option { + let nested = serde_json::from_str::(raw) + .ok() + .map(|n| n.error); + if matches!(&nested, Some(em) if em.name == MANUAL_FAILURE_ERROR_NAME) { + return nested; + } + if let Ok(em) = serde_json::from_str::(raw) { + return Some(em); + } + nested +} + +/// Returns the post-processing `success` value (after any `wm_failure` +/// override). Callers use this to make worker-loop decisions that depend on +/// whether the job ultimately succeeded — e.g. the init-script killpill. async fn process_jc( - jc: JobCompleted, + mut jc: JobCompleted, worker_name: &str, base_internal_url: &str, db: &DB, @@ -73,7 +103,32 @@ async fn process_jc( killpill_rx: &tokio::sync::broadcast::Receiver<()>, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, #[cfg(feature = "benchmark")] bench_infos: &mut BenchmarkInfo, -) { +) -> bool { + // Parse `wm_labels` and `wm_failure` together (single `from_str`) + // so we don't deserialize the whole result twice on every job. + let metadata = jc.result.result_metadata(); + + // If the script returned a `wm_failure: ` field in its + // result, tag the run as a failure. Inject an `error: { name, message }` + // at the top level so error handlers / UI / OTel see the standard error + // shape, while preserving sibling fields (`windmill_status_code`, + // `windmill_content_type`, `windmill_headers`, the user's data) at the + // top level so sync webhook responses still honor them. + if jc.success { + if let Some(failure_msg) = metadata.wm_failure.as_ref() { + if let Ok(Value::Object(mut map)) = serde_json::from_str::(jc.result.get()) { + map.insert( + "error".to_string(), + json!({ "name": MANUAL_FAILURE_ERROR_NAME, "message": failure_msg }), + ); + if let Ok(raw) = serde_json::value::to_raw_value(&Value::Object(map)) { + jc.result = Arc::new(raw); + } + } + jc.success = false; + } + } + let success: bool = jc.success; let span = if success { @@ -125,7 +180,7 @@ async fn process_jc( jc.job.id }; - if let Some(labels) = jc.result.wm_labels() { + if let Some(labels) = metadata.wm_labels.as_ref() { if !labels.is_empty() { span.record("labels", labels.join(",")); } @@ -163,7 +218,7 @@ async fn process_jc( span.record("script_hash", script_hash.to_string().as_str()); } if !success { - if let Ok(result_error) = serde_json::from_str::(jc.result.get()) { + if let Some(result_error) = extract_error_message(jc.result.get()) { span.record("error.message", result_error.message.as_str()); span.record("error.name", result_error.name.as_str()); span.record( @@ -218,6 +273,8 @@ async fn process_jc( ) .await; } + + success } enum JobCompletedRx { @@ -311,8 +368,7 @@ pub fn start_background_processor( result: SendResultPayload::JobCompleted(jc), time, }) => { - let is_init_script_and_failure = - !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG; + let is_init_script = jc.job.tag.as_str() == INIT_SCRIPT_TAG; let is_dependency_job = matches!( jc.job.kind, JobKind::Dependencies | JobKind::FlowDependencies @@ -322,7 +378,10 @@ pub fn start_background_processor( #[cfg(feature = "benchmark")] let is_top_level_job = jc.job.parent_job.is_none(); - process_jc( + // process_jc returns the post-override success value so a + // job that flipped to failure via `wm_failure` still + // triggers the init-script killpill. + let final_success = process_jc( jc, &worker_name, &base_internal_url, @@ -340,7 +399,7 @@ pub fn start_background_processor( .warn_after_seconds(10) .await; - if is_init_script_and_failure { + if is_init_script && !final_success { tracing::error!("init script errored, exiting"); killpill_tx.send(); break; @@ -793,19 +852,44 @@ pub async fn process_completed_job( } } } else { - let result = add_completed_job_error( - db, - &job, - mem_peak.to_owned(), - canceled_by.clone(), - serde_json::from_str(result.get()).unwrap_or_else( - |_| json!({ "message": format!("Non serializable error: {}", result.get()) }), - ), - worker_name, - false, - None, - ) - .await?; + // The result already carries our injected + // `error: { name: "ManualFailure", ... }` marker when process_jc + // retagged a successful run as a failure — store it as-is to preserve + // sibling fields like `windmill_status_code`. We check for the + // injected marker specifically (not just the presence of a + // `wm_failure` field) so a real runtime failure whose raw + // result happens to contain a `wm_failure` field still goes + // through the standard `WrappedError { error: ... }` wrap path. + let downstream_result: Arc> = if is_pre_shaped_wm_failure_result(result.get()) + { + windmill_queue::add_completed_job_pre_shaped_failure( + db, + &job, + mem_peak.to_owned(), + canceled_by.clone(), + Json(&*result), + worker_name, + false, + None, + ) + .await?; + result.clone() + } else { + let wrapped = add_completed_job_error( + db, + &job, + mem_peak.to_owned(), + canceled_by.clone(), + serde_json::from_str(result.get()).unwrap_or_else( + |_| json!({ "message": format!("Non serializable error: {}", result.get()) }), + ), + worker_name, + false, + None, + ) + .await?; + Arc::new(serde_json::value::to_raw_value(&wrapped).unwrap()) + }; if job.is_flow_step() { if let Some(parent_job) = job.parent_job { tracing::error!(parent_flow = %parent_job, subflow = %job.id, "process completed job error, updating flow status"); @@ -817,7 +901,7 @@ pub async fn process_completed_job( &job.workspace_id, false, canceled_by, - Arc::new(serde_json::value::to_raw_value(&result).unwrap()), + downstream_result, duration.and_then(|d| { job.started_at.map(|started_at| FlowJobDuration { started_at: started_at, @@ -855,13 +939,12 @@ pub async fn process_completed_job( .fetch_optional(db) .await?; if let Some(Some(job_ids)) = job_ids_json { - let err_result = Arc::new(serde_json::value::to_raw_value(&result).unwrap()); if let Ok(Some(_)) = handle_wac_child_completion( db, &job.id, parent_job, &job.workspace_id, - err_result, + downstream_result, false, job_ids, )