diff options
Diffstat (limited to 'src/compaction/records.rs')
| -rw-r--r-- | src/compaction/records.rs | 428 |
1 files changed, 0 insertions, 428 deletions
diff --git a/src/compaction/records.rs b/src/compaction/records.rs deleted file mode 100644 index ccd52c3..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::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>, -} - -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 - ); - } -} |