diff options
Diffstat (limited to 'src/remake/records.rs')
| -rw-r--r-- | src/remake/records.rs | 166 |
1 files changed, 87 insertions, 79 deletions
diff --git a/src/remake/records.rs b/src/remake/records.rs index 720aea1..0c4c444 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -28,7 +28,6 @@ use std::collections::BTreeMap; -use automerge::{AutoCommit, ChangeHash}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; use tracing::warn; @@ -39,28 +38,14 @@ use crate::clock::UnixMillis; use crate::document::meta::{self, REMAKES, VOTES}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; +use crate::document::{ChangeHash, ReadDoc, Transactable}; /// One day, for retention defaults. const DAY_MS: u64 = 24 * 60 * 60 * 1000; // --- 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(), - } - } -} +pub(crate) use crate::document::replica::EpochRecord; impl EpochRecord { pub(crate) fn to_epoch(&self) -> Option<Epoch> { @@ -137,6 +122,11 @@ pub(crate) struct Proposal { /// interval if absent. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) switch_delay_ms: Option<u64>, + /// The coordinator's grace period (ms, its `dormant_after`): every node judges this + /// remake's expiry, and when others may abort it, by it, whatever its own keepalive + /// (`Record::grace`). Absent in records from before: each node's own. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) grace_ms: Option<u64>, /// Members whose votes count (sorted), fixed at proposal time (`vote`). pub(crate) electorate: Vec<String>, /// Votes needed to commit: a majority of the electorate. @@ -326,14 +316,23 @@ impl Record { }) } - pub(crate) fn phase(&self, now: UnixMillis, grace_ms: u64) -> Phase { + /// The phase at `now`, expiry judged by the record's grace period (`grace`; `own`, + /// this node's, for a record without one). + pub(crate) fn phase(&self, now: UnixMillis, own: u64) -> Phase { match &self.decision { Some(Decision::Committed { .. }) => Phase::Committed, Some(Decision::Aborted { .. }) => Phase::Aborted, - None if now > self.proposal.settle_until.plus_ms(grace_ms) => Phase::Expired, + None if now > self.proposal.settle_until.plus_ms(self.grace(own)) => Phase::Expired, None => Phase::Proposed, } } + + /// The grace period this remake is judged by (ms): its coordinator's, the same on + /// every node, so nodes with different keepalives agree on when it expired; `own` + /// for a record from before it was written down. + pub(crate) fn grace(&self, own: u64) -> u64 { + self.proposal.grace_ms.unwrap_or(own) + } } // --- document access --------------------------------------------------------- @@ -352,13 +351,13 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> { } /// The record with this ID, if it exists and is readable, with its votes. -pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> { +pub(crate) fn load(doc: &impl ReadDoc, id: &str) -> Option<Record> { let record = load_record(doc, id)?; Some(with_votes(id, record, &load_register(doc))) } /// All readable records, with their votes. -pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> { +pub(crate) fn load_all(doc: &impl ReadDoc) -> Vec<(String, Record)> { let register = load_register(doc); store::keys_at(doc, &records_dir()) .unwrap_or_default() @@ -381,7 +380,7 @@ fn with_votes(id: &str, mut record: Record, register: &Register) -> Record { record } -fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> { +fn load_record(doc: &impl ReadDoc, id: &str) -> Option<Record> { let value = match store::read_path(doc, &record_path(id, "")) { Ok(Some(value)) => value, Ok(None) => return None, @@ -405,7 +404,7 @@ fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> { } /// A commit among concurrent decisions of remake `id` that disagree, if they do. -fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> { +fn contested_decision(doc: &impl ReadDoc, id: &str) -> Option<Decision> { let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision")) .ok()? .into_iter() @@ -433,7 +432,7 @@ fn register_path(voter: &str) -> Vec<String> { } /// The votes register; unreadable entries are skipped with a warning. -pub(crate) fn load_register(doc: &AutoCommit) -> Register { +pub(crate) fn load_register(doc: &impl ReadDoc) -> Register { store::keys_at(doc, &meta::path(&[VOTES])) .unwrap_or_default() .into_iter() @@ -461,7 +460,7 @@ pub(crate) fn load_register(doc: &AutoCommit) -> Register { } fn write_part<T: Serialize>( - doc: &mut AutoCommit, + doc: &mut impl Transactable, id: &str, part: &str, value: &T, @@ -471,7 +470,7 @@ fn write_part<T: Serialize>( } pub(crate) fn write_proposal( - doc: &mut AutoCommit, + doc: &mut impl Transactable, id: &str, proposal: &Proposal, ) -> Result<(), StoreError> { @@ -479,7 +478,7 @@ pub(crate) fn write_proposal( } pub(crate) fn write_decision( - doc: &mut AutoCommit, + doc: &mut impl Transactable, id: &str, decision: &Decision, ) -> Result<(), StoreError> { @@ -487,7 +486,7 @@ pub(crate) fn write_decision( } pub(crate) fn write_abort_request( - doc: &mut AutoCommit, + doc: &mut impl Transactable, id: &str, request: &AbortRequest, ) -> Result<(), StoreError> { @@ -495,7 +494,11 @@ pub(crate) fn write_abort_request( } /// Writes `voter`'s entry in the votes register (replacing its earlier one). -pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> { +pub(crate) fn write_vote( + doc: &mut impl Transactable, + voter: &str, + vote: &Vote, +) -> Result<(), StoreError> { let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?; store::write_path(doc, ®ister_path(voter), &json) } @@ -514,6 +517,8 @@ pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> { #[cfg(test)] mod tests { use super::*; + use crate::document::Document; + use crate::document::testing::change; #[test] fn retention_covers_until_every_limit_passed() { @@ -573,6 +578,7 @@ mod tests { proposed_at: UnixMillis(settle_until.saturating_sub(1_000)), settle_until: UnixMillis(settle_until), switch_delay_ms: None, + grace_ms: None, electorate: vec!["node-a".into(), "node-b".into()], quorum: 2, kind: DEFAULT_KIND.to_owned(), @@ -583,18 +589,8 @@ mod tests { #[test] fn round_trip_through_the_document() { - let mut doc = AutoCommit::new(); + let mut doc = Document::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: UnixMillis(5), - }, - ) - .unwrap(); let decision = Decision::Committed { to_epoch: EpochRecord { generation: 1, @@ -604,14 +600,25 @@ mod tests { snapshot_bytes: 123, decided_at: UnixMillis(11_000), }; - write_decision(&mut doc, "c-1", &decision).unwrap(); let vote = |remake: &str| Vote { remake: remake.into(), at: UnixMillis(7), }; - write_vote(&mut doc, "node-b", &vote("c-1")).unwrap(); - write_vote(&mut doc, "node-c", &vote("other")).unwrap(); - doc.commit(); + change(&mut doc, |doc| { + write_proposal(doc, "c-1", &p).unwrap(); + write_abort_request( + doc, + "c-1", + &AbortRequest { + by: "x".into(), + at: UnixMillis(5), + }, + ) + .unwrap(); + write_decision(doc, "c-1", &decision).unwrap(); + write_vote(doc, "node-b", &vote("c-1")).unwrap(); + write_vote(doc, "node-c", &vote("other")).unwrap(); + }); let all = load_all(&doc); assert_eq!(all.len(), 1); @@ -627,19 +634,16 @@ mod tests { #[test] fn concurrent_votes_are_kept_apart() { - let mut a = AutoCommit::new(); + let mut a = Document::new(); let vote = |remake: &str, at: u64| Vote { remake: remake.into(), at: UnixMillis(at), }; - write_vote(&mut a, "node-a", &vote("x", 1)).unwrap(); - a.commit(); + change(&mut a, |a| write_vote(a, "node-a", &vote("x", 1)).unwrap()); let mut b = a.fork(); // node-b votes in both, never seeing its other vote (a lost vote). - write_vote(&mut a, "node-b", &vote("x", 2)).unwrap(); - write_vote(&mut b, "node-b", &vote("y", 3)).unwrap(); - a.commit(); - b.commit(); + change(&mut a, |a| write_vote(a, "node-b", &vote("x", 2)).unwrap()); + change(&mut b, |b| write_vote(b, "node-b", &vote("y", 3)).unwrap()); a.merge(&mut b).unwrap(); let register = load_register(&a); @@ -649,8 +653,7 @@ mod tests { assert_eq!(twice, [vote("x", 2), vote("y", 3)]); // Seen, a vote is replaced. - write_vote(&mut a, "node-b", &vote("z", 4)).unwrap(); - a.commit(); + change(&mut a, |a| write_vote(a, "node-b", &vote("z", 4)).unwrap()); assert_eq!(load_register(&a)["node-b"], [vote("z", 4)]); } @@ -666,6 +669,10 @@ mod tests { assert_eq!(record.phase(UnixMillis(10_000), 500), Phase::Proposed); assert_eq!(record.phase(UnixMillis(10_500), 500), Phase::Proposed); assert_eq!(record.phase(UnixMillis(10_501), 500), Phase::Expired); + // The coordinator's grace period, written down, wins over this node's. + record.proposal.grace_ms = Some(2_000); + assert_eq!(record.phase(UnixMillis(10_501), 500), Phase::Proposed); + assert_eq!(record.phase(UnixMillis(12_001), 500), Phase::Expired); record.decision = Some(Decision::Aborted { reason: "test".into(), decided_at: UnixMillis(1), @@ -700,9 +707,10 @@ mod tests { #[test] fn disagreeing_decisions_are_contested() { - let mut base = AutoCommit::new(); - write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); - base.commit(); + let mut base = Document::new(); + change(&mut base, |base| { + write_proposal(base, "c-1", &proposal(10_000)).unwrap() + }); let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork()); let committed = Decision::Committed { to_epoch: EpochRecord { @@ -713,21 +721,20 @@ mod tests { snapshot_bytes: 1, decided_at: UnixMillis(2), }; - write_decision(&mut a, "c-1", &committed).unwrap(); - write_decision( - &mut b, - "c-1", - &Decision::Aborted { - reason: "test".into(), - decided_at: UnixMillis(3), - }, - ) - .unwrap(); + change(&mut a, |a| write_decision(a, "c-1", &committed).unwrap()); + change(&mut b, |b| { + write_decision( + b, + "c-1", + &Decision::Aborted { + reason: "test".into(), + decided_at: UnixMillis(3), + }, + ) + .unwrap() + }); // The same decision, concurrently: they agree. - write_decision(&mut c, "c-1", &committed).unwrap(); - a.commit(); - b.commit(); - c.commit(); + change(&mut c, |c| write_decision(c, "c-1", &committed).unwrap()); let mut agreeing = a.fork(); agreeing.merge(&mut c).unwrap(); assert!(!load(&agreeing, "c-1").unwrap().contested); @@ -740,15 +747,16 @@ mod tests { #[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 mut doc = Document::new(); + change(&mut doc, |doc| { + store::write_path( + doc, + &record_path("broken", ""), + &serde_json::json!({ "proposal": "not an object" }), + ) + .unwrap(); + write_proposal(doc, "fine", &proposal(1)).unwrap(); + }); let all = load_all(&doc); assert_eq!(all.len(), 1); assert_eq!(all[0].0, "fine"); |