fix(exec): propagate skips transitively under keep-going; UTF8-safe display_name; reap ephemeral images; fix cache clean (#130)

Author: markovejnovic

crates/hm-exec/src/local/runner/mod.rs

@@ -11,7 +11,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use anyhow::Result;
-use hm_plugin_protocol::{ExecutorInput, StepResult};
+use hm_plugin_protocol::{ExecutorInput, SnapshotRef, StepResult};
 use tokio_util::sync::CancellationToken;
 
 use crate::local::archive::ArchiveStore;
@@ -55,6 +55,21 @@ pub trait StepRunner: Send + Sync + fmt::Debug {
         ctx: &StepContext,
         input: ExecutorInput,
     ) -> Pin<Box<dyn Future<Output = Result<StepResult>> + Send + '_>>;
+
+    /// Reap transient snapshots once a run has finished.
+    ///
+    /// Ephemeral (uncached) leaf steps commit a snapshot purely so a
+    /// downstream `BuildsIn` child can restore from it; nothing else holds a
+    /// reference and the cache registry never tracks them. The scheduler
+    /// collects every such snapshot and calls this at run end (best-effort)
+    /// so they don't leak in the backend store. The default is a no-op for
+    /// runners that produce no reapable snapshots.
+    fn reap_snapshots<'a>(
+        &'a self,
+        _snapshots: Vec<SnapshotRef>,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
+        Box::pin(async {})
+    }
 }
 
 /// Maps runner names to [`StepRunner`] implementations.

crates/hm-exec/src/local/runner/vm.rs

@@ -49,6 +49,21 @@ impl StepRunner for VmRunner {
         let vm = Arc::clone(&self.vm);
         Box::pin(async move { run_step_vm(&vm, &ctx, input).await })
     }
+
+    fn reap_snapshots<'a>(
+        &'a self,
+        snapshots: Vec<SnapshotRef>,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
+        let vm = Arc::clone(&self.vm);
+        Box::pin(async move {
+            for snap in snapshots {
+                let id = SnapshotId::new(snap.0);
+                if let Err(e) = vm.remove_snapshot(&id).await {
+                    tracing::warn!(snapshot = %id, error = %e, "failed to reap ephemeral snapshot");
+                }
+            }
+        })
+    }
 }
 
 #[tracing::instrument(skip(vm, ctx), fields(step_key = %input.step.key))]

crates/hm-exec/src/local/scheduler.rs

@@ -60,6 +60,12 @@ struct StepOutcome {
     /// `None` only for steps short-circuited because a predecessor failed
     /// or the build was cancelled before they could run.
     summary: Option<StepResultSummary>,
+    /// Set when this step did not complete successfully — it failed, timed
+    /// out, was cancelled, or was itself skipped. Descendants gate on this
+    /// (not on `exit_code`) so a skip propagates transitively: a skipped
+    /// step reports `exit_code == 0`, so the exit code alone cannot
+    /// distinguish "passed" from "skipped" and the cascade would break.
+    failed_or_skipped: bool,
 }
 
 type StepFuture = futures::future::Shared<BoxFuture<'static, StepOutcome>>;
@@ -186,8 +192,12 @@ pub(crate) async fn run(
             let pred_outcomes: Vec<StepOutcome> =
                 join_all(preds.iter().map(|(_, f)| f.clone())).await;
 
-            // Early exit if any predecessor failed or the build was cancelled.
-            if cancel.is_cancelled() || pred_outcomes.iter().any(|o| o.exit_code != 0) {
+            // Early exit if any predecessor failed/was skipped, or the build
+            // was cancelled. Gating on `failed_or_skipped` (not `exit_code`)
+            // is what makes the skip propagate transitively: a skipped
+            // predecessor reports `exit_code == 0`, so an exit-code-only gate
+            // would let a skipped step's descendants run anyway.
+            if cancel.is_cancelled() || pred_outcomes.iter().any(|o| o.failed_or_skipped) {
                 let status = if cancel.is_cancelled() {
                     StepStatus::Canceled
                 } else {
@@ -203,6 +213,7 @@ pub(crate) async fn run(
                         exit_code: None,
                         duration_ms: 0,
                     }),
+                    failed_or_skipped: true,
                 };
             }
 
@@ -249,6 +260,7 @@ pub(crate) async fn run(
                             exit_code: Some(1),
                             duration_ms: 0,
                         }),
+                        failed_or_skipped: true,
                     }
                 }
             }
@@ -285,6 +297,22 @@ pub(crate) async fn run(
     let outcomes: Vec<StepOutcome> = join_all(pending).await;
     let any_failed = outcomes.iter().any(|o| o.exit_code != 0);
 
+    // Reap ephemeral leaf snapshots. Uncached steps commit an `ephemeral:*`
+    // image for downstream container lineage; the cache registry never tracks
+    // them, so once the run is over nothing else will. Collect every such
+    // snapshot the steps produced and ask the default runner to remove them
+    // (best-effort — failures are logged, not fatal).
+    let ephemeral: Vec<SnapshotRef> = outcomes
+        .iter()
+        .filter_map(|o| o.snapshot.clone())
+        .filter(|s| s.0.starts_with("ephemeral:"))
+        .collect();
+    if !ephemeral.is_empty()
+        && let Some(runner) = runner_registry.resolve(None)
+    {
+        runner.reap_snapshots(ephemeral).await;
+    }
+
     // Derive the overall verdict. Timeout wins (it also fired cancellation);
     // then cancellation; then any failed step; otherwise the build passed.
     let status = if timed_out {
@@ -363,10 +391,13 @@ async fn execute_step(
     let step_key = step_wire.key.clone();
     let display_name = step_wire.label.clone().unwrap_or_else(|| {
         let cmd = step_wire.cmd.trim();
-        if cmd.len() <= 40 {
+        if cmd.chars().count() <= 40 {
             cmd.to_owned()
         } else {
-            format!("{}…", &cmd[..39])
+            // Truncate on a char boundary, not a byte offset: `&cmd[..39]`
+            // panics if byte 39 falls inside a multibyte UTF-8 sequence.
+            let truncated: String = cmd.chars().take(39).collect();
+            format!("{truncated}…")
         }
     });
     let env_map = transition.env;
@@ -476,6 +507,7 @@ async fn execute_step(
                             exit_code: Some(124),
                             duration_ms: dur_ms,
                         }),
+                        failed_or_skipped: true,
                     });
                 }
             }
@@ -522,6 +554,7 @@ async fn execute_step(
                     exit_code: Some(sr.exit_code),
                     duration_ms: dur_ms,
                 }),
+                failed_or_skipped: sr.exit_code != 0,
             })
         }
         Err(e) => {

crates/hm-vm/src/registry.rs

@@ -154,6 +154,25 @@ impl ImageRegistry {
         snapshot.map(SnapshotId::new)
     }
 
+    /// Return every stored snapshot ID.
+    ///
+    /// Used by `hm cache clean` to remove the backing backend images before
+    /// the registry DB is deleted — without this the images orphan beyond
+    /// recovery by key. Order is unspecified.
+    #[must_use]
+    pub fn all_snapshot_ids(&self) -> Vec<SnapshotId> {
+        let Ok(conn) = self.conn.lock() else {
+            return Vec::new();
+        };
+        let Ok(mut stmt) = conn.prepare("SELECT snapshot_id FROM snapshots") else {
+            return Vec::new();
+        };
+        stmt.query_map([], |row| row.get::<_, String>(0).map(SnapshotId::new))
+            .ok()
+            .map(|rows| rows.filter_map(Result::ok).collect())
+            .unwrap_or_default()
+    }
+
     /// Returns the number of cached entries.
     #[must_use]
     pub fn len(&self) -> u64 {
@@ -315,6 +334,23 @@ mod tests {
         assert_eq!(got, Some(SnapshotId::new("snap-persist")));
     }
 
+    #[test]
+    fn all_snapshot_ids_returns_every_entry() {
+        let (reg, _dir) = open_temp(10);
+        assert!(reg.all_snapshot_ids().is_empty());
+
+        reg.put("k1", &SnapshotId::new("forever-a"));
+        reg.put("k2", &SnapshotId::new("forever-b"));
+
+        let mut ids: Vec<String> = reg
+            .all_snapshot_ids()
+            .into_iter()
+            .map(|s| s.to_string())
+            .collect();
+        ids.sort();
+        assert_eq!(ids, vec!["forever-a".to_string(), "forever-b".to_string()]);
+    }
+
     #[test]
     fn invalidate_returns_removed_snapshot() {
         let (reg, _dir) = open_temp(10);

crates/hm-vm/src/vm.rs

@@ -8,7 +8,8 @@ use tracing::{instrument, warn};
 use crate::backend::VmBackend;
 use crate::registry::ImageRegistry;
 use crate::types::{
-    Action, CachingPolicy, ExecutionResult, ImageSource, OutputSink, SnapshotLabel, VmConfig,
+    Action, CachingPolicy, ExecutionResult, ImageSource, OutputSink, SnapshotId, SnapshotLabel,
+    VmConfig,
 };
 
 /// High-level orchestrator that drives the VM lifecycle.
@@ -81,6 +82,20 @@ impl HmVm {
         result
     }
 
+    /// Remove a snapshot from the backend store.
+    ///
+    /// Used to reap ephemeral (uncached) leaf snapshots once a run finishes —
+    /// `CachingPolicy::None` commits a transient `ephemeral:*` image purely for
+    /// downstream container lineage, and nothing in the registry ever evicts
+    /// it. The scheduler reaps these explicitly at run end.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the backend fails to remove the snapshot.
+    pub async fn remove_snapshot(&self, snapshot: &SnapshotId) -> Result<()> {
+        self.backend.remove_snapshot(snapshot).await
+    }
+
     /// Inner lifecycle: inject, exec, snapshot. Separated so the caller
     /// can guarantee `vm.destroy()` runs regardless of outcome.
     async fn run_in_vm(

crates/hm/src/commands/cache/clean.rs

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use hm_vm::VmBackend as _;
 
 /// # Errors
 /// Returns an error if workspace cache removal fails.
@@ -21,6 +22,15 @@ pub async fn handle_clean() -> Result<i32> {
     let db_cleaned = if let Some(cache_dir) = hm_util::dirs::hm_cache_dir() {
         let db_path = cache_dir.join("registry.db");
         if db_path.exists() {
+            // Remove the backing Docker images BEFORE deleting registry.db.
+            // The registry is the only index from a cache key to its tagged
+            // image (`forever-*`, etc.); once the DB is gone the images can't
+            // be located by key, and `docker image prune` only reclaims
+            // *dangling* images, so a tagged snapshot survives it. So we
+            // enumerate the registry, remove each image via the Docker
+            // backend (best-effort), then drop the DB.
+            remove_registered_images(&db_path).await;
+
             std::fs::remove_file(&db_path)?;
             tracing::info!(path = %db_path.display(), "removed VM image registry");
             true
@@ -31,19 +41,60 @@ pub async fn handle_clean() -> Result<i32> {
         false
     };
 
-    if db_cleaned {
-        tracing::warn!(
-            "Docker images from previous runs may still exist — run `docker image prune` to reclaim disk"
-        );
-    }
-
     if !ws_cleaned && !db_cleaned {
         tracing::info!("nothing to clean");
     }
 
     Ok(0)
 }
 
+/// Remove every Docker image tracked by the registry at `db_path`.
+///
+/// Best-effort: a missing Docker daemon or an already-deleted image is logged
+/// and skipped, never fatal — `clean` must still delete the registry DB so the
+/// cache index is reset.
+async fn remove_registered_images(db_path: &std::path::Path) {
+    // Capacity here is irrelevant — we only read existing rows, never insert.
+    let registry = match hm_vm::ImageRegistry::open(db_path, u64::MAX) {
+        Ok(r) => r,
+        Err(e) => {
+            tracing::warn!(error = %e, "could not open image registry; skipping image removal");
+            return;
+        }
+    };
+
+    let snapshots = registry.all_snapshot_ids();
+    if snapshots.is_empty() {
+        return;
+    }
+
+    let backend = match hm_vm::docker::DockerBackend::connect() {
+        Ok(b) => b,
+        Err(e) => {
+            tracing::warn!(
+                error = %e,
+                "could not connect to Docker; {} cached image(s) may remain — remove them with `docker image rm`",
+                snapshots.len(),
+            );
+            return;
+        }
+    };
+
+    let mut removed = 0usize;
+    for snap in &snapshots {
+        match backend.remove_snapshot(snap).await {
+            Ok(()) => removed += 1,
+            Err(e) => {
+                tracing::warn!(image = %snap, error = %e, "failed to remove cached image");
+            }
+        }
+    }
+    tracing::info!(
+        "removed {removed} of {} cached Docker image(s)",
+        snapshots.len()
+    );
+}
+
 fn dir_size(path: &std::path::Path) -> u64 {
     fn walk(p: &std::path::Path) -> u64 {
         std::fs::read_dir(p)