🐛 fix(storage): honor reclaim-guard expiry on admission

A reclaim guard carries a 300-second lease. Arming honors it, so a lapsed
row is available to the next collector, but reference admission read no
clock and rejected a commit whenever a row was present. A confirmed orphan
purge that died between deleting the bytes and disarming its guard left a
row nothing ever clears, and every later publication of that digest failed
with BlobReclaiming for as long as the store lived.

The admission check now takes the store's clock and blocks only while the
lease holds, dropping the lapsed row in the same write transaction it
admits past so an abandoned guard cannot outlive its lease. MetaStore
gained the clock as an injectable field rather than reading wall time at
the call site, which is what lets a test step across a lease boundary
instead of waiting one out.

The release pass in the purge no longer asks the blob store whether the
bytes are present. An expired lease already proves its collector let go,
and requiring absence was exactly what kept a failed delete's guard alive
forever. Dropping the presence probe also removes the digest parse and the
head request that only existed to answer it.

Guarding the guard table's existence keeps the check from creating it, so
publishing a blob reference on a fresh store no longer materializes a
distributed domain table.

Author: gaborbernat

crates/peryx-ha-distributed/src/reclaim_guard.rs

@@ -51,7 +51,7 @@ pub fn purge_orphaned_blobs(
         ));
     }
 
-    release_absent_expired_guards(meta, blobs, now)?;
+    release_expired_guards(meta, now)?;
     let guard = ReclaimGuard {
         expires_at_unix: now.saturating_add(RECLAIM_GUARD_LEASE_SECS),
     };
@@ -84,25 +84,12 @@ pub fn purge_orphaned_blobs(
     Ok(report(selected))
 }
 
-fn release_absent_expired_guards(meta: &MetaStore, blobs: &BlobStorage, now: i64) -> Result<(), OrphanPurgeError> {
-    for (encoded, guard) in meta.reclaim_guards()? {
-        if !guard.is_expired_at(now) {
-            continue;
-        }
-        let digest = Digest::from_hex(&encoded).ok_or_else(|| OrphanPurgeError::Blob {
-            operation: "read orphan reclaim guard",
-            reason: format!("invalid SHA-256 digest {encoded:?}"),
-        })?;
-        let absent = blobs
-            .blocking()
-            .head(&digest)
-            .map_err(|error| OrphanPurgeError::Blob {
-                operation: "inspect guarded blob",
-                reason: error.to_string(),
-            })?
-            .is_none();
-        if absent {
-            meta.compare_and_disarm_reclaim_guard(&encoded, guard)?;
+/// A lapsed lease proves its collector no longer holds the blob, whether or not the bytes survived
+/// the purge that armed it, so the row goes regardless of what the store reports.
+fn release_expired_guards(meta: &MetaStore, now: i64) -> Result<(), OrphanPurgeError> {
+    for (digest, guard) in meta.reclaim_guards()? {
+        if guard.is_expired_at(now) {
+            meta.compare_and_disarm_reclaim_guard(&digest, guard)?;
         }
     }
     Ok(())

crates/peryx-ha-distributed/tests/unit/reclaim_guard_tests.rs

@@ -1,4 +1,6 @@
 use std::collections::BTreeSet;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicI64, Ordering};
 
 use peryx_ha::{ReclaimGuard, ReclaimGuardArm, ReclaimGuardStore as _};
 use peryx_storage::blob::BlobStorage;
@@ -118,30 +120,33 @@ fn expired_owner_is_replaced_before_deletion() {
 }
 
 #[test]
-fn expired_guard_for_an_absent_blob_is_released() {
+fn expired_guard_for_a_present_blob_is_released() {
     let (_directory, _path, meta, blobs) = stores();
-    let digest = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
-    meta.compare_and_arm_reclaim_guards(&[digest], 0, 0, ReclaimGuard { expires_at_unix: 10 })
+    let kept = blobs.blocking().put_bytes(b"kept").unwrap();
+    meta.compare_and_arm_reclaim_guards(&[kept.as_str()], 0, 0, ReclaimGuard { expires_at_unix: 10 })
         .unwrap();
 
-    let report = purge_orphaned_blobs(&meta, &blobs, true, 10, || Ok(BTreeSet::new())).unwrap();
+    let report = purge_orphaned_blobs(&meta, &blobs, true, 10, || {
+        Ok(BTreeSet::from([kept.as_str().to_owned()]))
+    })
+    .unwrap();
 
     assert!(report.blobs.is_empty());
-    assert_eq!(meta.reclaim_guard(digest).unwrap(), None);
+    assert!(blobs.blocking().head(&kept).unwrap().is_some());
+    assert_eq!(meta.reclaim_guard(kept.as_str()).unwrap(), None);
 }
 
 #[test]
-fn invalid_expired_guard_stops_the_purge() {
+fn expired_guard_for_an_absent_blob_is_released() {
     let (_directory, _path, meta, blobs) = stores();
-    meta.compare_and_arm_reclaim_guards(&["invalid"], 0, 0, ReclaimGuard { expires_at_unix: 10 })
+    let digest = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+    meta.compare_and_arm_reclaim_guards(&[digest], 0, 0, ReclaimGuard { expires_at_unix: 10 })
         .unwrap();
 
-    let error = purge_orphaned_blobs(&meta, &blobs, true, 10, || Ok(BTreeSet::new())).unwrap_err();
+    let report = purge_orphaned_blobs(&meta, &blobs, true, 10, || Ok(BTreeSet::new())).unwrap();
 
-    assert_eq!(
-        error.to_string(),
-        "read orphan reclaim guard: invalid SHA-256 digest \"invalid\""
-    );
+    assert!(report.blobs.is_empty());
+    assert_eq!(meta.reclaim_guard(digest).unwrap(), None);
 }
 
 #[test]
@@ -207,24 +212,49 @@ fn deletion_failure_keeps_the_guard_armed() {
 }
 
 #[test]
-fn guarded_blob_inspection_failure_stops_the_purge() {
+fn an_interrupted_purge_stops_rejecting_references_once_its_lease_lapses() {
     let directory = tempfile::tempdir().unwrap();
-    let root = directory.path().join("blobs");
-    std::fs::create_dir_all(root.join("sha256")).unwrap();
-    std::fs::write(root.join("sha256/01"), b"not a directory").unwrap();
-    let meta = MetaStore::open(directory.path().join("peryx.redb")).unwrap();
-    let digest = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
-    meta.compare_and_arm_reclaim_guards(&[digest], 0, 0, ReclaimGuard { expires_at_unix: 10 })
-        .unwrap();
+    let ticks = Arc::new(AtomicI64::new(10));
+    let clock = Arc::clone(&ticks);
+    let meta = MetaStore::open(directory.path().join("peryx.redb"))
+        .unwrap()
+        .with_clock(Arc::new(move || clock.load(Ordering::Relaxed)));
+    let blobs = BlobStorage::filesystem(directory.path().join("blobs"));
+    let orphan = blobs.blocking().put_bytes(b"orphan").unwrap();
+    let orphan_path = purge_orphaned_blobs(&meta, &blobs, false, 10, || Ok(BTreeSet::new()))
+        .unwrap()
+        .blobs
+        .pop()
+        .unwrap()
+        .path;
+    let mut scans = 0;
+    purge_orphaned_blobs(&meta, &blobs, true, 10, || {
+        scans += 1;
+        if scans == 2 {
+            std::fs::remove_file(&orphan_path).unwrap();
+            std::fs::create_dir(&orphan_path).unwrap();
+            std::fs::write(orphan_path.join("entry"), b"occupied").unwrap();
+        }
+        Ok(BTreeSet::new())
+    })
+    .unwrap_err();
+    let lease = ReclaimGuard {
+        expires_at_unix: 10 + RECLAIM_GUARD_LEASE_SECS,
+    };
+    assert_eq!(meta.reclaim_guard(orphan.as_str()).unwrap(), Some(lease));
 
-    let error =
-        purge_orphaned_blobs(&meta, &BlobStorage::filesystem(root), true, 10, || Ok(BTreeSet::new())).unwrap_err();
+    ticks.store(lease.expires_at_unix - 1, Ordering::Relaxed);
+    let rejected = republish(&meta, orphan.as_str()).unwrap_err();
+    ticks.store(lease.expires_at_unix, Ordering::Relaxed);
+    republish(&meta, orphan.as_str()).unwrap();
 
-    assert!(matches!(
-        error,
-        OrphanPurgeError::Blob {
-            operation: "inspect guarded blob",
-            ..
-        }
-    ));
+    assert!(matches!(rejected, MetaError::BlobReclaiming { digest } if digest == orphan.as_str()));
+    assert_eq!(meta.reclaim_guard(orphan.as_str()).unwrap(), None);
+}
+
+fn republish(meta: &MetaStore, digest: &str) -> Result<(), MetaError> {
+    meta.commit_driver_txn(|txn| {
+        txn.reference_blob(digest, 6);
+        Ok::<_, MetaError>(((), vec![b"{}".to_vec()]))
+    })
 }

crates/peryx-storage/src/meta/index.rs

@@ -1,5 +1,6 @@
-use peryx_ha::ArtifactPlacement;
-use redb::ReadableTable as _;
+use peryx_core::Clock;
+use peryx_ha::{ArtifactPlacement, ReclaimGuard};
+use redb::{ReadableTable as _, TableHandle as _};
 
 use super::error::{MetaError, MetaScanError};
 use super::policy_decision::advance_repository_generations;
@@ -402,7 +403,7 @@ impl MetaStore {
         }
         Self::enqueue_webhook_events(&txn, &webhooks).map_err(E::from)?;
         Self::write_artifact_placements(&txn, &placements).map_err(E::from)?;
-        check_blob_reclaim_guard(&txn, expected_serial, &journal).map_err(E::from)?;
+        check_blob_reclaim_guard(&txn, expected_serial, &journal, &self.clock).map_err(E::from)?;
         let journal_commit = commit_journal(&txn, &journal)?;
         advance_repository_generations(&txn, &policy_inputs).map_err(E::from)?;
         if let Some((repository, catalog)) = catalog_generation {
@@ -450,28 +451,47 @@ fn finish_journal(journal: PendingJournal, driver: DriverTxn<'_>) -> Result<Vec<
     }
 }
 
+/// Rejects a reference only while a collector still holds the blob's lease. A lapsed guard is the
+/// residue of a purge that died between arming and disarming, so it is dropped here rather than left
+/// to reject every later publication of that digest.
 fn check_blob_reclaim_guard(
     txn: &redb::WriteTransaction,
     expected_serial: Option<u64>,
     journal: &[JournalEntry],
+    clock: &Clock,
 ) -> Result<(), MetaError> {
     let Some(last) = journal.last() else {
         return Ok(());
     };
     if expected_serial.is_some() || last.blobs.is_empty() {
         return Ok(());
     }
-    let guards = txn.open_table(BLOB_RECLAIM_GUARD)?;
+    if !table_exists(txn, &BLOB_RECLAIM_GUARD)? {
+        return Ok(());
+    }
+    let now = clock();
+    let mut guards = txn.open_table(BLOB_RECLAIM_GUARD)?;
     for blob in &last.blobs {
-        if guards.get(blob.sha256.as_str())?.is_some() {
+        let held = guards.get(blob.sha256.as_str())?.map(|value| ReclaimGuard {
+            expires_at_unix: value.value(),
+        });
+        let Some(guard) = held else {
+            continue;
+        };
+        if !guard.is_expired_at(now) {
             return Err(MetaError::BlobReclaiming {
                 digest: blob.sha256.clone(),
             });
         }
+        guards.remove(blob.sha256.as_str())?;
     }
     Ok(())
 }
 
+fn table_exists(txn: &redb::WriteTransaction, table: &impl redb::TableHandle) -> Result<bool, MetaError> {
+    Ok(txn.list_tables()?.any(|handle| handle.name() == table.name()))
+}
+
 fn update_policy_generation(txn: &redb::WriteTransaction, repository: &str, catalog: u64) -> Result<(), MetaError> {
     let mut generations = txn.open_table(POLICY_INPUT_GENERATION)?;
     let mut generation = generations

crates/peryx-storage/src/meta/mod.rs

@@ -3,7 +3,9 @@
 
 use std::path::Path;
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 
+use peryx_core::Clock;
 use redb::{Database, ReadOnlyDatabase, ReadableDatabase as _, TableDefinition};
 
 mod analytics;
@@ -220,9 +222,19 @@ impl DriverBatch {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct MetaStore {
     db: Arc<MetaDatabase>,
+    clock: Clock,
+}
+
+impl std::fmt::Debug for MetaStore {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("MetaStore")
+            .field("db", &self.db)
+            .finish_non_exhaustive()
+    }
 }
 
 enum MetaDatabase {
@@ -326,9 +338,17 @@ impl MetaStore {
         txn.commit()?;
         Ok(Self {
             db: Arc::new(MetaDatabase::ReadWrite(db)),
+            clock: system_clock(),
         })
     }
 
+    /// Replaces the wall clock that decides whether a blob's reclaim-guard lease has lapsed.
+    #[must_use]
+    pub fn with_clock(mut self, clock: Clock) -> Self {
+        self.clock = clock;
+        self
+    }
+
     /// Validates distributed persistence without creating domain tables.
     ///
     /// # Errors
@@ -365,6 +385,7 @@ impl MetaStore {
     pub fn open_existing(path: impl AsRef<Path>) -> Result<Self, MetaError> {
         Ok(Self {
             db: Arc::new(MetaDatabase::ReadWrite(Database::open(path)?)),
+            clock: system_clock(),
         })
     }
 
@@ -375,6 +396,16 @@ impl MetaStore {
     pub fn open_existing_read_only(path: impl AsRef<Path>) -> Result<Self, MetaError> {
         Ok(Self {
             db: Arc::new(MetaDatabase::ReadOnly(ReadOnlyDatabase::open(path)?)),
+            clock: system_clock(),
         })
     }
 }
+
+/// A host clock that predates the epoch must not stop the store from opening.
+fn system_clock() -> Clock {
+    Arc::new(|| {
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_or(0, |elapsed| i64::try_from(elapsed.as_secs()).unwrap_or(i64::MAX))
+    })
+}

crates/peryx-storage/tests/unit/meta/fault.rs

@@ -119,12 +119,14 @@ pub fn create(
     write.commit().unwrap();
     MetaStore {
         db: Arc::new(MetaDatabase::ReadWrite(database)),
+        clock: super::system_clock(),
     }
 }
 
 pub fn reopen(inner: &Arc<InMemoryBackend>, fault: &Arc<Fault>) -> MetaStore {
     MetaStore {
         db: Arc::new(MetaDatabase::ReadWrite(database(inner, fault))),
+        clock: super::system_clock(),
     }
 }
 

crates/peryx-storage/tests/unit/tests/meta/integration_tests.rs

@@ -178,6 +178,23 @@ fn test_reclaim_guard_first_write_creates_only_its_table() {
     assert_distributed_tables(&path, &["blob_reclaim_guard"]);
 }
 
+#[test]
+fn test_referencing_a_blob_creates_no_reclaim_guard_table() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("peryx.redb");
+    let store = MetaStore::open(&path).unwrap();
+    store
+        .commit_driver_txn(|txn| {
+            txn.put("ref/1", b"points-here")?;
+            txn.reference_blob("sha256:blob", 4);
+            Ok::<_, MetaError>(((), vec![b"{}".to_vec()]))
+        })
+        .unwrap();
+    drop(store);
+
+    assert_distributed_tables(&path, &["journal", "journal_blobs", "journal_mutations"]);
+}
+
 #[test]
 fn test_ingress_first_write_creates_only_its_tables() {
     let dir = tempfile::tempdir().unwrap();