feat(spurd): survive agent restart without interrupting running jobs

spurd's SIGTERM handler unconditionally deregistered the node, force-evicting
every running job — so any graceful restart (e.g. a binary upgrade) failed
in-flight jobs even though the job process could survive it. This makes a
graceful restart transparent to running jobs, with the controller's heartbeat
timeout as the backstop if the restart takes too long.

- SIGTERM deregisters only when the agent has no active jobs (none running and
  none mid-launch); a reservation-only node still deregisters so the controller
  reclaims it.
- Per-job manifest split into identity (pid, start time, cgroup, exit sentinel)
  and an obligation ledger (epilog, completion report), each cleared as it is
  discharged. Reconcile drives the ledger to empty, persisting progress so a
  lost completion report retried on the next restart resends only the report,
  never re-running a non-idempotent epilog; the resolved exit is recorded so it
  survives the sentinel/rootfs being cleaned up.
- Completion detection without waitpid: tri-state cgroup liveness with a /proc
  fallback when cgroup.events is unreadable (non-v2 host, or a process never
  moved into the cgroup), so an adopted job is not misreported as complete.
- Resource restore is all-or-nothing, so a restarted agent refuses to adopt a
  job whose manifest names a GPU another adopted job already holds.
- Exit sentinel opened O_NONBLOCK|O_NOFOLLOW with a regular-file/nlink check and
  a bounded read, so a job-planted FIFO or oversized file can't block or bloat
  startup; the wrapper uses a top-level trap on EXIT (not a subshell) so a later
  syntax error still runs the lines above it.
- cgroups and spool dirs are keyed by (job_id, run_attempt); the spool dir stays
  root-owned with a sticky bit so a co-located user can't replace the manifest.
- spurctld warns at startup if heartbeat_timeout_secs is below 30s.

Author: yansun1996

crates/spur-sched/src/cons_tres.rs

@@ -66,6 +66,11 @@ impl NodeAllocation {
         }
     }
 
+    /// Whether any job is mid-launch (reserved, not yet committed to `running`).
+    pub fn has_launching(&self) -> bool {
+        !self.launching.is_empty()
+    }
+
     /// Available (unallocated) CPU count.
     pub fn free_cpus(&self) -> u32 {
         self.allocated_cpus.iter().filter(|&&a| !a).count() as u32
@@ -184,7 +189,7 @@ impl NodeAllocation {
             }
         }
 
-        self.allocated_memory_mb += memory_mb;
+        self.allocated_memory_mb = self.allocated_memory_mb.saturating_add(memory_mb);
         for &idx in &gpu_indices {
             self.gpu_allocated[idx] = true;
         }
@@ -208,6 +213,49 @@ impl NodeAllocation {
         self.owners.contains_key(&job_id)
     }
 
+    /// Re-adopt an already-committed allocation (e.g. after an agent restart),
+    /// pinning the exact ids rather than re-picking via first-fit. Idempotent:
+    /// a second call for the same job_id is a no-op, so a duplicate manifest
+    /// can't double-count memory (the cpu/gpu bitmaps are naturally
+    /// idempotent, but the memory counter isn't).
+    ///
+    /// GPUs are validated all-or-nothing before any mutation: a manifest naming
+    /// a device already held by another job is rejected with `GpusUnavailable`
+    /// so the caller can refuse the adoption rather than double-book the device
+    /// (releasing one owner would then free a GPU the other still holds).
+    pub fn restore_committed(
+        &mut self,
+        job_id: u32,
+        alloc: AllocationResult,
+    ) -> Result<(), AllocError> {
+        if self.owners.contains_key(&job_id) {
+            return Ok(());
+        }
+        let mut gpu_indices = Vec::with_capacity(alloc.gpu_ids.len());
+        for &device_id in &alloc.gpu_ids {
+            let idx = self
+                .gpus
+                .iter()
+                .position(|g| g.device_id == device_id)
+                .ok_or(AllocError::GpusUnavailable)?;
+            if self.gpu_allocated[idx] || gpu_indices.contains(&idx) {
+                return Err(AllocError::GpusUnavailable);
+            }
+            gpu_indices.push(idx);
+        }
+        for &cpu in &alloc.cpu_ids {
+            if let Some(a) = self.allocated_cpus.get_mut(cpu as usize) {
+                *a = true;
+            }
+        }
+        self.allocated_memory_mb = self.allocated_memory_mb.saturating_add(alloc.memory_mb);
+        for idx in gpu_indices {
+            self.gpu_allocated[idx] = true;
+        }
+        self.owners.insert(job_id, alloc);
+        Ok(())
+    }
+
     /// Release a job's allocation by id. Idempotent: releasing an unknown or
     /// already-released job is a no-op returning false.
     pub fn release_job(&mut self, job_id: u32) -> bool {
@@ -556,6 +604,130 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_restore_committed_pins_exact_ids_not_first_fit() {
+        let mut node = make_node_with_ids(4, 8_000, vec![0, 1], "mi300x");
+        // Simulate a spurd restart: nothing allocated yet in this fresh
+        // NodeAllocation, but a manifested job actually holds cpu 3 and gpu 1
+        // (not the first-fit picks allocate_for_job would make).
+        let alloc = AllocationResult {
+            cpu_ids: vec![3],
+            gpu_ids: vec![1],
+            memory_mb: 2_000,
+        };
+        node.restore_committed(42, alloc.clone()).unwrap();
+
+        assert_eq!(node.free_cpus(), 3);
+        assert_eq!(node.free_gpus(None), 1);
+        assert_eq!(node.free_memory_mb(), 6_000);
+        assert_eq!(node.allocated_gpu_ids(), vec![1]);
+
+        // A subsequent allocation must not double-book cpu 3 / gpu 1.
+        let next = node.allocate_for_job(43, 1, 0, &[0]).unwrap();
+        assert_eq!(next.cpu_ids, vec![0]);
+        assert!(!next.cpu_ids.contains(&3));
+
+        // Committed (not launching), so reconcile treats it like any other
+        // live job rather than sparing it as mid-launch.
+        let live: HashSet<u32> = [42, 43].into_iter().collect();
+        assert!(node
+            .reconcile(&live, Instant::now(), Duration::from_secs(120))
+            .is_empty());
+        assert!(node.release_job(42));
+        assert_eq!(node.free_memory_mb(), 8_000);
+    }
+
+    #[test]
+    fn test_has_launching_tracks_in_flight_launch() {
+        let mut node = make_node_with_ids(4, 8_000, vec![0], "mi300x");
+        assert!(!node.has_launching());
+        node.allocate_for_job(1, 1, 0, &[0]).unwrap();
+        assert!(
+            node.has_launching(),
+            "reserved-but-not-committed job is launching"
+        );
+        node.commit_job(1);
+        assert!(
+            !node.has_launching(),
+            "committed job is no longer launching"
+        );
+    }
+
+    #[test]
+    fn test_restore_committed_memory_saturates_on_overflow() {
+        // A corrupt/tampered manifest with a huge memory_mb must not wrap the
+        // accounting counter (which would silently corrupt free-memory math).
+        let mut node = make_node_with_ids(4, 8_000, vec![0], "mi300x");
+        node.restore_committed(
+            1,
+            AllocationResult {
+                cpu_ids: vec![],
+                gpu_ids: vec![],
+                memory_mb: u64::MAX,
+            },
+        )
+        .unwrap();
+        node.restore_committed(
+            2,
+            AllocationResult {
+                cpu_ids: vec![],
+                gpu_ids: vec![],
+                memory_mb: u64::MAX,
+            },
+        )
+        .unwrap();
+        // Saturated rather than wrapped; free memory floors at 0.
+        assert_eq!(node.free_memory_mb(), 0);
+    }
+
+    #[test]
+    fn test_restore_committed_rejects_gpu_double_book() {
+        // Two manifests naming the same device: the second adoption must be
+        // rejected, not silently double-book the GPU — otherwise releasing the
+        // first frees a device the second still holds.
+        let mut node = make_node_with_ids(4, 8_000, vec![0, 1], "mi300x");
+        node.restore_committed(
+            1,
+            AllocationResult {
+                cpu_ids: vec![0],
+                gpu_ids: vec![0],
+                memory_mb: 1_000,
+            },
+        )
+        .unwrap();
+        let conflict = node.restore_committed(
+            2,
+            AllocationResult {
+                cpu_ids: vec![1],
+                gpu_ids: vec![0],
+                memory_mb: 1_000,
+            },
+        );
+        assert_eq!(conflict, Err(AllocError::GpusUnavailable));
+        // The rejected adoption left nothing behind: gpu 0 still belongs only to
+        // job 1, and its cpu/memory were not partially applied.
+        assert_eq!(node.allocated_gpu_ids(), vec![0]);
+        assert_eq!(node.free_memory_mb(), 7_000);
+        assert!(!node.release_job(2));
+    }
+
+    #[test]
+    fn test_restore_committed_rejects_unknown_gpu() {
+        // A manifest naming a device this node doesn't have is rejected rather
+        // than silently dropped, so the caller can refuse the adoption.
+        let mut node = make_node_with_ids(4, 8_000, vec![0], "mi300x");
+        let bad = node.restore_committed(
+            1,
+            AllocationResult {
+                cpu_ids: vec![0],
+                gpu_ids: vec![99],
+                memory_mb: 1_000,
+            },
+        );
+        assert_eq!(bad, Err(AllocError::GpusUnavailable));
+        assert_eq!(node.free_memory_mb(), 8_000);
+    }
+
     #[test]
     fn test_memory_released_symmetrically_with_zero_cpus() {
         // A job with 0 cpus must still have its memory reserved and released

crates/spurctld/src/main.rs

@@ -22,7 +22,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use clap::Parser;
-use tracing::info;
+use tracing::{info, warn};
 
 use cluster::ClusterManager;
 use rpc_stats::RpcStatsCollector;
@@ -250,6 +250,14 @@ async fn main() -> anyhow::Result<()> {
 
     // Start node health checker (only on leader).
     let hb_timeout = config.controller.heartbeat_timeout_secs.unwrap_or(90);
+    if hb_timeout < 30 {
+        warn!(
+            hb_timeout,
+            "heartbeat_timeout_secs is below 30s; a graceful spurd restart slower than this \
+             will fail its running jobs — this is now the only safety net for a restart, \
+             not just crash detection"
+        );
+    }
     let health_cluster = cluster.clone();
     let health_raft = raft_handle.clone();
     tokio::spawn(async move {