allow superseding expired remakes

1 files changed, 86 insertions, 3 deletions

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/<pk>`: 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<Decision>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub(crate) abort_requested: Option<AbortRequest>,
+    /// 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<Record> {
             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<Decision> {
+    let decisions: Vec<Decision> = 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<String> {
@@ -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);
@@ -564,6 +608,45 @@ 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 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();
         store::write_path(