fix catch-up

7 files changed, 427 insertions, 94 deletions

diff --git a/src/compaction/api.rs b/src/compaction/api.rs
index f5c01f2..e55d018 100644
--- a/src/compaction/api.rs
+++ b/src/compaction/api.rs
@@ -121,9 +121,25 @@ async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> {
 /// Waits until the compaction leaves `proposed`. Bounded by the proposal's own expiry,
 /// at which point it is reported as `expired`.
 async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
+    // A record that was readable may be briefly unreadable or absent while concurrent
+    // changes merge; give up only if that persists.
+    const MAX_MISSES: u32 = 5;
+
     let mut changes = host.subscribe_changes();
+    let mut seen = false;
+    let mut misses = 0;
     loop {
-        let report = load_report(host, id).await?;
+        let Some(report) = load_report(host, id).await else {
+            misses += 1;
+            warn!(compaction = %id, seen, misses, "Compaction record not readable while waiting");
+            if !seen || misses >= MAX_MISSES {
+                return None;
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            continue;
+        };
+        seen = true;
+        misses = 0;
         if report.phase.is_final() {
             return Some(report);
         }
diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs
index f166f56..4694fbc 100644
--- a/src/compaction/catchup.rs
+++ b/src/compaction/catchup.rs
@@ -2,9 +2,9 @@
 //!
 //! Peers on another epoch don't sync, so a node that missed a switch (new, late, or
 //! one whose own build failed) can't learn the new epoch through ordinary sync. Instead,
-//! once an established peer advertises a newer generation, the node asks it for its
-//! whole document (`SnapshotRequest`), checks that the document starts with the
-//! advertised snapshot change, and switches to it:
+//! once an active peer advertises a newer generation, the node downloads that peer's
+//! document (`SnapshotRequest`/`SnapshotReply`), checks that it starts with the advertised
+//! snapshot change, and switches to it:
 //!
 //! * a node that wrote nothing the snapshot lacks (e.g. a brand-new node) adopts the
 //!   document as is, whatever the gap;
@@ -13,17 +13,24 @@
 //! * a node further behind with such changes stays where it is and reports it; that is
 //!   the returning-node case (replay window or quarantine, step 7).
 //!
-//! Only one fetch runs at a time, and the worker retries after `RETRY`. The reply is the
-//! peer's full document in one message; chunking comes later.
+//! The document travels in chunks of `CHUNK_BYTES`, one request per chunk, so a large
+//! document neither needs one huge transport message nor one long timeout. The serving
+//! node saves its document once per download and serves every chunk from that copy; a
+//! SHA-256 digest of the whole copy is in every reply and checked at the end.
+//!
+//! Peers refuse explicitly (non-members, busy, other epoch), so the requester moves on
+//! to the next peer right away instead of timing out.
 
+use std::sync::Arc;
 use std::time::Duration;
 
 use automerge::{ActorId, AutoCommit};
+use sha2::{Digest, Sha256};
 use tokio::time::Instant;
 use tracing::{debug, error, info, warn};
 
 use super::delta::{self, ReplayStats};
-use super::local::{CatchUp, LocalStatus};
+use super::local::{CatchUp, LocalStatus, Served};
 use super::records::unix_millis;
 use super::snapshot;
 use super::switch::swap_in;
@@ -31,21 +38,31 @@ use super::Host;
 use crate::replica::Epoch;
 use crate::transport::{ParsedMessage, SnapshotReply, SnapshotRequest};
 
+/// Pause before the next attempt (next peer) after a failed download.
 const RETRY: Duration = Duration::from_secs(10);
-const FETCH_TIMEOUT: Duration = Duration::from_secs(60);
+/// Payload per reply.
+pub(crate) const CHUNK_BYTES: usize = 1024 * 1024;
+/// How long to wait for one chunk, and how often to ask for it.
+const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
+const CHUNK_ATTEMPTS: u32 = 3;
+/// A served copy older than this is replaced when a new download starts.
+const SERVED_FRESH_FOR: Duration = Duration::from_secs(60);
 
 /// Called by the switch worker when there is nothing to switch to locally.
 pub(super) async fn check<H: Host>(host: &H) {
     let local = host.current_epoch();
 
-    let ahead: Vec<(String, Epoch)> = host
+    // Any active peer will do: it proved liveness recently, and whether it accepts us
+    // as a member is its decision (it refuses explicitly otherwise).
+    let mut ahead: Vec<(String, Epoch)> = host
         .peer_summaries()
         .await
         .into_iter()
-        .filter(|peer| peer.active && peer.established)
+        .filter(|peer| peer.active)
         .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));
 
     let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else {
         if host.compaction().catch_up().is_some() {
@@ -81,7 +98,7 @@ pub(super) async fn check<H: Host>(host: &H) {
     }
     let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
 
-    // Rotate through the peers on the target epoch.
+    // Rotate through the peers on the target epoch (sorted, so the rotation is stable).
     let sources: Vec<&String> = ahead
         .iter()
         .filter(|(_, epoch)| *epoch == target)
@@ -98,7 +115,7 @@ pub(super) async fn check<H: Host>(host: &H) {
 
     info!(local = %local, target = %target, peer = %source, attempt = attempts, "Fetching newer epoch from peer");
 
-    let result = match fetch(host, &source, local, target).await {
+    let result = match download(host, &source, local, target).await {
         Ok(document) => adopt(host, local, target, document).await,
         Err(error) => Err(error),
     };
@@ -117,38 +134,95 @@ pub(super) async fn check<H: Host>(host: &H) {
     }
 }
 
-async fn fetch<H: Host>(
+/// Downloads `peer`'s document of `target`, chunk by chunk.
+async fn download<H: Host>(
     host: &H,
     peer: &str,
     local: Epoch,
     target: Epoch,
 ) -> Result<Vec<u8>, String> {
-    let reply = host.compaction().await_reply(peer, target);
-    let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
-        epoch: local,
-        wanted: target,
-    });
-    if let Err(error) = host.send_message(peer, request).await {
-        host.compaction().stop_awaiting();
-        return Err(format!("sending the request failed: {error:#}"));
+    let mut document: Vec<u8> = Vec::new();
+    let mut expected: Option<([u8; 32], u64)> = None;
+    let mut last_progress = 0;
+
+    loop {
+        let offset = document.len() as u64;
+        let reply = request_chunk(host, peer, local, target, offset).await?;
+
+        if !reply.refused.is_empty() {
+            return Err(format!("peer refused: {}", reply.refused));
+        }
+        if reply.epoch != target {
+            return Err(format!("peer is on {} instead of {target}", reply.epoch));
+        }
+        match expected {
+            None => {
+                info!(peer = %peer, document_bytes = reply.total, "Downloading document");
+                expected = Some((reply.digest, reply.total));
+            }
+            Some((digest, total)) if digest != reply.digest || total != reply.total => {
+                return Err("the peer's document changed during the download".to_owned());
+            }
+            Some(_) => {}
+        }
+        let total = reply.total;
+        if reply.data.is_empty() && offset < total {
+            return Err(format!("empty chunk at offset {offset} of {total}"));
+        }
+        document.extend_from_slice(&reply.data);
+
+        let percent = (document.len() as u64 * 100)
+            .checked_div(total)
+            .unwrap_or(100);
+        if percent >= last_progress + 10 {
+            last_progress = percent;
+            debug!(peer = %peer, percent, "Download progress");
+        }
+
+        if document.len() as u64 >= total {
+            break;
+        }
+    }
+
+    let (digest, total) = expected.expect("set by the first reply");
+    if document.len() as u64 != total || sha256(&document) != digest {
+        return Err("downloaded document does not match its digest".to_owned());
     }
+    Ok(document)
+}
 
-    let reply = match tokio::time::timeout(FETCH_TIMEOUT, reply).await {
-        Ok(Ok(reply)) => reply,
-        Ok(Err(_)) => return Err("request superseded".to_owned()),
-        Err(_) => {
+async fn request_chunk<H: Host>(
+    host: &H,
+    peer: &str,
+    local: Epoch,
+    target: Epoch,
+    offset: u64,
+) -> Result<SnapshotReply, String> {
+    for attempt in 1..=CHUNK_ATTEMPTS {
+        let reply = host.compaction().await_reply(peer, target, offset);
+        let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
+            epoch: local,
+            wanted: target,
+            offset,
+        });
+        if let Err(error) = host.send_message(peer, request).await {
             host.compaction().stop_awaiting();
-            return Err(format!("no reply within {}s", FETCH_TIMEOUT.as_secs()));
+            return Err(format!("sending the request failed: {error:#}"));
         }
-    };
 
-    if reply.epoch != target || reply.document.is_empty() {
-        return Err(format!(
-            "peer is on {} and did not send {target}",
-            reply.epoch
-        ));
+        match tokio::time::timeout(CHUNK_TIMEOUT, reply).await {
+            Ok(Ok(reply)) => return Ok(reply),
+            Ok(Err(_)) => return Err("request superseded".to_owned()),
+            Err(_) => {
+                host.compaction().stop_awaiting();
+                debug!(peer = %peer, offset, attempt, "No reply for chunk; asking again");
+            }
+        }
     }
-    Ok(reply.document)
+    Err(format!(
+        "no reply for the chunk at offset {offset} after {CHUNK_ATTEMPTS} attempts of {}s",
+        CHUNK_TIMEOUT.as_secs()
+    ))
 }
 
 /// Verifies the fetched document, carries over this node's own missing changes, and
@@ -250,58 +324,125 @@ struct Prepared {
     metadata: snapshot::EpochMetadata,
 }
 
-/// Answers a peer's snapshot request with this node's document, if it is on the wanted
-/// epoch. Only for members of our cluster; one at a time.
+/// Answers a peer's chunk request from a saved copy of this node's document, if this
+/// node is on the wanted epoch and the peer is a member of our cluster. Refuses
+/// explicitly otherwise.
 pub(crate) async fn serve_snapshot<H: Host>(host: H, peer: String, request: SnapshotRequest) {
-    if !host.is_cluster_member(&peer).await {
-        debug!(peer = %peer, "Ignoring snapshot request from a non-member");
-        return;
-    }
-    if !host.compaction().try_start_serving() {
-        debug!(peer = %peer, "Already serving a snapshot; the peer will retry");
-        return;
-    }
-
     let current = host.current_epoch();
     let wanted = request.wanted;
 
-    let document = if current == wanted {
-        // Clone under the read lock, save without one. The history check guards against
-        // a switch between reading the epoch and cloning.
-        let mut doc = host.read_doc(|doc| doc.clone()).await;
-        tokio::task::spawn_blocking(move || {
-            if doc
-                .get_change_by_hash(&automerge::ChangeHash(wanted.id))
-                .is_some()
-            {
-                doc.save()
-            } else {
-                Vec::new()
+    let reply = match served_copy(&host, &peer, current, wanted, request.offset).await {
+        Ok(served) => {
+            let start = (request.offset as usize).min(served.bytes.len());
+            let end = (start + CHUNK_BYTES).min(served.bytes.len());
+            if request.offset == 0 {
+                info!(peer = %peer, wanted = %wanted, document_bytes = served.bytes.len(), "Serving document");
             }
-        })
+            SnapshotReply {
+                epoch: served.epoch,
+                wanted,
+                digest: served.digest,
+                total: served.bytes.len() as u64,
+                offset: request.offset,
+                data: served.bytes[start..end].to_vec(),
+                refused: String::new(),
+            }
+        }
+        Err(reason) => {
+            info!(peer = %peer, wanted = %wanted, reason = %reason, "Refusing document request");
+            SnapshotReply {
+                epoch: current,
+                wanted,
+                digest: [0; 32],
+                total: 0,
+                offset: request.offset,
+                data: Vec::new(),
+                refused: reason,
+            }
+        }
+    };
+
+    if let Err(error) = host
+        .send_message(&peer, ParsedMessage::SnapshotReply(reply))
         .await
-        .unwrap_or_else(|error| {
+    {
+        warn!(peer = %peer, error = %format!("{error:#}"), "Sending document chunk failed");
+    }
+}
+
+/// The saved document to serve chunks from: reused while a download is in progress,
+/// saved anew when a download starts and the copy is stale.
+async fn served_copy<H: Host>(
+    host: &H,
+    peer: &str,
+    current: Epoch,
+    wanted: Epoch,
+    offset: u64,
+) -> Result<Arc<Served>, String> {
+    if !host.is_cluster_member(peer).await {
+        return Err("not a member of this cluster (no completed handshake)".to_owned());
+    }
+    if current != wanted {
+        return Err(format!("this node is on {current}"));
+    }
+
+    if let Some(served) = host.compaction().served(wanted) {
+        if offset > 0 || served.saved_at.elapsed() < SERVED_FRESH_FOR {
+            return Ok(served);
+        }
+    } else if offset > 0 {
+        return Err("download expired; start again".to_owned());
+    }
+
+    if !host.compaction().try_start_serving() {
+        return Err("busy saving the document; try again".to_owned());
+    }
+
+    // Clone under the read lock, save without one. The history check guards against
+    // a switch between reading the epoch and cloning.
+    let mut doc = host.read_doc(|doc| doc.clone()).await;
+    let saved = tokio::task::spawn_blocking(move || {
+        if doc
+            .get_change_by_hash(&automerge::ChangeHash(wanted.id))
+            .is_some()
+        {
+            Some(doc.save())
+        } else {
+            None
+        }
+    })
+    .await;
+    host.compaction().done_serving();
+
+    let bytes = match saved {
+        Ok(Some(bytes)) => bytes,
+        Ok(None) => return Err("this node switched epochs".to_owned()),
+        Err(error) => {
             error!(error = %error, "Saving the document for a peer failed");
-            Vec::new()
-        })
-    } else {
-        Vec::new()
+            return Err("saving the document failed".to_owned());
+        }
     };
-    host.compaction().done_serving();
 
-    info!(peer = %peer, wanted = %wanted, document_bytes = document.len(), "Serving snapshot");
-    let reply = ParsedMessage::SnapshotReply(SnapshotReply {
-        epoch: if document.is_empty() { current } else { wanted },
-        document,
+    let served = Arc::new(Served {
+        epoch: wanted,
+        digest: sha256(&bytes),
+        bytes,
+        saved_at: Instant::now(),
     });
-    if let Err(error) = host.send_message(&peer, reply).await {
-        warn!(peer = %peer, error = %format!("{error:#}"), "Sending snapshot failed");
-    }
+    host.compaction().set_served(Arc::clone(&served));
+    Ok(served)
+}
+
+fn sha256(bytes: &[u8]) -> [u8; 32] {
+    let mut digest = [0; 32];
+    digest.copy_from_slice(&Sha256::digest(bytes));
+    digest
 }
 
-/// Hands a peer's snapshot reply to the waiting fetch.
+/// Hands a peer's chunk to the waiting download.
 pub(crate) fn receive_snapshot_reply<H: Host>(host: &H, peer: &str, reply: SnapshotReply) {
+    let offset = reply.offset;
     if !host.compaction().deliver_reply(peer, reply) {
-        debug!(peer = %peer, "Unexpected snapshot reply; dropped");
+        debug!(peer = %peer, offset, "Unexpected document chunk; dropped");
     }
 }
diff --git a/src/compaction/local.rs b/src/compaction/local.rs
index 869c94e..adf8764 100644
--- a/src/compaction/local.rs
+++ b/src/compaction/local.rs
@@ -6,7 +6,8 @@
 
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Mutex, MutexGuard};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
 
 use automerge::AutoCommit;
 use serde::Serialize;
@@ -34,11 +35,24 @@ struct Inner {
     first_seen: HashMap<String, Instant>,
     status: HashMap<String, LocalStatus>,
     archive: Vec<Archived>,
-    /// The snapshot request this node is waiting for (peer, wanted epoch).
-    awaiting: Option<(String, Epoch, oneshot::Sender<SnapshotReply>)>,
+    /// The chunk this node is waiting for: (peer, wanted epoch, offset).
+    awaiting: Option<(String, Epoch, u64, oneshot::Sender<SnapshotReply>)>,
     catch_up: Option<CatchUp>,
+    /// The saved document chunks are served from (one download source at a time).
+    served: Option<Arc<Served>>,
 }
 
+/// A saved copy of this node's document, served to downloading peers.
+pub(crate) struct Served {
+    pub(crate) epoch: Epoch,
+    pub(crate) bytes: Vec<u8>,
+    pub(crate) digest: [u8; 32],
+    pub(crate) saved_at: Instant,
+}
+
+/// A served copy is dropped this long after it was saved.
+const SERVED_KEPT_FOR: Duration = Duration::from_secs(15 * 60);
+
 /// Fetching a newer epoch from a peer.
 #[derive(Clone, Debug)]
 pub(crate) struct CatchUp {
@@ -140,26 +154,41 @@ impl Compaction {
         prune(&mut inner.archive, current_generation);
     }
 
+    /// Drops archives past their retention and a served copy nobody needs any more.
     pub(crate) fn prune_archive(&self, current_generation: u64) {
-        prune(&mut self.lock().archive, current_generation);
+        let mut inner = self.lock();
+        prune(&mut inner.archive, current_generation);
+        if inner
+            .served
+            .as_ref()
+            .is_some_and(|served| served.saved_at.elapsed() > SERVED_KEPT_FOR)
+        {
+            inner.served = None;
+        }
     }
 
-    /// Registers interest in a snapshot reply from `peer`, replacing any earlier one.
+    /// Registers interest in one chunk from `peer`, replacing any earlier request.
     pub(crate) fn await_reply(
         &self,
         peer: &str,
         wanted: Epoch,
+        offset: u64,
     ) -> oneshot::Receiver<SnapshotReply> {
         let (sender, receiver) = oneshot::channel();
-        self.lock().awaiting = Some((peer.to_owned(), wanted, sender));
+        self.lock().awaiting = Some((peer.to_owned(), wanted, offset, sender));
         receiver
     }
 
-    /// Hands a reply to the waiting fetch. False if nobody asked that peer.
+    /// Hands a reply to the waiting download. False unless it answers exactly the
+    /// pending request (late replies to earlier attempts are dropped).
     pub(crate) fn deliver_reply(&self, peer: &str, reply: SnapshotReply) -> bool {
         let mut inner = self.lock();
         match inner.awaiting.take() {
-            Some((expected, _, sender)) if expected == peer => sender.send(reply).is_ok(),
+            Some((expected, wanted, offset, sender))
+                if expected == peer && wanted == reply.wanted && offset == reply.offset =>
+            {
+                sender.send(reply).is_ok()
+            }
             other => {
                 inner.awaiting = other;
                 false
@@ -167,6 +196,19 @@ impl Compaction {
         }
     }
 
+    /// The copy served for `epoch`, if any.
+    pub(crate) fn served(&self, epoch: Epoch) -> Option<Arc<Served>> {
+        self.lock()
+            .served
+            .as_ref()
+            .filter(|served| served.epoch == epoch)
+            .cloned()
+    }
+
+    pub(crate) fn set_served(&self, served: Arc<Served>) {
+        self.lock().served = Some(served);
+    }
+
     pub(crate) fn stop_awaiting(&self) {
         self.lock().awaiting = None;
     }
diff --git a/src/mycelium.rs b/src/mycelium.rs
index 22d49c1..96b9e6d 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -576,6 +576,21 @@ impl Transport for MyceliumTransport {
             }
             .into());
         }
+
+        // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing
+        // a message that never arrived.
+        if tracing::enabled!(tracing::Level::DEBUG) {
+            if let Ok(body) = response
+                .bytes()
+                .await
+                .map_err(anyhow::Error::from)
+                .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?))
+            {
+                if let Some(id) = body.get("id").and_then(|id| id.as_str()) {
+                    tracing::debug!(peer = %message.peer, message_id = id, "Mycelium accepted message");
+                }
+            }
+        }
         Ok(())
     }
 
@@ -919,18 +934,25 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
 struct SnapshotRequestShadow {
     epoch: EpochShadow,
     wanted: EpochShadow,
+    offset: u64,
 }
 
 #[derive(Serialize, Deserialize)]
 struct SnapshotReplyShadow {
     epoch: EpochShadow,
-    document: ByteBuf,
+    wanted: EpochShadow,
+    digest: ByteBuf,
+    total: u64,
+    offset: u64,
+    data: ByteBuf,
+    refused: String,
 }
 
 fn snapshot_request_to_bytes(msg: &SnapshotRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&SnapshotRequestShadow {
         epoch: epoch_to_shadow(&msg.epoch),
         wanted: epoch_to_shadow(&msg.wanted),
+        offset: msg.offset,
     })
 }
 
@@ -939,22 +961,37 @@ fn bytes_to_snapshot_request(bytes: &[u8]) -> Result<SnapshotRequest, TransportE
     Ok(SnapshotRequest {
         epoch: epoch_from_shadow(shadow.epoch)?,
         wanted: epoch_from_shadow(shadow.wanted)?,
+        offset: shadow.offset,
     })
 }
 
 fn snapshot_reply_to_bytes(msg: &SnapshotReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&SnapshotReplyShadow {
         epoch: epoch_to_shadow(&msg.epoch),
-        // One copy of the document; replies are rare.
-        document: ByteBuf::from(msg.document.clone()),
+        wanted: epoch_to_shadow(&msg.wanted),
+        digest: ByteBuf::from(msg.digest.to_vec()),
+        total: msg.total,
+        offset: msg.offset,
+        data: ByteBuf::from(msg.data.clone()),
+        refused: msg.refused.clone(),
     })
 }
 
 fn bytes_to_snapshot_reply(bytes: &[u8]) -> Result<SnapshotReply, TransportError> {
     let shadow: SnapshotReplyShadow = rmp_serde::from_slice(bytes)?;
+    let digest = shadow
+        .digest
+        .as_slice()
+        .try_into()
+        .map_err(|_| TransportError::InvalidPayload("digest must be 32 bytes".into()))?;
     Ok(SnapshotReply {
         epoch: epoch_from_shadow(shadow.epoch)?,
-        document: shadow.document.into_vec(),
+        wanted: epoch_from_shadow(shadow.wanted)?,
+        digest,
+        total: shadow.total,
+        offset: shadow.offset,
+        data: shadow.data.into_vec(),
+        refused: shadow.refused,
     })
 }
 
diff --git a/src/state.rs b/src/state.rs
index b3a496c..4c7282f 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -102,6 +102,9 @@ pub(crate) struct PeerState {
 
     /// Epoch the peer advertised in its most recent message.
     pub(crate) epoch: Option<Epoch>,
+
+    /// Consecutive sync exchanges that changed nothing (livelock detection).
+    pub(crate) stalled_exchanges: u32,
 }
 
 impl Default for PeerState {
@@ -120,6 +123,7 @@ impl Default for PeerState {
             last_ping_at: None,
             last_sync_rx: None,
             last_resync_at: None,
+            stalled_exchanges: 0,
             epoch: None,
         }
     }
@@ -186,6 +190,19 @@ impl PeerState {
         due
     }
 
+    /// Schedules a fresh handshake, which resets sync on both sides: the peer resets on
+    /// our HELLO, we reset on its reply. The peer worker sends the HELLO on its next tick.
+    /// Usable without awaiting (e.g. under a blocking lock).
+    pub(crate) fn request_resync(&mut self) {
+        if matches!(self.hello_state, HelloState::Sending { .. }) {
+            return; // a HELLO is already on its way
+        }
+        self.hello_state = HelloState::AwaitingReply {
+            next_attempt: Instant::now(),
+            next_delay: Duration::from_millis(500),
+        };
+    }
+
     /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
     pub(crate) fn needs_heal(
         &mut self,
@@ -511,10 +528,7 @@ impl AppState {
             .map(|(id, peer)| (id.clone(), Arc::clone(peer)))
             .collect();
 
-        for (_, peer) in &peers {
-            peer.lock().await.reset_sync();
-        }
-
+        // Sync states were reset under the document lock by the swap itself.
         self.publish_change();
 
         for (id, _) in peers {
@@ -652,6 +666,12 @@ impl Host for AppState {
                     // Updated under the same lock, so the cached epoch never disagrees
                     // with the document for anyone holding the lock.
                     state.epoch_tx.send_replace(epoch);
+                    // Reset every peer's sync state before the lock is released, so no
+                    // message is generated from or applied to a state that describes the
+                    // old document. (Lock order doc -> peer, as in sync generation.)
+                    for peer in state.peer_states.blocking_read().values() {
+                        peer.blocking_lock().reset_sync();
+                    }
                     Ok(old_doc)
                 })
                 .await
diff --git a/src/sync.rs b/src/sync.rs
index 07ed1a0..407e88e 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -13,7 +13,7 @@ use tokio::time::Instant;
 use tracing::{debug, info, instrument, trace, warn};
 
 use crate::peer::{peer_cluster, send_hello_once};
-use crate::state::{AppState, Liveness, PeerMembership, PendingOutbound};
+use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
 use crate::transport::{self, ParsedMessage};
 
 const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500);
@@ -268,6 +268,7 @@ pub(crate) async fn handle_sync(
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
         let epoch = envelope.epoch;
         let message = envelope.message;
+        let peer = peer.to_owned();
 
         // Some(caught_up) when applied, None when the message belongs to another epoch.
         tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
@@ -281,6 +282,9 @@ pub(crate) async fn handle_sync(
 
             let mut peer_state = peer_state_mutex.blocking_lock();
 
+            let heads_before = replica.doc.get_heads();
+            let their_heads_before = peer_state.sync_state.their_heads.clone();
+
             replica
                 .doc
                 .sync()
@@ -288,6 +292,12 @@ pub(crate) async fn handle_sync(
                 .context("Automerge synchronization failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
 
+            let heads_after = replica.doc.get_heads();
+            let unproductive = heads_before == heads_after
+                && peer_state.sync_state.their_heads == their_heads_before
+                && peer_state.sync_state.their_heads.as_ref() != Some(&heads_after);
+            break_sync_stall(&peer, &mut peer_state, unproductive);
+
             Ok(Some(
                 check_caught_up
                     && peer_state
@@ -324,6 +334,60 @@ pub(crate) async fn handle_sync(
     Ok(())
 }
 
+/// Exchanges in a row that changed nothing on either side before `sent_hashes` is
+/// cleared, and before a full two-sided reset (re-handshake).
+const STALL_CLEAR_SENT: u32 = 8;
+const STALL_RESET: u32 = 32;
+
+/// Breaks sync livelocks.
+///
+/// Automerge never resends a change it believes the peer has received (`sent_hashes`).
+/// If that change was lost on the way (a dropped or undeliverable message, an epoch
+/// check, a reset of only one side's state), the two sides keep exchanging messages
+/// with differing heads and no changes, as fast as the transport allows. The keepalive
+/// heal never fires then, because sync traffic keeps flowing.
+///
+/// * A change the peer explicitly needs is never withheld as "already sent".
+/// * After a few unproductive exchanges, forget what we sent: the next message carries
+///   everything the peer's bloom filter lacks.
+/// * If that doesn't help either, reset both sides through a fresh handshake.
+fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) {
+    let sync_state = &mut peer_state.sync_state;
+    if let Some(need) = &sync_state.their_need {
+        let before = sync_state.sent_hashes.len();
+        sync_state.sent_hashes.retain(|hash| !need.contains(hash));
+        let resend = before - sync_state.sent_hashes.len();
+        if resend > 0 {
+            debug!(peer = %peer, resend, "Peer needs changes we had sent; sending them again");
+        }
+    }
+
+    if !unproductive {
+        peer_state.stalled_exchanges = 0;
+        return;
+    }
+
+    peer_state.stalled_exchanges += 1;
+    match peer_state.stalled_exchanges {
+        STALL_CLEAR_SENT => {
+            warn!(
+                peer = %peer,
+                sent = sync_state.sent_hashes.len(),
+                their_heads = ?sync_state.their_heads,
+                shared_heads = ?sync_state.shared_heads,
+                "Sync is stalled; forgetting which changes were sent"
+            );
+            sync_state.sent_hashes.clear();
+        }
+        STALL_RESET => {
+            warn!(peer = %peer, "Sync is still stalled; resetting it on both sides");
+            peer_state.stalled_exchanges = 0;
+            peer_state.request_resync();
+        }
+        _ => {}
+    }
+}
+
 pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
     let mut dirty_peers = HashSet::<String>::new();
     let mut retry_peers = HashSet::<String>::new();
diff --git a/src/transport.rs b/src/transport.rs
index b7a4094..51faf16 100644
--- a/src/transport.rs
+++ b/src/transport.rs
@@ -68,22 +68,35 @@ pub struct SnapshotRequest {
     pub epoch: Epoch,
     /// The epoch whose document is wanted.
     pub wanted: Epoch,
+    /// Byte offset of the wanted chunk in the saved document.
+    pub offset: u64,
 }
 
+/// One chunk of the sender's saved document, or a refusal.
 #[derive(Clone)]
 pub struct SnapshotReply {
-    /// The sender's epoch, which is the epoch of `document`.
+    /// The sender's epoch; equals `wanted` unless refused.
     pub epoch: Epoch,
-    /// `AutoCommit::save()` of the sender's document; empty if it isn't on the wanted
-    /// epoch (any more).
-    pub document: Vec<u8>,
+    pub wanted: Epoch,
+    /// SHA-256 of the whole saved document; identifies the copy chunks come from.
+    pub digest: [u8; 32],
+    /// Length of the whole saved document.
+    pub total: u64,
+    pub offset: u64,
+    pub data: Vec<u8>,
+    /// Non-empty if the sender won't serve the document, with the reason.
+    pub refused: String,
 }
 
 impl std::fmt::Debug for SnapshotReply {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SnapshotReply")
             .field("epoch", &self.epoch)
-            .field("document_bytes", &self.document.len())
+            .field("wanted", &self.wanted)
+            .field("total", &self.total)
+            .field("offset", &self.offset)
+            .field("data_bytes", &self.data.len())
+            .field("refused", &self.refused)
             .finish()
     }
 }