replace snapshots by quarantine & replay

1 files changed, 28 insertions, 0 deletions

diff --git a/src/compaction/records.rs b/src/compaction/records.rs
index 6150a75..d61a7f0 100644
--- a/src/compaction/records.rs
+++ b/src/compaction/records.rs
@@ -71,6 +71,17 @@ pub(crate) struct Retention {
     pub(crate) seconds: Option<u64>,
 }
 
+impl Retention {
+    /// True while at least one given limit has not passed: `epochs` epochs and
+    /// `age_ms` milliseconds after the fact.
+    pub(crate) fn covers(&self, epochs: u64, age_ms: u64) -> bool {
+        self.epochs.is_some_and(|limit| epochs <= limit)
+            || self
+                .seconds
+                .is_some_and(|limit| age_ms <= limit.saturating_mul(1000))
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct Policy {
@@ -192,6 +203,7 @@ impl Record {
             previous: self.proposal.from_epoch.to_epoch()?,
             base_heads: self.proposal.base_change_hashes()?,
             policy: serde_json::to_value(&self.proposal.policy).ok()?,
+            proposed_at: self.proposal.proposed_at,
         })
     }
 
@@ -297,6 +309,22 @@ fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn retention_covers_until_every_limit_passed() {
+        let both = Retention {
+            epochs: Some(1),
+            seconds: Some(10),
+        };
+        assert!(both.covers(1, 60_000)); // epochs still within
+        assert!(both.covers(5, 5_000)); // time still within
+        assert!(!both.covers(2, 10_001));
+        let none = Retention {
+            epochs: None,
+            seconds: None,
+        };
+        assert!(!none.covers(0, 0));
+    }
+
     fn proposal(settle_until: u64) -> Proposal {
         Proposal {
             coordinator: "node-a".into(),