fix(k8s): recover a node the controller has forgotten

crates/spur-k8s/src/agent.rs

@@ -115,9 +115,17 @@ impl VirtualAgent {
         let job = match result {
             Ok(Ok(job)) => job,
             Ok(Err(status)) => return Err(status),
+            // The retry above swallows NotFound while it waits for a label that
+            // may still be propagating. When the budget runs out the answer is
+            // that no SpurJob exists, which is an explicit rejection and not a
+            // transport failure: deadline_exceeded here made the controller
+            // report "agent unreachable" for a running, reachable operator.
             Err(_elapsed) => {
-                return Err(Status::deadline_exceeded(format!(
-                    "namespace lookup for spur.amd.com/job-id={job_id} timed out after {}s",
+                return Err(Status::not_found(format!(
+                    "no SpurJob carries the label spur.amd.com/job-id={job_id} after {}s. In Pod \
+                     mode the operator makes Pods only for a SpurJob custom resource, so a job \
+                     submitted with the CLI (sbatch, spur submit) has nothing to launch. Submit \
+                     it with `kubectl apply` of a SpurJob instead.",
                     NS_LOOKUP_BUDGET.as_secs()
                 )))
             }

crates/spur-k8s/src/heartbeat.rs

@@ -4,7 +4,7 @@
 use std::collections::HashMap;
 
 use tokio::sync::RwLock;
-use tracing::{debug, warn};
+use tracing::{debug, info, warn};
 
 use spur_proto::proto::slurm_controller_client::SlurmControllerClient;
 use spur_proto::proto::{HeartbeatRequest, RegisterAgentRequest};
@@ -67,6 +67,28 @@ impl HeartbeatManager {
                         };
                         match client.heartbeat(req).await {
                             Ok(_) => debug!(node = %name, "heartbeat sent"),
+                            // A controller that lost its state, for example a Raft
+                            // cluster that was built again, answers NOT_FOUND for a
+                            // node it once knew. The node watcher registers on its
+                            // initial list and on a change, and neither happens
+                            // again, so the node would stay unknown and the cluster
+                            // would schedule nothing. Register it again here.
+                            Err(e) if e.code() == tonic::Code::NotFound => {
+                                let stored = self.registry.read().await.get(name).cloned();
+                                match stored {
+                                    Some(reg) => match client.register_agent(reg).await {
+                                        Ok(_) => info!(
+                                            node = %name,
+                                            "spurctld did not know this node; registered it again"
+                                        ),
+                                        Err(e) => warn!(
+                                            node = %name, error = %e,
+                                            "failed to register the node again"
+                                        ),
+                                    },
+                                    None => warn!(node = %name, "node no longer tracked"),
+                                }
+                            }
                             Err(e) => warn!(node = %name, error = %e, "heartbeat failed"),
                         }
                     }

crates/spurctld/src/scheduler_loop.rs

@@ -1224,6 +1224,11 @@ async fn dispatch_to_agent(
             tonic::Code::Unavailable | tonic::Code::DeadlineExceeded => {
                 DispatchError::Unreachable(s.into())
             }
+            // The agent answered and refused. Reporting this as unreachable sent
+            // the operator looking for a network fault that does not exist.
+            tonic::Code::NotFound | tonic::Code::FailedPrecondition => {
+                DispatchError::AgentRejected(s.message().to_string())
+            }
             _ => DispatchError::Other(s.into()),
         })?;