1 files changed, 78 insertions, 5 deletions

diff --git a/src/peer.rs b/src/peer.rs
index 3dd344e..238c341 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -11,6 +11,7 @@ use tokio::sync::Mutex;
 use tokio::time::{Instant, MissedTickBehavior};
 use tracing::{debug, info, instrument, trace, warn};
 
+use crate::replica::{Epoch, EpochRelation};
 use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState};
 use crate::store::{self, StoreError};
 use crate::sync::{handle_sync, SyncEvent};
@@ -46,7 +47,7 @@ async fn apply_forget_records(state: &AppState) {
         return;
     }
 
-    let forgotten = forgotten_peers(&*state.doc.read().await);
+    let forgotten = forgotten_peers(&state.doc.read().await.doc);
     if forgotten.is_empty() {
         return;
     }
@@ -109,8 +110,12 @@ fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest {
 
 async fn local_keepalive(state: &AppState) -> transport::Keepalive {
     // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
-    let heads_digest = heads_digest(&mut *state.doc.write().await);
-    transport::Keepalive { heads_digest }
+    // Digest and epoch come from the same locked replica, so they always match.
+    let mut replica = state.doc.write().await;
+    transport::Keepalive {
+        heads_digest: heads_digest(&mut replica.doc),
+        epoch: replica.epoch,
+    }
 }
 
 async fn heal_if_diverged(
@@ -119,6 +124,11 @@ async fn heal_if_diverged(
     local: &transport::Keepalive,
     remote: &transport::Keepalive,
 ) {
+    // Digests of documents from different epochs are never equal and never converge.
+    if local.epoch != remote.epoch {
+        return;
+    }
+
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
         return;
     };
@@ -224,7 +234,12 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
 async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> {
     let result = state
         .transport
-        .send_message(peer, &ParsedMessage::Hello(transport::HelloRequest {}))
+        .send_message(
+            peer,
+            &ParsedMessage::Hello(transport::HelloRequest {
+                epoch: state.local_epoch(),
+            }),
+        )
         .await;
 
     if let Some(peer_state_mutex) = state.get_peer_state(peer).await {
@@ -308,7 +323,11 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
         }
     }
 
-    let reply = transport::HelloReply { cluster_id, peers };
+    let reply = transport::HelloReply {
+        cluster_id,
+        peers,
+        epoch: state.local_epoch(),
+    };
 
     let ready = state.is_ready().await;
 
@@ -482,6 +501,54 @@ pub(crate) async fn peer_cluster(state: &AppState, peer: &str) -> Option<String>
     }
 }
 
+// --- epochs ------------------------------------------------------------------
+
+/// Records the epoch a peer advertised and compares it with ours. Returns true when both
+/// are on the same epoch, the only case in which the documents may be synced.
+async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool {
+    let local = state.local_epoch();
+    let relation = local.relation_to(&remote);
+
+    let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
+        return relation == EpochRelation::Same;
+    };
+
+    let previous = {
+        let mut peer_state = peer_state_mutex.lock().await;
+        let previous = peer_state.epoch.replace(remote);
+        if previous.is_some_and(|previous| previous != remote) {
+            // The peer switched documents; our sync state describes its old one.
+            peer_state.reset_sync();
+        }
+        previous
+    };
+
+    // Log transitions only, not every keepalive.
+    if previous != Some(remote) {
+        match relation {
+            EpochRelation::Same => {
+                if previous.is_some() {
+                    info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync");
+                    let _ = state
+                        .sync_events
+                        .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+                }
+            }
+            other => {
+                warn!(
+                    peer = %peer,
+                    local_epoch = %local,
+                    peer_epoch = %remote,
+                    relation = ?other,
+                    "Peer is on a different epoch; not syncing with it"
+                );
+            }
+        }
+    }
+
+    relation == EpochRelation::Same
+}
+
 // --- inbound dispatch --------------------------------------------------------
 
 #[instrument(skip(state, message))]
@@ -509,6 +576,8 @@ pub(crate) async fn handle_message(
         info!(peer = %peer, "Dormant peer is active again; re-handshaking");
     }
 
+    let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await;
+
     debug!(message = ?message, "Received message");
 
     match message {
@@ -564,6 +633,10 @@ pub(crate) async fn handle_message(
             if reactivated {
                 ensure_handshake(state, &peer).await;
             }
+            if !same_epoch {
+                debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch");
+                return Ok(());
+            }
             handle_sync(state, &peer, envelope).await?;
         }
         transport::ParsedMessage::Ping(remote) => {