add epochs

10 files changed, 431 insertions, 43 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index 8038f16..cb0d701 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -145,6 +145,7 @@ fn response_for_value(
             peers::PeerStatus,
             peers::PeerListResponse,
             peers::PeerLivenessStatus,
+            peers::EpochInfo,
         )
     ),
     tags(
diff --git a/src/api/peers.rs b/src/api/peers.rs
index b2658a5..b12e3bf 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -15,6 +15,7 @@ use utoipa::ToSchema;
 
 use super::store_status;
 use crate::peer::forgotten_path;
+use crate::replica::Epoch;
 use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState};
 use crate::store;
 
@@ -27,6 +28,23 @@ pub(crate) enum PeerLivenessStatus {
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "camelCase")]
+pub(crate) struct EpochInfo {
+    generation: u64,
+    /// Hex-encoded snapshot hash; all zeros for the genesis epoch.
+    id: String,
+}
+
+impl From<Epoch> for EpochInfo {
+    fn from(epoch: Epoch) -> Self {
+        Self {
+            generation: epoch.generation,
+            id: epoch.id_hex(),
+        }
+    }
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
 pub(crate) struct PeerStatus {
     public_key: String,
     validated_cluster: Option<String>,
@@ -35,11 +53,16 @@ pub(crate) struct PeerStatus {
     last_seen_secs_ago: Option<u64>,
     dormant_for_secs: Option<u64>,
     has_pending_delivery: bool,
+    /// Epoch the peer last advertised; absent until it has sent a message.
+    epoch: Option<EpochInfo>,
+    /// Whether that epoch equals ours, i.e. whether we sync with this peer.
+    same_epoch: Option<bool>,
 }
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerListResponse {
+    local_epoch: EpochInfo,
     peers: Vec<PeerStatus>,
 }
 
@@ -73,6 +96,7 @@ pub(crate) async fn list_peers(
         .collect();
 
     let now = Instant::now();
+    let local_epoch = state.local_epoch();
     let mut peers = Vec::with_capacity(snapshot.len());
 
     for (public_key, peer_state_mutex) in snapshot {
@@ -99,11 +123,16 @@ pub(crate) async fn list_peers(
                 .map(|t| now.saturating_duration_since(t).as_secs()),
             dormant_for_secs,
             has_pending_delivery: ps.pending_outbound_sync.is_some(),
+            epoch: ps.epoch.map(EpochInfo::from),
+            same_epoch: ps.epoch.map(|epoch| epoch == local_epoch),
         });
     }
 
     peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
-    Ok(Json(PeerListResponse { peers }))
+    Ok(Json(PeerListResponse {
+        local_epoch: local_epoch.into(),
+        peers,
+    }))
 }
 
 #[utoipa::path(
diff --git a/src/bootstrap.rs b/src/bootstrap.rs
index 53f473f..df47ff6 100644
--- a/src/bootstrap.rs
+++ b/src/bootstrap.rs
@@ -72,24 +72,28 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
                     PeerMembership::Member { cluster_id } => Some(cluster_id.as_str()),
                 };
                 peer_diagnostics.push(format!(
-                    "{peer}: validated_cluster={:?}, pending_inbound_sync={}",
+                    "{peer}: validated_cluster={:?}, epoch={:?}, pending_inbound_sync={}",
                     validated_cluster,
+                    peer_state.epoch,
                     peer_state.pending_inbound_sync.len()
                 ))
             }
             let local_cluster = state.cluster_id().await;
+            let local_epoch = state.local_epoch();
             let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire);
             error!(
                 local_cluster = ?local_cluster,
+                local_epoch = %local_epoch,
                 initial_sync_done,
                 peers = %peer_diagnostics.join("; "),
                 timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(),
                 "Bootstrap timed out"
             );
             bail!(
-                "Bootstrap timed out after {:?}; local_cluster={:?}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.",
+                "Bootstrap timed out after {:?}; local_cluster={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.",
                 BOOTSTRAP_TIMEOUT,
                 local_cluster,
+                local_epoch,
                 initial_sync_done,
                 peer_diagnostics.join("; ")
             );
diff --git a/src/main.rs b/src/main.rs
index 9774d78..61e484e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -24,6 +24,7 @@ mod bootstrap;
 mod firewall;
 mod mycelium;
 mod peer;
+mod replica;
 mod state;
 mod store;
 mod sync;
@@ -32,6 +33,7 @@ mod transport;
 use auth::AuthState;
 use firewall::{Firewall, FirewallTransport};
 use mycelium::{MyceliumSetup, MyceliumSpawnConfig};
+use replica::{Epoch, Replica};
 use state::{AppState, NodeLifecycle, Timing};
 use transport::Transport;
 
@@ -80,6 +82,11 @@ struct Args {
     /// Keepalive interval; dormancy, resync and probe timings derive from it.
     #[arg(long, value_name = "SECONDS", default_value_t = 60)]
     keepalive_secs: u64,
+
+    /// Test hook: start on a non-genesis epoch of this generation (with a fixed dummy
+    /// id), to check that nodes on different epochs refuse to sync.
+    #[arg(long, hide = true, requires = "debug_api", value_name = "GENERATION")]
+    debug_epoch: Option<u64>,
 }
 
 #[tokio::main]
@@ -117,6 +124,16 @@ async fn main() -> Result<()> {
     let (change_tx, _) = watch::channel(0_u64);
     let (change_events, _) = broadcast::channel(256);
 
+    let initial_epoch = match args.debug_epoch {
+        Some(generation) => Epoch {
+            generation,
+            id: [0xee; 32],
+        },
+        None => Epoch::GENESIS,
+    };
+    let (epoch_tx, _) = watch::channel(initial_epoch);
+    info!(epoch = %initial_epoch, "Local epoch");
+
     let firewall = Arc::new(Firewall::default());
 
     let mycelium_transport =
@@ -124,7 +141,8 @@ async fn main() -> Result<()> {
     info!(local_id = mycelium_transport.local_id(), "Transport ready");
 
     let state = AppState {
-        doc: Arc::new(RwLock::new(AutoCommit::new())),
+        doc: Arc::new(RwLock::new(Replica::new(initial_epoch, AutoCommit::new()))),
+        epoch_tx,
         lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining {
             discovered_cluster_id: None,
         })),
@@ -228,3 +246,6 @@ async fn main() -> Result<()> {
     info!("HTTP API server stopped");
     Ok(())
 }
+
+// === Footnotes ===============================================================
+// [concurrent]: these comments are in place so we remember what to do if we go from async to concurrent.
diff --git a/src/mycelium.rs b/src/mycelium.rs
index 94c5a3d..3471f71 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -19,7 +19,7 @@ use url::Url;
 
 use crate::transport::EncodedMessage;
 use crate::transport::Keepalive;
-use crate::transport::{HelloReply, HelloRequest, SyncEnvelope};
+use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
 use crate::transport::{MessageHandler, ParsedMessage, Transport};
 
 const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
@@ -739,14 +739,54 @@ fn format_message(
     })
 }
 
+// =====================================================================
+// 1. WIRE SHADOWS
+// =====================================================================
+//
+// rmp_serde encodes structs as arrays. Fields added later are appended at the end and
+// marked #[serde(default)], so a message from a node that predates them still decodes
+// (an absent epoch is the genesis epoch).
+
+/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch.
+#[derive(Serialize, Deserialize, Default)]
+struct EpochShadow {
+    generation: u64,
+    id: ByteBuf,
+}
+
+fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow {
+    EpochShadow {
+        generation: epoch.generation,
+        id: ByteBuf::from(epoch.id.to_vec()),
+    }
+}
+
+fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> {
+    if shadow.generation == 0 && shadow.id.is_empty() {
+        return Ok(Epoch::GENESIS);
+    }
+    let id = shadow
+        .id
+        .as_slice()
+        .try_into()
+        .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?;
+    Ok(Epoch {
+        generation: shadow.generation,
+        id,
+    })
+}
+
 #[derive(Serialize, Deserialize)]
 struct KeepaliveShadow {
     heads_digest: ByteBuf,
+    #[serde(default)]
+    epoch: EpochShadow,
 }
 
 fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&KeepaliveShadow {
         heads_digest: ByteBuf::from(msg.heads_digest.to_vec()),
+        epoch: epoch_to_shadow(&msg.epoch),
     })
 }
 
@@ -757,11 +797,17 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
         .as_slice()
         .try_into()
         .map_err(|_| TransportError::InvalidPayload("heads digest must be 32 bytes".into()))?;
-    Ok(Keepalive { heads_digest })
+    Ok(Keepalive {
+        heads_digest,
+        epoch: epoch_from_shadow(shadow.epoch)?,
+    })
 }
 
 #[derive(Serialize, Deserialize)]
-pub struct HelloRequestShadow {}
+pub struct HelloRequestShadow {
+    #[serde(default)]
+    epoch: EpochShadow,
+}
 
 #[derive(Serialize, Deserialize)]
 pub struct HelloReplyShadow<'a> {
@@ -769,6 +815,8 @@ pub struct HelloReplyShadow<'a> {
     pub cluster_id: Cow<'a, str>,
     #[serde(default)]
     pub peers: Vec<String>,
+    #[serde(default)]
+    epoch: EpochShadow,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -776,6 +824,8 @@ pub struct SyncEnvelopeShadow<'a> {
     pub cluster_id: Cow<'a, str>,
     #[serde(with = "automerge_msg_codec")]
     pub message: automerge::sync::Message,
+    #[serde(default)]
+    epoch: EpochShadow,
 }
 
 // =====================================================================
@@ -808,12 +858,16 @@ mod automerge_msg_codec {
 // =====================================================================
 
 // --- HelloRequest ---
-pub fn hello_request_to_bytes(_msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    rmp_serde::to_vec(&HelloRequestShadow {})
+pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    rmp_serde::to_vec(&HelloRequestShadow {
+        epoch: epoch_to_shadow(&msg.epoch),
+    })
 }
-pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, rmp_serde::decode::Error> {
-    let _: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
-    Ok(HelloRequest {})
+pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
+    let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
+    Ok(HelloRequest {
+        epoch: epoch_from_shadow(shadow.epoch)?,
+    })
 }
 
 // --- HelloReply ---
@@ -821,14 +875,16 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
     let shadow = HelloReplyShadow {
         cluster_id: Cow::Borrowed(&msg.cluster_id),
         peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
+        epoch: epoch_to_shadow(&msg.epoch),
     };
     rmp_serde::to_vec(&shadow)
 }
-pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, rmp_serde::decode::Error> {
+pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
     let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
     Ok(HelloReply {
         cluster_id: shadow.cluster_id.into_owned(),
         peers: shadow.peers,
+        epoch: epoch_from_shadow(shadow.epoch)?,
     })
 }
 
@@ -837,13 +893,15 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::
     let shadow = SyncEnvelopeShadow {
         cluster_id: Cow::Borrowed(&msg.cluster_id),
         message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
+        epoch: epoch_to_shadow(&msg.epoch),
     };
     rmp_serde::to_vec(&shadow)
 }
-pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, rmp_serde::decode::Error> {
+pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
     let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
     Ok(SyncEnvelope {
         cluster_id: shadow.cluster_id.into_owned(),
+        epoch: epoch_from_shadow(shadow.epoch)?,
         message: shadow.message,
     })
 }
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) => {
diff --git a/src/replica.rs b/src/replica.rs
new file mode 100644
index 0000000..4579029
--- /dev/null
+++ b/src/replica.rs
@@ -0,0 +1,130 @@
+//! The local replica: the Automerge document together with the epoch it belongs to.
+//!
+//! Compaction (later steps) replaces the document with a snapshot whose history is
+//! unrelated to the old one. Syncing two unrelated histories would make Automerge merge
+//! them, turning every key into a conflict with its own copy, so nodes on different
+//! epochs must never sync. Keeping the epoch and the document under one lock makes it
+//! impossible to label a sync message from one document with the other's epoch.
+
+use std::fmt;
+use std::ops::{Deref, DerefMut};
+
+use automerge::AutoCommit;
+
+/// One generation of the cluster document.
+///
+/// * `generation` orders epochs: every compaction increments it, so a node can tell
+///   whether a peer is behind or ahead.
+/// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same
+///   generation apart, which only happens if two compactions committed concurrently.
+#[derive(Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Epoch {
+    pub generation: u64,
+    pub id: [u8; 32],
+}
+
+/// How a peer's epoch relates to ours.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum EpochRelation {
+    Same,
+    /// The peer is on an older generation and has to migrate.
+    PeerBehind,
+    /// The peer is on a newer generation; we have to migrate.
+    PeerAhead,
+    /// Same generation, different snapshot: competing compactions. Never sync.
+    Diverged,
+}
+
+impl Epoch {
+    /// Clusters that have never been compacted.
+    pub const GENESIS: Epoch = Epoch {
+        generation: 0,
+        id: [0; 32],
+    };
+
+    pub fn relation_to(&self, peer: &Epoch) -> EpochRelation {
+        use std::cmp::Ordering;
+        match peer.generation.cmp(&self.generation) {
+            Ordering::Less => EpochRelation::PeerBehind,
+            Ordering::Greater => EpochRelation::PeerAhead,
+            Ordering::Equal if peer.id == self.id => EpochRelation::Same,
+            Ordering::Equal => EpochRelation::Diverged,
+        }
+    }
+
+    pub fn id_hex(&self) -> String {
+        self.id.iter().map(|byte| format!("{byte:02x}")).collect()
+    }
+}
+
+impl fmt::Display for Epoch {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if *self == Epoch::GENESIS {
+            write!(f, "0:genesis")
+        } else {
+            write!(f, "{}:{}", self.generation, &self.id_hex()[..12])
+        }
+    }
+}
+
+impl fmt::Debug for Epoch {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(self, f)
+    }
+}
+
+/// The document and its epoch, always locked and swapped together.
+///
+/// Derefs to the document, so code that only needs the `AutoCommit` keeps working
+/// with a guard of `Replica`.
+pub struct Replica {
+    pub epoch: Epoch,
+    pub doc: AutoCommit,
+}
+
+impl Replica {
+    pub fn new(epoch: Epoch, doc: AutoCommit) -> Self {
+        Self { epoch, doc }
+    }
+}
+
+impl Deref for Replica {
+    type Target = AutoCommit;
+
+    fn deref(&self) -> &AutoCommit {
+        &self.doc
+    }
+}
+
+impl DerefMut for Replica {
+    fn deref_mut(&mut self) -> &mut AutoCommit {
+        &mut self.doc
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn epoch(generation: u64, byte: u8) -> Epoch {
+        Epoch {
+            generation,
+            id: [byte; 32],
+        }
+    }
+
+    #[test]
+    fn relation() {
+        let local = epoch(3, 1);
+        assert_eq!(local.relation_to(&epoch(3, 1)), EpochRelation::Same);
+        assert_eq!(local.relation_to(&epoch(2, 9)), EpochRelation::PeerBehind);
+        assert_eq!(local.relation_to(&epoch(4, 9)), EpochRelation::PeerAhead);
+        assert_eq!(local.relation_to(&epoch(3, 2)), EpochRelation::Diverged);
+    }
+
+    #[test]
+    fn display() {
+        assert_eq!(Epoch::GENESIS.to_string(), "0:genesis");
+        assert_eq!(epoch(2, 0xab).to_string(), "2:abababababab");
+    }
+}
diff --git a/src/state.rs b/src/state.rs
index 3fe1790..e5fe25c 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -12,6 +12,7 @@ use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
 use crate::firewall::Firewall;
+use crate::replica::{Epoch, Replica};
 use crate::sync::{ChangeSet, SyncEvent};
 use crate::transport;
 
@@ -94,6 +95,9 @@ pub(crate) struct PeerState {
 
     /// Last self-healing re-handshake (cooldown).
     pub(crate) last_resync_at: Option<Instant>,
+
+    /// Epoch the peer advertised in its most recent message.
+    pub(crate) epoch: Option<Epoch>,
 }
 
 impl Default for PeerState {
@@ -112,6 +116,7 @@ impl Default for PeerState {
             last_ping_at: None,
             last_sync_rx: None,
             last_resync_at: None,
+            epoch: None,
         }
     }
 }
@@ -245,7 +250,12 @@ impl Timing {
 
 #[derive(Clone)]
 pub(crate) struct AppState {
-    pub(crate) doc: Arc<RwLock<AutoCommit>>,
+    /// The document and its epoch. Derefs to `AutoCommit`.
+    pub(crate) doc: Arc<RwLock<Replica>>,
+
+    /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers).
+    /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`.
+    pub(crate) epoch_tx: watch::Sender<Epoch>,
 
     pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>,
 
@@ -272,6 +282,12 @@ impl AppState {
         self.transport.local_id()
     }
 
+    /// The local epoch. Code that labels or applies sync messages must use the epoch
+    /// of the locked `Replica` instead, so label and document can't disagree.
+    pub(crate) fn local_epoch(&self) -> Epoch {
+        *self.epoch_tx.borrow()
+    }
+
     pub(crate) async fn is_ready(&self) -> bool {
         matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. })
     }
@@ -405,7 +421,7 @@ impl AppState {
         tokio::task::spawn_blocking(move || {
             let data = {
                 let guard = doc.blocking_read();
-                read(&guard)
+                read(&guard.doc)
             };
             finish(data)
         })
@@ -422,7 +438,7 @@ impl AppState {
         T: Send + 'static,
     {
         let doc = Arc::clone(&self.doc);
-        tokio::task::spawn_blocking(move || f(&mut doc.blocking_write()))
+        tokio::task::spawn_blocking(move || f(&mut doc.blocking_write().doc))
             .await
             .expect("document write task panicked")
     }
@@ -442,16 +458,16 @@ impl AppState {
 
         tokio::task::spawn_blocking(move || {
             let result = {
-                let mut doc = state.doc.blocking_write();
-                let result = f(&mut doc);
+                let mut replica = state.doc.blocking_write();
+                let result = f(&mut replica.doc);
                 match &result {
                     // One API write = one Automerge change.
                     Ok(_) => {
-                        doc.commit();
+                        replica.doc.commit();
                     }
                     // Discard partial ops (e.g. old value deleted, new one not written).
                     Err(_) => {
-                        doc.rollback();
+                        replica.doc.rollback();
                     }
                 }
                 result
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");
     }
diff --git a/src/transport.rs b/src/transport.rs
index 95d3046..9cfb026 100644
--- a/src/transport.rs
+++ b/src/transport.rs
@@ -5,11 +5,15 @@ use std::boxed::Box;
 use std::future::Future;
 use std::pin::Pin;
 
+pub use crate::replica::Epoch;
+
 pub type HeadsDigest = [u8; 32];
 
 #[derive(Debug, Clone)]
 pub struct Keepalive {
     pub heads_digest: HeadsDigest,
+    /// The digest is only comparable between nodes on the same epoch.
+    pub epoch: Epoch,
 }
 
 #[derive(Debug)]
@@ -21,18 +25,35 @@ pub enum ParsedMessage {
     Pong(Keepalive),
 }
 
-#[derive(Debug)]
-pub struct HelloRequest {}
+impl ParsedMessage {
+    /// Every message carries the sender's epoch.
+    pub fn epoch(&self) -> Epoch {
+        match self {
+            ParsedMessage::Hello(hello) => hello.epoch,
+            ParsedMessage::HelloReply(reply) => reply.epoch,
+            ParsedMessage::Sync(envelope) => envelope.epoch,
+            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct HelloRequest {
+    pub epoch: Epoch,
+}
 
 #[derive(Debug, Clone)]
 pub struct HelloReply {
     pub cluster_id: String,
     pub peers: Vec<String>,
+    pub epoch: Epoch,
 }
 
 #[derive(Debug, Clone)]
 pub struct SyncEnvelope {
     pub cluster_id: String,
+    /// Epoch of the document the message was generated from.
+    pub epoch: Epoch,
     pub message: Message,
 }