remake: introduce follow-the-majority rule

1 files changed, 139 insertions, 27 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 7269d8a..43cc390 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -26,6 +26,11 @@
 //!
 //! If no peer completes a sync in time, or the check fails, the node restores its old
 //! document and epoch and tries again later.
+//!
+//! Two remakes committed for one epoch make two epochs of the next generation, which
+//! never sync. A node behind follows the one more peers advertise (`most_followed`);
+//! a ready node on the other rejoins *sideways* once more ready peers are on it than on
+//! its own (`sideways`), its writes on the epoch it leaves going to the quarantine.
 
 use std::collections::{BTreeMap, HashMap};
 use std::convert::Infallible;
@@ -89,13 +94,17 @@ pub(super) async fn check<H: Host>(host: &H) {
         .count();
     // Only ready members count: a rejoining peer advertises an epoch it doesn't hold
     // yet, and a node of another swarm (or a stranger) advertises one that isn't ours.
-    let mut ahead: Vec<(String, Epoch)> = peers
+    let mut serving: Vec<(String, Epoch)> = peers
         .into_iter()
         .filter(|peer| peer.serves())
         .filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch)))
+        .collect();
+    serving.sort_by(|a, b| a.0.cmp(&b.0));
+    let mut ahead: Vec<(String, Epoch)> = serving
+        .iter()
         .filter(|(_, epoch)| epoch.generation > local.generation)
+        .cloned()
         .collect();
-    ahead.sort_by(|a, b| a.0.cmp(&b.0));
     ahead.retain(|(peer, epoch)| {
         let plausible = epoch.generation - local.generation <= MAX_GENERATION_GAP;
         if !plausible {
@@ -107,7 +116,8 @@ pub(super) async fn check<H: Host>(host: &H) {
     // One peer's word is not enough for a ready node to give up its document: a buggy
     // or hostile member could otherwise keep the swarm rejoining an epoch nobody
     // serves. A newer epoch is followed once enough distinct peers advertise it.
-    let required = if host.is_node_ready().await {
+    let ready = host.is_node_ready().await;
+    let required = if ready {
         active_members.clamp(1, CORROBORATION)
     } else {
         1
@@ -116,37 +126,46 @@ pub(super) async fn check<H: Host>(host: &H) {
     for (_, epoch) in &ahead {
         *advertised.entry(*epoch).or_default() += 1;
     }
-    let corroborated: Vec<Epoch> = advertised
+    let corroborated: Vec<(Epoch, usize)> = advertised
         .into_iter()
         .filter(|(_, count)| *count >= required)
-        .map(|(epoch, _)| epoch)
         .collect();
 
-    let Some(generation) = corroborated.iter().map(|epoch| epoch.generation).max() else {
-        if host.remake().catch_up().is_some() {
-            host.remake().set_catch_up(None);
+    let mut aside = false;
+    let target = match corroborated.iter().map(|(epoch, _)| epoch.generation).max() {
+        Some(generation) => {
+            let targets: Vec<(Epoch, usize)> = corroborated
+                .into_iter()
+                .filter(|(epoch, _)| epoch.generation == generation)
+                .collect();
+            // Several epochs of the newest generation (two remakes committed for one
+            // epoch): the one most peers are on, as `sideways` would pick.
+            let target = most_followed(&targets).expect("corroborated has this generation");
+            if targets.len() > 1 {
+                let listed: Vec<String> =
+                    targets.iter().map(|(epoch, _)| epoch.to_string()).collect();
+                warn!(epochs = ?listed, chosen = %target, "Peers disagree about the newest epoch; following the one most of them are on");
+            }
+            target
         }
-        if !ahead.is_empty() {
-            debug!(
-                peers = ahead.len(),
-                required, "A newer epoch is advertised, but not by enough peers yet"
-            );
+        None => {
+            if !ahead.is_empty() {
+                debug!(
+                    peers = ahead.len(),
+                    required, "A newer epoch is advertised, but not by enough peers yet"
+                );
+            }
+            let sideways = ready.then(|| sideways(local, &serving, required)).flatten();
+            let Some(target) = sideways else {
+                if host.remake().catch_up().is_some() {
+                    host.remake().set_catch_up(None);
+                }
+                return;
+            };
+            aside = true;
+            target
         }
-        return;
-    };
-
-    let mut targets: Vec<Epoch> = corroborated
-        .into_iter()
-        .filter(|epoch| epoch.generation == generation)
-        .collect();
-    targets.sort_by_key(|epoch| epoch.id);
-    targets.dedup();
-    let [target] = targets.as_slice() else {
-        let listed: Vec<String> = targets.iter().map(ToString::to_string).collect();
-        warn!(epochs = ?listed, "Peers disagree about the newest epoch; not catching up");
-        return;
     };
-    let target = *target;
 
     // Back off between attempts for the same target.
     let previous = host
@@ -161,9 +180,50 @@ pub(super) async fn check<H: Host>(host: &H) {
     }
     let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
 
+    if aside {
+        warn!(
+            local = %local,
+            theirs = %target,
+            "More peers are on another epoch of this generation than on ours; rejoining theirs (writes made on ours since go to the quarantine)"
+        );
+    }
     start(host, local, target, attempts).await;
 }
 
+/// For a ready node, another epoch of its own generation to rejoin, if any. Two remakes
+/// committed for one epoch (`remakes.md`, Limits: a coordinator cut off after it
+/// committed, electorates of an old partition, a vote lost with a restart) leave two
+/// epochs of a generation that never sync. The one more ready members are on wins:
+/// a node on the other rejoins it, with `required` of them at least; on a tie, the
+/// epoch with the lower ID. Each node judges by the peers it sees, so the side that
+/// moves is the one that sees itself outnumbered.
+fn sideways(local: Epoch, serving: &[(String, Epoch)], required: usize) -> Option<Epoch> {
+    let mut counts: HashMap<Epoch, usize> = HashMap::new();
+    for (_, epoch) in serving {
+        if epoch.generation == local.generation {
+            *counts.entry(*epoch).or_default() += 1;
+        }
+    }
+    // This node is on its own epoch too.
+    let ours = counts.remove(&local).unwrap_or(0) + 1;
+    let others: Vec<(Epoch, usize)> = counts
+        .into_iter()
+        .filter(|(epoch, count)| {
+            *count >= required && (*count > ours || (*count == ours && epoch.id < local.id))
+        })
+        .collect();
+    most_followed(&others)
+}
+
+/// Of epochs with how many peers are on each, the one most are on; of those, the one
+/// with the lowest ID, so every node picks the same.
+fn most_followed(epochs: &[(Epoch, usize)]) -> Option<Epoch> {
+    epochs
+        .iter()
+        .max_by(|a, b| a.1.cmp(&b.1).then(b.0.id.cmp(&a.0.id)))
+        .map(|(epoch, _)| *epoch)
+}
+
 async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
     host.remake().set_catch_up(Some(CatchUp {
         target,
@@ -494,6 +554,15 @@ fn prepare(
     );
 
     let plan = match (&next, translator) {
+        // Rejoined sideways (`sideways`): whatever was written on our epoch, the swarm
+        // went another way; nothing says where it would go.
+        _ if gap == 0 => Plan::Quarantine {
+            paths,
+            reason: format!(
+                "{} local changes were made on {previous}, which the swarm left for {target}, another epoch of the same generation",
+                missing.own
+            ),
+        },
         (Some(_), Err(reason)) => Plan::Quarantine {
             paths,
             reason: format!("cannot translate {} local changes: {reason}", missing.own),
@@ -576,6 +645,49 @@ mod tests {
     use automerge::{ActorId, ROOT, transaction::Transactable};
     use serde_json::{Value as JsonValue, json};
 
+    #[test]
+    fn sideways_follows_the_epoch_more_peers_are_on() {
+        let epoch = |id: u8| Epoch {
+            generation: 3,
+            id: [id; 32],
+        };
+        let on = |peers: &[(&str, Epoch)]| -> Vec<(String, Epoch)> {
+            peers
+                .iter()
+                .map(|(peer, epoch)| ((*peer).to_owned(), *epoch))
+                .collect()
+        };
+        let (ours, theirs) = (epoch(5), epoch(9));
+        // Alone against two: theirs.
+        let serving = on(&[("b", theirs), ("c", theirs)]);
+        assert_eq!(sideways(ours, &serving, 2), Some(theirs));
+        // Two (with us) against two: the lower ID, ours; they come to us.
+        let serving = on(&[("b", theirs), ("c", theirs), ("d", ours)]);
+        assert_eq!(sideways(ours, &serving, 2), None);
+        assert_eq!(
+            sideways(theirs, &on(&[("a", ours), ("d", ours), ("b", theirs)]), 2),
+            Some(ours)
+        );
+        // Fewer than required: nothing, even outnumbered.
+        assert_eq!(sideways(ours, &on(&[("b", theirs)]), 2), None);
+        // Another generation isn't sideways.
+        let newer = Epoch {
+            generation: 4,
+            id: [1; 32],
+        };
+        assert_eq!(sideways(ours, &on(&[("b", newer), ("c", newer)]), 2), None);
+        // Several: the one most are on.
+        let other = epoch(2);
+        let serving = on(&[
+            ("b", theirs),
+            ("c", theirs),
+            ("d", theirs),
+            ("e", other),
+            ("f", other),
+        ]);
+        assert_eq!(sideways(ours, &serving, 2), Some(theirs));
+    }
+
     /// An old-epoch document with one own change after the base, and the swarm's
     /// document of the next epoch (snapshot plus one foreign change).
     fn scenario(policy: &JsonValue) -> (Document, Document, Epoch) {