epoch tag in sync

5 files changed, 117 insertions, 17 deletions

diff --git a/docs/peers.md b/docs/peers.md
index e86e530..7213e70 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -104,7 +104,10 @@ epoch, and lost or stale sync state is repaired by the reply timeout, the `need`
 the heal. Keepalives only inform.
 
 Every message carries the sender's epoch, and with it how many times the sender changed
-epoch in its current run. A node records a peer's epoch only from a message whose count
+epoch in its current run. Sync messages, which go often and only between peers on one
+epoch, carry a tag of it instead: its generation and the first 8 bytes of its ID. The
+receiver takes it for its own epoch, or the sender's as it last knew it, whichever it
+names; one that names neither is dropped. A node records a peer's epoch only from a message whose count
 is at least the newest it has seen from that peer, so a message delayed from before
 the peer switched (or rejoined, or gave up a rejoin) can't flip its recorded epoch
 back. The count starts over when the peer restarts, which its HELLO or HELLO_REPLY
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);
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 3fd4490..32be29b 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -389,12 +389,14 @@ impl ParsedMessage {
     }
 
     /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them
-    /// but WHO, ME, PING and PONG (which the node never sees).
+    /// but WHO, ME, PING and PONG (which the node never sees), and SYNC, which carries a
+    /// tag of it only (`EpochTag`).
     pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> {
         match self {
             ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)),
             ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)),
-            ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)),
+            // Only a tag (`EpochTag`): resolved by the receiver, which knows the epochs.
+            ParsedMessage::Sync(_) => None,
             ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => {
                 Some((status.epoch, status.epoch_seq))
             }
@@ -656,9 +658,9 @@ impl std::fmt::Debug for HelloReply {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(crate) struct SyncEnvelope {
-    /// Epoch of the document the message was generated from.
-    #[serde(with = "crate::transport::wire::codec::epoch")]
-    pub(crate) epoch: Epoch,
+    /// Epoch of the document the message was generated from, by a tag (`EpochTag`): the
+    /// receiver knows the epoch it names, as ours or as the sender's.
+    pub(crate) epoch: EpochTag,
     /// How many times the sender changed epoch in its current run, when it sent this
     /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver
     /// ignores an epoch older than the newest it has seen in the same session.
@@ -682,6 +684,46 @@ pub(crate) struct SyncEnvelope {
     pub(crate) sent: crate::clock::UnixMillis,
 }
 
+/// An epoch, as sync messages name it: its generation and the first bytes of its ID,
+/// about a third of the whole. Sync messages go often, and only between peers on one
+/// epoch, which each knows from the other's handshake and keepalives: the tag tells which
+/// of the epochs the receiver knows it is, and one it knows of no other is dropped (an
+/// epoch of the same generation agreeing in 8 bytes, but not the rest, would take
+/// competing remakes that also collide there).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct EpochTag {
+    pub(crate) generation: u64,
+    #[serde(with = "crate::transport::wire::codec::bytes8")]
+    pub(crate) prefix: [u8; 8],
+}
+
+impl EpochTag {
+    pub(crate) fn of(epoch: Epoch) -> Self {
+        let mut prefix = [0; 8];
+        prefix.copy_from_slice(&epoch.id[..8]);
+        Self {
+            generation: epoch.generation,
+            prefix,
+        }
+    }
+
+    /// Whether this tags `epoch`.
+    pub(crate) fn names(&self, epoch: &Epoch) -> bool {
+        *self == Self::of(*epoch)
+    }
+
+    /// The first of `known` it names, if any.
+    pub(crate) fn resolve(&self, known: impl IntoIterator<Item = Epoch>) -> Option<Epoch> {
+        known.into_iter().find(|epoch| self.names(epoch))
+    }
+}
+
+impl std::fmt::Display for EpochTag {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}:{}…", self.generation, Hex(&self.prefix))
+    }
+}
+
 /// What a sync envelope carries.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(crate) enum SyncBody {
@@ -1042,4 +1084,21 @@ mod tests {
         }
         assert_eq!(seen.len(), heads.len());
     }
+
+    /// A tag names its epoch, not one of another generation or another ID, and resolves
+    /// to the epoch it names among those known.
+    #[test]
+    fn epoch_tags_name_their_epoch() {
+        let epoch = |generation, byte| Epoch {
+            generation,
+            id: [byte; 32],
+        };
+        let ours = epoch(3, 1);
+        let tag = EpochTag::of(ours);
+        assert!(tag.names(&ours));
+        assert!(!tag.names(&epoch(4, 1)), "another generation");
+        assert!(!tag.names(&epoch(3, 2)), "another remake of it");
+        assert_eq!(tag.resolve([epoch(3, 2), ours]), Some(ours));
+        assert_eq!(tag.resolve([epoch(2, 1)]), None);
+    }
 }
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 3f7736d..d1ac14d 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 29;
+pub(crate) const VERSION: u8 = 30;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -166,6 +166,19 @@ pub(crate) mod codec {
         }
     }
 
+    /// 8 bytes (`EpochTag::prefix`).
+    pub(crate) mod bytes8 {
+        use super::*;
+
+        pub(crate) fn serialize<S: Serializer>(value: &[u8; 8], s: S) -> Result<S::Ok, S::Error> {
+            Bytes::new(value).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 8], D::Error> {
+            fixed(&ByteBuf::deserialize(d)?, "an epoch tag")
+        }
+    }
+
     /// A 64-byte signature.
     pub(crate) mod bytes64 {
         use super::*;
@@ -401,7 +414,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             sender_run_id: [6; 16],
         }),
         ParsedMessage::Sync(SyncEnvelope {
-            epoch,
+            epoch: crate::transport::EpochTag::of(epoch),
             epoch_seq: 1,
             body: SyncBody::Delta(DeltaSpec {
                 from: Vec::new(),