3 files changed, 76 insertions, 2 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index db25251..42b6526 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -7,8 +7,10 @@
 //! 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):
@@ -50,6 +52,10 @@ const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60);
 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 {
@@ -147,6 +153,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.
diff --git a/src/remake/local.rs b/src/remake/local.rs
index d2ea2c2..787bd5e 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -34,6 +34,8 @@ struct Inner {
     catch_up: Option<CatchUp>,
     /// A rejoin in progress: the node syncs an empty document of a newer epoch.
     rejoin: Option<Rejoin>,
+    /// While the rejoin downloads the target epoch's snapshot (`catchup`), sync waits.
+    fetching_snapshot: Option<Epoch>,
     quarantine: Vec<Quarantined>,
 }
 
@@ -211,6 +213,23 @@ impl RemakeState {
         self.lock().rejoin.take()
     }
 
+    /// Whether sync should wait: a rejoin is downloading its epoch's snapshot.
+    pub(crate) fn fetching_snapshot(&self) -> bool {
+        self.lock().fetching_snapshot.is_some()
+    }
+
+    pub(crate) fn set_fetching_snapshot(&self, target: Option<Epoch>) {
+        self.lock().fetching_snapshot = target;
+    }
+
+    /// The rejoin of `target`, if still in progress, gets its full time for the sync
+    /// that follows the download.
+    pub(crate) fn restart_rejoin_clock(&self, target: Epoch) {
+        if let Some(rejoin) = self.lock().rejoin.as_mut().filter(|r| r.target == target) {
+            rejoin.started = Instant::now();
+        }
+    }
+
     pub(crate) fn add_quarantine(&self, quarantined: Quarantined) {
         self.lock().quarantine.push(quarantined);
     }
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 0895931..1c2be36 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -68,6 +68,16 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// alive recently enough, read from `doc`.
     fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String>;
 
+    /// Downloads a change from peers on the local epoch, in chunks from several at once,
+    /// checked against its hash.
+    fn fetch_change(
+        &self,
+        hash: automerge::ChangeHash,
+    ) -> impl Future<Output = anyhow::Result<automerge::Change>> + Send;
+
+    /// Starts sync with every peer (e.g. after holding it back).
+    fn request_sync(&self);
+
     fn is_node_ready(&self) -> impl Future<Output = bool> + Send;
 
     /// Known (not forgotten) peers.