feat(reservations): persist CRUD via raft WAL and enforce scheduling parity

crates/spur-cli/src/format_engine.rs

@@ -193,6 +193,7 @@ pub fn squeue_header(spec: char) -> &'static str {
         'N' => "NODELIST",
         'b' => "GRES",
         'L' => "TIME_LEFT",
+        'v' => "RESERVATION",
         _ => "?",
     }
 }

crates/spur-cli/src/scontrol.rs

@@ -87,6 +87,9 @@ pub enum ScontrolCommand {
         /// Comma-separated users (optional)
         #[arg(long, default_value = "")]
         users: String,
+        /// Comma-separated flags (maint, ignore_jobs, no_hold_jobs, overlap)
+        #[arg(long, default_value = "")]
+        flags: String,
     },
     /// Update a reservation
     #[command(name = "update-reservation")]
@@ -218,6 +221,7 @@ pub async fn main_with_args(args: Vec<String>) -> Result<()> {
             nodes,
             accounts,
             users,
+            flags,
         } => {
             create_reservation(
                 &args.controller,
@@ -227,6 +231,7 @@ pub async fn main_with_args(args: Vec<String>) -> Result<()> {
                 &nodes,
                 &accounts,
                 &users,
+                &flags,
             )
             .await
         }
@@ -371,6 +376,9 @@ async fn show(controller: &str, entity: &str, name: Option<&str>) -> Result<()>
                     label_str.sort();
                     println!("   Labels={}", label_str.join(","));
                 }
+                if !node.active_reservation.is_empty() {
+                    println!("   ActiveReservation={}", node.active_reservation);
+                }
                 println!("   CpuLoad={}", node.cpu_load as f64 / 100.0);
                 println!();
             }
@@ -421,6 +429,12 @@ async fn show(controller: &str, entity: &str, name: Option<&str>) -> Result<()>
                 println!("   StartTime={}", res.start_time);
                 println!("   EndTime={}", res.end_time);
                 println!("   Nodes={}", res.nodes);
+                if !res.state.is_empty() {
+                    println!("   State={}", res.state);
+                }
+                if !res.flags.is_empty() {
+                    println!("   Flags={}", res.flags);
+                }
                 if !res.accounts.is_empty() {
                     println!("   Accounts={}", res.accounts);
                 }
@@ -657,6 +671,7 @@ async fn update_node(
 }
 
 /// Create a reservation via the controller.
+#[allow(clippy::too_many_arguments)]
 async fn create_reservation(
     controller: &str,
     name: &str,
@@ -665,6 +680,7 @@ async fn create_reservation(
     nodes: &str,
     accounts: &str,
     users: &str,
+    flags: &str,
 ) -> Result<()> {
     let mut client = SlurmControllerClient::connect(controller.to_string())
         .await
@@ -685,6 +701,11 @@ async fn create_reservation(
         .map(|s| s.trim().to_string())
         .filter(|s| !s.is_empty())
         .collect();
+    let flag_list: Vec<String> = flags
+        .split(',')
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+        .collect();
 
     client
         .create_reservation(spur_proto::proto::CreateReservationRequest {
@@ -694,6 +715,7 @@ async fn create_reservation(
             nodes: node_list,
             accounts: account_list,
             users: user_list,
+            flags: flag_list,
         })
         .await
         .context("failed to create reservation")?;

crates/spur-cli/src/sinfo.rs

@@ -288,6 +288,9 @@ fn effective_state_str(node: &NodeInfo) -> String {
     if !node.active_reservation.is_empty()
         && node.state == spur_proto::proto::NodeState::NodeIdle as i32
     {
+        if node.reservation_maint {
+            return "maint".into();
+        }
         return "resv".into();
     }
     spur_core::node::NodeState::from_proto_i32(node.state)

crates/spur-cli/src/smd.rs

@@ -109,6 +109,9 @@ pub async fn main_with_args(args: Vec<String>) -> Result<()> {
 
 fn node_state_str(node: &NodeInfo) -> &'static str {
     if !node.active_reservation.is_empty() && node.state == NodeState::NodeIdle as i32 {
+        if node.reservation_maint {
+            return "maint";
+        }
         return "resv";
     }
     spur_core::node::NodeState::from_proto_i32(node.state)

crates/spur-cli/src/squeue.rs

@@ -152,6 +152,7 @@ fn resolve_job_field(job: &spur_proto::proto::JobInfo, spec: char) -> String {
         'o' => job.command.clone(),
         'S' => format_timestamp(job.start_time.as_ref()),
         'V' => format_timestamp(job.submit_time.as_ref()),
+        'v' => job.reservation.clone(),
         'e' => format_timestamp(job.end_time.as_ref()),
         _ => "?".into(),
     }

crates/spur-core/src/config.rs

@@ -283,6 +283,15 @@ pub struct ControllerConfig {
     /// Defaults to 90 when absent.
     #[serde(default)]
     pub heartbeat_timeout_secs: Option<u64>,
+
+    /// Maximum automatic requeues before a job is held with `JobHoldMaxRequeue`.
+    /// Configured in TOML as `[controller] max_batch_requeue` (default: 5).
+    #[serde(default = "default_max_batch_requeue")]
+    pub max_batch_requeue: u32,
+}
+
+fn default_max_batch_requeue() -> u32 {
+    5
 }
 
 fn default_listen_addr() -> String {
@@ -317,6 +326,7 @@ impl Default for ControllerConfig {
             node_id: None,
             raft_listen_addr: "[::]:6821".into(),
             heartbeat_timeout_secs: None,
+            max_batch_requeue: default_max_batch_requeue(),
         }
     }
 }
@@ -381,6 +391,9 @@ pub struct SchedulerConfig {
     /// Max seconds to wait in COMPLETING before force-finishing the job.
     #[serde(default = "default_complete_wait")]
     pub complete_wait_secs: u32,
+    /// Grace minutes after a reservation ends before cancelling its running jobs.
+    #[serde(default)]
+    pub resv_overrun_minutes: u32,
 }
 
 fn default_scheduler_plugin() -> String {
@@ -408,6 +421,7 @@ impl Default for SchedulerConfig {
             fairshare_halflife_days: 14,
             default_time_limit_minutes: 60,
             complete_wait_secs: 300,
+            resv_overrun_minutes: 0,
         }
     }
 }
@@ -790,6 +804,12 @@ impl SlurmConfig {
         if self.cluster_name.is_empty() {
             return Err(ConfigError::MissingField("cluster_name".into()));
         }
+        if self.controller.max_batch_requeue == 0 {
+            return Err(ConfigError::InvalidValue {
+                field: "controller.max_batch_requeue".into(),
+                value: "0 (must be at least 1)".into(),
+            });
+        }
         Ok(())
     }
 
@@ -1163,6 +1183,34 @@ memory_mb = 1024000
     fn controller_config_defaults_for_new_fields() {
         let config = ControllerConfig::default();
         assert_eq!(config.heartbeat_timeout_secs, None);
+        assert_eq!(config.max_batch_requeue, 5);
+    }
+
+    #[test]
+    fn controller_config_parses_max_batch_requeue() {
+        let toml = r#"
+cluster_name = "test"
+
+[controller]
+max_batch_requeue = 7
+"#;
+        let config = SlurmConfig::load_from_str(toml).unwrap();
+        assert_eq!(config.controller.max_batch_requeue, 7);
+    }
+
+    #[test]
+    fn controller_config_rejects_zero_max_batch_requeue() {
+        let toml = r#"
+cluster_name = "test"
+
+[controller]
+max_batch_requeue = 0
+"#;
+        let err = SlurmConfig::load_from_str(toml).unwrap_err();
+        assert!(
+            err.to_string().contains("max_batch_requeue"),
+            "unexpected error: {err}"
+        );
     }
 
     #[test]

crates/spur-core/src/job.rs

@@ -251,9 +251,20 @@ pub enum PendingReason {
     QosGrpNodeLimit,
     BurstBufferResources,
     BurstBufferStageIn,
+    ReservedMaintenance,
+    ReservationDeleted,
+    JobHoldMaxRequeue,
 }
 
 impl PendingReason {
+    /// True when the job is held and must not be scheduled until released.
+    pub fn is_scheduling_hold(&self) -> bool {
+        matches!(
+            self,
+            Self::Held | Self::JobHeldAdmin | Self::JobHoldMaxRequeue | Self::ReservationDeleted
+        )
+    }
+
     pub fn display(&self) -> &'static str {
         match self {
             Self::None => "None",
@@ -294,6 +305,9 @@ impl PendingReason {
             Self::QosGrpNodeLimit => "QOSGrpNodeLimit",
             Self::BurstBufferResources => "BurstBufferResources",
             Self::BurstBufferStageIn => "BurstBufferStageIn",
+            Self::ReservedMaintenance => "ReqNodeNotAvail, Reserved for maintenance",
+            Self::ReservationDeleted => "ReservationDeleted",
+            Self::JobHoldMaxRequeue => "JobHoldMaxRequeue",
         }
     }
 }
@@ -778,6 +792,7 @@ impl Job {
             // Requeue transitions: terminal → Pending (for --requeue jobs)
             (JobState::Timeout, JobState::Pending) => true,
             (JobState::Preempted, JobState::Pending) => true,
+            (JobState::Preempted, JobState::Cancelled) => true,
             (JobState::NodeFail, JobState::Pending) => true,
             (JobState::Failed, JobState::Pending) => true,
             _ => false,
@@ -1170,6 +1185,7 @@ mod tests {
         (PendingReason::QosGrpNodeLimit, "QOSGrpNodeLimit"),
         (PendingReason::BurstBufferResources, "BurstBufferResources"),
         (PendingReason::BurstBufferStageIn, "BurstBufferStageIn"),
+        (PendingReason::JobHoldMaxRequeue, "JobHoldMaxRequeue"),
     ];
 
     #[test]