From e6cb4be03510f0c7db88e4e16a3df62964bcd067 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 24 Oct 2026 22:08:31 +0200 Subject: allow superseding expired remakes --- src/remake/records.rs | 89 +++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 86 insertions(+), 3 deletions(-) (limited to 'src/remake/records.rs') diff --git a/src/remake/records.rs b/src/remake/records.rs index 37af491..cd3e5be 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,6 +8,12 @@ //! * `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 +//! a record is *contested*: it reads as committed, and nodes don't switch to it on +//! their own. +//! //! Votes are kept apart, in the votes register `_continuum/votes/`: one entry per //! member for the whole epoch, written by that member only (`vote`). Reading a record //! fills in the votes that name it. @@ -224,6 +230,14 @@ pub(crate) struct Record { pub(crate) decision: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) abort_requested: Option, + /// Decisions written concurrently disagree; `decision` is then the commit. Not + /// stored: found when the record is read. + #[serde( + default, + skip_deserializing, + skip_serializing_if = "std::ops::Not::not" + )] + pub(crate) contested: bool, /// The register's votes naming this remake, by member public key. Not stored in /// the record: filled in when it is read (`load`). #[serde(default, skip_deserializing)] @@ -340,13 +354,42 @@ fn load_record(doc: &AutoCommit, id: &str) -> Option { return None; } }; - match serde_json::from_value(value) { - Ok(record) => Some(record), + let mut record: Record = match serde_json::from_value(value) { + Ok(record) => record, Err(error) => { warn!(remake = %id, error = %error, "Skipping unreadable remake record"); - None + return None; } + }; + if let Some(commit) = contested_decision(doc, id) { + record.decision = Some(commit); + record.contested = true; + } + Some(record) +} + +/// A commit among concurrent decisions of remake `id` that disagree, if they do. +fn contested_decision(doc: &AutoCommit, id: &str) -> Option { + let decisions: Vec = store::read_candidates(doc, &record_path(id, "decision")) + .ok()? + .into_iter() + .filter_map(|value| serde_json::from_value(value).ok()) + .collect(); + let committed: Vec<&EpochRecord> = decisions + .iter() + .filter_map(|decision| match decision { + Decision::Committed { to_epoch, .. } => Some(to_epoch), + Decision::Aborted { .. } => None, + }) + .collect(); + let agree = committed.is_empty() + || (committed.len() == decisions.len() && committed.iter().all(|e| *e == committed[0])); + if agree { + return None; } + decisions + .into_iter() + .find(|decision| matches!(decision, Decision::Committed { .. })) } fn register_path(voter: &str) -> Vec { @@ -551,6 +594,7 @@ mod tests { proposal: proposal(10_000), decision: None, abort_requested: None, + contested: false, votes: BTreeMap::new(), }; assert_eq!(record.phase(10_000, 500), Phase::Proposed); @@ -563,6 +607,45 @@ mod tests { assert_eq!(record.phase(0, 0), Phase::Aborted); } + #[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 a, mut b, mut c) = (base.fork(), base.fork(), base.fork()); + let committed = Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + snapshot_bytes: 1, + decided_at: 2, + }; + write_decision(&mut a, "c-1", &committed).unwrap(); + write_decision( + &mut b, + "c-1", + &Decision::Aborted { + reason: "test".into(), + decided_at: 3, + }, + ) + .unwrap(); + // The same decision, concurrently: they agree. + write_decision(&mut c, "c-1", &committed).unwrap(); + a.commit(); + b.commit(); + c.commit(); + let mut agreeing = a.fork(); + agreeing.merge(&mut c).unwrap(); + assert!(!load(&agreeing, "c-1").unwrap().contested); + + a.merge(&mut b).unwrap(); + let record = load(&a, "c-1").unwrap(); + assert!(record.contested); + assert_eq!(record.phase(0, 0), Phase::Committed); + } + #[test] fn unreadable_records_are_skipped() { let mut doc = AutoCommit::new(); -- cgit v1.3.1