1 files changed, 96 insertions, 0 deletions
diff --git a/src/swarm/peer/epoch.rs b/src/swarm/peer/epoch.rs
new file mode 100644
index 0000000..56711b1
--- /dev/null
+++ b/src/swarm/peer/epoch.rs
@@ -0,0 +1,96 @@
+//! The epochs peers advertise: recorded per peer, late messages aside, and compared
+//! with ours.
+
+use tokio::time::Instant;
+use tracing::{debug, info, warn};
+
+use crate::document::replica::{Epoch, EpochRelation};
+use crate::state::AppState;
+use crate::swarm::sync::SyncEvent;
+use crate::transport;
+
+/// Records the epoch a peer advertised and compares it with ours. Returns true when the
+/// message's epoch is ours, the only case in which its documents may be synced.
+///
+/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was
+/// sent before the peer's last epoch change and delivered late: its epoch isn't
+/// recorded, so it can't flip our view of the peer back. The count starts over when the
+/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`).
+pub(super) async fn observe_peer_epoch(
+    state: &AppState,
+    peer: &str,
+    remote: Epoch,
+    seq: u32,
+    run: Option<transport::RunId>,
+) -> 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;
+        if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) {
+            peer_state.run_id = Some(run);
+            peer_state.epoch_seq = None;
+        }
+        if peer_state.epoch_seq.is_some_and(|newest| seq < newest) {
+            debug!(
+                peer = %peer,
+                epoch = %remote,
+                seq,
+                newest = ?peer_state.epoch_seq,
+                "Ignoring a late message's epoch"
+            );
+            return relation == EpochRelation::Same;
+        }
+        peer_state.epoch_seq = Some(seq);
+        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. Count
+            // the restart as sync activity, so the keepalive heal doesn't re-handshake
+            // (resetting both sides again) while this fresh sync is under way.
+            peer_state.reset_sync();
+            peer_state.last_sync_rx = Some(Instant::now());
+            // Whether it would sync with us was said about its old epoch.
+            peer_state.syncs_with_us = None;
+        }
+        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()));
+                }
+            }
+            // Expected after a remake: either side switches or catches up.
+            EpochRelation::PeerAhead | EpochRelation::PeerBehind => {
+                info!(
+                    peer = %peer,
+                    local_epoch = %local,
+                    peer_epoch = %remote,
+                    relation = ?relation,
+                    "Peer is on another epoch; not syncing with it until both are on the same"
+                );
+            }
+            EpochRelation::Diverged => {
+                warn!(
+                    peer = %peer,
+                    local_epoch = %local,
+                    peer_epoch = %remote,
+                    "Peer is on a different epoch of the same generation; not syncing with it"
+                );
+            }
+        }
+    }
+
+    relation == EpochRelation::Same
+}