1 files changed, 104 insertions, 15 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index db25251..cad5a22 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -2,13 +2,17 @@
 //!
 //! 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`;
 //! 2. switches to an empty document labelled with the newer epoch: an empty history
-//!    conflicts with nothing, so ordinary sync fills it from peers on that epoch,
-//!    exactly like a new node joining;
+//!    conflicts with nothing. It downloads the epoch's snapshot change in chunks from
+//!    several peers on that epoch at once (`swarm::blob`; sync waits meanwhile), then
+//!    ordinary sync adds the changes made since, exactly like a new node joining. If
+//!    the download fails, sync transfers the snapshot too, in one message;
 //! 3. once caught up with a peer, checks that the history starts with the epoch's
 //!    snapshot change;
 //! 4. decides about its own changes the swarm never saw (none, for a new node):
@@ -23,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};
@@ -44,12 +48,23 @@ 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.
 pub(super) async fn check<H: Host>(host: &H) {
     if let Some((target, started)) = host.remake().rejoin_progress() {
         let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT);
+        if host.remake().fetching_snapshot() {
+            // The download has its own stall timeout; the clock restarts after it.
+            return;
+        }
         if host.initial_sync_done() {
             finish(host).await;
         } else if started.elapsed() > timeout {
@@ -65,27 +80,62 @@ pub(super) async fn check<H: Host>(host: &H) {
     }
 
     let local = host.current_epoch();
-    let mut ahead: Vec<(String, Epoch)> = host
-        .peer_summaries()
-        .await
+    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)> = peers
         .into_iter()
-        .filter(|peer| peer.active)
+        .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();
@@ -147,6 +197,45 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
         started: Instant::now(),
         was_ready,
     });
+
+    // Sync waits until the snapshot is here (or its download failed).
+    host.remake().set_fetching_snapshot(Some(target));
+    tokio::spawn(fetch_snapshot(host.clone(), target));
+}
+
+/// Downloads the snapshot change of `target` and applies it, then lets sync go on.
+async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
+    let outcome = host.fetch_change(ChangeHash(target.id)).await;
+    let still_rejoining = host
+        .remake()
+        .rejoin_progress()
+        .is_some_and(|(rejoining, _)| rejoining == target)
+        && host.current_epoch() == target;
+    if still_rejoining {
+        match outcome {
+            Ok(change) => {
+                let applied = host
+                    .commit_doc(move |doc| doc.apply_changes([change]))
+                    .await;
+                match applied {
+                    Ok(()) => {
+                        info!(epoch = %target, "Applied the epoch's snapshot; syncing the changes made since")
+                    }
+                    Err(error) => {
+                        warn!(epoch = %target, error = %error, "Couldn't apply the downloaded snapshot; syncing it instead")
+                    }
+                }
+            }
+            Err(error) => warn!(
+                epoch = %target,
+                error = %format!("{error:#}"),
+                "Couldn't download the epoch's snapshot; syncing it instead"
+            ),
+        }
+        host.remake().restart_rejoin_clock(target);
+    }
+    host.remake().set_fetching_snapshot(None);
+    host.request_sync();
 }
 
 /// What to do with this node's own changes the new epoch lacks.
@@ -465,7 +554,7 @@ mod tests {
                 policy,
                 proposed_at: 0,
                 kind: "compact",
-                content: &JsonValue::Null,
+                content: None,
                 translation: &JsonValue::Null,
             },
         )