1 files changed, 51 insertions, 6 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 4aa66e7..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):
@@ -31,13 +33,13 @@ use automerge::{AutoCommit, ChangeHash};
 use tokio::time::Instant;
 use tracing::{error, info, warn};
 
+use super::Host;
 use super::delta::{self, ReplayStats};
 use super::kind::Translator;
 use super::local::{CatchUp, LocalStatus, Rejoin};
 use super::quarantine;
 use super::records::unix_millis;
 use super::snapshot::{self, EpochMetadata};
-use super::Host;
 use crate::document::replica::Epoch;
 
 /// Pause before trying the same target again after a failed rejoin.
@@ -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.
@@ -444,9 +489,9 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::remake::snapshot::{build, SnapshotParams};
-    use automerge::{transaction::Transactable, ActorId, ROOT};
-    use serde_json::{json, Value as JsonValue};
+    use crate::remake::snapshot::{SnapshotParams, build};
+    use automerge::{ActorId, ROOT, transaction::Transactable};
+    use serde_json::{Value as JsonValue, json};
 
     /// 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).