diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index cd8923d408..82065e8a37 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1068,12 +1068,87 @@ pub async fn start_child_process( Ok(child) } +/// Which limit a resolved job timeout comes from. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TimeoutSource { + /// The timeout set on the script or the flow step. + Custom, + /// What is left of a wall clock that the phases of one job spend down together. + RemainingBudget, + /// The instance `job_default_timeout` setting. + InstanceDefault, + /// The instance ceiling (`TIMEOUT`), which no other timeout can exceed. + InstanceMax, +} + +impl TimeoutSource { + /// How to name a limit of `secs` seconds in a message shown to the job's author. + pub fn describe(&self, secs: u64) -> String { + match self { + TimeoutSource::Custom => { + format!("the {secs}s custom timeout set on the script or flow step") + } + TimeoutSource::RemainingBudget => format!( + "the {secs}s left of the job's timeout, which its phases share" + ), + TimeoutSource::InstanceDefault => format!( + "the {secs}s instance default timeout (instance setting 'Default timeout'); set a longer custom timeout on the script or flow step to raise it" + ), + TimeoutSource::InstanceMax => format!( + "the {secs}s instance maximum job duration (TIMEOUT env variable of the worker), which no custom or default timeout can exceed" + ), + } + } +} + +lazy_static! { + static ref RUNNING_JOB_CUSTOM_TIMEOUTS: std::sync::Mutex>> = + std::sync::Mutex::new(HashMap::new()); +} + +/// Makes a running job's custom timeout visible to the phases that are not handed it (dependency +/// resolution, install, build), for as long as the guard lives. +pub struct RunningJobCustomTimeout(Uuid); + +impl RunningJobCustomTimeout { + pub fn register(job_id: Uuid, custom_timeout_secs: Option) -> Self { + if let Ok(mut m) = RUNNING_JOB_CUSTOM_TIMEOUTS.lock() { + m.insert( + job_id, + windmill_common::runnable_settings::none_if_non_positive(custom_timeout_secs), + ); + } + Self(job_id) + } +} + +impl Drop for RunningJobCustomTimeout { + fn drop(&mut self) { + if let Ok(mut m) = RUNNING_JOB_CUSTOM_TIMEOUTS.lock() { + m.remove(&self.0); + } + } +} + +/// `None` when the job is not registered, `Some(None)` when it runs without a custom timeout. +fn registered_custom_timeout(job_id: Uuid) -> Option> { + RUNNING_JOB_CUSTOM_TIMEOUTS + .lock() + .ok() + .and_then(|m| m.get(&job_id).copied()) +} + +/// Resolves the time limit of one phase of a job and where that limit comes from. +/// +/// A phase called without a custom timeout runs under the instance default. A job's custom +/// timeout may only lengthen such a phase, never shorten it: a short run timeout must not kill +/// a dependency install that the instance default allows. pub async fn resolve_job_timeout( _conn: &Connection, _w_id: &str, - _job_id: Uuid, + job_id: Uuid, custom_timeout_secs: Option, -) -> (Duration, Option, bool) { +) -> (Duration, Option, TimeoutSource) { let mut warn_msg: Option = None; #[cfg(feature = "cloud")] let cloud_premium_workspace = @@ -1091,6 +1166,7 @@ pub async fn resolve_job_timeout( let global_max_timeout_duration = Duration::from_secs(max_job_duration_secs(cloud_premium_workspace)); + let max_secs = global_max_timeout_duration.as_secs(); // A `custom_timeout_secs <= 0` is not a 0-second limit but "unset": fall through to the // default/global-max timeout instead of killing the job immediately. @@ -1098,34 +1174,95 @@ pub async fn resolve_job_timeout( Some(timeout_secs) if Duration::from_secs(timeout_secs as u64) < global_max_timeout_duration => { - (Duration::from_secs(timeout_secs as u64), warn_msg, true) + // A phase handed anything but the job's own custom timeout was handed the rest of a + // shared wall clock, whatever limit that clock was started from. + let source = + if registered_custom_timeout(job_id).is_none_or(|t| t == Some(timeout_secs)) { + TimeoutSource::Custom + } else { + TimeoutSource::RemainingBudget + }; + (Duration::from_secs(timeout_secs as u64), warn_msg, source) } Some(timeout_secs) => { - warn_msg = Some(format!("WARNING: Custom job timeout of {timeout_secs} seconds was greater than the maximum timeout. It will be ignored and the max timeout will be used instead")); + warn_msg = Some(format!("WARNING: Custom job timeout of {timeout_secs} seconds was greater than the instance maximum job duration of {max_secs} seconds (TIMEOUT env variable of the worker). It will be ignored and the max timeout will be used instead")); tracing::warn!(warn_msg); - (global_max_timeout_duration, warn_msg, false) + ( + global_max_timeout_duration, + warn_msg, + TimeoutSource::InstanceMax, + ) } None => { // fallback to default timeout or max if not set - let default_timeout = match JOB_DEFAULT_TIMEOUT.read().await.clone() { - None => global_max_timeout_duration, + let (default_timeout, source) = match JOB_DEFAULT_TIMEOUT.read().await.clone() { + None => (global_max_timeout_duration, TimeoutSource::InstanceMax), Some(default_timeout_secs) if Duration::from_secs(default_timeout_secs as u64) < global_max_timeout_duration => { - Duration::from_secs(default_timeout_secs as u64) + ( + Duration::from_secs(default_timeout_secs as u64), + TimeoutSource::InstanceDefault, + ) } Some(default_timeout_secs) => { - warn_msg = Some(format!("WARNING: Default job timeout of {default_timeout_secs} seconds was greater than the maximum timeout. It will be ignored and the global max timeout will be used instead")); + warn_msg = Some(format!("WARNING: Default job timeout of {default_timeout_secs} seconds was greater than the instance maximum job duration of {max_secs} seconds (TIMEOUT env variable of the worker). It will be ignored and the max timeout will be used instead")); tracing::warn!(warn_msg); - global_max_timeout_duration + (global_max_timeout_duration, TimeoutSource::InstanceMax) } }; - (default_timeout, warn_msg, false) + let job_custom_timeout = registered_custom_timeout(job_id) + .flatten() + .map(|secs| Duration::from_secs(secs as u64)); + let (timeout, source) = lengthen_with_job_custom_timeout( + default_timeout, + source, + job_custom_timeout, + global_max_timeout_duration, + ); + (timeout, warn_msg, source) } } } +fn lengthen_with_job_custom_timeout( + phase_timeout: Duration, + phase_source: TimeoutSource, + job_custom_timeout: Option, + max_timeout: Duration, +) -> (Duration, TimeoutSource) { + match job_custom_timeout { + Some(custom) if custom > phase_timeout && custom < max_timeout => { + (custom, TimeoutSource::Custom) + } + Some(custom) if custom > phase_timeout => (max_timeout, TimeoutSource::InstanceMax), + _ => (phase_timeout, phase_source), + } +} + +#[cfg(test)] +mod job_timeout_tests { + use super::{lengthen_with_job_custom_timeout, TimeoutSource}; + use std::time::Duration; + + // A phase that is not handed the job's custom timeout (dependency install, build) runs under + // the instance default. The custom timeout may extend it up to the instance max, but a + // shorter one must leave it alone. + #[test] + fn job_custom_timeout_only_lengthens_a_setup_phase() { + let s = Duration::from_secs; + let default = (s(120), TimeoutSource::InstanceDefault); + let resolve = |custom: Option| { + lengthen_with_job_custom_timeout(default.0, default.1, custom.map(s), s(600)) + }; + assert_eq!(resolve(None), default); + assert_eq!(resolve(Some(240)), (s(240), TimeoutSource::Custom)); + assert_eq!(resolve(Some(30)), default); + assert_eq!(resolve(Some(900)), (s(600), TimeoutSource::InstanceMax)); + } +} + /// Compute the nsjail timeout (in seconds) with a 15s buffer so handle_child fires first. pub async fn resolve_nsjail_timeout( conn: &Connection, diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 8119897d6d..90db1b9795 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -779,6 +779,7 @@ pub(crate) async fn dbt_dep( // A dependency job carries no per-job timeout of its own, but its phases // still share one wall clock rather than each getting the instance-wide // one. + let _custom_timeout = crate::common::RunningJobCustomTimeout::register(*job_id, None); let deadline = JobDeadline::start(&conn, w_id, *job_id, None).await; let mut ctx = JobCtx { mem_peak, canceled_by, occupancy_metrics, worker_name, deadline }; let prepared = prepare_project( @@ -3560,11 +3561,10 @@ async fn attach_column_index( .iter() .map(|n| n.unique_id.as_str()) .collect(); - let index = - crate::dbt_column_index::collect( - p, descriptor, inv, command, ctx, job_id, w_id, conn, &kept, - ) - .await?; + let index = crate::dbt_column_index::collect( + p, descriptor, inv, command, ctx, job_id, w_id, conn, &kept, + ) + .await?; drop(kept); let Some(index) = index else { return Ok(()); diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 3c9be1f4c5..22a2bf4e8b 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -49,7 +49,7 @@ use futures::{ stream, StreamExt, }; -use crate::common::{resolve_job_timeout, OccupancyMetrics, StreamNotifier}; +use crate::common::{resolve_job_timeout, OccupancyMetrics, StreamNotifier, TimeoutSource}; use crate::job_logger::{append_job_logs, append_result_stream, append_with_limit, strip_nul}; use crate::job_logger_oss::process_streaming_log_lines; use crate::worker_utils::{ping_job_status, update_worker_ping_from_job}; @@ -181,7 +181,7 @@ pub async fn handle_child( enum KillReason { TooManyLogs, - Timeout { is_job_specific: bool }, + Timeout { phase: String, source: TimeoutSource, secs: u64 }, Cancelled(Option), AlreadyCompleted, } @@ -190,11 +190,9 @@ pub async fn handle_child( fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { match self { KillReason::TooManyLogs => f.write_str("too many logs (max size: 2MB)"), - KillReason::Timeout { is_job_specific } => f.write_str(if *is_job_specific { - "timeout after exceeding job-specific duration limit" - } else { - "timeout after exceeding instance-wide job duration limit" - }), + KillReason::Timeout { phase, source, secs } => { + write!(f, "timeout: `{phase}` exceeded {}", source.describe(*secs)) + } KillReason::Cancelled(canceled_by) => { let mut reason = "cancelled".to_string(); if let Some(canceled_by) = canceled_by { @@ -212,7 +210,7 @@ pub async fn handle_child( } } - let (timeout_duration, timeout_warn_msg, is_job_specific) = + let (timeout_duration, timeout_warn_msg, timeout_source) = resolve_job_timeout(&conn, w_id, job_id, custom_timeout).await; if let Some(msg) = timeout_warn_msg { append_logs(&job_id, w_id, msg.as_str(), conn).await; @@ -224,7 +222,11 @@ pub async fn handle_child( biased; result = Box::into_pin(child.wait()) => return result.map(Ok), Ok(()) = too_many_logs.changed() => KillReason::TooManyLogs, - _ = sleep(timeout_duration) => KillReason::Timeout { is_job_specific }, + _ = sleep(timeout_duration) => KillReason::Timeout { + phase: child_name.to_string(), + source: timeout_source, + secs: timeout_duration.as_secs(), + }, ex = update_job, if job_id != Uuid::nil() => match ex { UpdateJobPollingExit::Done(canceled_by) => KillReason::Cancelled(canceled_by), UpdateJobPollingExit::AlreadyCompleted => KillReason::AlreadyCompleted, @@ -233,17 +235,16 @@ pub async fn handle_child( tx.send(()).expect("rx should never be dropped"); drop(tx); + let timeout_reason = format!( + "`{child_name}` exceeded {}", + timeout_source.describe(timeout_duration.as_secs()) + ); let set_reason = async { if matches!(kill_reason, KillReason::Timeout { .. }) { match conn { Connection::Sql(db) => { - if let Err(err) = set_job_cancelled_query( - job_id, - db, - "timeout", - &format!("duration > {}", timeout_duration.as_secs()), - ) - .await + if let Err(err) = + set_job_cancelled_query(job_id, db, "timeout", &timeout_reason).await { tracing::error!(%job_id, %err, "error setting cancelation reason for job {job_id}: {err}"); } @@ -255,7 +256,7 @@ pub async fn handle_child( None, &JobCancelled { canceled_by: "timeout".to_string(), - reason: format!("duration > {}", timeout_duration.as_secs()), + reason: timeout_reason.clone(), }, ) .await diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b88d3341d3..80ca37acb8 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -5401,6 +5401,7 @@ async fn handle_code_execution_job( precomputed_agent_info: Option, has_stream: &mut bool, ) -> error::Result> { + let _custom_timeout = crate::common::RunningJobCustomTimeout::register(job.id, job.timeout); let script_hash = || { job.runnable_id .ok_or_else(|| Error::internal_err("expected script hash"))