diff options
Diffstat (limited to 'src/compaction/records.rs')
| -rw-r--r-- | src/compaction/records.rs | 398 |
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 + ); + } +} |