diff options
Diffstat (limited to 'src/remake/records.rs')
| -rw-r--r-- | src/remake/records.rs | 95 |
1 files changed, 80 insertions, 15 deletions
diff --git a/src/remake/records.rs b/src/remake/records.rs index cd3e5be..283f612 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,9 +8,10 @@ //! * `abortRequested`: written by any node on an operator's request. It is an input to //! the coordinator's decision, not a decision itself. //! -//! Once a remake has expired, its coordinator no longer commits it, and any node may -//! abort it on an operator's request (`api`). Should clocks disagree by more than the -//! grace margin, a commit and such an abort could still be written concurrently. Such +//! Once a remake has expired, its coordinator no longer commits it, and a grace period +//! later any node may abort it on an operator's request (`api`). Should clocks disagree +//! by more than that grace period, a commit and such an abort could still be written +//! concurrently. Such //! a record is *contested*: it reads as committed, and nodes don't switch to it on //! their own. //! @@ -130,7 +131,9 @@ 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>, + /// None for a kind whose base is fixed when it commits (`Decision::Committed`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) base_heads: Option<Vec<String>>, pub(crate) policy: Policy, pub(crate) settle_ms: u64, pub(crate) proposed_at: u64, @@ -181,13 +184,19 @@ fn default_kind() -> String { DEFAULT_KIND.to_owned() } -impl Proposal { - pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> { - self.base_heads - .iter() - .map(|hex| decode_hex32(hex).map(ChangeHash)) - .collect() - } +/// Hex change hashes as hashes; None if one is unreadable. +pub(crate) fn decode_heads(heads: &[String]) -> Option<Vec<ChangeHash>> { + heads + .iter() + .map(|hex| decode_hex32(hex).map(ChangeHash)) + .collect() +} + +/// Hashes as sorted hex, as records keep them. +pub(crate) fn encode_heads(heads: &[ChangeHash]) -> Vec<String> { + let mut hex: Vec<String> = heads.iter().map(ChangeHash::to_string).collect(); + hex.sort(); + hex } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -196,6 +205,9 @@ pub(crate) enum Decision { #[serde(rename_all = "camelCase")] Committed { to_epoch: EpochRecord, + /// The base, when the proposal leaves it to the commit (sorted hex). + #[serde(default, skip_serializing_if = "Option::is_none")] + base_heads: Option<Vec<String>>, snapshot_bytes: u64, decided_at: u64, }, @@ -275,13 +287,39 @@ impl Record { } } + /// The base: the proposal's, or for a kind that fixes it at the commit, the + /// decision's. None until committed then, or if unreadable. + pub(crate) fn base(&self) -> Option<Vec<ChangeHash>> { + let heads = match (&self.proposal.base_heads, &self.decision) { + (Some(heads), _) => heads, + ( + None, + Some(Decision::Committed { + base_heads: Some(heads), + .. + }), + ) => heads, + _ => return None, + }; + decode_heads(heads) + } + /// Everything needed to (re)build this remake's snapshot but the content (which the - /// caller adds, `content::ensure`); None if the proposal is unreadable. + /// caller adds, `content::ensure`); None if the proposal or base is unreadable. pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { + self.snapshot_request_at(id, self.base()?) + } + + /// The same, at `base` (the coordinator fixing a base at the commit). + pub(crate) fn snapshot_request_at( + &self, + id: &str, + base: Vec<ChangeHash>, + ) -> Option<SnapshotRequest> { Some(SnapshotRequest { remake_id: id.to_owned(), previous: self.proposal.from_epoch.to_epoch()?, - base_heads: self.proposal.base_change_hashes()?, + base_heads: base, policy: serde_json::to_value(&self.proposal.policy).ok()?, proposed_at: self.proposal.proposed_at, kind: self.proposal.kind.clone(), @@ -499,7 +537,7 @@ mod tests { Proposal { coordinator: "node-a".into(), from_epoch: Epoch::GENESIS.into(), - base_heads: vec!["ab".repeat(32)], + base_heads: Some(vec!["ab".repeat(32)]), policy: Policy { replay_window: Policy::default_replay_window(), archive_retention: Policy::default_archive_retention(), @@ -535,6 +573,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 123, decided_at: 11_000, }; @@ -552,7 +591,7 @@ mod tests { 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_eq!(record.base().unwrap().len(), 1); assert!(record.abort_requested.is_some()); assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]); assert_eq!(record.proposal.quorum, 2); @@ -608,6 +647,31 @@ mod tests { } #[test] + fn a_base_fixed_at_the_commit() { + let mut record = Record { + proposal: Proposal { + base_heads: None, + ..proposal(10_000) + }, + decision: None, + abort_requested: None, + contested: false, + votes: BTreeMap::new(), + }; + assert_eq!(record.base(), None); + record.decision = Some(Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + base_heads: Some(vec!["ef".repeat(32)]), + snapshot_bytes: 1, + decided_at: 2, + }); + assert_eq!(record.base(), Some(vec![ChangeHash([0xef; 32])])); + } + + #[test] fn disagreeing_decisions_are_contested() { let mut base = AutoCommit::new(); write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); @@ -618,6 +682,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 1, decided_at: 2, }; |