fix(spurctld): pin jwt_key and interval_secs to startup; scope reconfigure to leader

Three correctness fixes on top of the live-reload work:

- auth.jwt_key is now restart-only. Node tokens are HS256-signed with it
  (7-day TTL) and verified per-RPC; swapping the key live would instantly
  reject every outstanding node token and silently partition healthy nodes.
  ControllerService captures the key once at startup (resolve_startup_jwt_key)
  and verification uses that pinned value, mirroring Slurm's restart-only
  AuthType. admission.mode stays live (flipping it invalidates nothing).

- scheduler.interval_secs is now pinned everywhere. The scheduler loop
  already captured it once at boot; the preemption requeue hold read it live
  from config(), so post-reconfigure the hold window drifted while the loop
  cadence did not. Capture it in ClusterManager at construction and read the
  pinned value, keeping interval_secs truly restart-only.

- reconfigure is documented as leader-only. It runs on the Raft leader and
  swaps only the leader's in-memory config; no WAL entry carries config, so
  followers keep their startup config until restart (in k8s they re-read the
  same ConfigMap). Even a partition edit that propagates via WAL makes
  followers reconcile against their own stale config().nodes, so they get new
  partition membership but old node features until restart. Clarified in the
  reconfigure doc comment, CLI help/output, partitioning.rst, and
  kubernetes.rst. WAL-propagating config is a planned follow-up.

Tests: reconfigure_does_not_adopt_new_jwt_key builds a real ControllerService,
reconfigures the key, and proves a token minted with the startup key still
verifies (and would fail under the new key). The max_batch_requeue test now
drives maybe_requeue past the cap and asserts the consumer requeues after the
new cap is applied, rather than only checking the swapped config value.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: yansun1996

crates/spur-cli/src/scontrol.rs

@@ -221,8 +221,9 @@ pub enum ScontrolCommand {
         #[arg(long)]
         name: String,
     },
-    /// Re-read spur.conf and apply it live (partitions, nodes, licenses,
-    /// hooks, scheduler tunables, etc.; ports/DB/raft need a restart)
+    /// Re-read spur.conf and apply it live on the leader (partitions, nodes,
+    /// licenses, hooks, scheduler tunables, etc.). Ports/DB/raft/jwt_key need a
+    /// restart; followers converge on restart.
     Reconfigure,
     /// Create a reservation
     #[command(name = "create-reservation")]
@@ -1480,7 +1481,7 @@ async fn reconfigure(controller: &str) -> Result<()> {
     client.reconfigure(()).await.context("reconfigure failed")?;
 
     println!(
-        "Reconfiguration complete (listen ports, accounting DB, and raft peers still require a controller restart)"
+        "Reconfiguration complete on the leader (followers converge on restart; listen ports, accounting DB, raft peers, and jwt_key still require a controller restart)"
     );
     Ok(())
 }

crates/spurctld/src/cluster.rs

@@ -223,6 +223,12 @@ pub struct ClusterManager {
     /// values live; sections captured once at startup (bound sockets, DB pool,
     /// scheduler loop interval) remain restart-only — see `reconfigure`.
     config: RwLock<Arc<SlurmConfig>>,
+    /// Scheduler tick interval captured at startup. The scheduler loop's cadence
+    /// is fixed once at boot (restart-only), so the preemption requeue hold —
+    /// which is sized to that cadence — must read this pinned value, not the
+    /// live `config()`, or the hold window would drift after `reconfigure`
+    /// while the loop keeps ticking at the old rate.
+    scheduler_interval_secs: u32,
     /// Path to the spur.conf file, re-read by `reconfigure()`. spur.conf is
     /// never written back — the Raft WAL is the sole source of runtime truth.
     /// None when running without a config file (e.g. in tests).
@@ -272,11 +278,13 @@ impl ClusterManager {
         let burst_buffer_total_gb = config.burst_buffer.total_gb;
         let fairshare_cache = Arc::new(FairshareCache::new());
         let first_job_id = config.controller.first_job_id;
+        let scheduler_interval_secs = config.scheduler.interval_secs;
         let qos_cache = Arc::new(QosCache::new());
         let association_cache = Arc::new(AssociationCache::new());
 
         let cm = Self {
             config: RwLock::new(Arc::new(config)),
+            scheduler_interval_secs,
             config_path,
             jobs: RwLock::new(HashMap::new()),
             nodes: RwLock::new(HashMap::new()),
@@ -941,7 +949,7 @@ impl ClusterManager {
                 // the maybe_requeue MAX_REQUEUE cap: Slurm always requeues a
                 // preempted job regardless of its --requeue flag. This is a
                 // deliberate divergence from the ordinary requeue path.
-                let hold_secs = (self.config().scheduler.interval_secs as i64 * 2 + 3).max(5);
+                let hold_secs = (self.scheduler_interval_secs as i64 * 2 + 3).max(5);
                 let hold = Utc::now() + chrono::Duration::seconds(hold_secs);
                 // Honor a later user --begin: compute the max on the leader so
                 // followers apply one verbatim instant (no per-replica clock).
@@ -2440,19 +2448,32 @@ impl ClusterManager {
     /// semantics in Slurm: runtime-only changes not reflected in the conf are
     /// overwritten by the incoming conf values.
     ///
-    /// Reloaded live (readers take a fresh snapshot via `config()`):
-    /// `[[partitions]]`, `[[nodes]]` features/weight, `licenses`,
+    /// **Leader-only.** The command is forwarded to the Raft leader, and this
+    /// swaps only the leader's in-memory config — no WAL entry carries the new
+    /// config. Followers keep the config they read at startup until they
+    /// restart (in Kubernetes they re-read the same ConfigMap). Do not rely on
+    /// reconfigured non-partition state surviving an immediate failover.
+    /// Partition edits DO propagate (via partition WAL ops), but a follower
+    /// re-runs `reconcile_partitions` against its own stale `config().nodes`,
+    /// so after a partition edit followers pick up new partition membership but
+    /// keep old node features until restart. WAL-propagating config is a
+    /// planned follow-up.
+    ///
+    /// Reloaded live on the leader (readers take a fresh snapshot via
+    /// `config()`): `[[partitions]]`, `[[nodes]]` features/weight, `licenses`,
     /// `burst_buffer`, `scheduler` tunables (`complete_wait`, `resv_overrun`),
     /// `controller.max_batch_requeue`, `hooks`, `notifications`, `federation`,
-    /// `power` suspend/resume commands, `admission.mode`, `auth.jwt_key`, and
+    /// `power` suspend/resume commands, `admission.mode`, and
     /// `metrics.high_cardinality`.
     ///
     /// Restart-only (baked in at startup — mirrors Slurm's restart-required set
-    /// of ports/plugins/StateSaveLocation): bind addresses and ports
+    /// of ports/plugins/StateSaveLocation/AuthType): bind addresses and ports
     /// (`controller.listen_addr`, `metrics`/`rest_api` listeners), the
     /// accounting database pool (`accounting.database_url`), Raft identity/peers,
-    /// `controller.first_job_id`, and the scheduler loop cadence
-    /// (`scheduler.interval_secs`, `max_jobs_per_cycle`, `topology`).
+    /// `controller.first_job_id`, `auth.jwt_key` (swapping it live would
+    /// instantly invalidate every outstanding node token), and the scheduler
+    /// loop cadence (`scheduler.interval_secs`, `max_jobs_per_cycle`,
+    /// `topology`).
     pub fn reconfigure(&self) -> Result<(), anyhow::Error> {
         let Some(ref path) = self.config_path else {
             anyhow::bail!("reconfigure requires a config file path, but none is configured");
@@ -2557,7 +2578,7 @@ impl ClusterManager {
             self.reconcile_partitions(&mut nodes);
         }
 
-        info!("reconfigure: applied spur.conf; restart-only sections (listen ports, accounting DB, raft peers, scheduler cadence) unchanged until controller restart");
+        info!("reconfigure: applied spur.conf on this leader (followers converge on restart); restart-only sections (listen ports, accounting DB, raft peers, jwt_key, scheduler cadence) unchanged until controller restart");
         Ok(())
     }
 
@@ -4973,27 +4994,54 @@ mod tests {
         (cm, conf_path)
     }
 
-    #[tokio::test]
-    async fn reconfigure_reloads_max_batch_requeue_live() {
+    /// Consumer-driven: `maybe_requeue` must honor the new `max_batch_requeue`
+    /// after reconfigure, not just the swapped config value. A job whose
+    /// `requeue_count` sits between the old and new caps is a no-op under the
+    /// old cap but requeues to Pending under the new one.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn reconfigure_max_batch_requeue_changes_consumer_behavior() {
+        let conf = |cap: u32| {
+            format!(
+                "cluster_name = \"test\"\n\
+                 [controller]\nmax_batch_requeue = {cap}\n\
+                 [[partitions]]\nname = \"default\"\ndefault = true\nstate = \"UP\"\nnodes = \"ALL\"\n"
+            )
+        };
         let dir = TempDir::new().unwrap();
-        let (cm, conf_path) = test_cluster_with_conf_file(
-            &dir,
-            "cluster_name = \"test\"\n[controller]\nmax_batch_requeue = 3\n",
-        )
-        .await;
-        assert_eq!(cm.config().controller.max_batch_requeue, 3);
+        let (cm, conf_path) = test_cluster_with_conf_file(&dir, &conf(3)).await;
+        register_node(&cm, "worker1", 8, 16000);
 
-        std::fs::write(
-            &conf_path,
-            "cluster_name = \"test\"\n[controller]\nmax_batch_requeue = 9\n",
-        )
-        .unwrap();
+        let job_id = run_job_on(&cm, "requeue-cap", "worker1");
+        // Put the job in a terminal, requeue-eligible state (Failed → Pending is
+        // a valid requeue transition and is NOT in the max-requeue hold set, so
+        // over-cap is a clean no-op) with 5 attempts already recorded.
+        {
+            let mut jobs = cm.jobs.write();
+            let job = jobs.get_mut(&job_id).unwrap();
+            job.state = JobState::Failed;
+            job.spec.requeue = true;
+            job.requeue_count = 5;
+        }
+
+        // Cap = 3, count = 5 → over cap → maybe_requeue is a no-op (stays Failed).
+        cm.maybe_requeue(job_id).unwrap();
+        assert_eq!(
+            cm.get_job(job_id).unwrap().state,
+            JobState::Failed,
+            "over-cap job must not requeue before reconfigure"
+        );
+
+        // Raise the cap past the attempt count and reconfigure.
+        std::fs::write(&conf_path, conf(9)).unwrap();
         cm.reconfigure().unwrap();
 
+        // Cap = 9, count = 5 → under cap → maybe_requeue returns it to Pending.
+        cm.maybe_requeue(job_id).unwrap();
+        settle(&cm, job_id, JobState::Pending);
         assert_eq!(
-            cm.config().controller.max_batch_requeue,
-            9,
-            "reconfigure must apply the new max_batch_requeue to the live config"
+            cm.get_job(job_id).unwrap().state,
+            JobState::Pending,
+            "after reconfigure raised the cap, the consumer must requeue the job"
         );
     }
 

crates/spurctld/src/server.rs

@@ -34,6 +34,11 @@ pub struct ControllerService {
     client_addrs: BTreeMap<u64, String>,
     rpc_stats: Arc<RpcStatsCollector>,
     sched_stats: Arc<SchedStatsCollector>,
+    /// JWT signing key for node tokens, captured at startup. Deliberately NOT
+    /// re-read on `scontrol reconfigure`: swapping it live would instantly fail
+    /// verification of every outstanding node token (7-day TTL), silently
+    /// partitioning healthy nodes. Like Slurm's AuthType, it is restart-only.
+    jwt_key: String,
 }
 
 struct LeaderProxy {
@@ -89,6 +94,18 @@ impl LeaderProxy {
     }
 }
 
+/// Resolve the node-token signing key from config at startup. Captured once by
+/// `serve` into `ControllerService::jwt_key`; deliberately not re-read on
+/// `reconfigure` (see the field doc). Falls back to a shared default so
+/// key-less dev clusters interoperate.
+fn resolve_startup_jwt_key(config: &spur_core::config::SlurmConfig) -> String {
+    config
+        .auth
+        .jwt_key
+        .clone()
+        .unwrap_or_else(|| "spur-default-key".to_string())
+}
+
 impl ControllerService {
     // tonic::Status is 176 bytes (over clippy's 128-byte threshold); fixed upstream in tonic 0.13+
     #[allow(clippy::result_large_err)]
@@ -140,8 +157,7 @@ impl ControllerService {
     fn validate_admission(&self, join_token: &str, hostname: &str) -> Result<String, Status> {
         use spur_core::config::AdmissionMode;
 
-        let config = self.cluster.config();
-        if !matches!(config.admission.mode, AdmissionMode::Token) {
+        if !matches!(self.cluster.config().admission.mode, AdmissionMode::Token) {
             return Ok(String::new());
         }
 
@@ -156,9 +172,7 @@ impl ControllerService {
         spur_core::admission::validate_token(token_id, secret, &token_store)
             .map_err(|e| Status::permission_denied(e.to_string()))?;
 
-        let jwt_key = config.auth.jwt_key.as_deref().unwrap_or("spur-default-key");
-
-        spur_core::admission::generate_node_token(hostname, jwt_key.as_bytes())
+        spur_core::admission::generate_node_token(hostname, self.jwt_key.as_bytes())
             .map_err(|e| Status::internal(e.to_string()))
     }
 }
@@ -623,17 +637,15 @@ impl SlurmController for ControllerService {
         }
         let req = request.into_inner();
 
-        let config = self.cluster.config();
         if matches!(
-            config.admission.mode,
+            self.cluster.config().admission.mode,
             spur_core::config::AdmissionMode::Token
         ) {
             if req.node_token.is_empty() {
                 return Err(Status::unauthenticated("node token required"));
             }
-            let jwt_key = config.auth.jwt_key.as_deref().unwrap_or("spur-default-key");
             let identity =
-                spur_core::admission::verify_node_token(&req.node_token, jwt_key.as_bytes())
+                spur_core::admission::verify_node_token(&req.node_token, self.jwt_key.as_bytes())
                     .map_err(|e| Status::unauthenticated(e.to_string()))?;
             if identity.hostname != req.hostname {
                 return Err(Status::permission_denied("node token hostname mismatch"));
@@ -963,17 +975,15 @@ impl SlurmController for ControllerService {
 
         let req = request.into_inner();
 
-        let config = self.cluster.config();
         if matches!(
-            config.admission.mode,
+            self.cluster.config().admission.mode,
             spur_core::config::AdmissionMode::Token
         ) {
             if req.node_token.is_empty() {
                 return Err(Status::unauthenticated("node token required"));
             }
-            let jwt_key = config.auth.jwt_key.as_deref().unwrap_or("spur-default-key");
             let identity =
-                spur_core::admission::verify_node_token(&req.node_token, jwt_key.as_bytes())
+                spur_core::admission::verify_node_token(&req.node_token, self.jwt_key.as_bytes())
                     .map_err(|e| Status::unauthenticated(e.to_string()))?;
             if identity.hostname != req.hostname {
                 return Err(Status::permission_denied("node token hostname mismatch"));
@@ -1746,13 +1756,16 @@ pub async fn serve(
 
     let leader_proxy = LeaderProxy::new(raft_handle.clone(), client_addrs.clone());
 
+    let jwt_key = resolve_startup_jwt_key(&cluster.config());
+
     let service = ControllerService {
         cluster,
         client_addrs,
         raft: raft_handle.clone(),
         leader_proxy,
         rpc_stats: rpc_stats.clone(),
         sched_stats: sched_stats.clone(),
+        jwt_key,
     };
 
     let stats_layer = RpcStatsLayer::new(rpc_stats, raft_handle);
@@ -2327,6 +2340,73 @@ mod tests {
         assert_eq!(status.message(), "partition 'gpu' not found");
     }
 
+    /// GATE: `auth.jwt_key` is captured at startup and must NOT change on
+    /// `reconfigure`. Swapping it live would instantly invalidate every
+    /// outstanding node token. This drives the real capture path
+    /// (`resolve_startup_jwt_key`) and the real `reconfigure`, then proves the
+    /// running controller still verifies a token minted with the startup key.
+    #[tokio::test]
+    async fn reconfigure_does_not_adopt_new_jwt_key() {
+        use spur_core::admission::{generate_node_token, verify_node_token};
+
+        let dir = tempfile::TempDir::new().unwrap();
+        let conf_path = dir.path().join("spur.conf");
+        std::fs::write(
+            &conf_path,
+            "cluster_name = \"test\"\n[auth]\nplugin = \"jwt\"\njwt_key = \"old-secret\"\n",
+        )
+        .unwrap();
+
+        let config = spur_core::config::SlurmConfig::load_from_file(&conf_path).unwrap();
+        let cluster = Arc::new(
+            ClusterManager::new_with_config_path(config, dir.path(), Some(conf_path.clone()))
+                .unwrap(),
+        );
+        let handle = crate::raft::start_raft(1, &["[::1]:0".into()], dir.path(), cluster.clone())
+            .await
+            .unwrap();
+        handle
+            .raft
+            .wait(Some(std::time::Duration::from_secs(5)))
+            .metrics(|m| m.current_leader == Some(1), "leader elected")
+            .await
+            .unwrap();
+        cluster.set_raft(handle.raft);
+
+        // The controller captures the signing key exactly here, at startup.
+        let startup_key = resolve_startup_jwt_key(&cluster.config());
+        assert_eq!(startup_key, "old-secret");
+        let token = generate_node_token("node-1", startup_key.as_bytes()).unwrap();
+
+        // Operator edits jwt_key and reconfigures.
+        std::fs::write(
+            &conf_path,
+            "cluster_name = \"test\"\n[auth]\nplugin = \"jwt\"\njwt_key = \"new-secret\"\n",
+        )
+        .unwrap();
+        cluster.reconfigure().unwrap();
+
+        // The live config reflects the new key (proving reconfigure did swap
+        // config)...
+        assert_eq!(
+            cluster.config().auth.jwt_key.as_deref(),
+            Some("new-secret"),
+            "reconfigure must swap the live config"
+        );
+        // ...but the controller's captured key is unchanged, so tokens minted
+        // with the startup key still verify. A live-reloaded key would reject
+        // this token.
+        assert_eq!(startup_key, "old-secret", "captured key must not change");
+        assert!(
+            verify_node_token(&token, startup_key.as_bytes()).is_ok(),
+            "outstanding node token must still verify against the startup key"
+        );
+        assert!(
+            verify_node_token(&token, b"new-secret").is_err(),
+            "sanity: the token would fail under the new key (proving the key matters)"
+        );
+    }
+
     #[test]
     fn submit_rpc_status_maps_internal() {
         use crate::cluster::SubmitError;

docs/deployment/kubernetes.rst

@@ -79,8 +79,12 @@ Raft peers use StatefulSet DNS names. The node ID is auto-detected from the pod
 Adjust partition definitions and node resources to match your cluster hardware.
 Once the controller is running, ``scontrol reconfigure`` applies most section
 changes (partitions, nodes, licenses, hooks, scheduler tunables) without a
-restart; listen ports, the accounting database, and Raft peers still require
-restarting the controller. See :doc:`partitioning` for the full breakdown.
+restart; listen ports, the accounting database, Raft peers, and ``jwt_key``
+still require restarting the controller. ``reconfigure`` runs on the Raft
+leader only — followers keep their startup config until restarted, at which
+point they re-read this same ConfigMap and converge. To roll all controllers
+onto an updated ConfigMap, restart the StatefulSet pods. See
+:doc:`partitioning` for the full breakdown.
 
 Submitting Jobs
 ---------------