late messages don't flip peer's epoch back

1 files changed, 16 insertions, 1 deletions

diff --git a/src/state.rs b/src/state.rs
index ce47673..d14c1ea 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -127,8 +127,14 @@ pub(crate) struct PeerState {
     /// from it in between (None while they match).
     pub(crate) diverged_since: Option<Instant>,
 
-    /// Epoch the peer advertised in its most recent message.
+    /// Epoch the peer advertised in its most recent message (late ones aside: see
+    /// `epoch_seq`).
     pub(crate) epoch: Option<Epoch>,
+    /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run.
+    /// An epoch with a lower one is older than what we have, delivered late.
+    pub(crate) epoch_seq: Option<u32>,
+    /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over.
+    pub(crate) run_id: Option<crate::transport::RunId>,
 
     /// Whether the peer was ready according to its most recent sync message or keepalive.
     pub(crate) reports_ready: Option<bool>,
@@ -330,6 +336,8 @@ impl PeerState {
             last_resync_at: None,
             diverged_since: None,
             epoch: None,
+            epoch_seq: None,
+            run_id: None,
             reports_ready: None,
             syncs_with_us: None,
             reported_heads: None,
@@ -676,6 +684,12 @@ impl AppState {
         *self.epoch_tx.borrow()
     }
 
+    /// The local epoch and `Replica::epoch_seq`, read together.
+    pub(crate) async fn epoch_stamp(&self) -> (Epoch, u32) {
+        let replica = self.doc.read().await;
+        (replica.epoch, replica.epoch_seq)
+    }
+
     pub(crate) async fn is_ready(&self) -> bool {
         matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. })
     }
@@ -1213,6 +1227,7 @@ impl Host for AppState {
                     let new_doc = f(&mut replica.doc)?;
                     let old_doc = std::mem::replace(&mut replica.doc, new_doc);
                     replica.epoch = epoch;
+                    replica.epoch_seq += 1;
                     replica.refresh_heads();
                     // Updated under the same lock, so the cached epoch never disagrees
                     // with the document for anyone holding the lock.