Serialize startup prepared artifact reclamation

crates/lazaret-host-coordinator/src/recovery/reconciliation.rs

@@ -232,20 +232,8 @@ where
         .recover_preparations()
         .map_err(|error| error.to_string())?;
     failpoint(RecoveryBoundary::ArtifactGc, GenerationId::new(0))?;
-    let catalog_snapshot = catalog.snapshot().map_err(|error| error.to_string())?;
-    let selected = catalog_snapshot
-        .prepared_artifacts
-        .iter()
-        .filter(|artifact| {
-            !catalog_snapshot
-                .artifact_leases
-                .iter()
-                .any(|lease| lease.prepared_artifact_id == artifact.prepared_artifact_id)
-        })
-        .map(|artifact| artifact.prepared_artifact_id)
-        .collect::<Vec<_>>();
     prepared
-        .gc_artifacts(&selected)
+        .gc_unleased_artifacts()
         .map_err(|error| error.to_string())?;
     Ok(())
 }

crates/lazaret-prepared-lower/src/store.rs

@@ -13,7 +13,9 @@ use lazaret_core::identity::{
     DeviceId, HostPath, IdMapDigest, InodeId, PolicyId, PrepareOperationId, PreparedArtifactId,
     SelectedManifestDigest, SourceIdentity,
 };
-use lazaret_core::records::{ArtifactState, PrepareOperation, PrepareStageState, PreparedArtifact};
+use lazaret_core::records::{
+    ArtifactState, PrepareOperation, PrepareStageState, PreparedArtifact, StorageCatalog,
+};
 use lazaret_linux::filesystem::{
     ExclusiveFileLock, SharedFileLock, try_exclusive_flock, try_shared_flock,
 };
@@ -522,12 +524,14 @@ impl PreparedLowerStore {
         &self,
         selected: &[PreparedArtifactId],
     ) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
-        self.gc_artifacts_inner(selected)
+        let physical_lock = self.physical_lock()?;
+        self.gc_artifacts_inner(selected, &physical_lock)
     }
 
     fn gc_artifacts_inner(
         &self,
         selected: &[PreparedArtifactId],
+        physical_lock: &ExclusiveFileLock,
     ) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
         let mut unique = std::collections::HashSet::new();
         if selected.iter().any(|id| !unique.insert(*id)) {
@@ -539,6 +543,40 @@ impl PreparedLowerStore {
             .catalog
             .snapshot()
             .map_err(PreparedLowerError::catalog)?;
+        self.gc_artifacts_from_snapshot(selected, &snapshot, physical_lock)
+    }
+
+    /// Reclaims the exact unleased artifacts observed while holding physical authority.
+    ///
+    /// # Errors
+    /// Refuses catalog drift, leases, physical ownership mismatch, foreign paths, and
+    /// interrupted reclaim without exact durable evidence.
+    pub fn gc_unleased_artifacts(&self) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
+        let physical_lock = self.physical_lock()?;
+        let snapshot = self
+            .catalog
+            .snapshot()
+            .map_err(PreparedLowerError::catalog)?;
+        let selected = snapshot
+            .prepared_artifacts
+            .iter()
+            .filter(|artifact| {
+                !snapshot
+                    .artifact_leases
+                    .iter()
+                    .any(|lease| lease.prepared_artifact_id == artifact.prepared_artifact_id)
+            })
+            .map(|artifact| artifact.prepared_artifact_id)
+            .collect::<Vec<_>>();
+        self.gc_artifacts_from_snapshot(&selected, &snapshot, &physical_lock)
+    }
+
+    fn gc_artifacts_from_snapshot(
+        &self,
+        selected: &[PreparedArtifactId],
+        snapshot: &StorageCatalog,
+        physical_lock: &ExclusiveFileLock,
+    ) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
         let mut results = Vec::with_capacity(selected.len());
         for id in selected {
             let artifact = snapshot
@@ -548,8 +586,8 @@ impl PreparedLowerStore {
                 .context("prepared-artifact GC target is absent")
                 .map_err(PreparedLowerError::reclaim)?;
             let outcome = match artifact.state {
-                ArtifactState::Available => self.reclaim(artifact)?,
-                ArtifactState::Reclaiming => self.resume_reclaim(artifact)?,
+                ArtifactState::Available => self.reclaim_inner(artifact, physical_lock)?,
+                ArtifactState::Reclaiming => self.resume_reclaim_locked(artifact, physical_lock)?,
             };
             results.push(PreparedArtifactGcResult {
                 prepared_artifact_id: *id,
@@ -566,7 +604,15 @@ impl PreparedLowerStore {
     /// Refuses leases, forged catalog rows, physical identity mismatch, foreign paths, and
     /// filesystem or catalog failures.
     pub fn reclaim(&self, expected: &PreparedArtifact) -> PreparedLowerResult<ReclaimOutcome> {
-        let _physical_lock = self.physical_lock()?;
+        let physical_lock = self.physical_lock()?;
+        self.reclaim_inner(expected, &physical_lock)
+    }
+
+    fn reclaim_inner(
+        &self,
+        expected: &PreparedArtifact,
+        _physical_lock: &ExclusiveFileLock,
+    ) -> PreparedLowerResult<ReclaimOutcome> {
         let artifact = self
             .open_relative(&artifact_path(expected.prepared_artifact_id))
             .map_err(PreparedLowerError::reclaim)?
@@ -598,7 +644,15 @@ impl PreparedLowerStore {
         &self,
         reclaiming: &PreparedArtifact,
     ) -> PreparedLowerResult<ReclaimOutcome> {
-        let _physical_lock = self.physical_lock()?;
+        let physical_lock = self.physical_lock()?;
+        self.resume_reclaim_locked(reclaiming, &physical_lock)
+    }
+
+    fn resume_reclaim_locked(
+        &self,
+        reclaiming: &PreparedArtifact,
+        _physical_lock: &ExclusiveFileLock,
+    ) -> PreparedLowerResult<ReclaimOutcome> {
         let _claim = self
             .lock_reclaiming_artifact(reclaiming)
             .map_err(PreparedLowerError::reclaim)?;

crates/lazaret-prepared-lower/tests/prepared_lower/publication.rs

@@ -449,6 +449,60 @@ fn selected_gc_is_idempotent_only_at_the_durable_reclaim_boundary() {
     assert!(fixture.store.gc_artifacts(&[id]).is_err());
 }
 
+#[test]
+fn concurrent_unleased_gc_serializes_observation_with_reclaim() {
+    let fixture = Fixture::new();
+    let artifact = fixture
+        .store
+        .prepare(&fixture.request(
+            75,
+            &[(
+                layer(&[Entry {
+                    path: "payload",
+                    kind: EntryType::Regular,
+                    data: b"concurrent-gc",
+                    target: None,
+                    mode: 0o644,
+                }]),
+                OCI_TAR,
+            )],
+        ))
+        .unwrap();
+    let (_, first) = fixture.independent_store();
+    let (_, second) = fixture.independent_store();
+    let barrier = Arc::new(Barrier::new(3));
+    let first_barrier = Arc::clone(&barrier);
+    let first = thread::spawn(move || {
+        first_barrier.wait();
+        first.gc_unleased_artifacts()
+    });
+    let second_barrier = Arc::clone(&barrier);
+    let second = thread::spawn(move || {
+        second_barrier.wait();
+        second.gc_unleased_artifacts()
+    });
+
+    barrier.wait();
+    let results = [
+        first.join().unwrap().unwrap(),
+        second.join().unwrap().unwrap(),
+    ];
+    assert_eq!(results.iter().map(Vec::len).sum::<usize>(), 1);
+    assert_eq!(
+        results.iter().flatten().next().unwrap().outcome,
+        ReclaimOutcome::Removed
+    );
+    assert!(
+        fixture
+            .catalog
+            .snapshot()
+            .unwrap()
+            .prepared_artifacts
+            .is_empty()
+    );
+    assert!(!fixture.artifact_root(&artifact).exists());
+}
+
 #[test]
 fn selected_artifact_excludes_gc_until_the_claim_is_released() {
     let fixture = Fixture::new();