Implement exact bootstrap recovery and artifact GC

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

@@ -55,6 +55,12 @@ pub enum ReclaimOutcome {
     ReplacementPreserved,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PreparedArtifactGcResult {
+    pub prepared_artifact_id: PreparedArtifactId,
+    pub outcome: ReclaimOutcome,
+}
+
 #[derive(Clone, Debug)]
 pub struct PreparedLowerStore {
     root: Arc<File>,
@@ -270,6 +276,55 @@ impl PreparedLowerStore {
         }
     }
 
+    /// Recovers the complete catalog snapshot of interrupted preparation operations.
+    ///
+    /// # Errors
+    /// Stops at the first foreign, mismatched, unreadable, or changed operation without treating
+    /// it as owned absence.
+    pub fn recover_preparations(&self) -> Result<Vec<(PrepareOperationId, PrepareRecovery)>> {
+        let operations = self.catalog.snapshot()?.prepare_operations;
+        operations
+            .iter()
+            .map(|operation| {
+                self.recover_prepare(operation)
+                    .map(|outcome| (operation.operation_id, outcome))
+            })
+            .collect()
+    }
+
+    /// Reclaims only the caller-selected exact catalog artifacts.
+    ///
+    /// # Errors
+    /// Refuses an unknown target, duplicate target, any lease, catalog drift, or physical
+    /// ownership mismatch. Reclaiming rows resume their existing exact operation.
+    pub fn gc_artifacts(
+        &self,
+        selected: &[PreparedArtifactId],
+    ) -> Result<Vec<PreparedArtifactGcResult>> {
+        let mut unique = std::collections::HashSet::new();
+        if selected.iter().any(|id| !unique.insert(*id)) {
+            bail!("prepared-artifact GC target is duplicated");
+        }
+        let snapshot = self.catalog.snapshot()?;
+        let mut results = Vec::with_capacity(selected.len());
+        for id in selected {
+            let artifact = snapshot
+                .prepared_artifacts
+                .iter()
+                .find(|artifact| artifact.prepared_artifact_id == *id)
+                .context("prepared-artifact GC target is absent")?;
+            let outcome = match artifact.state {
+                ArtifactState::Available => self.reclaim(artifact)?,
+                ArtifactState::Reclaiming => self.resume_reclaim(artifact)?,
+            };
+            results.push(PreparedArtifactGcResult {
+                prepared_artifact_id: *id,
+                outcome,
+            });
+        }
+        Ok(results)
+    }
+
     /// Moves an exact zero-lease artifact to quarantine, deletes it, and closes reclaim intent.
     ///
     /// # Errors