stop: verify absence in ANY state (running+exited) then rm, second running-only loop; serve: up to 3 re-probes on short-free verdict

src/docker/stop.rs

@@ -13,13 +13,12 @@ use tracing::info;
 
 /// Builds the per-node remote stop command.
 ///
-/// `wait = true` (default): after `docker stop`, explicitly remove any
-/// stopped-but-still-present labeled containers (there is no `--rm`, so
-/// an Exited container lingers), then poll every 1 s (60 s max) until
-/// `docker ps` shows no labeled container running; if any remain, exit 1 —
-/// `run_ssh` surfaces that as an error. Without the explicit removal and
-/// the running-only polling, a graceful-shutdown straggler could keep the
-/// node reported as `occupied` for the immediately following serve.
+/// `wait = true` (default): after `docker stop`, poll until `docker ps` shows
+/// no container with the `sm121.recipe` / `sm121.cluster` labels in ANY state
+/// (running or exited-but-not-yet-removed) for 30 s, then explicitly `rm` the
+/// stragglers and poll running-only for another 30 s. Reporting success while
+/// any labeled container still existed is what made an immediately following
+/// `serve` see the node as `occupied` and bail out.
 pub fn remote_command(recipe_name: &str, cluster_name: &str, rank: u32, wait: bool) -> String {
     let filter = format!(
         "--filter {} --filter {}",
@@ -28,15 +27,25 @@ pub fn remote_command(recipe_name: &str, cluster_name: &str, rank: u32, wait: bo
     );
     let wait_block = if wait {
         "\
+for _ in $(seq 1 30); do \
+    n=$(docker ps -aq {filter}); \
+    [ -z \"$n\" ] && break; \
+    sleep 1; \
+done; \
 docker ps -aq {filter} | xargs -r docker rm >/dev/null 2>&1; \
-for _ in $(seq 1 60); do \
+for _ in $(seq 1 30); do \
     n=$(docker ps -q {filter}); \
     [ -z \"$n\" ] && break; \
     sleep 1; \
 done; \
+n=$(docker ps -aq {filter}); \
+if [ -n \"$n\" ]; then \
+    echo \"sm121 stop: containers still present after wait: $n\" >&2; \
+    exit 1; \
+fi; \
 n=$(docker ps -q {filter}); \
 if [ -n \"$n\" ]; then \
-    echo \"sm121 stop: containers still running after 60 s: $n\" >&2; \
+    echo \"sm121 stop: containers still running after wait: $n\" >&2; \
     exit 1; \
 fi; "
             .to_string()
@@ -204,8 +213,8 @@ mod tests {
     fn default_includes_verification_loop() {
         let cmd = remote_command("r", "c", 0, true);
         assert!(cmd.contains("xargs -r docker stop"));
+        assert!(cmd.contains("docker ps -aq"));
         assert!(cmd.contains("xargs -r docker rm"));
-        assert!(cmd.contains("seq 1 60"));
         assert!(cmd.contains("exit 1"));
         assert!(cmd.ends_with("true"));
     }
@@ -213,7 +222,8 @@ mod tests {
     #[test]
     fn no_wait_omits_verification_loop() {
         let cmd = remote_command("r", "c", 1, false);
-        assert!(!cmd.contains("seq 1 60"));
+        assert!(!cmd.contains("seq 1 30"));
+        assert!(!cmd.contains("docker rm"));
         assert!(!cmd.contains("exit 1"));
         assert!(cmd.contains("/tmp/sm121_r_1"));
     }

src/orchestrate/mod.rs

@@ -155,39 +155,46 @@ where
     if let Some(required) = recipe.required_nodes {
         let actual = resolved.0.nodes.len() as u32;
         if actual < required && !req.force_occupied {
-            tokio::time::sleep(Duration::from_secs(req.boot_timing.retry_grace_secs)).await;
-            let state2 = req.prober.probe(config).await;
-            if let Ok((c2, n2)) =
-                select_cluster(config, cluster_name, recipe.required_nodes, &state2)
-            {
-                let avail2: Vec<Node> = c2
-                    .nodes
-                    .iter()
-                    .filter(|n| {
-                        state2
-                            .nodes
-                            .get(&n.name)
-                            .is_some_and(|ns| ns.reachable && !ns.occupied)
-                    })
-                    .cloned()
-                    .collect();
-                if avail2.len() as u32 >= required {
-                    info!(
-                        "cluster re-probe recovered '{}': {} free node(s) (first pass saw {})",
-                        c2.name,
-                        avail2.len(),
-                        actual
-                    );
-                    resolved = (
-                        Cluster {
-                            nodes: avail2,
-                            ..c2
-                        },
-                        n2.cloned(),
-                    );
-                } else {
-                    last_state = state2;
+            // A container still tearing down (graceful shutdown can take
+            // tens of seconds) reads as `occupied` for a while after stop
+            // returns — re-probe up to 3 times with a grace window between
+            // attempts before declaring the cluster short on free nodes.
+            for attempt in 1..=3 {
+                tokio::time::sleep(Duration::from_secs(req.boot_timing.retry_grace_secs)).await;
+                let state2 = req.prober.probe(config).await;
+                if let Ok((c2, n2)) =
+                    select_cluster(config, cluster_name, recipe.required_nodes, &state2)
+                {
+                    let avail2: Vec<Node> = c2
+                        .nodes
+                        .iter()
+                        .filter(|n| {
+                            state2
+                                .nodes
+                                .get(&n.name)
+                                .is_some_and(|ns| ns.reachable && !ns.occupied)
+                        })
+                        .cloned()
+                        .collect();
+                    if avail2.len() as u32 >= required {
+                        info!(
+                            "cluster re-probe #{} recovered '{}': {} free node(s) (first pass saw {})",
+                            attempt,
+                            c2.name,
+                            avail2.len(),
+                            actual
+                        );
+                        resolved = (
+                            Cluster {
+                                nodes: avail2,
+                                ..c2
+                            },
+                            n2.cloned(),
+                        );
+                        break;
+                    }
                 }
+                last_state = state2;
             }
         }
         let actual = resolved.0.nodes.len() as u32;