add epochs

1 files changed, 23 insertions, 7 deletions

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