rename compaction to remake

1 files changed, 0 insertions, 428 deletions

diff --git a/src/compaction/records.rs b/src/compaction/records.rs
deleted file mode 100644
index f533ee2..0000000
--- a/src/compaction/records.rs
+++ /dev/null
@@ -1,428 +0,0 @@
-//! 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::document::replica::Epoch;
-use crate::document::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>,
-}
-
-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 {
-    /// How stale a returning node's changes may be and still be replayed automatically;
-    /// beyond it they are quarantined.
-    #[serde(default = "Policy::default_replay_window")]
-    pub(crate) replay_window: Retention,
-    /// How long nodes keep the old epoch's document.
-    #[serde(default = "Policy::default_archive_retention")]
-    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()?,
-            proposed_at: self.proposal.proposed_at,
-        })
-    }
-
-    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::*;
-
-    #[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(),
-            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
-        );
-    }
-}