fix(hive-priv): a stop that succeeds must actually release the machine name

`nixos-container stop` exiting 0 does not mean machined has dropped the
registration. A process sitting in the machine cgroup without being a
child of the container's init never receives the shutdown's SIGTERM if
it has been SIGSTOP'd, so the registration outlives the "successful"
stop. Every later start then fails with "Failed to register machine:
already exists", and machined re-persists the stale record across its
own restart, so there is no cleaning it up afterwards.

StopContainer now asks for the stop, waits for machined to release the
name, escalates to SIGKILL if it hasn't, re-verifies, and fails loudly
if the name is still held — so a caller is never told the stop worked
and then walks into the confusing registration error.

The probe resolves the name through machined's GetMachine, the same
lookup that later rejects the registration, rather than checking the
container's systemd unit: the unit going inactive while the name is
still held is precisely the case being caught.

Verify-and-escalate lives in the helper, not at a call site, so every
stop gets it — dashboard, reconcile, destroy, cold-start fallback. The
start path already distrusts its own exit code the same way; this is
the missing half of that pair.

hive-priv/src/main.rs

@@ -176,7 +176,7 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
 
         PrivRequest::StopContainer { ref name } => {
             validate_container_name(name)?;
-            container_run(&["stop", &container_system_name(name)]).await
+            stop_and_release(&container_system_name(name)).await
         }
 
         PrivRequest::KillContainer { ref name } => {
@@ -1486,6 +1486,126 @@ async fn machinectl_run(args: &[&str]) -> Result<(String, String)> {
     Ok((stdout, stderr))
 }
 
+/// How long to wait for machined to drop a machine's registration after a
+/// shutdown has been asked for. Generous: a container with slow-stopping
+/// units legitimately takes a while, and escalating early would SIGKILL a
+/// shutdown that was going to finish on its own.
+const NAME_RELEASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
+
+/// Poll cadence while waiting for the registration to go away.
+const NAME_RELEASE_POLL: std::time::Duration = std::time::Duration::from_millis(500);
+
+/// Cap on a single registration probe. The probe is a D-Bus round trip to
+/// machined; if machined itself is wedged the call would otherwise sit on
+/// the D-Bus method timeout, which is far longer than the whole stop
+/// sequence should take.
+const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+
+/// True when machined still holds a registration for `machine`.
+///
+/// This asks machined the same question the registration itself answers —
+/// `machinectl show` resolves the name through `GetMachine`, the very lookup
+/// that makes a later boot fail with `Failed to register machine: already
+/// exists`. So a positive answer here is exactly the condition that breaks
+/// the next start, not a proxy for it.
+///
+/// Deliberately NOT the container's systemd unit state: the unit can be
+/// `inactive` while the registration is still held, and that gap is the
+/// whole bug this probe exists to catch.
+///
+/// Errors and timeouts answer "still registered". Being wrong that way costs
+/// a redundant SIGKILL to something already gone; being wrong the other way
+/// hands back a stop that silently leaked the name.
+async fn machine_registered(machine: &str) -> bool {
+    let probe = Command::new("machinectl")
+        .args(["show", machine, "--property=Name"])
+        // Don't leave a probe behind when the timeout below fires.
+        .kill_on_drop(true)
+        .output();
+    match tokio::time::timeout(MACHINE_PROBE_TIMEOUT, probe).await {
+        Ok(Ok(out)) => out.status.success(),
+        Ok(Err(e)) => {
+            tracing::warn!(%machine, error = %e, "machinectl show failed to run; assuming still registered");
+            true
+        }
+        Err(_) => {
+            tracing::warn!(%machine, "machinectl show timed out; assuming still registered");
+            true
+        }
+    }
+}
+
+/// Poll [`machine_registered`] until the name is free or the timeout expires.
+/// Returns true once the registration is gone.
+async fn wait_name_released(machine: &str) -> bool {
+    let deadline = tokio::time::Instant::now() + NAME_RELEASE_TIMEOUT;
+    loop {
+        if !machine_registered(machine).await {
+            return true;
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return false;
+        }
+        tokio::time::sleep(NAME_RELEASE_POLL).await;
+    }
+}
+
+/// Stop `machine` and only report success once machined has actually released
+/// the name.
+///
+/// `nixos-container stop` exiting 0 does not mean the machine is gone. A
+/// process that sits in the machine's cgroup without being a child of the
+/// container's init — a shell exec'd in from outside, say — never receives
+/// the shutdown's SIGTERM if it has been stopped with SIGSTOP, so the
+/// registration outlives the "successful" stop. Every later start of that
+/// container then fails with `Failed to register machine: already exists`,
+/// and machined re-persists the stale record across its own restart, so
+/// there is no cleaning it up after the fact. The only place to catch it is
+/// here, in the stop.
+///
+/// So: ask for the stop, wait for the name, and escalate to SIGKILL if it is
+/// still held. If even that doesn't free it, fail loudly — a caller that is
+/// told the stop worked will go on to start the container and hit the
+/// confusing registration error instead of this one.
+///
+/// The verify-then-escalate lives in the helper rather than at a call site so
+/// that every stop gets it: dashboard, reconcile, destroy, cold-start
+/// fallback. The start path already distrusts its own exit code the same way;
+/// this is the missing half of that pair.
+async fn stop_and_release(machine: &str) -> Result<(String, String)> {
+    let stop = container_run(&["stop", machine]).await;
+    if wait_name_released(machine).await {
+        // Happy path, and also the path where a stop that reported failure
+        // nonetheless brought the machine down. Either way the caller gets
+        // the original result untouched.
+        return stop;
+    }
+
+    tracing::warn!(
+        %machine,
+        "stop finished but machined still holds the registration; escalating to SIGKILL"
+    );
+    let kill = machinectl_run(&["kill", machine, "--signal=SIGKILL"]).await;
+    if wait_name_released(machine).await {
+        if let Err(e) = &kill {
+            // The name is free, which is what was asked for. machinectl
+            // complaining on the way there is worth a line, not a failure.
+            tracing::warn!(%machine, error = %e, "SIGKILL reported an error but the name was released");
+        }
+        return Ok(kill.unwrap_or_default());
+    }
+
+    let detail = kill.err().map_or_else(
+        || "SIGKILL was delivered".to_owned(),
+        |e| format!("SIGKILL also failed: {e}"),
+    );
+    bail!(
+        "stop {machine}: machined still holds the machine name after {}s ({detail}). \
+         Starting this container again will fail to register.",
+        NAME_RELEASE_TIMEOUT.as_secs() * 2
+    )
+}
+
 /// Invoke `nixos-container` with the given args and forward output lines
 /// to the caller as `PrivEvent::Line` messages in real time, logging each
 /// line to journald as it arrives. Returns `(String::new(), String::new())`