#[cfg(feature = "otel")] use opentelemetry::trace::FutureExt; use serde::{Deserialize, Serialize}; use sqlx::types::Json; use std::{ collections::HashMap, sync::{ atomic::{AtomicBool, AtomicU16, Ordering}, Arc, }, }; use tracing::{field, Instrument}; #[cfg(not(feature = "otel"))] use windmill_common::otel_oss::FutureExt; use uuid::Uuid; /// Set by the result processor when a WAC child completion makes suspend reach 0, /// signaling the worker main loop to check for suspended jobs immediately. pub static WAC_SUSPEND_READY: AtomicBool = AtomicBool::new(false); use windmill_common::{ add_time, error::{self, Error}, flow_status::FlowJobDuration, jobs::JobKind, utils::WarnAfterExt, worker::{error_to_value, to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, KillpillSender, DB, }; #[cfg(feature = "benchmark")] use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ append_logs, asset_dispatch, 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}; use tokio::{sync::Notify, task::JoinHandle}; use windmill_queue::{add_completed_job, add_completed_job_error}; use crate::{ bash_executor::ANSI_ESCAPE_RE, 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, UpdateFlow, SAME_WORKER_REQUIREMENTS, }; use windmill_common::client::AuthedClient; #[derive(Debug, Deserialize)] struct ErrorMessage { message: String, 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( mut jc: JobCompleted, worker_name: &str, base_internal_url: &str, db: &DB, worker_dir: &str, same_worker_tx: Option<&SameWorkerSender>, job_completed_sender: &JobCompletedSender, stats_map: &JobStatsMap, 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 { tracing::span!( tracing::Level::INFO, "job_postprocessing", job_id = %jc.job.id, root_job = field::Empty, workspace_id = %jc.job.workspace_id, worker = %worker_name,tag = %jc.job.tag, // hostname = %hostname, language = field::Empty, script_path = field::Empty, flow_step_id = field::Empty, parent_job = field::Empty, job_kind = %jc.job.kind.as_str(), created_by = %jc.job.created_by, trigger_kind = field::Empty, trigger = field::Empty, script_hash = field::Empty, otel.name = field::Empty, success = %success, labels = field::Empty, ) } else { tracing::span!( tracing::Level::INFO, "job_postprocessing", job_id = %jc.job.id, root_job = field::Empty, workspace_id = %jc.job.workspace_id, worker = %worker_name,tag = %jc.job.tag, // hostname = %hostname, language = field::Empty, script_path = field::Empty, flow_step_id = field::Empty, parent_job = field::Empty, job_kind = %jc.job.kind.as_str(), created_by = %jc.job.created_by, trigger_kind = field::Empty, trigger = field::Empty, script_hash = field::Empty, otel.name = field::Empty, otel.status_code = "ERROR", otel.status_message = field::Empty, success = %success, error.message = field::Empty, error.name = field::Empty, labels = field::Empty, ) }; let rj = if let Some(root_job) = jc.job.flow_innermost_root_job { root_job } else { jc.job.id }; if let Some(labels) = metadata.wm_labels.as_ref() { if !labels.is_empty() { span.record("labels", labels.join(",")); } } // The secondary `job_postprocessing` span stays on the UUID-derived context // (MiniCompletedJob carries no args, so the inbound traceparent isn't // available here); the primary job span is relocated in `create_span_with_name`. windmill_common::otel_oss::set_span_parent(&span, &rj); if let Some(lg) = jc.job.script_lang.as_ref() { span.record("language", lg.as_str()); } if let Some(step_id) = jc.job.flow_step_id.as_ref() { span.record( "otel.name", format!("job_postprocessing {}", step_id).as_str(), ); span.record("flow_step_id", step_id.as_str()); } else { span.record("otel.name", "job postprocessing"); } if let Some(parent_job) = jc.job.parent_job.as_ref() { span.record("parent_job", parent_job.to_string().as_str()); } if let Some(script_path) = jc.job.runnable_path.as_ref() { span.record("script_path", script_path.as_str()); } if let Some(root_job) = jc.job.flow_innermost_root_job.as_ref() { span.record("root_job", root_job.to_string().as_str()); } if let Some(trigger_kind) = jc.job.trigger_kind.as_ref() { span.record("trigger_kind", trigger_kind.to_string().as_str()); } if let Some(trigger) = jc.job.trigger.as_ref() { span.record("trigger", trigger.as_str()); } if let Some(script_hash) = jc.job.runnable_id.as_ref() { span.record("script_hash", script_hash.to_string().as_str()); } if !success { 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( "otel.status_message", crate::worker::truncate_description(&result_error.message).as_str(), ); } else { span.record("otel.status_message", "Job failed"); } } // Extract stats info before moving jc let duration_ms = jc.duration.clone(); let script_lang = jc.job.script_lang.clone(); let workspace_id = jc.job.workspace_id.clone(); let root_job = handle_receive_completed_job( jc, &base_internal_url, &db, worker_dir, same_worker_tx, &worker_name, job_completed_sender.clone(), killpill_rx, #[cfg(feature = "benchmark")] bench, ) .instrument(span) .warn_after_seconds(10) .await; if let Some(root_job) = root_job { add_root_flow_job_to_otlp(&root_job, success); #[cfg(feature = "benchmark")] if bench_infos.count_top_level(root_job.id) { bench_infos .shared_iters .fetch_add(1, std::sync::atomic::Ordering::Relaxed); } } // Accumulate job stats if duration is available if let Some(duration_ms) = duration_ms { accumulate_job_stats( stats_map, &*WORKER_GROUP, script_lang, &workspace_id, duration_ms, ) .await; } success } enum JobCompletedRx { JobCompleted(SendResult), Killpill, WakeUp, } pub fn start_background_processor( job_completed_rx: JobCompletedReceiver, job_completed_sender: JobCompletedSender, same_worker_queue_size: Arc, job_completed_processor_is_done: Arc, wake_up_notify: Arc, last_processing_duration: Arc, base_internal_url: String, db: DB, worker_dir: String, same_worker_tx: SameWorkerSender, worker_name: String, killpill_tx: KillpillSender, is_dedicated_worker: bool, // True when this processor runs inside the agent-worker API server, relaying // completions on behalf of many remote agent workers. Such a processor must // never kill itself: dropping its receiver would disconnect the shared // job-completed channel and make every future /send_result fail until the // whole server is restarted. is_agent_server: bool, stats_map: JobStatsMap, ) -> JoinHandle<()> { tokio::spawn(async move { let mut has_been_killed = false; let JobCompletedReceiver { bounded_rx, mut killpill_rx, unbounded_rx } = job_completed_rx; #[cfg(feature = "benchmark")] let mut infos = BenchmarkInfo::new(windmill_common::bench::shared_bench_iters()); // Start periodic stats flush task let db_clone = db.clone(); let stats_map_clone = stats_map.clone(); let mut killpill_rx_clone = killpill_rx.resubscribe(); let flush_handle = tokio::spawn(async move { let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(900)); // Flush every 15 min loop { tokio::select! { _ = interval.tick() => { if let Err(e) = flush_stats_to_db(&db_clone, &stats_map_clone).await { tracing::error!("Failed to flush worker group job stats: {}", e); } } _ = killpill_rx_clone.recv() => { tracing::info!("bg processor received killpill signal, flushing remaining stats"); break; } } } }); //if we have been killed, we want to drain the queue of jobs while let Some(sr) = { if has_been_killed { tracing::info!("bg processor is killed, draining. same_worker_queue_size: {}, unbounded_rx: {}, bounded_rx: {}", same_worker_queue_size.load(Ordering::SeqCst), unbounded_rx.len(), bounded_rx.len()) } if has_been_killed && same_worker_queue_size.load(Ordering::SeqCst) == 0 { unbounded_rx .try_recv() .ok() .map(JobCompletedRx::JobCompleted) .or_else(|| bounded_rx.try_recv().ok().map(JobCompletedRx::JobCompleted)) } else { tokio::select! { biased; result = unbounded_rx.recv_async() => { result.ok().map(JobCompletedRx::JobCompleted) } result = bounded_rx.recv_async() => { result.ok().map(JobCompletedRx::JobCompleted) }, _ = wake_up_notify.notified() => { tracing::info!("bg processor received wake up signal, checking if same worker queue is empty"); Some(JobCompletedRx::WakeUp) }, _ = killpill_rx.recv() => { tracing::info!("bg processor received killpill signal, queuing killpill job"); Some(JobCompletedRx::Killpill) } } } } { #[cfg(feature = "benchmark")] let mut bench = BenchmarkIter::new(); match sr { JobCompletedRx::JobCompleted(SendResult { result: SendResultPayload::JobCompleted(jc), time, }) => { let is_init_script = jc.job.tag.as_str() == INIT_SCRIPT_TAG; let is_dependency_job = matches!( jc.job.kind, JobKind::Dependencies | JobKind::FlowDependencies ); let jc_id = jc.job.id; #[cfg(feature = "benchmark")] let bench_job_id = jc.job.id; #[cfg(feature = "benchmark")] let is_top_level_job = jc.job.parent_job.is_none(); // 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, &db, &worker_dir, Some(&same_worker_tx), &job_completed_sender, &stats_map, &killpill_rx, #[cfg(feature = "benchmark")] &mut bench, #[cfg(feature = "benchmark")] &mut infos, ) .warn_after_seconds(10) .await; if is_init_script && !final_success { if is_agent_server { // The failed init script belongs to a remote agent // worker, not to this server. That worker handles its // own restart; killing the server relay here would // strand every other agent worker's completions. tracing::error!( job_id = %jc_id, "agent worker init script errored; failure recorded, keeping server bg processor alive" ); } else { tracing::error!("init script errored, exiting"); killpill_tx.send(); break; } } if is_dependency_job && is_dedicated_worker { tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted."); sqlx::query!( "UPDATE config SET config = config WHERE name = $1", format!("worker__{}", *WORKER_GROUP) ) .execute(&db) .await .expect("update config to trigger restart of all dedicated workers at that config"); killpill_tx.send(); } add_time!(bench, "job completed processed"); #[cfg(feature = "benchmark")] { if infos.add_iter(bench, bench_job_id, is_top_level_job) { infos.shared_iters.fetch_add(1, Ordering::Relaxed); } } last_processing_duration .store(time.elapsed().as_secs() as u16, Ordering::SeqCst); } JobCompletedRx::JobCompleted(SendResult { result: SendResultPayload::UpdateFlow(UpdateFlow { flow, w_id, success, result, worker_dir, stop_early_override, token, }), time, }) => { // let r; tracing::info!(parent_flow = %flow, "updating flow status after job completion"); if let Err(e) = update_flow_status_after_job_completion( &db, &AuthedClient::new( base_internal_url.to_string(), w_id.clone(), token.clone(), None, ), flow, &Uuid::nil(), &w_id, success, None, Arc::new(result), None, true, &same_worker_tx, &worker_dir, stop_early_override, &worker_name, job_completed_sender.clone(), None, &killpill_rx, #[cfg(feature = "benchmark")] &mut bench, ) .await { tracing::error!("Error updating flow status after job completion for {flow} on {worker_name}: {e:#}"); } #[cfg(feature = "benchmark")] { if infos.add_iter(bench, flow, true) { infos.shared_iters.fetch_add(1, Ordering::Relaxed); } } last_processing_duration .store(time.elapsed().as_secs() as u16, Ordering::SeqCst); } JobCompletedRx::Killpill => { tracing::info!("killpill job received, processing only same worker jobs"); has_been_killed = true; } JobCompletedRx::WakeUp => {} } } // Flush any remaining stats before shutting down tracing::info!("flushing remaining stats before shutting down"); let flush_result = tokio::time::timeout(std::time::Duration::from_secs(10), flush_handle).await; match flush_result { Ok(Ok(())) => tracing::info!("Stats flushed successfully"), Ok(Err(join_err)) => tracing::error!("Stats flush task failed: {}", join_err), Err(_) => tracing::error!("Stats flush timed out after 10 seconds"), } job_completed_processor_is_done.store(true, Ordering::SeqCst); tracing::info!("finished processing all completed jobs"); #[cfg(feature = "benchmark")] { infos .write_to_file("profiling_result_processor.json") .expect("write to file profiling"); } }) } async fn send_job_completed(job_completed_tx: JobCompletedSender, jc: JobCompleted) { if let Err(e) = job_completed_tx .send_job(jc, true) .with_context(windmill_common::otel_oss::otel_ctx()) .await { tracing::error!("send job completed failed, triggering worker shutdown: {e:#}"); job_completed_tx.send_worker_killpill(); } } pub async fn process_result( job: MiniCompletedJob, result: error::Result>>, job_dir: &str, job_completed_tx: JobCompletedSender, mem_peak: i32, canceled_by: Option, cached_res_path: Option, token: &str, result_columns: Option>, preprocessed_args: Option>>, conn: &Connection, duration: Option, has_stream: bool, flow_runners: Option>, ) -> error::Result { match result { Ok(result) => { send_job_completed( job_completed_tx, JobCompleted { job, preprocessed_args, result, result_columns, mem_peak, canceled_by, success: true, cached_res_path, token: token.to_string(), duration, has_stream: Some(has_stream), from_cache: None, flow_runners, done_tx: None, }, ) .with_context(windmill_common::otel_oss::otel_ctx()) .await; Ok(crate::worker::JobOutcome::Completed) } Err(e) => { let error_value = match e { Error::ExitStatus(program, i) => { let res = read_result(job_dir, None).await.ok(); if res.as_ref().is_some_and(|x| !x.get().is_empty()) { res.unwrap() } else { match conn { Connection::Sql(db) => { let last_10_log_lines = sqlx::query_scalar!( "SELECT right(logs, 600) FROM job_logs WHERE job_id = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1", &job.id, &job.workspace_id ).fetch_one(db).await.ok().flatten().unwrap_or("".to_string()); let log_lines = last_10_log_lines .split("CODE EXECUTION ---") .last() .unwrap_or(&last_10_log_lines); extract_error_value( &program, log_lines, i, job.flow_step_id.clone(), ) } Connection::Http(_) => { to_raw_value(&"See logs for more details".to_string()) } } } } Error::ExecutionRawError(e) => to_raw_value(&e), err @ _ => to_raw_value(&SerializedError { message: format!("execution error:\n{err:#}",), name: "ExecutionErr".to_string(), step_id: job.flow_step_id.clone(), exit_code: None, }), }; // Use the structured error message that was just extracted (the // user-facing script error) rather than the generic Error string. // Pull `.message` out of the JSON object if present, otherwise // accept a bare string (e.g. agent-worker "See logs for more // details"), and only fall back to "Job failed" when the value // carries no readable description. let description = serde_json::from_str::(error_value.get()) .ok() .and_then(|value| { value .get("message") .and_then(|m| m.as_str()) .or_else(|| value.as_str()) .map(|m| crate::worker::truncate_description(m)) }) .unwrap_or_else(|| "Job failed".to_string()); send_job_completed( job_completed_tx, JobCompleted { job, result: Arc::new(to_raw_value(&error_value)), result_columns: None, preprocessed_args: None, mem_peak, canceled_by, success: false, cached_res_path, token: token.to_string(), duration, has_stream: Some(has_stream), from_cache: None, flow_runners, done_tx: None, }, ) .with_context(windmill_common::otel_oss::otel_ctx()) .await; Ok(crate::worker::JobOutcome::Failed { description }) } } } pub async fn handle_receive_completed_job( jc: JobCompleted, base_internal_url: &str, db: &DB, worker_dir: &str, same_worker_tx: Option<&SameWorkerSender>, worker_name: &str, job_completed_tx: JobCompletedSender, killpill_rx: &tokio::sync::broadcast::Receiver<()>, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> Option> { let token = jc.token.clone(); let workspace = jc.job.workspace_id.clone(); let client = AuthedClient::new(base_internal_url.to_string(), workspace, token, None); let job = jc.job.clone(); let mem_peak = jc.mem_peak.clone(); let canceled_by = jc.canceled_by.clone(); let processed_completed_job = process_completed_job( jc, &client, db, &worker_dir, same_worker_tx.clone(), worker_name, job_completed_tx.clone(), killpill_rx, #[cfg(feature = "benchmark")] bench, ) .warn_after_seconds(10) .await; match processed_completed_job { // The job was already completed by another worker (e.g. this worker was // declared a zombie and the job restarted+finished elsewhere, then this // worker caught up). The job genuinely succeeded; routing this through // `handle_job_error` would propagate a spurious "AlreadyCompleted" // failure up the parent flow. Drop it instead, mirroring the // `JobOutcome::AlreadyCompleted` guard on the execution path. Err(err @ Error::AlreadyCompleted(_)) => { tracing::info!( job_id = %job.id, "job already completed by another worker, skipping result processing: {err:#}" ); None } Err(err) => { handle_job_error( db, &client, &job, mem_peak, canceled_by, err, false, same_worker_tx.clone(), &worker_dir, worker_name, job_completed_tx, killpill_rx, #[cfg(feature = "benchmark")] bench, ) .await; None } Ok(r) => r, } } /// A git-sync check run threaded through a pull job: the PR diff preview (phase 4) /// or the live deploy status (phase 6). Both markers carry the same shape. #[cfg(all(feature = "enterprise", feature = "private"))] #[derive(serde::Deserialize)] struct GitSyncCheck { check_run_id: i64, repo_url: String, #[serde(default)] pr_number: Option, #[serde(default)] head_sha: Option, /// Whether the PR itself modifies wmill.yaml (None = undetermined); picks /// the wording for a settings difference in the diff summary. #[serde(default)] wmill_yaml_changed: Option, } /// Parsed diff summary from a (dry-run or real) pull result. `None` when the /// result can't be parsed into the expected shape. #[cfg(all(feature = "enterprise", feature = "private"))] fn parse_git_sync_changes(result_raw: &str) -> Option<(Vec<(String, String)>, bool)> { use serde::Deserialize; #[derive(Deserialize)] struct Change { #[serde(rename = "type")] change_type: String, path: String, } #[derive(Deserialize)] struct SettingsDiff { #[serde(rename = "hasChanges", default)] has_changes: bool, } #[derive(Deserialize)] struct SyncResponse { changes: Option>, #[serde(rename = "settingsDiffResult")] settings_diff_result: Option, } let resp = serde_json::from_str::(result_raw).ok()?; // A result carrying neither field isn't a recognizable diff; return None so the // caller falls back to the unsummarized path instead of a false "in sync". if resp.changes.is_none() && resp.settings_diff_result.is_none() { return None; } let settings_changed = resp .settings_diff_result .map(|s| s.has_changes) .unwrap_or(false); Some(( resp.changes .unwrap_or_default() .into_iter() .map(|c| (c.change_type, c.path)) .collect(), settings_changed, )) } #[cfg(all(feature = "enterprise", feature = "private"))] fn format_change_list(changes: &[(String, String)]) -> Vec { let mut lines = Vec::new(); for (change_type, path) in changes.iter().take(100) { lines.push(format!("- `{}` {}", change_type, path)); } if changes.len() > 100 { lines.push(format!("- ... and {} more", changes.len() - 100)); } lines } #[cfg(all(test, feature = "enterprise", feature = "private"))] mod git_sync_check_tests { use super::{format_change_list, parse_git_sync_changes}; #[test] fn parse_empty_changes_is_in_sync() { // Present-but-empty diff → a real "in sync" result, not None. let (changes, settings) = parse_git_sync_changes(r#"{"changes":[]}"#).unwrap(); assert!(changes.is_empty()); assert!(!settings); } #[test] fn parse_missing_fields_is_none() { // Neither field present → unrecognizable, falls back to the caller's path. assert!(parse_git_sync_changes("{}").is_none()); } #[test] fn parse_unparseable_is_none() { assert!(parse_git_sync_changes("not json").is_none()); } #[test] fn parse_changes_and_settings() { let (changes, settings) = parse_git_sync_changes( r#"{"changes":[{"type":"edited","path":"f/a"}],"settingsDiffResult":{"hasChanges":true}}"#, ) .unwrap(); assert_eq!(changes, vec![("edited".to_string(), "f/a".to_string())]); assert!(settings); } #[test] fn format_truncates_over_100() { let changes: Vec<(String, String)> = (0..150) .map(|i| ("edited".to_string(), format!("f/{i}"))) .collect(); let lines = format_change_list(&changes); assert_eq!(lines.len(), 101); assert_eq!(lines.last().unwrap(), "- ... and 50 more"); } } /// When an auto-pull job (carrying `__git_sync_auto_pull`) fails, roll the /// optimistic `last_synced_sha` advance back to the pre-pull value so the commit /// is retried instead of being silently treated as synced, and record the failure. #[cfg(all(feature = "enterprise", feature = "private"))] async fn maybe_reconcile_git_sync_auto_pull( db: &DB, job_id: &uuid::Uuid, workspace_id: &str, success: bool, ) { if success { return; // the optimistic synced state is already correct } let marker: Option = match sqlx::query_scalar!( "SELECT args->'__git_sync_auto_pull' FROM v2_job WHERE id = $1", job_id ) .fetch_optional(db) .await { Ok(v) => v.flatten(), Err(e) => { tracing::error!("git auto-pull: failed to read job args: {e:#}"); return; } }; let Some(marker) = marker else { return; }; #[derive(serde::Deserialize)] struct AutoPullMarker { repo_resource_path: String, #[serde(default)] prev_synced: std::collections::HashMap, } let Ok(m) = serde_json::from_value::(marker) else { return; }; windmill_git_sync::record_auto_pull_failure( db, workspace_id, &m.repo_resource_path, &m.prev_synced, "auto-pull job failed".to_string(), ) .await; } /// Branch a git-sync push job deployed to, mirroring the hub script's /// derivation: a dev workspace deploys to its environment-label branch /// (`dev`/`staging`), other fork workspaces to `wm-fork//`, /// else the promotion `wm_deploy/**` formula (per-folder or per-item form). /// A dev workspace in promotion mode is the exception: it takes the promotion /// `wm_deploy/**` formula (per-item PRs into the parent) instead of its label /// branch. `None` when the deploy stays on the base branch (workspace-wide /// mode) and has no PR to open. #[cfg(all(feature = "enterprise", feature = "private"))] fn git_sync_deploy_pr_head_branch( workspace_id: &str, parent_workspace_id: Option<&str>, dev_workspace_label: Option<&str>, base: &str, use_individual_branch: bool, group_by_folder: bool, item_path: &str, item_parent_path: &str, path_type: &str, ) -> Option { let is_dev = dev_workspace_label.is_some(); // A dev workspace with promotion on falls through to the wm_deploy/** // formula below; the label/fork branches only apply when promotion is off. if !(is_dev && use_individual_branch) { if is_dev { return Some(windmill_common::workspaces::dev_workspace_branch( dev_workspace_label, )); } let is_fork = parent_workspace_id.is_some() || workspace_id.starts_with(windmill_common::workspaces::WM_FORK_PREFIX); if is_fork { let suffix = workspace_id .strip_prefix(windmill_common::workspaces::WM_FORK_PREFIX) .unwrap_or(workspace_id); return Some(format!("wm-fork/{base}/{suffix}")); } } if !use_individual_branch { return None; } // Mirrors the CLI's computeGitSyncDeployBranch: user/group objects are // pushed to the base branch and never get their own wm_deploy branch. if path_type == "user" || path_type == "group" { return None; } let git_ref = if !item_path.is_empty() { item_path } else { item_parent_path }; if git_ref.is_empty() { return None; } Some(if group_by_folder { format!( "wm_deploy/{workspace_id}/{}", git_ref.split('/').take(2).collect::>().join("__") ) } else { format!( "wm_deploy/{workspace_id}/{}/{}", path_type, git_ref.replace('/', "__") ) }) } /// Whether the push job's result says a commit was actually pushed. `None` /// when the result doesn't carry the flag (hub script versions predating it). #[cfg(all(feature = "enterprise", feature = "private"))] fn git_sync_push_result_pushed(result: &str) -> Option { serde_json::from_str::(result) .ok()? .get("pushed")? .as_bool() } /// When a git-sync push job carrying `__git_sync_open_pr` succeeds, open (or /// reopen) the PR for the branch it pushed: `wm-fork//` for a fork /// deploy, `wm_deploy/**` for a promotion deploy. Runs outbound with the /// installation token, so it works regardless of webhook reachability. /// Best-effort: failures are logged, never propagated. #[cfg(all(feature = "enterprise", feature = "private"))] async fn maybe_open_git_sync_deploy_pr( db: &DB, job_id: &uuid::Uuid, workspace_id: &str, result: &str, ) { // A no-op push (workspace already matches the repo — e.g. the deploy was // itself caused by an auto-pull) must not ensure a PR: it would recreate // PRs the user closed and spam creation attempts with no diff. if git_sync_push_result_pushed(result) == Some(false) { return; } let row = match sqlx::query!( r#"SELECT args->'__git_sync_open_pr' as "marker", args->>'repo_url_resource_path' as "repo_path", args->>'parent_workspace_id' as "parent_workspace_id", args->>'dev_workspace_label' as "dev_workspace_label", args->>'parent_dev_workspace_label' as "parent_dev_workspace_label", COALESCE((args->'use_individual_branch')::bool, false) as "use_individual_branch!", COALESCE((args->'group_by_folder')::bool, false) as "group_by_folder!", COALESCE(args->'items'->0->>'path', args->>'path', '') as "item_path!", COALESCE(args->'items'->0->>'parent_path', args->>'parent_path', '') as "item_parent_path!", COALESCE(args->'items'->0->>'path_type', args->>'path_type', '') as "path_type!", COALESCE(args->'items'->0->>'commit_msg', args->>'commit_msg', '') as "commit_msg!" FROM v2_job WHERE id = $1"#, job_id ) .fetch_optional(db) .await { Ok(Some(r)) => r, Ok(None) => return, Err(e) => { tracing::error!("git sync PR: failed to read job args: {e:#}"); return; } }; if row.marker.is_none() { return; } // Runtime Enterprise gate, like the poller: the toggles may have been set // while a license was active (or written directly), and this hook drives // GitHub API calls with the installation token. if !matches!( windmill_common::ee_oss::get_license_plan().await, windmill_common::ee_oss::LicensePlan::Enterprise ) { tracing::warn!( "git sync PR: skipping PR creation for {workspace_id}: requires an Enterprise license" ); return; } let Some(repo_path) = row.repo_path else { return; }; // Base = the tracked branch (resource branch, else the repo default). Also // acts as the app-backed gate: PR creation needs the installation token. let base = match windmill_common::git_sync_ee::get_app_repo_head_for_autopull( db, workspace_id, &repo_path, ) .await { Ok(Some((branch, _))) => branch, Ok(None) => { tracing::warn!( "git sync PR: repo {repo_path} in {workspace_id} has a PR-on-deploy toggle set but is not GitHub-App-backed; skipping (connect the repo through the GitHub App, or use the open-pr-on-commit workflow)" ); return; } Err(e) => { tracing::warn!("git sync PR: could not resolve base branch for {repo_path}: {e:#}"); return; } }; let Some(head) = git_sync_deploy_pr_head_branch( workspace_id, row.parent_workspace_id.as_deref(), row.dev_workspace_label.as_deref(), &base, row.use_individual_branch, row.group_by_folder, &row.item_path, &row.item_parent_path, &row.path_type, ) else { return; }; let repo_url = match windmill_common::git_sync_ee::resolve_repo_url_interpolated( db, workspace_id, &repo_path, ) .await { Ok(url) => url, Err(e) => { tracing::warn!("git sync PR: could not resolve repo url for {repo_path}: {e:#}"); return; } }; // A fork of a dev workspace diverged from the dev's label branch, so its PR // merges back there; everything else targets the tracked branch. let pr_base = row.parent_dev_workspace_label.as_deref().unwrap_or(&base); match windmill_common::git_sync_ee::ensure_pull_request( db, workspace_id, &repo_url, &head, pr_base, &row.commit_msg, ) .await { Ok(()) => { persist_git_sync_open_pr_error(db, workspace_id, &repo_path, None).await; } Err(e) => { tracing::warn!( "git sync PR: failed to open PR {head} -> {pr_base} for {repo_path}: {e:#}" ); let msg: String = format!("{e:#}").chars().take(400).collect(); persist_git_sync_open_pr_error(db, workspace_id, &repo_path, Some(msg)).await; } } } /// Best-effort: record (or clear) the last PR-creation failure on the repo's /// settings so the toggle can explain a silent no-op in the UI (the usual /// cause is a GitHub App installation that hasn't approved the pull-request /// permission yet). Merges into a fresh read of the row, only writes on change. #[cfg(all(feature = "enterprise", feature = "private"))] async fn persist_git_sync_open_pr_error( db: &DB, workspace_id: &str, repo_path: &str, error: Option, ) { // Single targeted update (like the EE auto-pull status writer): a full // read-modify-write of the column would race the poller's concurrent // last_synced_sha/last_pull_status writes and silently clobber them. let bare_path = repo_path.trim_start_matches("$res:"); let result: error::Result<()> = async { sqlx::query!( r#" UPDATE workspace_settings SET git_sync = jsonb_set( git_sync, '{repositories}', (SELECT jsonb_agg( CASE WHEN elem->>'git_repo_resource_path' IN ($2, '$res:' || $2) THEN CASE WHEN $3::text IS NULL THEN elem - 'open_pr_error' ELSE jsonb_set(elem, '{open_pr_error}', to_jsonb($3::text), true) END ELSE elem END) FROM jsonb_array_elements(git_sync->'repositories') AS elem) ) WHERE workspace_id = $1 AND jsonb_typeof(git_sync->'repositories') = 'array' "#, workspace_id, bare_path, error.as_deref(), ) .execute(db) .await?; Ok(()) } .await; if let Err(e) = result { tracing::warn!("git sync PR: failed to persist open_pr_error for {repo_path}: {e:#}"); } } /// When a git-sync pull job carrying a check marker completes, post the outcome /// to its GitHub check run: the PR diff preview (`__git_sync_pr_check`, phase 4) /// or the live deploy status (`__git_sync_deploy_check`, phase 6). /// A one-line description of the repo's sync filters, appended to an "in sync" /// PR verdict: a PR that only touches files outside these paths deploys /// nothing on merge, which otherwise looks like a wrong verdict. #[cfg(all(feature = "enterprise", feature = "private"))] fn format_git_sync_scope_note(include: &[String], exclude: &[String]) -> Option { if include.is_empty() { return None; } let fmt = |paths: &[String]| { paths .iter() .map(|p| format!("`{p}`")) .collect::>() .join(", ") }; let mut note = format!( "\n\nOnly files matching this repository's sync filters deploy on merge: {}", fmt(include) ); if !exclude.is_empty() { note.push_str(&format!(" (excluding {})", fmt(exclude))); } note.push('.'); Some(note) } #[cfg(all(feature = "enterprise", feature = "private"))] async fn git_sync_repo_scope_note( db: &DB, workspace_id: &str, repo_path: Option<&str>, ) -> Option { let repo_path = repo_path?; let settings = sqlx::query_scalar!( "SELECT git_sync FROM workspace_settings WHERE workspace_id = $1", workspace_id ) .fetch_optional(db) .await .ok()? .flatten()?; let settings: windmill_common::workspaces::WorkspaceGitSyncSettings = serde_json::from_value(settings).ok()?; // Job args carry the bare resource path; stored settings keep the $res: prefix. let repo = settings.repositories.iter().find(|r| { r.git_repo_resource_path.trim_start_matches("$res:") == repo_path.trim_start_matches("$res:") })?; let s = repo.settings.as_ref()?; format_git_sync_scope_note( &s.include_path, s.exclude_path.as_deref().unwrap_or_default(), ) } #[cfg(all(feature = "enterprise", feature = "private"))] async fn maybe_post_git_sync_check( db: &DB, job_id: &uuid::Uuid, workspace_id: &str, success: bool, result_raw: &str, ) { // Only git-sync pull jobs carry one of these markers; everything else no-ops. let row = match sqlx::query!( r#"SELECT args->'__git_sync_pr_check' AS "pr", args->'__git_sync_deploy_check' AS "deploy", args->>'repo_url_resource_path' AS "repo_path" FROM v2_job WHERE id = $1"#, job_id ) .fetch_optional(db) .await { Ok(r) => r, Err(e) => { tracing::error!("git sync-check: failed to read job args: {e:#}"); return; } }; let Some(row) = row else { return; }; // A PR dry-run and a deploy pull are mutually exclusive markers. let (is_deploy, marker) = match (row.pr, row.deploy) { (Some(pr), _) => (false, pr), (None, Some(deploy)) => (true, deploy), (None, None) => return, }; let Ok(mut check) = serde_json::from_value::(marker) else { return; }; // Markers carry the literal resource URL (job args are persisted, so a // `$var:`-resolved URL must not land there); interpolate before calling // GitHub. check.repo_url = match windmill_common::variables::get_variable_or_self(check.repo_url, db, workspace_id) .await { Ok(u) => u, Err(e) => { tracing::error!("git sync-check: cannot interpolate repo url: {e:#}"); return; } }; // "In sync" on a PR that visibly changes files reads as a bug when those // files are outside the repo's sync filters — say what the scope is. let scope_note = if !is_deploy && success { git_sync_repo_scope_note(db, workspace_id, row.repo_path.as_deref()).await } else { None }; let (conclusion, title, summary): (&str, String, String) = if is_deploy { // Phase 6: real deploy pull -> "Deployed N changes" / "In sync" / failure. if !success { ( "failure", format!("Deploy to {} failed", workspace_id), "Deploying the latest commit failed. See the job in Windmill for details." .to_string(), ) } else { match parse_git_sync_changes(result_raw) { Some((changes, settings_changed)) if changes.is_empty() && !settings_changed => ( "success", format!("In sync with {}", workspace_id), format!( "No changes to deploy to `{}` from this commit.", workspace_id ), ), Some((changes, settings_changed)) => { let mut lines = vec![format!( "Deployed {} change(s) to `{}`:\n", changes.len(), workspace_id )]; lines.extend(format_change_list(&changes)); if settings_changed { lines.push("\nWorkspace settings also changed.".to_string()); } ( "success", format!("Deployed {} change(s) to {}", changes.len(), workspace_id), lines.join("\n"), ) } None => ( "success", format!("Deployed to {}", workspace_id), format!("Windmill deployed the latest commit to `{}`.", workspace_id), ), } } } else { // Phase 4: dry-run diff preview for a PR. if !success { ( "failure", "Windmill diff failed".to_string(), "The dry-run pull to compute the diff failed. See the job in Windmill for details." .to_string(), ) } else { match parse_git_sync_changes(result_raw) { Some((changes, settings_changed)) if changes.is_empty() && !settings_changed => ( "success", "In sync".to_string(), format!( "Merging this PR would make no changes to the workspace.{}", scope_note.as_deref().unwrap_or_default() ), ), Some((changes, settings_changed)) => { let mut lines = vec![format!( "Merging this PR would apply {} change(s) to the workspace:\n", changes.len() )]; lines.extend(format_change_list(&changes)); if settings_changed { lines.push(match check.wmill_yaml_changed { Some(true) => "\nThis PR changes wmill.yaml: pulling also applies the updated workspace settings.".to_string(), Some(false) => "\nIndependent of this PR, the workspace's git-sync settings differ from the repo's wmill.yaml and a pull updates them to match.".to_string(), None => "\nA pull also updates the workspace's git-sync settings to match the repo's wmill.yaml.".to_string(), }); } ( "neutral", format!("{} change(s) to deploy", changes.len()), lines.join("\n"), ) } None => ( "neutral", "Diff computed".to_string(), "Windmill computed a diff but could not summarize it.".to_string(), ), } } }; if let Err(e) = windmill_common::git_sync_ee::update_check_run( db, workspace_id, &check.repo_url, check.check_run_id, conclusion, &title, &summary, ) .await { tracing::error!("git sync-check: failed to update check run: {e:#}"); } // Phase 4 also maintains ONE managed comment on the PR (Cloudflare // deploy-preview style): upserted on every synchronize, so reviewers see the // current diff without opening the Checks tab. if !is_deploy { if let Some(pr_number) = check.pr_number { let marker = ""; let head = check .head_sha .as_deref() .map(|s| &s[..s.len().min(7)]) .unwrap_or("latest"); let body = format!( "{marker}\n### Windmill deploy preview\n\n| | |\n|---|---|\n| **Workspace** | `{workspace_id}` |\n| **Status** | {title} |\n| **Commit** | `{head}` |\n\n
Details\n\n{summary}\n\n
" ); if let Err(e) = windmill_common::git_sync_ee::upsert_pr_comment( db, workspace_id, &check.repo_url, pr_number, marker, &body, ) .await { tracing::warn!("git sync-check: failed to upsert PR diff comment: {e:#}"); } } } } pub async fn process_completed_job( JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, duration, result_columns, preprocessed_args, from_cache, flow_runners, done_tx, .. }: JobCompleted, client: &AuthedClient, db: &DB, worker_dir: &str, same_worker_tx: Option<&SameWorkerSender>, worker_name: &str, job_completed_tx: JobCompletedSender, killpill_rx: &tokio::sync::broadcast::Receiver<()>, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> error::Result>> { if success { // println!("bef completed job{:?}", SystemTime::now()); if let Some(cached_path) = cached_res_path { save_in_cache(db, client, &job, cached_path, result.clone()).await; } let is_flow_step = job.is_flow_step(); let parent_job = job.parent_job.clone(); let job_id = job.id.clone(); let workspace_id = job.workspace_id.clone(); let started_at = job.started_at.clone(); if job.flow_step_id.as_deref() == Some("preprocessor") { // Do this before inserting to `v2_job_completed` for backwards compatibility // when we set `flow_status->_metadata->preprocessed_args` to true. sqlx::query!( r#"UPDATE v2_job SET args = '{"reason":"PREPROCESSOR_ARGS_ARE_DISCARDED"}'::jsonb, preprocessed = TRUE WHERE id = $1 AND preprocessed = FALSE"#, job.id ) .execute(db) .await .map_err(|e| { Error::InternalErr(format!( "error while deleting args of preprocessing step: {e:#}" )) })?; } else if let Some(preprocessed_args) = preprocessed_args { // Update script args to preprocessed args, but preserve a // resolved pipeline `partition` (injected before the body ran // by resolve_partition_for_job). Run identity is immutable — // the preprocessor must not change or drop it, or the asset // cascade would read no partition for this producer. windmill_common::partition::merge_args_preserving_partition( db, job.id, preprocessed_args, ) .await?; } add_time!(bench, "pre add_completed_job"); let (_, duration, wac_job_ids) = add_completed_job( db, &job, true, false, Json(&result), result_columns, mem_peak.to_owned(), canceled_by.clone(), false, duration, from_cache.unwrap_or(false), ) .await?; #[cfg(all(feature = "enterprise", feature = "private"))] if job.kind == JobKind::DeploymentCallback { maybe_post_git_sync_check(db, &job_id, &workspace_id, true, result.get()).await; maybe_open_git_sync_deploy_pr(db, &job_id, &workspace_id, result.get()).await; } // Asset-trigger fan-out: best-effort, never propagates errors. // Internal eligibility checks gate to top-level Script/Preview runs; // see windmill_queue::asset_dispatch. asset_dispatch::dispatch_asset_triggers(db, &job).await; drop(job); add_time!(bench, "add_completed_job END"); if is_flow_step { if let Some(parent_job) = parent_job { // tracing::info!(parent_flow = %parent_job, subflow = %job_id, "updating flow status (2)"); let r = update_flow_status_after_job_completion( db, client, parent_job, &job_id, &workspace_id, true, canceled_by, result, started_at.map(|x| FlowJobDuration { started_at: x, duration_ms: duration }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, None, worker_name, job_completed_tx, flow_runners, killpill_rx, #[cfg(feature = "benchmark")] bench, ) .warn_after_seconds(10) .await?; add_time!(bench, "updated flow status END"); if let Some(done_tx) = done_tx { done_tx .send(()) .expect("done receiver should still be alive"); } return Ok(r); } } else if let Some(parent_job) = parent_job { // wac_job_ids is piggybacked from the duration write in // add_completed_job — no extra query needed. if let Some(job_ids) = wac_job_ids { if let Ok(Some(_)) = handle_wac_child_completion( db, &job_id, parent_job, &workspace_id, result, true, job_ids, ) .await { if let Some(done_tx) = done_tx { done_tx .send(()) .expect("done receiver should still be alive"); } return Ok(None); } } } } else { // 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()) }; #[cfg(all(feature = "enterprise", feature = "private"))] if job.kind == JobKind::DeploymentCallback { maybe_post_git_sync_check(db, &job.id, &job.workspace_id, false, result.get()).await; maybe_reconcile_git_sync_auto_pull(db, &job.id, &job.workspace_id, false).await; } 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"); let r = update_flow_status_after_job_completion( db, client, parent_job, &job.id, &job.workspace_id, false, canceled_by, downstream_result, duration.and_then(|d| { job.started_at.map(|started_at| FlowJobDuration { started_at: started_at, duration_ms: d, }) }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, None, worker_name, job_completed_tx, flow_runners, killpill_rx, #[cfg(feature = "benchmark")] bench, ) .warn_after_seconds(10) .await?; if let Some(done_tx) = done_tx { done_tx .send(()) .expect("done receiver should still be alive"); } return Ok(r); } } else if let Some(parent_job) = job.parent_job { // WAC child failed — query job_ids from parent (errors are rare, // so the extra read is acceptable here). let job_ids_json: Option> = sqlx::query_scalar( "SELECT workflow_as_code_status->'_checkpoint'->'pending_steps'->'job_ids' \ FROM v2_job_status WHERE id = $1", ) .bind(&parent_job) .fetch_optional(db) .await?; if let Some(Some(job_ids)) = job_ids_json { if let Ok(Some(_)) = handle_wac_child_completion( db, &job.id, parent_job, &job.workspace_id, downstream_result, false, job_ids, ) .await { if let Some(done_tx) = done_tx { done_tx .send(()) .expect("done receiver should still be alive"); } return Ok(None); } } } } return Ok(None); } /// Handle a WAC v2 child job completion. /// Returns Ok(Some(())) if the parent was a WAC job and was handled, /// Ok(None) if the parent is not a WAC job (caller should fall through). /// /// CONCURRENCY: Multiple parallel children may complete simultaneously on /// different workers. We use atomic SQL operations throughout: /// - `completed_steps` is merged via `jsonb_set(... || jsonb_build_object(...))` /// — PostgreSQL serialises concurrent UPDATEs on the same row, so each /// worker sees the previous worker's writes. /// - The suspend counter (set to N at dispatch time) is decremented atomically /// with `RETURNING` to determine the "all done" condition. pub(crate) async fn handle_wac_child_completion( db: &DB, child_job_id: &Uuid, parent_job_id: Uuid, workspace_id: &str, result: Arc>, success: bool, job_ids_value: Value, ) -> error::Result> { let job_ids = match job_ids_value { Value::Object(m) => m, _ => return Ok(None), // Not a WAC parent or no pending steps }; let child_id_str = child_job_id.to_string(); let step_key = job_ids.iter().find_map(|(key, val)| { if val.as_str() == Some(&child_id_str) { Some(key.clone()) } else { None } }); let step_key = match step_key { Some(k) => k, None => { if !success { // No step key and failed — can't store error, fail parent immediately tracing::error!( parent_job = %parent_job_id, child_job = %child_job_id, "WAC v2 child job failed but no step key found, failing parent" ); sqlx::query!( "UPDATE v2_job_queue SET suspend = 0, suspend_until = NULL WHERE id = $1", parent_job_id, ) .execute(db) .await?; let parent_mini = get_mini_completed_job(&parent_job_id, workspace_id, db).await?; if let Some(parent_mini) = parent_mini { let child_err: Value = serde_json::from_str(result.get()).unwrap_or(Value::Null); let err_value = json!({ "message": format!("WAC child job {} failed (no step key)", child_job_id), "error": child_err, }); let _ = windmill_queue::add_completed_job_error( db, &parent_mini, 0, None, err_value, "wac_child_handler", false, None, ) .await; } return Ok(Some(())); } tracing::warn!( parent_job = %parent_job_id, child_job = %child_job_id, "WAC v2 child completed but no matching step key found in checkpoint, decrementing suspend to avoid parent hang" ); // Still decrement suspend so the parent doesn't hang indefinitely let _ = sqlx::query_scalar!( "UPDATE v2_job_queue \ SET suspend = GREATEST(suspend - 1, 0) \ WHERE id = $1 \ RETURNING suspend", parent_job_id, ) .fetch_optional(db) .await?; return Ok(Some(())); } }; // Build result — wrap errors with _error marker so workflow try/catch can handle them let result_value: Value = if success { serde_json::from_str(result.get()).unwrap_or(Value::Null) } else { let child_err: Value = serde_json::from_str(result.get()).unwrap_or(Value::Null); tracing::info!( parent_job = %parent_job_id, child_job = %child_job_id, step_key = %step_key, "WAC v2 child job failed, storing error for workflow try/catch" ); windmill_common::wac::wac_failure_record( &step_key, Some(&child_job_id.to_string()), &child_err, ) }; tracing::info!( parent_job = %parent_job_id, child_job = %child_job_id, step_key = %step_key, success = success, "WAC v2 child job completed" ); // Use a transaction to ensure completed_steps merge + suspend decrement // are atomic. Without this, a crash between the two could strand the parent. let result_json = serde_json::to_value(&result_value) .map_err(|e| error::Error::InternalErr(format!("Failed to serialize step result: {e}")))?; let mut tx = db.begin().await?; // Merge the completed step into the checkpoint. // Uses `|| jsonb_build_object(key, value)` so concurrent children on // different workers don't overwrite each other — PostgreSQL serialises // concurrent UPDATEs on the same row and each sees the previous write. sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( workflow_as_code_status, '{_checkpoint,completed_steps}', COALESCE(workflow_as_code_status->'_checkpoint'->'completed_steps', '{}'::jsonb) || jsonb_build_object($2::text, $3::jsonb) ) WHERE id = $1", ) .bind(&parent_job_id) .bind(&step_key) .bind(&result_json) .execute(&mut *tx) .await .map_err(|e| error::Error::InternalErr(format!("Failed to add WAC completed step: {e}")))?; // Decrement the suspend counter. The counter was set to N (number of // children) at dispatch time. When it reaches 0 all children are done. // Keep suspend_until non-null so the suspended pull query // (`WHERE suspend_until IS NOT NULL AND suspend <= 0`) picks up the parent. let new_suspend: Option = sqlx::query_scalar!( "UPDATE v2_job_queue \ SET suspend = GREATEST(suspend - 1, 0) \ WHERE id = $1 \ RETURNING suspend", parent_job_id, ) .fetch_optional(&mut *tx) .await?; let all_done = new_suspend == Some(0); if all_done { // Clear pending_steps from checkpoint since all children are complete. // This is cosmetic — the next replay will overwrite it anyway — but // keeps the checkpoint clean for frontend display. let _ = sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = \ workflow_as_code_status #- '{_checkpoint,pending_steps}' \ WHERE id = $1", ) .bind(&parent_job_id) .execute(&mut *tx) .await; } tx.commit().await?; if all_done { tracing::info!( parent_job = %parent_job_id, "WAC v2 all child jobs completed, unsuspending parent" ); WAC_SUSPEND_READY.store(true, Ordering::Relaxed); } Ok(Some(())) } pub async fn handle_non_flow_job_error( db: &DB, job: &MiniCompletedJob, mem_peak: i32, canceled_by: Option, err_string: String, err_json: Value, worker_name: &str, ) -> Result { append_logs( &job.id, &job.workspace_id, format!("Unexpected error during job execution:\n{err_string}"), &db.into(), ) .await; add_completed_job_error( db, job, mem_peak, canceled_by, err_json, worker_name, false, None, ) .await } #[tracing::instrument(name = "job_error", level = "info", skip_all, fields(job_id = %job.id))] pub async fn handle_job_error( db: &DB, client: &AuthedClient, job: &MiniCompletedJob, mem_peak: i32, canceled_by: Option, err: Error, unrecoverable: bool, same_worker_tx: Option<&SameWorkerSender>, worker_dir: &str, worker_name: &str, job_completed_tx: JobCompletedSender, killpill_rx: &tokio::sync::broadcast::Receiver<()>, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) { let err_string = format!("{}: {}", err.name(), err.to_string()); let err_json = error_to_value(&err); let update_job_future = || async { handle_non_flow_job_error( db, job, mem_peak, canceled_by.clone(), err_string, err_json.clone(), worker_name, ) .warn_after_seconds(10) .await }; let update_job_future = if job.is_flow_step() || job.is_flow() { let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job { if let Err(e) = update_job_future().await { tracing::error!( "error updating job future for job {} for handle_job_error: {e:#}", job.id ); } (parent_job_id, job.id) } else { (job.id, Uuid::nil()) }; let wrapped_error = WrappedError { error: err_json.clone() }; tracing::error!(parent_flow = %flow, subflow = %job_status_to_update, "handle job error, updating flow status: {err_json:?}"); let updated_flow = update_flow_status_after_job_completion( db, client, flow, &job_status_to_update, &job.workspace_id, false, canceled_by.clone(), Arc::new(serde_json::value::to_raw_value(&wrapped_error).unwrap()), None, unrecoverable, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).clone(), worker_dir, None, worker_name, job_completed_tx.clone(), None, killpill_rx, #[cfg(feature = "benchmark")] bench, ) .await; if let Err(err) = updated_flow { if let Some(parent_job_id) = job.parent_job { if let Ok(Some(parent_job)) = get_mini_completed_job(&parent_job_id, &job.workspace_id, db) .warn_after_seconds(10) .await { let e = json!({"message": err.to_string(), "name": "InternalErr"}); append_logs( &parent_job.id, &job.workspace_id, format!("Unexpected error during flow job error handling:\n{err}"), &db.into(), ) .await; let _ = add_completed_job_error( db, &parent_job, mem_peak, canceled_by.clone(), e, worker_name, false, None, ) .warn_after_seconds(10) .await; } } } None } else { Some(update_job_future) }; if let Some(f) = update_job_future { let _ = f().await; } } #[derive(Debug, Serialize)] pub struct SerializedError { pub message: String, pub name: String, #[serde(skip_serializing_if = "Option::is_none")] pub step_id: Option, #[serde(skip_serializing_if = "Option::is_none")] pub exit_code: Option, } pub fn extract_error_value( program: &str, log_lines: &str, i: i32, step_id: Option, ) -> Box { return to_raw_value(&SerializedError { message: format!( "exit code for \"{program}\": {i}, last log lines:\n{}", ANSI_ESCAPE_RE.replace_all(log_lines.trim(), "").to_string() ), name: "ExecutionErr".to_string(), step_id, exit_code: Some(i), }); } #[cfg(all(test, feature = "enterprise", feature = "private"))] mod git_sync_pr_tests { use super::{git_sync_deploy_pr_head_branch, git_sync_push_result_pushed}; #[test] fn user_and_group_items_get_no_promotion_branch() { for path_type in ["user", "group"] { assert_eq!( git_sync_deploy_pr_head_branch( "ws", None, None, "main", true, false, "u/someone", "", path_type ), None ); } } #[test] fn scope_note_lists_filters() { use super::format_git_sync_scope_note; assert_eq!(format_git_sync_scope_note(&[], &[]), None); assert_eq!( format_git_sync_scope_note(&["f/**".into()], &[]).unwrap(), "\n\nOnly files matching this repository's sync filters deploy on merge: `f/**`." ); assert_eq!( format_git_sync_scope_note(&["f/**".into(), "u/**".into()], &["f/pat/**".into()]) .unwrap(), "\n\nOnly files matching this repository's sync filters deploy on merge: `f/**`, `u/**` (excluding `f/pat/**`)." ); } #[test] fn push_result_pushed_flag() { assert_eq!( git_sync_push_result_pushed(r#"{"pushed": true}"#), Some(true) ); assert_eq!( git_sync_push_result_pushed(r#"{"pushed": false}"#), Some(false) ); // Older hub script versions return null / no flag: undetermined. assert_eq!(git_sync_push_result_pushed("null"), None); assert_eq!(git_sync_push_result_pushed(r#"{"other": 1}"#), None); assert_eq!(git_sync_push_result_pushed("not json"), None); } #[test] fn fork_branch_wins_and_strips_the_id_prefix() { // Generated fork id: branch suffix drops the wm-fork- prefix. assert_eq!( git_sync_deploy_pr_head_branch( "wm-fork-abc", Some("prod"), None, "main", false, false, "", "", "" ), Some("wm-fork/main/abc".to_string()) ); // Dev workspace (prefix-less id, detected via parent): verbatim suffix. assert_eq!( git_sync_deploy_pr_head_branch( "staging", Some("prod"), None, "main", false, false, "", "", "" ), Some("wm-fork/main/staging".to_string()) ); // Orphaned fork (parent deleted): the id prefix still identifies it. assert_eq!( git_sync_deploy_pr_head_branch( "wm-fork-abc", None, None, "main", true, false, "f/x/y", "", "script" ), Some("wm-fork/main/abc".to_string()) ); } #[test] fn promotion_branch_matches_the_hub_script_formula() { // Per-item form: wm_deploy/// __>. assert_eq!( git_sync_deploy_pr_head_branch( "dev", None, None, "main", true, false, "f/folder/my_script", "", "script" ), Some("wm_deploy/dev/script/f__folder__my_script".to_string()) ); // Grouped-by-folder form: first two path segments joined by __. assert_eq!( git_sync_deploy_pr_head_branch( "dev", None, None, "main", true, true, "f/folder/my_script", "", "script" ), Some("wm_deploy/dev/f__folder".to_string()) ); // Renamed object: falls back to the parent path. assert_eq!( git_sync_deploy_pr_head_branch( "dev", None, None, "main", true, false, "", "f/folder/old", "script" ), Some("wm_deploy/dev/script/f__folder__old".to_string()) ); } #[test] fn no_branch_when_deploy_stays_on_base() { // Workspace-wide mode commits straight to the tracked branch. assert_eq!( git_sync_deploy_pr_head_branch( "dev", None, None, "main", false, false, "f/x/y", "", "script" ), None ); // Promotion mode but no per-item ref (e.g. user/group objects). assert_eq!( git_sync_deploy_pr_head_branch("dev", None, None, "main", true, false, "", "", "user"), None ); } #[test] fn dev_workspace_label_branch_wins() { // Dev workspaces deploy to their environment-label branch verbatim. assert_eq!( git_sync_deploy_pr_head_branch( "staging-ws", Some("prod"), Some("staging"), "main", false, false, "", "", "" ), Some("staging".to_string()) ); // Label present even on a wm-fork-prefixed id: label still wins. assert_eq!( git_sync_deploy_pr_head_branch( "wm-fork-x", Some("prod"), Some("dev"), "main", false, false, "", "", "" ), Some("dev".to_string()) ); } #[test] fn dev_workspace_promotion_uses_wm_deploy_branch() { // Promotion on: a dev workspace gets per-item wm_deploy/** branches // (namespaced by its own id), not its env-label branch. assert_eq!( git_sync_deploy_pr_head_branch( "dev", Some("prod"), Some("dev"), "main", true, false, "f/folder/my_script", "", "script" ), Some("wm_deploy/dev/script/f__folder__my_script".to_string()) ); // Per-folder form still honored for a promotion dev workspace. assert_eq!( git_sync_deploy_pr_head_branch( "dev", Some("prod"), Some("dev"), "main", true, true, "f/folder/my_script", "", "script" ), Some("wm_deploy/dev/f__folder".to_string()) ); // Promotion off: the env-label branch still wins. assert_eq!( git_sync_deploy_pr_head_branch( "dev", Some("prod"), Some("dev"), "main", false, false, "f/x/y", "", "script" ), Some("dev".to_string()) ); } #[test] fn dev_promotion_user_group_items_open_no_pr() { // User/group objects get no wm_deploy branch even on a dev workspace; the // CLI isolates them to the env-label branch, so the backend opens no PR // (never a PR from the env-label branch into the parent for these). for path_type in ["user", "group"] { assert_eq!( git_sync_deploy_pr_head_branch( "dev", Some("prod"), Some("dev"), "main", true, false, "u/alice", "", path_type ), None ); } } }