add epoch catch-up / snapshots

1 files changed, 48 insertions, 24 deletions

diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs
index 5eb1962..9f17df2 100644
--- a/src/compaction/switch.rs
+++ b/src/compaction/switch.rs
@@ -15,20 +15,22 @@
 //! The write lock is held only for step 3, which is proportional to the changes made
 //! during steps 1-2.
 //!
-//! A node whose build doesn't match the committed epoch stays on the old epoch and
-//! reports the mismatch; fetching the snapshot from peers comes with step 6.
+//! A node whose build doesn't match the committed epoch, or that never saw the commit,
+//! fetches the new epoch from a peer instead (`catchup`).
 
 use std::time::Duration;
 
-use automerge::{ActorId, AutoCommit};
+use automerge::{ActorId, AutoCommit, ChangeHash};
 use tokio::sync::broadcast;
 use tracing::{error, info, warn};
 
+use super::catchup;
 use super::delta::{self, ReplayStats};
 use super::local::LocalStatus;
 use super::records::{self, unix_millis, EpochRecord, Record};
 use super::snapshot::{self, COMPACTIONS, META_ROOT};
 use super::Host;
+use crate::replica::Epoch;
 use crate::store;
 use crate::sync::paths_related;
 
@@ -79,13 +81,15 @@ async fn check<H: Host>(host: &H) {
         .collect();
 
     let (id, record) = match committed.as_slice() {
-        [] => return,
+        // Nothing committed here; maybe peers moved on without us.
+        [] => return catchup::check(host).await,
         [(id, record)] => (id.clone(), record.clone()),
         several => {
-            // The settle protocol prevents this; never guess which one is right.
+            // The settle protocol prevents this; never guess which one is right. Peers
+            // that did switch can still be followed.
             let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect();
-            error!(compactions = ?ids, "Several compactions committed for this epoch; not switching");
-            return;
+            error!(compactions = ?ids, "Several compactions committed for this epoch; not switching on my own");
+            return catchup::check(host).await;
         }
     };
 
@@ -94,7 +98,9 @@ async fn check<H: Host>(host: &H) {
         .status(&id)
         .is_some_and(|status| status.error.is_some())
     {
-        return; // already failed; reported in the compaction's status
+        // Our own switch failed (reported in the compaction's status): fetch the epoch
+        // from a peer that made it.
+        return catchup::check(host).await;
     }
 
     let switch_delay = host.dormant_after() / 4; // one keepalive interval
@@ -148,7 +154,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
 
     // 2. Build (or reuse), verify and replay without holding any lock.
     let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking(
-        move || -> Result<(AutoCommit, Vec<automerge::ChangeHash>, ReplayStats), String> {
+        move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
             let mut copy = copy;
             let snapshot = match prebuilt {
                 Some(snapshot) => snapshot,
@@ -179,21 +185,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
     .map_err(|error| format!("switch task failed: {error}"))??;
 
     // 3. Swap, replaying what arrived since the copy.
-    let old = host
-        .switch_epoch(expected, move |live| {
-            let mut new_doc = new_doc;
-            let live_heads = live.get_heads();
-            let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc);
-            if late.replayed + late.failed > 0 {
-                info!(
-                    replayed = late.replayed,
-                    failed = late.failed,
-                    "Replayed late changes"
-                );
-            }
-            Ok::<_, String>(new_doc)
-        })
-        .await?;
+    let old = swap_in(host, expected, new_doc, copy_heads).await?;
 
     // 4. Keep the old document as configured.
     host.compaction()
@@ -208,3 +200,35 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
     );
     Ok(stats)
 }
+
+/// Under the write lock: replays into `new_doc` whatever the live document received
+/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it
+/// in as `target`. Returns the old document.
+pub(super) async fn swap_in<H: Host>(
+    host: &H,
+    target: Epoch,
+    new_doc: AutoCommit,
+    copy_heads: Vec<ChangeHash>,
+) -> Result<AutoCommit, String> {
+    host.switch_epoch(target, move |live| {
+        // Only one worker switches, but be sure the copy still describes the live document.
+        if !copy_heads
+            .iter()
+            .all(|head| live.get_change_by_hash(head).is_some())
+        {
+            return Err("the document was replaced while preparing the switch".to_owned());
+        }
+        let mut new_doc = new_doc;
+        let live_heads = live.get_heads();
+        let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc);
+        if late.replayed + late.failed > 0 {
+            info!(
+                replayed = late.replayed,
+                failed = late.failed,
+                "Replayed late changes"
+            );
+        }
+        Ok(new_doc)
+    })
+    .await
+}