fix: honor custom timeout in setup phases and name the limit hit (#11557)

* fix: honor custom timeout in setup phases and name the limit hit

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* fix: label a shared-deadline phase timeout as the job's remaining budget

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* fix: register dbt deploy jobs so their phase timeout is not called custom

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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<HashMap<Uuid, Option<i32>>> =
+        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<i32>) -> 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<Option<i32>> {
+    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<i32>,
-) -> (Duration, Option<String>, bool) {
+) -> (Duration, Option<String>, TimeoutSource) {
     let mut warn_msg: Option<String> = 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<Duration>,
+    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<u64>| {
+            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,

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(());

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<CanceledBy>),
         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

backend/windmill-worker/src/worker.rs

@@ -5401,6 +5401,7 @@ async fn handle_code_execution_job(
     precomputed_agent_info: Option<PrecomputedAgentInfo>,
     has_stream: &mut bool,
 ) -> error::Result<Box<RawValue>> {
+    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"))