fix(spurctld): report wall-time expiry as TIMEOUT

crates/spur-core/src/job.rs

@@ -245,6 +245,9 @@ pub enum PendingReason {
     ReqNodeNotAvail,
     BeginTime,
     DeadLine,
+    /// Slurm's `FAIL_TIMEOUT`: the run ended because it exhausted its wall-clock
+    /// time limit. Sibling of `DeadLine`, which fires before the job ever starts.
+    TimeLimit,
     Licenses,
     NonZeroExitCode,
     RaisedSignal,
@@ -331,6 +334,7 @@ impl PendingReason {
             Self::ReqNodeNotAvail => "ReqNodeNotAvail",
             Self::BeginTime => "BeginTime",
             Self::DeadLine => "DeadLine",
+            Self::TimeLimit => "TimeLimit",
             Self::Licenses => "Licenses",
             Self::NonZeroExitCode => "NonZeroExitCode",
             Self::RaisedSignal => "RaisedSignal",
@@ -718,6 +722,15 @@ pub struct Job {
     #[serde(default)]
     pub srun_step_dispatch: bool,
 
+    /// Wall-clock instant the controller signalled this run for exceeding its
+    /// time limit, cleared on requeue. Replicated rather than kept in the
+    /// watchdog's memory for two reasons: the completion path runs inside the
+    /// Raft apply and must reach the same verdict on every replica, and a
+    /// leadership change mid-grace-period would otherwise restart the grace
+    /// period from scratch.
+    #[serde(default)]
+    pub time_limit_signaled_at: Option<DateTime<Utc>>,
+
     /// Wall-clock instant the job entered Suspended (None unless currently suspended).
     #[serde(default)]
     pub suspended_at: Option<DateTime<Utc>>,
@@ -780,6 +793,7 @@ impl Job {
             het_job_id: None,
             het_group: None,
             node_completions: HashMap::new(),
+            time_limit_signaled_at: None,
             suspended_at: None,
             suspended_secs: 0,
             bb_stage_state: BbStageState::None,
@@ -829,6 +843,39 @@ impl Job {
         }
     }
 
+    /// Reconcile the per-node completion verdict with what the controller
+    /// already knows about the run, yielding its final `(state, reason)`.
+    ///
+    /// A run signalled for exceeding its time limit reports `Timeout` however
+    /// the process itself ended: the exit signal records how it was stopped,
+    /// not why. An OOM kill outranks even that, being direct kernel evidence of
+    /// a distinct failure the user has to act on.
+    pub fn completion_verdict(
+        &self,
+        derived_state: JobState,
+        exit_code: i32,
+        signal: i32,
+        oom: bool,
+    ) -> (JobState, PendingReason) {
+        let state = if oom {
+            JobState::OutOfMemory
+        } else if self.time_limit_signaled_at.is_some() {
+            JobState::Timeout
+        } else {
+            derived_state
+        };
+
+        let reason = match state {
+            JobState::OutOfMemory => PendingReason::OutOfMemory,
+            JobState::Timeout => PendingReason::TimeLimit,
+            _ if signal != 0 => PendingReason::RaisedSignal,
+            _ if exit_code != 0 => PendingReason::NonZeroExitCode,
+            _ => PendingReason::None,
+        };
+
+        (state, reason)
+    }
+
     pub fn all_nodes_completed(&self) -> bool {
         !self.allocated_nodes.is_empty()
             && self.node_completions.len() == self.allocated_nodes.len()
@@ -1096,6 +1143,10 @@ impl Job {
             (JobState::Completing, JobState::Completed) => true,
             (JobState::Completing, JobState::Failed) => true,
             (JobState::Completing, JobState::Cancelled) => true,
+            // A job signalled for exceeding its time limit routes through
+            // Completing like any other, so the final verdict lands from there
+            // (Slurm's JOB_TIMEOUT | JOB_COMPLETING).
+            (JobState::Completing, JobState::Timeout) => true,
             (JobState::Completing, JobState::NodeFail) => true,
             (JobState::Completing, JobState::OutOfMemory) => true,
             (JobState::Suspended, JobState::Running) => true,
@@ -1519,6 +1570,68 @@ mod tests {
         assert_eq!(signal, 11);
     }
 
+    /// A job whose run the watchdog signalled for exhausting its time limit.
+    fn timed_out_job() -> Job {
+        let mut job = make_job();
+        job.time_limit_signaled_at = Some(Utc::now());
+        job
+    }
+
+    #[test]
+    fn completion_verdict_reports_timeout_for_a_job_killed_by_its_time_limit() {
+        // The regression: SIGTERM from the watchdog looks exactly like any other
+        // signal death to derived_completion, which reports Failed.
+        let (state, reason) = timed_out_job().completion_verdict(JobState::Failed, 0, 15, false);
+        assert_eq!(state, JobState::Timeout);
+        assert_eq!(reason, PendingReason::TimeLimit);
+    }
+
+    #[test]
+    fn completion_verdict_reports_timeout_when_the_job_exits_cleanly_on_sigterm() {
+        // A script that traps SIGTERM, checkpoints, and exits 0 still ran out of
+        // time; the run's outcome is not the handler's exit status.
+        let (state, reason) = timed_out_job().completion_verdict(JobState::Completed, 0, 0, false);
+        assert_eq!(state, JobState::Timeout);
+        assert_eq!(reason, PendingReason::TimeLimit);
+    }
+
+    #[test]
+    fn completion_verdict_leaves_an_unsignalled_death_alone() {
+        // Nothing to do with the time limit: a job killed by an external SIGKILL
+        // must keep reporting Failed / RaisedSignal.
+        let (state, reason) = make_job().completion_verdict(JobState::Failed, 0, 9, false);
+        assert_eq!(state, JobState::Failed);
+        assert_eq!(reason, PendingReason::RaisedSignal);
+
+        let (state, reason) = make_job().completion_verdict(JobState::Failed, 42, 0, false);
+        assert_eq!(state, JobState::Failed);
+        assert_eq!(reason, PendingReason::NonZeroExitCode);
+
+        let (state, reason) = make_job().completion_verdict(JobState::Completed, 0, 0, false);
+        assert_eq!(state, JobState::Completed);
+        assert_eq!(reason, PendingReason::None);
+    }
+
+    #[test]
+    fn completion_verdict_lets_an_oom_kill_outrank_the_time_limit() {
+        // Kernel evidence of a specific failure the user must act on beats the
+        // controller's own reason for signalling the job.
+        let (state, reason) = timed_out_job().completion_verdict(JobState::Failed, 0, 9, true);
+        assert_eq!(state, JobState::OutOfMemory);
+        assert_eq!(reason, PendingReason::OutOfMemory);
+    }
+
+    #[test]
+    fn a_timed_out_job_finalizes_from_completing() {
+        // The completion path routes every job through Completing, so without
+        // this transition a timed-out job could not reach its verdict.
+        let mut job = make_job();
+        job.transition(JobState::Running).unwrap();
+        job.transition(JobState::Completing).unwrap();
+        job.transition(JobState::Timeout).unwrap();
+        assert_eq!(job.state, JobState::Timeout);
+    }
+
     #[test]
     fn completion_state_for_exit_code_maps_expected_states() {
         assert_eq!(
@@ -1810,6 +1923,7 @@ mod tests {
         (PendingReason::BurstBufferResources, "BurstBufferResources"),
         (PendingReason::BurstBufferStageIn, "BurstBufferStageIn"),
         (PendingReason::JobHoldMaxRequeue, "JobHoldMaxRequeue"),
+        (PendingReason::TimeLimit, "TimeLimit"),
         (PendingReason::AssocMaxJobsLimit, "AssocMaxJobsLimit"),
         (
             PendingReason::AssocMaxSubmitJobLimit,

crates/spur-core/src/wal.rs

@@ -68,6 +68,15 @@ pub enum WalOperation {
         exit_code: i32,
         signal: i32,
     },
+    /// The time-limit watchdog signalled a running job for exhausting its wall
+    /// clock. Durable so the grace period survives a leadership change and so
+    /// every replica finalizes the run as `Timeout` rather than reading the
+    /// terminating signal as an ordinary failure. `at` is stamped on the leader
+    /// so replicas share one instant instead of consulting their own clocks.
+    JobTimeLimitSignaled {
+        job_id: JobId,
+        at: chrono::DateTime<chrono::Utc>,
+    },
     /// An srun job step finished. Records the step's exit code durably so the
     /// job's DerivedExitCode (running max over steps) survives restart/replay.
     JobStepComplete {

crates/spur-tests/src/t55_format.rs

@@ -127,6 +127,7 @@ mod tests {
             ),
             (PendingReason::ReservationDeleted, "ReservationDeleted"),
             (PendingReason::JobHoldMaxRequeue, "JobHoldMaxRequeue"),
+            (PendingReason::TimeLimit, "TimeLimit"),
             (PendingReason::QosMaxCpuPerJobLimit, "QOSMaxCpuPerJobLimit"),
             (
                 PendingReason::QosMaxWallDurationPerJobLimit,

crates/spurctld/src/cluster.rs

@@ -991,6 +991,16 @@ impl ClusterManager {
         Ok(())
     }
 
+    /// Record that a running job has exhausted its time limit, before the
+    /// caller sends SIGTERM. Durably marking the run first is what lets the
+    /// completion path report `Timeout` instead of reading the terminating
+    /// signal as an ordinary failure — a job that exits promptly on SIGTERM
+    /// reports back long before the grace period is up.
+    pub fn signal_time_limit(&self, job_id: JobId, at: DateTime<Utc>) -> anyhow::Result<()> {
+        self.propose(WalOperation::JobTimeLimitSignaled { job_id, at })?;
+        Ok(())
+    }
+
     /// Preempt a running job per its partition's PreemptMode. Does the
     /// controller-side state change; the caller dispatches the signal named by
     /// the returned `PreemptOutcome`. `Off` is rejected.
@@ -3187,6 +3197,7 @@ impl ClusterManager {
         job.allocated_nodes.clear();
         job.allocated_resources = None;
         job.per_node_alloc.clear();
+        job.time_limit_signaled_at = None;
         job.set_pending_reason(PendingReason::None);
         // Stale after requeue (points at nodes the job left); next dispatch resets it.
         job.actual_stdout_path = None;
@@ -3589,11 +3600,8 @@ impl ClusterManager {
                         let (derived_state, final_exit, raw_signal) =
                             Job::derived_completion(&job.node_completions, &primary);
                         let final_signal = raw_signal & !spur_core::job::OOM_SIGNAL_FLAG;
-                        let final_state = if oom {
-                            JobState::OutOfMemory
-                        } else {
-                            derived_state
-                        };
+                        let (final_state, final_reason) =
+                            job.completion_verdict(derived_state, final_exit, final_signal, oom);
                         match job.transition(final_state) {
                             Ok(()) => {
                                 job.exit_code = Some(final_exit);
@@ -3602,15 +3610,7 @@ impl ClusterManager {
                                 // steps, accumulated live by JobStepComplete; a
                                 // job with no srun steps keeps 0 (Slurm parity),
                                 // not the batch exit. Left as-is here.
-                                job.set_pending_reason(if oom {
-                                    PendingReason::OutOfMemory
-                                } else if final_signal != 0 {
-                                    PendingReason::RaisedSignal
-                                } else if final_exit != 0 {
-                                    PendingReason::NonZeroExitCode
-                                } else {
-                                    PendingReason::None
-                                });
+                                job.set_pending_reason(final_reason);
                                 job.end_time = Some(timestamp);
                                 job.node_completions.clear();
                                 Some((final_state, final_exit))
@@ -3644,6 +3644,15 @@ impl ClusterManager {
                     };
                 }
             }
+            WalOperation::JobTimeLimitSignaled { job_id, at } => {
+                if let Some(job) = jobs.get_mut(job_id) {
+                    // A run that already ended keeps the verdict it finalized
+                    // with: the watchdog raced the job's own exit and lost.
+                    if job.state.is_active() && job.time_limit_signaled_at.is_none() {
+                        job.time_limit_signaled_at = Some(*at);
+                    }
+                }
+            }
             WalOperation::JobComplete {
                 job_id,
                 exit_code,
@@ -3678,6 +3687,11 @@ impl ClusterManager {
                     }
                     job.exit_code = Some(*exit_code);
                     job.end_time = Some(timestamp);
+                    // Derived from the replicated entry, so every replica reports
+                    // the same reason for a job the watchdog had to force-kill.
+                    if *state == JobState::Timeout {
+                        job.set_pending_reason(PendingReason::TimeLimit);
+                    }
                     // Suspended -> terminal: fold the final suspended interval in
                     // and clear suspended_at so it never lingers on a terminal job.
                     if let Some(since) = job.suspended_at.take() {
@@ -6142,6 +6156,103 @@ mod tests {
         assert_eq!(job.pending_reason, PendingReason::RaisedSignal);
     }
 
+    // A job that exits promptly on the watchdog's SIGTERM used to report FAILED:
+    // its completion reached the controller well before the grace period was up,
+    // and nothing durable recorded why it had been signalled.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn node_complete_after_a_time_limit_signal_reports_timeout() {
+        let dir = TempDir::new().unwrap();
+        let cm = test_cluster(&dir).await;
+        register_node(&cm, "worker1", 8, 16000);
+
+        let job_id = run_job_on(&cm, "time-limit-job", "worker1");
+        cm.signal_time_limit(job_id, Utc::now()).unwrap();
+        wait_for("time limit expiry recorded", || {
+            cm.get_job(job_id)
+                .is_some_and(|j| j.time_limit_signaled_at.is_some())
+        });
+
+        // What spurd reports for a script that dies on SIGTERM.
+        cm.node_complete(job_id, "worker1", 0, 15, 0).unwrap();
+
+        let job = cm.get_job(job_id).unwrap();
+        assert_eq!(job.state, JobState::Timeout);
+        assert_eq!(job.pending_reason, PendingReason::TimeLimit);
+        // The terminating signal is still reported, so ExitCode stays 0:15.
+        assert_eq!(job.exit_code, Some(0));
+        assert_eq!(job.exit_signal, 15);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn a_time_limit_signal_after_the_run_ended_is_a_noop() {
+        let dir = TempDir::new().unwrap();
+        let cm = test_cluster(&dir).await;
+        register_node(&cm, "worker1", 8, 16000);
+
+        // The watchdog can lose the race: the job finished on its own just as
+        // the deadline passed, so its verdict is already final.
+        let job_id = run_job_on(&cm, "raced-job", "worker1");
+        cm.node_complete(job_id, "worker1", 0, 0, 0).unwrap();
+        settle(&cm, job_id, JobState::Completed);
+
+        cm.signal_time_limit(job_id, Utc::now()).unwrap();
+
+        let job = cm.get_job(job_id).unwrap();
+        assert_eq!(job.state, JobState::Completed);
+        assert_eq!(job.pending_reason, PendingReason::None);
+        assert!(job.time_limit_signaled_at.is_none());
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn force_kill_after_the_grace_period_reports_the_time_limit_reason() {
+        let dir = TempDir::new().unwrap();
+        let cm = test_cluster(&dir).await;
+        register_node(&cm, "worker1", 8, 16000);
+
+        // A job that outlives the grace period is finalized by the watchdog
+        // itself rather than by an agent report.
+        let job_id = run_job_on(&cm, "grace-expired", "worker1");
+        cm.signal_time_limit(job_id, Utc::now()).unwrap();
+        cm.complete_job(job_id, -1, JobState::Timeout).unwrap();
+        settle(&cm, job_id, JobState::Timeout);
+
+        let job = cm.get_job(job_id).unwrap();
+        assert_eq!(job.pending_reason, PendingReason::TimeLimit);
+        assert_eq!(job.exit_code, Some(-1));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn requeue_after_a_time_limit_kill_clears_the_marker() {
+        let dir = TempDir::new().unwrap();
+        let cm = test_cluster(&dir).await;
+        register_node(&cm, "worker1", 8, 16000);
+
+        let mut spec = basic_spec("requeue-after-timeout");
+        spec.requeue = true;
+        let job_id = submit_and_wait(&cm, spec);
+        let alloc = scalar_alloc(2, 4000);
+        cm.start_job(
+            job_id,
+            vec!["worker1".into()],
+            alloc.clone(),
+            per_node_for(&["worker1"], alloc),
+        )
+        .unwrap();
+        settle(&cm, job_id, JobState::Running);
+
+        cm.signal_time_limit(job_id, Utc::now()).unwrap();
+        cm.node_complete(job_id, "worker1", 0, 15, 0).unwrap();
+
+        // Attributing the kill to the time limit is what routes a well-behaved
+        // job into the requeue path at all.
+        settle(&cm, job_id, JobState::Pending);
+        let job = cm.get_job(job_id).unwrap();
+        assert_eq!(job.requeue_count, 1);
+        // A marker left behind would make the next run report TIMEOUT the
+        // moment it ended, whatever its outcome.
+        assert!(job.time_limit_signaled_at.is_none());
+    }
+
     // Reproduces the two steps report_job_status performs (validate the wire
     // report, then node_complete) since ControllerService can't be built here.
     // A signaled job's report (Completed, exit_code=0, signal=9) must be accepted