replace snapshots by quarantine & replay

1 files changed, 48 insertions, 22 deletions

diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs
index 86caba9..0a89e23 100644
--- a/src/compaction/snapshot.rs
+++ b/src/compaction/snapshot.rs
@@ -46,6 +46,8 @@ pub(crate) struct SnapshotParams<'a> {
     pub(crate) base_heads: &'a [ChangeHash],
     /// Retention policy from the compaction request; stored verbatim (canonicalized).
     pub(crate) policy: &'a JsonValue,
+    /// When the compaction was proposed (unix ms); recorded for replay windows.
+    pub(crate) proposed_at: u64,
 }
 
 pub(crate) struct Snapshot {
@@ -79,6 +81,7 @@ pub(crate) struct SnapshotRequest {
     pub(crate) previous: Epoch,
     pub(crate) base_heads: Vec<ChangeHash>,
     pub(crate) policy: JsonValue,
+    pub(crate) proposed_at: u64,
 }
 
 impl SnapshotRequest {
@@ -88,6 +91,7 @@ impl SnapshotRequest {
             previous: self.previous,
             base_heads: &self.base_heads,
             policy: &self.policy,
+            proposed_at: self.proposed_at,
         }
     }
 }
@@ -107,13 +111,16 @@ pub(crate) async fn build_detached(
     .map_err(|error| SnapshotError::Task(error.to_string()))?
 }
 
-/// The protocol's view of an epoch, read at its snapshot change.
+/// The protocol's view of an epoch, as recorded in its snapshot.
 #[derive(Clone, Debug)]
 pub(crate) struct EpochMetadata {
     pub(crate) compaction_id: String,
     pub(crate) previous: Epoch,
+    /// Highest sequence number per actor of `previous` contained in the snapshot.
     pub(crate) clock: BTreeMap<String, u64>,
     pub(crate) policy: Policy,
+    /// When the compaction was proposed (unix ms): the point the cluster moved on.
+    pub(crate) proposed_at: u64,
 }
 
 #[derive(serde::Deserialize)]
@@ -124,6 +131,34 @@ struct MetadataRecord {
     previous: EpochRecord,
     clock: BTreeMap<String, u64>,
     policy: Policy,
+    #[serde(default)]
+    proposed_at: u64,
+}
+
+fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, String> {
+    let value = store::read_path(doc, &epoch_metadata_path(generation))
+        .map_err(|error| error.to_string())?
+        .ok_or_else(|| format!("no epoch metadata for generation {generation}"))?;
+    let record: MetadataRecord = serde_json::from_value(value)
+        .map_err(|error| format!("invalid epoch metadata: {error}"))?;
+    if record.generation != generation {
+        return Err(format!(
+            "epoch metadata names generation {} instead of {generation}",
+            record.generation
+        ));
+    }
+    let previous = record
+        .previous
+        .to_epoch()
+        .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
+
+    Ok(EpochMetadata {
+        compaction_id: record.compaction_id,
+        previous,
+        clock: record.clock,
+        policy: record.policy,
+        proposed_at: record.proposed_at,
+    })
 }
 
 /// Checks that `doc` is a document of `epoch`: its history starts with exactly the
@@ -144,28 +179,17 @@ pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<Epoc
     }
 
     let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?;
-    let value = store::read_path(&at_snapshot, &epoch_metadata_path(epoch.generation))
-        .map_err(|error| error.to_string())?
-        .ok_or_else(|| format!("no epoch metadata for generation {}", epoch.generation))?;
-    let record: MetadataRecord = serde_json::from_value(value)
-        .map_err(|error| format!("invalid epoch metadata: {error}"))?;
-    if record.generation != epoch.generation {
-        return Err(format!(
-            "epoch metadata names generation {} instead of {}",
-            record.generation, epoch.generation
-        ));
-    }
-    let previous = record
-        .previous
-        .to_epoch()
-        .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
+    parse_metadata(&at_snapshot, epoch.generation)
+}
 
-    Ok(EpochMetadata {
-        compaction_id: record.compaction_id,
-        previous,
-        clock: record.clock,
-        policy: record.policy,
-    })
+/// Metadata of an earlier generation, as the current document holds it. Snapshots
+/// carry the metadata of all earlier epochs forward (only compaction records stay
+/// behind), so a node many epochs behind can still find the epoch that followed its own.
+pub(crate) fn read_epoch_metadata(
+    doc: &AutoCommit,
+    generation: u64,
+) -> Result<EpochMetadata, String> {
+    parse_metadata(doc, generation)
 }
 
 /// Path of the metadata record for an epoch generation.
@@ -212,6 +236,7 @@ pub(crate) fn build(
         "baseHeads": base_heads,
         "clock": clock,
         "policy": params.policy,
+        "proposedAt": params.proposed_at,
     }));
     store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?;
 
@@ -288,6 +313,7 @@ mod tests {
             previous: Epoch::GENESIS,
             base_heads: heads,
             policy,
+            proposed_at: 0,
         }
     }