1 files changed, 53 insertions, 11 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 56aeee1..cad5a22 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -2,7 +2,9 @@
 //!
 //! Peers on another epoch don't sync, so a node that missed a switch (new, late,
 //! returning, or one whose own build failed) can't follow through ordinary sync. Once
-//! an active peer advertises a newer generation, the node
+//! active, ready members advertise a newer generation (two of them for a node that is
+//! ready itself, one for a node that never was; at most `MAX_GENERATION_GAP` ahead),
+//! the node
 //!
 //! 1. stops being ready and keeps its document aside; reads are answered from a copy
 //!    of it, marked stale (`X-Continuum-Stale`), writes get `503`;
@@ -25,13 +27,13 @@
 //! If no peer completes a sync in time, or the check fails, the node restores its old
 //! document and epoch and tries again later.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::convert::Infallible;
 use std::time::Duration;
 
 use automerge::{AutoCommit, ChangeHash};
 use tokio::time::Instant;
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, warn};
 
 use super::Host;
 use super::delta::{self, ReplayStats};
@@ -46,6 +48,13 @@ use crate::document::replica::Epoch;
 const RETRY: Duration = Duration::from_secs(30);
 /// Lower bound for how long a rejoin may take to complete a sync.
 const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60);
+/// A peer more than this many generations ahead is lying or broken (every generation
+/// is a committed remake); its epoch is ignored instead of followed.
+const MAX_GENERATION_GAP: u64 = 1000;
+/// Peers that must advertise a newer epoch before a ready node leaves its document for
+/// it (fewer if it has fewer active members). A node that was never ready has nothing
+/// to lose and follows the first peer it sees.
+const CORROBORATION: usize = 2;
 
 /// Called by the switch worker when there is nothing to switch to locally, and on every
 /// tick while a rejoin is in progress.
@@ -71,29 +80,62 @@ pub(super) async fn check<H: Host>(host: &H) {
     }
 
     let local = host.current_epoch();
+    let peers = host.peer_summaries().await;
+    let active_members = peers
+        .iter()
+        .filter(|peer| peer.active && peer.member)
+        .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)> = host
-        .peer_summaries()
-        .await
+    let mut ahead: Vec<(String, Epoch)> = peers
         .into_iter()
         .filter(|peer| peer.serves())
         .filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch)))
         .filter(|(_, epoch)| epoch.generation > local.generation)
         .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 {
+            warn!(peer = %peer, epoch = %epoch, local = %local, "Peer advertises an implausible epoch; ignoring it");
+        }
+        plausible
+    });
 
-    let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else {
+    // 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 {
+        active_members.clamp(1, CORROBORATION)
+    } else {
+        1
+    };
+    let mut advertised: HashMap<Epoch, usize> = HashMap::new();
+    for (_, epoch) in &ahead {
+        *advertised.entry(*epoch).or_default() += 1;
+    }
+    let corroborated: Vec<Epoch> = 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);
         }
+        if !ahead.is_empty() {
+            debug!(
+                peers = ahead.len(),
+                required, "A newer epoch is advertised, but not by enough peers yet"
+            );
+        }
         return;
     };
 
-    let mut targets: Vec<Epoch> = ahead
-        .iter()
-        .filter(|(_, epoch)| epoch.generation == generation)
-        .map(|(_, epoch)| *epoch)
+    let mut targets: Vec<Epoch> = corroborated
+        .into_iter()
+        .filter(|epoch| epoch.generation == generation)
         .collect();
     targets.sort_by_key(|epoch| epoch.id);
     targets.dedup();