add epochs

1 files changed, 50 insertions, 15 deletions

diff --git a/src/sync.rs b/src/sync.rs
index f1500cd..199f9fb 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -50,6 +50,15 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
             return Ok(());
         }
 
+        if peer_state
+            .epoch
+            .is_some_and(|epoch| epoch != state.local_epoch())
+        {
+            // Unrelated document histories must never be synced.
+            trace!("Skipping synchronization with peer on another epoch");
+            return Ok(());
+        }
+
         match &peer_state.membership {
             PeerMembership::Unknown => return Ok(()),
             PeerMembership::Member { cluster_id } => cluster_id.clone(),
@@ -94,18 +103,22 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
 
             let generated = tokio::task::spawn_blocking(
                 move || -> Result<Option<transport::EncodedMessage>> {
-                    let message = {
-                        let mut doc = doc.blocking_write();
+                    // Label the message with the epoch of the document it was generated
+                    // from, read under the same lock.
+                    let (epoch, message) = {
+                        let mut replica = doc.blocking_write();
                         let mut peer_state = psm.blocking_lock();
                         peer_state.needs_generate = false;
-                        let sync_doc = doc.sync();
-                        sync_doc.generate_sync_message(&mut peer_state.sync_state)
+                        let epoch = replica.epoch;
+                        let sync_doc = replica.doc.sync();
+                        (epoch, sync_doc.generate_sync_message(&mut peer_state.sync_state))
                     };
                     let Some(message) = message else {
                         return Ok(None);
                     };
                     let envelope = transport::SyncEnvelope {
                         cluster_id,
+                        epoch,
                         message,
                     };
                     transport
@@ -198,7 +211,9 @@ pub(crate) async fn handle_sync(
         match &peer_state.membership {
             PeerMembership::Unknown => {
                 let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| {
-                    pending.message == envelope.message && pending.cluster_id == envelope.cluster_id
+                    pending.message == envelope.message
+                        && pending.cluster_id == envelope.cluster_id
+                        && pending.epoch == envelope.epoch
                 });
                 if !duplicate {
                     peer_state.pending_inbound_sync.push(envelope.clone());
@@ -242,32 +257,52 @@ pub(crate) async fn handle_sync(
         );
     }
 
-    let caught_up = {
+    let epoch_label = envelope.epoch;
+
+    let applied = {
         let doc = Arc::clone(&state.doc);
         let peer_state_mutex = Arc::clone(&peer_state_mutex);
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
+        let epoch = envelope.epoch;
         let message = envelope.message;
 
-        tokio::task::spawn_blocking(move || -> Result<bool> {
-            let mut doc = doc.blocking_write();
+        // Some(caught_up) when applied, None when the message belongs to another epoch.
+        tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
+            let mut replica = doc.blocking_write();
+
+            // Authoritative check, under the same lock as the apply: the document may
+            // have switched epochs since the message was accepted.
+            if replica.epoch != epoch {
+                return Ok(None);
+            }
+
             let mut peer_state = peer_state_mutex.blocking_lock();
 
-            doc.sync()
+            replica
+                .doc
+                .sync()
                 .receive_sync_message(&mut peer_state.sync_state, message)
                 .context("Automerge synchronization failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
 
-            Ok(check_caught_up
-                && peer_state
-                    .sync_state
-                    .their_heads
-                    .as_ref()
-                    .is_some_and(|heads| heads.iter().all(|h| doc.get_change_by_hash(h).is_some())))
+            Ok(Some(
+                check_caught_up
+                    && peer_state.sync_state.their_heads.as_ref().is_some_and(|heads| {
+                        heads
+                            .iter()
+                            .all(|h| replica.doc.get_change_by_hash(h).is_some())
+                    }),
+            ))
         })
         .await
         .context("Sync task panicked")??
     };
 
+    let Some(caught_up) = applied else {
+        debug!(epoch = %epoch_label, "Dropped sync message from another epoch");
+        return Ok(());
+    };
+
     if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
         info!(peer = %peer, "Initial sync complete");
     }