fix(lifecycle): serialize per-box boots to stop the orphan-VM restart race (#146)

HIGH, from the concurrency audit. `boot_from_record` creates a real VM OUTSIDE
the state lock (the flock only spans the post-boot record write). A box that is
both the monitor's auto-restart target AND a valid user `restart`/`start` target
(a dead box) can be booted by both concurrently: each clones the record, boots a
VM unlocked, then races to persist. apply_boot_result has no already-running
guard, so the second write overwrites the first's pid — the first VM (shim +
overlay mount + network endpoint) becomes untracked and is never reaped.
ensure_network_connected is idempotent for the dup and the overlay stacks without
EBUSY, so the second boot doesn't fail on its own.

Fix: a per-box advisory `BootLock` (flock on `locks/<box_id>.boot.lock`) held
across the boot AND the record write, via a new `boot::boot_and_record` helper
used by both the monitor and `restart`. A loser that acquires the lock and finds
the box already running with a live, identity-matched shim returns
`AlreadyRunning` and does NOT boot a duplicate — so there is no orphan to tear
down (the winner records its pid before releasing the lock, so the loser observes
it). The removed-during-boot orphan teardown is centralized into the same helper
(both call sites previously needed it; restart lacked it).

Validated: full a3s-box-cli lib suite (597) green, fmt + clippy clean on the KVM
server. A cross-process boot serialization fix; the interleaving needs a
multi-process/real-VM harness to exercise (like the other concurrency fixes), but
the flock + re-check-under-lock is self-evidently correct.

NOTE: touches monitor.rs poll_once near PR #144's region — rebase if #144 merges
first (adjacent, non-overlapping edits).

Co-authored-by: Roy Lin <roylin@a3s.box>

Author: ZhiXiao-Lin

src/cli/src/boot.rs

@@ -9,7 +9,7 @@ use a3s_box_runtime::{prom::RuntimeMetrics, NetworkStore, VmManager, VolumeStore
 use std::path::PathBuf;
 
 use crate::commands::common;
-use crate::state::{BoxRecord, HealthCheck};
+use crate::state::{BoxRecord, HealthCheck, StateFile};
 
 /// Result of a successful box boot.
 pub struct BootResult {
@@ -126,6 +126,121 @@ pub fn apply_boot_result(
     }
 }
 
+/// Per-box advisory boot lock.
+///
+/// Serializes boots of the SAME box across processes (the `monitor` daemon vs a
+/// user `restart`/`start`). Without it both can run `boot_from_record` — which
+/// creates a real VM OUTSIDE the state lock — concurrently, and the second
+/// post-boot record write overwrites the first's pid, ORPHANING the first VM
+/// (untracked shim + overlay mount, never reaped). Held across the boot AND the
+/// record write so a waiting actor re-checks AFTER the winner records its pid and
+/// skips booting a duplicate. Lives on a per-box `locks/<box_id>.boot.lock`
+/// sibling; `flock` releases on drop or crash, never stranding the lock.
+struct BootLock {
+    #[cfg(unix)]
+    _file: std::fs::File,
+}
+
+impl BootLock {
+    #[cfg(unix)]
+    fn acquire(box_id: &str) -> std::io::Result<Self> {
+        use std::os::unix::io::AsRawFd;
+        let dir = a3s_box_core::dirs_home().join("locks");
+        std::fs::create_dir_all(&dir)?;
+        let path = dir.join(format!("{box_id}.boot.lock"));
+        let file = std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .open(&path)?;
+        // Blocking exclusive advisory lock; released when `_file` drops.
+        if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 {
+            return Err(std::io::Error::last_os_error());
+        }
+        Ok(Self { _file: file })
+    }
+
+    #[cfg(not(unix))]
+    fn acquire(_box_id: &str) -> std::io::Result<Self> {
+        Ok(Self {})
+    }
+}
+
+/// Outcome of [`boot_and_record`].
+pub enum BootOutcome {
+    /// Booted a VM and recorded it; carries the (possibly updated) restart count.
+    Restarted { restart_count: u32 },
+    /// Another actor already (re)started this box (observed running with a live
+    /// shim under the per-box lock); no VM was booted — a no-op for the caller.
+    AlreadyRunning,
+    /// The record was removed (concurrent `rm`) while the VM booted; the
+    /// just-booted orphan VM has been torn down here.
+    RemovedDuringBoot,
+}
+
+/// Whether `box_id` is currently running with a live, identity-matched shim.
+/// Loads fresh reconciled state, so a recorded-but-dead pid reads as not-live.
+fn box_already_live(box_id: &str) -> Result<bool, Box<dyn std::error::Error>> {
+    let state = StateFile::load_default()?;
+    Ok(state.find_by_id(box_id).is_some_and(|rec| {
+        rec.status == "running"
+            && rec.pid.is_some_and(|p| {
+                crate::process::is_process_alive_with_identity(p, rec.pid_start_time)
+            })
+    }))
+}
+
+/// Boot a box from its record and persist the result, SERIALIZED per box so
+/// concurrent actors (monitor auto-restart vs user restart/start) cannot boot the
+/// same box twice and orphan a VM. Holds a per-box [`BootLock`] across the boot
+/// and the record write; a loser that finds the box already live skips booting.
+pub async fn boot_and_record(
+    record: &BoxRecord,
+    count_update: RestartCountUpdate,
+) -> Result<BootOutcome, Box<dyn std::error::Error>> {
+    let box_id = record.id.clone();
+    let _lock = {
+        let id = box_id.clone();
+        tokio::task::spawn_blocking(move || BootLock::acquire(&id))
+            .await
+            .map_err(|e| -> Box<dyn std::error::Error> {
+                format!("boot lock task failed: {e}").into()
+            })??
+    };
+
+    // Re-check fresh state UNDER the per-box lock: if another actor already booted
+    // this box, skip — do not create a duplicate VM.
+    if box_already_live(&box_id)? {
+        return Ok(BootOutcome::AlreadyRunning);
+    }
+
+    let result = boot_from_record(record).await?;
+    let booted_pid = result.pid;
+    // Persist atomically; `None` if the record was removed (concurrent rm) mid-boot.
+    let restart_count = StateFile::modify(|s| {
+        Ok::<Option<u32>, std::io::Error>(s.find_by_id_mut(&box_id).map(|rec| {
+            apply_boot_result(rec, result, count_update);
+            rec.restart_count
+        }))
+    })?;
+
+    match restart_count {
+        Some(restart_count) => Ok(BootOutcome::Restarted { restart_count }),
+        None => {
+            // The box was removed while booting: the VM we started has no record.
+            // Tear it down so it doesn't leak as an orphan shim + overlay mount.
+            if let Some(pid) = booted_pid {
+                crate::process::graceful_stop(pid, libc::SIGTERM, 5).await;
+            }
+            crate::cleanup::cleanup_removed_box(record);
+            Ok(BootOutcome::RemovedDuringBoot)
+        }
+    }
+    // `_lock` drops here — AFTER the record write — so a waiting actor's re-check
+    // observes our recorded pid.
+}
+
 fn ensure_boot_resources(
     record: &BoxRecord,
 ) -> Result<BootResourceGuard, Box<dyn std::error::Error>> {

src/cli/src/commands/monitor.rs

@@ -306,44 +306,36 @@ async fn poll_once(tracker: &mut BackoffTracker) -> Result<(), Box<dyn std::erro
             println!("{}", restart_log_line(&record, RestartReason::Dead));
         }
 
-        // Attempt restart
-        match boot::boot_from_record(&record).await {
-            Ok(result) => {
-                // Capture the shim pid before `result` is moved into the closure,
-                // so we can tear the VM down if the record vanished mid-boot.
-                let booted_pid = result.pid;
-                // Re-load fresh under the lock and apply only this box's restart
-                // fields. Returns the new restart count, or None if the record is
-                // gone — a concurrent `rm` removed the box while we were booting.
-                let new_count = StateFile::modify(|s| {
-                    let count = s.find_by_id_mut(&box_id).map(|rec| {
-                        boot::apply_boot_result(rec, result, boot::RestartCountUpdate::Increment);
-                        rec.restart_count
-                    });
-                    Ok::<Option<u32>, std::io::Error>(count)
-                })?;
+        // Attempt restart, SERIALIZED per box via a per-box boot lock so a
+        // concurrent user `restart`/`start` and this monitor restart cannot both
+        // boot the same box (the second record write would overwrite the first's
+        // pid, orphaning a VM). The orphan-on-`rm`-during-boot teardown now lives
+        // inside `boot_and_record`.
+        match boot::boot_and_record(&record, boot::RestartCountUpdate::Increment).await {
+            Ok(boot::BootOutcome::Restarted { restart_count }) => {
                 tracker.record_attempt(&box_id);
-                match new_count {
-                    Some(new_count) => println!(
-                        "monitor: box {name} ({short_id}) restarted (count: {new_count})",
-                        name = record.name,
-                        short_id = record.short_id,
-                    ),
-                    None => {
-                        // The box was removed during the restart boot: the VM we
-                        // just started has no record tracking it. Tear it down so
-                        // it doesn't leak as an orphan shim + overlay mount.
-                        eprintln!(
-                            "monitor: box {name} ({short_id}) was removed during restart; tearing down the orphaned VM",
-                            name = record.name,
-                            short_id = record.short_id,
-                        );
-                        if let Some(pid) = booted_pid {
-                            crate::process::graceful_stop(pid, libc::SIGTERM, 5).await;
-                        }
-                        crate::cleanup::cleanup_removed_box(&record);
-                    }
-                }
+                println!(
+                    "monitor: box {name} ({short_id}) restarted (count: {restart_count})",
+                    name = record.name,
+                    short_id = record.short_id,
+                );
+            }
+            Ok(boot::BootOutcome::AlreadyRunning) => {
+                // Another actor (a user restart/start) already brought this box
+                // back under the per-box boot lock — nothing to do.
+                println!(
+                    "monitor: box {name} ({short_id}) already restarted by another actor; skipping",
+                    name = record.name,
+                    short_id = record.short_id,
+                );
+            }
+            Ok(boot::BootOutcome::RemovedDuringBoot) => {
+                tracker.record_attempt(&box_id);
+                eprintln!(
+                    "monitor: box {name} ({short_id}) was removed during restart; tore down the orphaned VM",
+                    name = record.name,
+                    short_id = record.short_id,
+                );
             }
             Err(e) => {
                 tracker.record_attempt(&box_id);

src/cli/src/commands/restart.rs

@@ -89,19 +89,16 @@ async fn restart_one(
     // (image, cmd, dirs) is immutable across the stop, so the in-memory handle is
     // fine to boot from; only the post-boot status write must be atomic.
     let record = resolve::resolve(state, &box_id)?;
-    let result = boot::boot_from_record(record).await?;
-
-    // Persist the boot result atomically; the closure re-resolves by id against
-    // fresh state so it never persists a stale snapshot (and Preserve keeps the
-    // freshly-loaded restart_count).
-    StateFile::modify(move |s| {
-        if let Some(record) = s.find_by_id_mut(&box_id) {
-            boot::apply_boot_result(record, result, boot::RestartCountUpdate::Preserve);
+    // Boot + persist under a per-box boot lock (see boot::boot_and_record): if the
+    // monitor (or another concurrent restart) already brought this box back, skip
+    // rather than boot a duplicate VM that orphans one of the two.
+    match boot::boot_and_record(record, boot::RestartCountUpdate::Preserve).await? {
+        boot::BootOutcome::Restarted { .. } => println!("{name}"),
+        boot::BootOutcome::AlreadyRunning => println!("{name} (already started)"),
+        boot::BootOutcome::RemovedDuringBoot => {
+            return Err(format!("{name} was removed during restart").into());
         }
-        Ok::<(), std::io::Error>(())
-    })?;
-
-    println!("{name}");
+    }
     Ok(())
 }