add epoch switch

1 files changed, 398 insertions, 0 deletions

diff --git a/src/compaction/records.rs b/src/compaction/records.rs
new file mode 100644
index 0000000..ce2b2a5
--- /dev/null
+++ b/src/compaction/records.rs
@@ -0,0 +1,398 @@
+//! Compaction records: `_continuum/compactions/<id>` in the replicated document.
+//!
+//! A record has up to three parts, each with a single kind of writer, so concurrent
+//! writes never fight over the same key:
+//!
+//! * `proposal`: written once, by the coordinator.
+//! * `decision`: written once, by the coordinator only (committed or aborted).
+//! * `abortRequested`: written by any node on an operator's request. It is an input to
+//!   the coordinator's decision, not a decision itself.
+//!
+//! A proposal whose coordinator never decides (crash, partition) is reported as
+//! `expired` once its settle period plus a grace margin has passed. Expiry is derived,
+//! never written, so a late but valid commit can't collide with someone's abort.
+//!
+//! Records edited into an unreadable shape through the store API are skipped with a
+//! warning.
+
+use automerge::{AutoCommit, ChangeHash};
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+use super::snapshot::{SnapshotRequest, COMPACTIONS, META_ROOT};
+use crate::replica::Epoch;
+use crate::store::{self, StoreError};
+
+/// One day, for retention defaults.
+const DAY_SECS: u64 = 24 * 60 * 60;
+
+pub(crate) fn unix_millis() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+// --- record types ------------------------------------------------------------
+
+/// An epoch in a record: generation plus hex-encoded snapshot hash.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct EpochRecord {
+    pub(crate) generation: u64,
+    pub(crate) id: String,
+}
+
+impl From<Epoch> for EpochRecord {
+    fn from(epoch: Epoch) -> Self {
+        Self {
+            generation: epoch.generation,
+            id: epoch.id_hex(),
+        }
+    }
+}
+
+impl EpochRecord {
+    pub(crate) fn to_epoch(&self) -> Option<Epoch> {
+        Some(Epoch {
+            generation: self.generation,
+            id: decode_hex32(&self.id)?,
+        })
+    }
+}
+
+/// How long something is kept. A value is kept until every given limit has passed.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Retention {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) epochs: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) seconds: Option<u64>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Policy {
+    /// How stale a returning node's changes may be and still be replayed automatically;
+    /// beyond it they are quarantined.
+    pub(crate) replay_window: Retention,
+    /// How long nodes keep the old epoch's document.
+    pub(crate) archive_retention: Retention,
+}
+
+impl Policy {
+    pub(crate) fn default_replay_window() -> Retention {
+        Retention {
+            epochs: Some(1),
+            seconds: Some(7 * DAY_SECS),
+        }
+    }
+
+    pub(crate) fn default_archive_retention() -> Retention {
+        Retention {
+            epochs: Some(2),
+            seconds: Some(30 * DAY_SECS),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Proposal {
+    pub(crate) coordinator: String,
+    pub(crate) from_epoch: EpochRecord,
+    /// Hex change hashes, sorted. The proposal record itself is not part of the base.
+    pub(crate) base_heads: Vec<String>,
+    pub(crate) policy: Policy,
+    pub(crate) settle_ms: u64,
+    pub(crate) proposed_at: u64,
+    pub(crate) settle_until: u64,
+    /// Peers that must stay active during the settle period.
+    pub(crate) peers_considered: Vec<String>,
+}
+
+impl Proposal {
+    pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> {
+        self.base_heads
+            .iter()
+            .map(|hex| decode_hex32(hex).map(ChangeHash))
+            .collect()
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "outcome", rename_all = "camelCase")]
+pub(crate) enum Decision {
+    #[serde(rename_all = "camelCase")]
+    Committed {
+        to_epoch: EpochRecord,
+        snapshot_bytes: u64,
+        decided_at: u64,
+    },
+    #[serde(rename_all = "camelCase")]
+    Aborted { reason: String, decided_at: u64 },
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct AbortRequest {
+    pub(crate) by: String,
+    pub(crate) at: u64,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Record {
+    pub(crate) proposal: Proposal,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) decision: Option<Decision>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) abort_requested: Option<AbortRequest>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) enum Phase {
+    /// Settling; the coordinator has not decided yet.
+    Proposed,
+    /// No decision long after the settle period: the coordinator is gone.
+    Expired,
+    Committed,
+    Aborted,
+}
+
+impl Phase {
+    /// Still able to become (or already) the next epoch.
+    pub(crate) fn is_open(self) -> bool {
+        matches!(self, Phase::Proposed | Phase::Committed)
+    }
+
+    pub(crate) fn is_final(self) -> bool {
+        !matches!(self, Phase::Proposed)
+    }
+}
+
+impl Record {
+    /// The epoch this compaction committed to, if committed and readable.
+    pub(crate) fn committed_epoch(&self) -> Option<Epoch> {
+        match &self.decision {
+            Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(),
+            _ => None,
+        }
+    }
+
+    /// Everything needed to (re)build this compaction's snapshot; None if the proposal
+    /// is unreadable.
+    pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
+        Some(SnapshotRequest {
+            compaction_id: id.to_owned(),
+            previous: self.proposal.from_epoch.to_epoch()?,
+            base_heads: self.proposal.base_change_hashes()?,
+            policy: serde_json::to_value(&self.proposal.policy).ok()?,
+        })
+    }
+
+    pub(crate) fn phase(&self, now_ms: u64, grace_ms: u64) -> Phase {
+        match &self.decision {
+            Some(Decision::Committed { .. }) => Phase::Committed,
+            Some(Decision::Aborted { .. }) => Phase::Aborted,
+            None if now_ms > self.proposal.settle_until.saturating_add(grace_ms) => Phase::Expired,
+            None => Phase::Proposed,
+        }
+    }
+}
+
+// --- document access ---------------------------------------------------------
+
+fn records_dir() -> Vec<String> {
+    vec![META_ROOT.to_owned(), COMPACTIONS.to_owned()]
+}
+
+pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
+    let mut path = records_dir();
+    path.push(id.to_owned());
+    if !part.is_empty() {
+        path.push(part.to_owned());
+    }
+    path
+}
+
+/// The record with this ID, if it exists and is readable.
+pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
+    let value = match store::read_path(doc, &record_path(id, "")) {
+        Ok(Some(value)) => value,
+        Ok(None) => return None,
+        Err(error) => {
+            warn!(compaction = %id, error = %error, "Failed to read compaction record");
+            return None;
+        }
+    };
+    match serde_json::from_value(value) {
+        Ok(record) => Some(record),
+        Err(error) => {
+            warn!(compaction = %id, error = %error, "Skipping unreadable compaction record");
+            None
+        }
+    }
+}
+
+/// All readable records.
+pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
+    store::keys_at(doc, &records_dir())
+        .unwrap_or_default()
+        .into_iter()
+        .filter_map(|id| load(doc, &id).map(|record| (id, record)))
+        .collect()
+}
+
+fn write_part<T: Serialize>(
+    doc: &mut AutoCommit,
+    id: &str,
+    part: &str,
+    value: &T,
+) -> Result<(), StoreError> {
+    let json = serde_json::to_value(value).map_err(anyhow::Error::from)?;
+    store::write_path(doc, &record_path(id, part), &json)
+}
+
+pub(crate) fn write_proposal(
+    doc: &mut AutoCommit,
+    id: &str,
+    proposal: &Proposal,
+) -> Result<(), StoreError> {
+    write_part(doc, id, "proposal", proposal)
+}
+
+pub(crate) fn write_decision(
+    doc: &mut AutoCommit,
+    id: &str,
+    decision: &Decision,
+) -> Result<(), StoreError> {
+    write_part(doc, id, "decision", decision)
+}
+
+pub(crate) fn write_abort_request(
+    doc: &mut AutoCommit,
+    id: &str,
+    request: &AbortRequest,
+) -> Result<(), StoreError> {
+    write_part(doc, id, "abortRequested", request)
+}
+
+fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
+    if hex.len() != 64 || !hex.is_ascii() {
+        return None;
+    }
+    let mut out = [0u8; 32];
+    for (i, byte) in out.iter_mut().enumerate() {
+        *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).ok()?;
+    }
+    Some(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn proposal(settle_until: u64) -> Proposal {
+        Proposal {
+            coordinator: "node-a".into(),
+            from_epoch: Epoch::GENESIS.into(),
+            base_heads: vec!["ab".repeat(32)],
+            policy: Policy {
+                replay_window: Policy::default_replay_window(),
+                archive_retention: Policy::default_archive_retention(),
+            },
+            settle_ms: 1_000,
+            proposed_at: settle_until.saturating_sub(1_000),
+            settle_until,
+            peers_considered: vec!["node-b".into()],
+        }
+    }
+
+    #[test]
+    fn round_trip_through_the_document() {
+        let mut doc = AutoCommit::new();
+        let p = proposal(10_000);
+        write_proposal(&mut doc, "c-1", &p).unwrap();
+        write_abort_request(
+            &mut doc,
+            "c-1",
+            &AbortRequest {
+                by: "x".into(),
+                at: 5,
+            },
+        )
+        .unwrap();
+        let decision = Decision::Committed {
+            to_epoch: EpochRecord {
+                generation: 1,
+                id: "cd".repeat(32),
+            },
+            snapshot_bytes: 123,
+            decided_at: 11_000,
+        };
+        write_decision(&mut doc, "c-1", &decision).unwrap();
+        doc.commit();
+
+        let all = load_all(&doc);
+        assert_eq!(all.len(), 1);
+        let (id, record) = &all[0];
+        assert_eq!(id, "c-1");
+        assert_eq!(record.proposal.policy, p.policy);
+        assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1);
+        assert!(record.abort_requested.is_some());
+        assert_eq!(record.phase(0, 0), Phase::Committed);
+    }
+
+    #[test]
+    fn phases() {
+        let mut record = Record {
+            proposal: proposal(10_000),
+            decision: None,
+            abort_requested: None,
+        };
+        assert_eq!(record.phase(10_000, 500), Phase::Proposed);
+        assert_eq!(record.phase(10_500, 500), Phase::Proposed);
+        assert_eq!(record.phase(10_501, 500), Phase::Expired);
+        record.decision = Some(Decision::Aborted {
+            reason: "test".into(),
+            decided_at: 1,
+        });
+        assert_eq!(record.phase(0, 0), Phase::Aborted);
+    }
+
+    #[test]
+    fn unreadable_records_are_skipped() {
+        let mut doc = AutoCommit::new();
+        store::write_path(
+            &mut doc,
+            &record_path("broken", ""),
+            &serde_json::json!({ "proposal": "not an object" }),
+        )
+        .unwrap();
+        write_proposal(&mut doc, "fine", &proposal(1)).unwrap();
+        doc.commit();
+        let all = load_all(&doc);
+        assert_eq!(all.len(), 1);
+        assert_eq!(all[0].0, "fine");
+    }
+
+    #[test]
+    fn epoch_record_round_trip() {
+        let epoch = Epoch {
+            generation: 3,
+            id: [0x5a; 32],
+        };
+        assert_eq!(EpochRecord::from(epoch).to_epoch(), Some(epoch));
+        assert_eq!(
+            EpochRecord {
+                generation: 1,
+                id: "zz".into()
+            }
+            .to_epoch(),
+            None
+        );
+    }
+}