From a02ed87abf6a0dc9d3f36bafcf333272586093aa Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Thu, 5 Nov 2026 03:50:38 +0100
Subject: epoch tag in sync

---
 src/swarm/peer/inbound.rs | 24 +++++++++++++++++++++---
 src/swarm/sync.rs         | 19 +++++++++++++------
 2 files changed, 34 insertions(+), 9 deletions(-)

(limited to 'src/swarm')

diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index f43a45a..c75f66b 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -38,8 +38,26 @@ pub(crate) async fn handle_message(
         note_alive(state, &peer, &message).await;
         return Ok(());
     }
-    // WHO and ME end at the identity layer (`transport::identity`).
-    let Some((epoch, epoch_seq)) = message.epoch_stamp() else {
+    // WHO and ME end at the identity layer (`transport::identity`). A sync message names
+    // its epoch by a tag (`EpochTag`): ours, or the sender's as we know it; one naming
+    // neither is of an epoch we don't sync on, and says nothing we could record.
+    let stamp = match &message {
+        transport::ParsedMessage::Sync(envelope) => {
+            let known = match state.get_peer_state(&peer).await {
+                Some(peer_state) => peer_state.lock().await.epoch,
+                None => None,
+            };
+            let epoch = envelope
+                .epoch
+                .resolve(std::iter::once(state.local_epoch()).chain(known));
+            if epoch.is_none() {
+                debug!(peer = %peer, epoch = %envelope.epoch, "Dropping a sync message of an epoch we don't know");
+            }
+            epoch.map(|epoch| (epoch, envelope.epoch_seq))
+        }
+        _ => message.epoch_stamp(),
+    };
+    let Some((epoch, epoch_seq)) = stamp else {
         return Ok(());
     };
 
@@ -145,7 +163,7 @@ pub(crate) async fn handle_message(
                 ensure_handshake(state, &peer).await;
             }
             if !same_epoch {
-                debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch");
+                debug!(peer = %peer, epoch = %epoch, "Dropping sync message from another epoch");
                 return Ok(());
             }
             crate::swarm::source::synced_from(state, &peer);
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 28f3ca0..394e589 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -239,14 +239,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     return (session, Ok(None));
                 };
                 // What the envelope adds to an Automerge message (`swarm::stats`), and of
-                // that, the epoch (msgpack: 36 bytes).
+                // that, the epoch's tag (msgpack: 12 bytes, up to 20 for generations past
+                // 127).
                 let inner = match &body {
                     transport::SyncBody::Message(payload) => Some(payload.0.len()),
                     transport::SyncBody::Delta(_) => None,
                 };
-                let ids = 36;
+                let ids = rmp_serde::encode::to_vec(&transport::EpochTag::of(epoch))
+                    .map_or(0, |tag| tag.len());
                 let envelope = transport::SyncEnvelope {
-                    epoch,
+                    epoch: transport::EpochTag::of(epoch),
                     epoch_seq,
                     body,
                     ready,
@@ -406,14 +408,20 @@ pub(crate) async fn handle_sync(
         return Ok(());
     }
 
-    let epoch_label = envelope.epoch;
+    // The tag names our epoch, or the message is of no epoch we sync on (the sender
+    // switched meanwhile, or we did).
+    let Some(epoch) = envelope.epoch.resolve([state.local_epoch()]) else {
+        debug!(peer = %peer, epoch = %envelope.epoch, "Dropped sync message from another epoch");
+        return Ok(());
+    };
+    let epoch_label = epoch;
     let from_ready = envelope.ready;
     let writer = envelope.writer;
     let delivered = envelope.delivered;
     let payload = match envelope.body {
         transport::SyncBody::Message(payload) => payload,
         transport::SyncBody::Delta(spec) => {
-            fetch_delta(state, peer, envelope.epoch, spec);
+            fetch_delta(state, peer, epoch, spec);
             return Ok(());
         }
     };
@@ -435,7 +443,6 @@ pub(crate) async fn handle_sync(
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
         // While catching up, everything arrives late; that says nothing about propagation.
         let measure_propagation = !check_caught_up;
-        let epoch = envelope.epoch;
         let message = payload;
         let peer = peer.to_owned();
         let clock = Arc::clone(&state.clock);
-- 
cgit v1.3.1