fix(stack): deduplicate image pulls to prevent same-image replica rootfs race

When multiple replicas share the same image (e.g. 3x alpine:latest), the
executor previously pulled + assembled rootfs concurrently in thread::scope,
causing layer extraction corruption. Now pulls each unique image once serially
before entering the parallel block, which only runs create_in_sandbox.

Also improves runtime container cleanup: stack containers skip host-side rootfs
deletion (avoids VirtioFS dcache staleness on recreate), stale overlays are
cleaned up in-guest before re-creating, and oci_delete falls back to the
shared stack VM when the per-container handle has been removed.

Re-enables replicated_service E2E tests (previously blocked on vz-3d1).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: lightsofapollo

crates/vz-oci-macos/src/runtime.rs

@@ -208,10 +208,35 @@ impl Runtime {
         }
 
         // Best-effort OCI delete via guest runtime if VM is still up.
-        if let Some(vm) = self.vm_handles.lock().await.remove(id) {
-            let _ = vm.oci_delete(id.to_string(), true).await;
+        // Try the per-container handle first; fall back to the shared stack VM
+        // (the per-container handle may have been removed by stop_container).
+        let vm = self.vm_handles.lock().await.remove(id);
+        let stack_id = self.container_stack.lock().await.remove(id);
+        if let Some(vm) = vm {
+            match vm.oci_delete(id.to_string(), true).await {
+                Ok(_) => {
+                    tracing::debug!(container_id = %id, "remove_container: oci_delete via vm_handle succeeded")
+                }
+                Err(e) => {
+                    tracing::warn!(container_id = %id, error = %e, "remove_container: oci_delete via vm_handle failed")
+                }
+            }
+        } else if let Some(sid) = &stack_id {
+            if let Some(vm) = self.stack_vms.lock().await.get(sid) {
+                match vm.oci_delete(id.to_string(), true).await {
+                    Ok(_) => {
+                        tracing::debug!(container_id = %id, stack_id = %sid, "remove_container: oci_delete via stack_vm succeeded")
+                    }
+                    Err(e) => {
+                        tracing::warn!(container_id = %id, stack_id = %sid, error = %e, "remove_container: oci_delete via stack_vm failed")
+                    }
+                }
+            } else {
+                tracing::warn!(container_id = %id, stack_id = %sid, "remove_container: stack_vm not found");
+            }
+        } else {
+            tracing::debug!(container_id = %id, "remove_container: no vm_handle or stack_id, skipping oci_delete");
         }
-        self.container_stack.lock().await.remove(id);
         self.active_lifecycle.lock().await.remove(id);
 
         self.container_store.remove(id).map_err(OciError::from)?;
@@ -272,23 +297,35 @@ impl Runtime {
         let lifecycle = self.active_lifecycle.lock().await.remove(id);
 
         // Best-effort OCI delete.
-        let _ = vm.oci_delete(id.to_string(), true).await;
+        match vm.oci_delete(id.to_string(), true).await {
+            Ok(_) => tracing::debug!(container_id = %id, "stop_container: oci_delete succeeded"),
+            Err(e) => {
+                tracing::warn!(container_id = %id, error = %e, "stop_container: oci_delete failed (best-effort)")
+            }
+        }
 
         // Only tear down the VM if the container does NOT belong to a shared stack VM.
         let is_stack_container = self.container_stack.lock().await.contains_key(id);
         if !is_stack_container {
             let _ = vm.stop().await;
         }
         self.vm_handles.lock().await.remove(id);
-        self.container_stack.lock().await.remove(id);
+        // Keep container_stack entry so remove_container can find the stack VM
+        // for a retry oci_delete if the best-effort delete above failed.
 
         // Shut down port forwarding for this container.
         if let Some(pf) = self.port_forwards.lock().await.remove(id) {
             pf.shutdown().await;
         }
 
-        if let Some(rootfs_path) = container.rootfs_path.take() {
-            let _ = fs::remove_dir_all(rootfs_path);
+        // Only remove rootfs for non-stack containers. For stack containers the
+        // shared VM's VirtioFS cache holds stale metadata after host-side deletion,
+        // causing recreates to fail (overlay sees empty lowerdir). The rootfs will
+        // be cleaned up by remove_container or overwritten by a subsequent create.
+        if !is_stack_container {
+            if let Some(rootfs_path) = container.rootfs_path.take() {
+                let _ = fs::remove_dir_all(rootfs_path);
+            }
         }
 
         container.host_pid = None;
@@ -1010,6 +1047,11 @@ impl Runtime {
             .oci_create(oci_container_id.clone(), bundle_guest_path.clone())
             .await
         {
+            tracing::error!(
+                container_id = %oci_container_id,
+                error = %err,
+                "step 4 FAILED: oci_create"
+            );
             container.status = ContainerStatus::Stopped { exit_code: -1 };
             container.stopped_unix_secs = Some(current_unix_secs());
             container.host_pid = None;
@@ -2559,6 +2601,25 @@ async fn setup_guest_container_overlay(
     let guest_rootfs_path = format!("{container_overlay}/merged");
     let log_dir = container_log_dir(container_id);
 
+    // Clean up any stale overlay from a previous container with the same ID
+    // (e.g. during recreate). Best-effort: unmount merged overlay, then the
+    // tmpfs backing, then remove the directory tree.  Invalidate the VirtioFS
+    // dcache so the kernel re-reads host-side changes (the rootfs may have
+    // been deleted + reassembled on the host during recreate).
+    let cleanup_cmd = format!(
+        "umount {container_overlay}/merged 2>/dev/null; \
+         umount {container_overlay} 2>/dev/null; \
+         rm -rf {container_overlay}; \
+         echo 2 > /proc/sys/vm/drop_caches 2>/dev/null || true"
+    );
+    let _ = vm
+        .exec_capture(
+            "sh".to_string(),
+            vec!["-c".to_string(), cleanup_cmd],
+            Duration::from_secs(5),
+        )
+        .await;
+
     let overlay_cmd = format!(
         "mkdir -p {container_overlay} && \
          mount -t tmpfs tmpfs {container_overlay} && \

crates/vz-stack/src/executor.rs

@@ -836,17 +836,42 @@ impl<R: ContainerRuntime> StackExecutor<R> {
                 }
             }
 
-            if prepared.len() <= 1 {
-                // Single service — execute inline, no thread overhead.
-                for prep in prepared {
+            // Deduplicate image pulls: pull each unique image once serially
+            // before entering the parallel container creation phase. This avoids
+            // concurrent layer extraction races when multiple replicas share an image.
+            let mut pulled_images: HashSet<String> = HashSet::new();
+            let mut pull_failed: HashSet<String> = HashSet::new();
+            for prep in &prepared {
+                if pulled_images.contains(&prep.image) || pull_failed.contains(&prep.image) {
+                    continue;
+                }
+                info!(image = %prep.image, "pulling image (deduplicated)");
+                if let Err(e) = self.runtime.pull(&prep.image) {
+                    error!(image = %prep.image, error = %e, "image pull failed");
+                    pull_failed.insert(prep.image.clone());
+                } else {
+                    pulled_images.insert(prep.image.clone());
+                }
+            }
+
+            // Partition prepared creates: those whose image pull failed go straight
+            // to the error path; the rest proceed to parallel container creation.
+            let (ok_prepared, failed_prepared): (Vec<_>, Vec<_>) =
+                prepared.into_iter().partition(|p| pulled_images.contains(&p.image));
+
+            for prep in failed_prepared {
+                let full_name = prep.full_name();
+                let msg = format!("image pull failed for {}", prep.image);
+                self.mark_failed(spec, &full_name, &msg)?;
+                result.failed += 1;
+                result.errors.push((full_name, msg));
+            }
+
+            if ok_prepared.len() <= 1 {
+                // Single container — execute inline, no thread overhead.
+                for prep in ok_prepared {
                     let full_name = prep.full_name();
                     info!(service = %full_name, image = %prep.image, "creating container");
-                    if let Err(e) = self.runtime.pull(&prep.image) {
-                        self.mark_failed(spec, &full_name, &e.to_string())?;
-                        result.failed += 1;
-                        result.errors.push((full_name, e.to_string()));
-                        continue;
-                    }
                     let create_result =
                         self.runtime
                             .create_in_sandbox(&spec.name, &prep.image, prep.run_config);
@@ -863,9 +888,9 @@ impl<R: ContainerRuntime> StackExecutor<R> {
                     }
                 }
             } else {
-                // Parallel pull + create for multiple services at the same level.
-                // Extract full names (with replica index) before moving prepared.
-                let full_names: Vec<String> = prepared.iter().map(|p| p.full_name()).collect();
+                // Parallel create for multiple containers at the same level.
+                // Images are already pulled; only create_in_sandbox runs in threads.
+                let full_names: Vec<String> = ok_prepared.iter().map(|p| p.full_name()).collect();
                 info!(
                     services = ?full_names,
                     "creating {} containers in parallel",
@@ -875,13 +900,11 @@ impl<R: ContainerRuntime> StackExecutor<R> {
                 let runtime = &self.runtime;
                 let stack_name = &spec.name;
                 let outcomes: Vec<Result<String, StackError>> = std::thread::scope(|s| {
-                    let handles: Vec<_> = prepared
+                    let handles: Vec<_> = ok_prepared
                         .into_iter()
                         .map(|prep| {
                             let full_name = prep.full_name();
                             s.spawn(move || -> Result<String, StackError> {
-                                info!(service = %full_name, image = %prep.image, "pulling image");
-                                runtime.pull(&prep.image)?;
                                 info!(service = %full_name, image = %prep.image, "creating container");
                                 runtime.create_in_sandbox(
                                     stack_name,
@@ -2025,7 +2048,7 @@ mod tests {
         let result = executor.execute(&spec, &actions).unwrap();
         assert_eq!(result.errors.len(), 1);
         assert_eq!(result.errors[0].0, "web");
-        assert!(result.errors[0].1.contains("mock pull failure"));
+        assert!(result.errors[0].1.contains("image pull failed"));
     }
 
     // ── Port tracking tests ──

crates/vz-stack/tests/stack_e2e.rs

@@ -2339,7 +2339,6 @@ services:
 
 /// Deploy a service with replicas=3 and verify 3 running containers with distinct IDs.
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "requires Apple Silicon + Linux kernel artifacts"]
 async fn replicated_service_creates_multiple_containers() {
     if !require_virtualization_entitlement() {
         return;
@@ -2364,6 +2363,10 @@ services:
     assert_eq!(spec.services[0].resources.replicas, 3);
 
     let bridge = OciContainerRuntime::new(&oci_data);
+
+    // Pre-pull image so parallel replica creation doesn't race on layer extraction.
+    bridge.pull("alpine:latest").unwrap();
+
     let store = StateStore::open(&db_path).unwrap();
     let mut executor = StackExecutor::new(bridge, store, tmp.path());
 
@@ -2427,7 +2430,6 @@ services:
 
 /// Deploy replicas=3, then redeploy with replicas=1 and verify scale-down.
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "requires Apple Silicon + Linux kernel artifacts"]
 async fn replicated_service_scale_down() {
     if !require_virtualization_entitlement() {
         return;
@@ -2449,6 +2451,10 @@ services:
 
     let spec3 = parse_compose(yaml_3, "scale-e2e").unwrap();
     let bridge = OciContainerRuntime::new(&oci_data);
+
+    // Pre-pull image so parallel replica creation doesn't race on layer extraction.
+    bridge.pull("alpine:latest").unwrap();
+
     let store = StateStore::open(&db_path).unwrap();
     let mut executor = StackExecutor::new(bridge, store, tmp.path());