From a02ed87abf6a0dc9d3f36bafcf333272586093aa Mon Sep 17 00:00:00 2001 From: tv 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 ++++++++----- src/transport/mod.rs | 69 +++++++++++++++++++++++++++++++++++++++++++---- src/transport/wire.rs | 17 ++++++++++-- 4 files changed, 113 insertions(+), 16 deletions(-) (limited to 'src') 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) -> Option { + 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(value: &[u8; 8], s: S) -> Result { + 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 { 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(), -- cgit v1.3.1