From 944bd3a9da9c14153247b483ba2cfe1457fc84d2 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 20:33:40 +0100 Subject: late messages don't flip peer's epoch back --- src/document/replica.rs | 4 ++++ src/state.rs | 17 ++++++++++++++- src/swarm/blob.rs | 8 +++++-- src/swarm/peer.rs | 53 ++++++++++++++++++++++++++++++++++++++++------- src/swarm/sync.rs | 7 ++++--- src/transport/mod.rs | 25 ++++++++++++++++++++++ src/transport/mycelium.rs | 40 ++++++++++++++++++++++++----------- 7 files changed, 129 insertions(+), 25 deletions(-) diff --git a/src/document/replica.rs b/src/document/replica.rs index e24f93c..db47223 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -86,6 +86,9 @@ impl fmt::Debug for Epoch { /// [`Replica::heads`]. pub struct Replica { pub epoch: Epoch, + /// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this + /// run: messages carry it, so a late one can't flip a peer's view of our epoch back. + pub epoch_seq: u32, pub doc: AutoCommit, heads: Vec, } @@ -94,6 +97,7 @@ impl Replica { pub fn new(epoch: Epoch, doc: AutoCommit) -> Self { let mut replica = Self { epoch, + epoch_seq: 0, doc, heads: Vec::new(), }; 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, - /// 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, + /// 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, + /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over. + pub(crate) run_id: Option, /// Whether the peer was ready according to its most recent sync message or keepalive. pub(crate) reports_ready: Option, @@ -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. diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 8c891ae..42cfbf3 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -171,8 +171,10 @@ async fn prepare(state: &AppState, id: BlobId) -> Option> { } async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { + let (epoch, epoch_seq) = state.epoch_stamp().await; let message = ParsedMessage::Blob(BlobMessage { - epoch: state.local_epoch(), + epoch, + epoch_seq, id, body, }); @@ -223,8 +225,10 @@ impl Drop for Registration<'_> { } async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec) { + let (epoch, epoch_seq) = state.epoch_stamp().await; let message = ParsedMessage::Blob(BlobMessage { - epoch: state.local_epoch(), + epoch, + epoch_seq, id, body: BlobBody::Request { chunks }, }); diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index daf1af4..8c1e7eb 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -110,11 +110,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { // Heads and epoch come from the same locked replica, so they always match. The heads // are cached (`Replica`), so the read lock is enough. - let (heads, epoch) = { + let (heads, epoch, epoch_seq) = { let replica = state.doc.read().await; ( transport::HeadsSummary::of(replica.heads().to_vec()), replica.epoch, + replica.epoch_seq, ) }; let syncs_with_you = match state.get_peer_state(peer).await { @@ -129,6 +130,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: transport::Keepalive { heads, epoch, + epoch_seq, ready: state.is_ready().await, syncs_with_you, } @@ -276,6 +278,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) .unwrap_or_default(); + let (epoch, epoch_seq) = state.epoch_stamp().await; state.lock_rtt().hello_sent(peer, Instant::now()); let result = state .transport @@ -283,7 +286,8 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { peer, &ParsedMessage::Hello(transport::HelloRequest { swarm_id, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready: state.is_ready().await, run_id: state.run_id, @@ -354,14 +358,17 @@ async fn send_hello_reply( .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); let ready = state.is_ready().await; + let (epoch, epoch_seq) = state.epoch_stamp().await; let reply = transport::HelloReply { swarm_id, peers, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready, run_id, away_ms: away.map(|away| away.as_millis() as u64), + sender_run_id: state.run_id, }; debug!(to = %peer, ready, "Sending HELLO_REPLY"); @@ -582,9 +589,20 @@ pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option { // --- epochs ------------------------------------------------------------------ -/// Records the epoch a peer advertised and compares it with ours. Returns true when both -/// are on the same epoch, the only case in which the documents may be synced. -async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool { +/// 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`). +async fn observe_peer_epoch( + state: &AppState, + peer: &str, + remote: Epoch, + seq: u32, + run: Option, +) -> bool { let local = state.local_epoch(); let relation = local.relation_to(&remote); @@ -594,6 +612,21 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool 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 @@ -758,7 +791,13 @@ pub(crate) async fn handle_message( info!(peer = %peer, "Dormant peer is active again; re-handshaking"); } - let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await; + let run = match &message { + transport::ParsedMessage::Hello(hello) => Some(hello.run_id), + transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id), + _ => None, + }; + let same_epoch = + observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await; //debug!(from = %peer, message = ?message, "Received message"); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 80d6d56..93af6f7 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -141,10 +141,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { tokio::task::spawn_blocking(move || -> Result> { // Label the message with the epoch of the document it was generated // from, read under the same lock. - let (epoch, message) = { + let (epoch, epoch_seq, message) = { let mut replica = doc.blocking_write(); let mut peer_state = psm.blocking_lock(); - let epoch = replica.epoch; + let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); let sent_before = peer_state.sync.state.sent_hashes.clone(); let message = { let sync_doc = replica.doc.sync(); @@ -178,7 +178,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { regenerate: false, }); } - (epoch, message) + (epoch, epoch_seq, message) }; let Some(message) = message else { return Ok(None); @@ -186,6 +186,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let envelope = transport::SyncEnvelope { swarm_id, epoch, + epoch_seq, message, ready, }; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index c5769ca..564aada 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -86,6 +86,8 @@ pub struct Keepalive { /// Only comparable between nodes on the same epoch. pub heads: HeadsSummary, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub ready: bool, /// The sender would sync with the recipient: counts it as an active member on its @@ -131,6 +133,8 @@ impl std::fmt::Display for BlobId { pub struct BlobMessage { /// The sender's epoch (every message carries it). pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, pub id: BlobId, pub body: BlobBody, } @@ -215,6 +219,17 @@ impl ParsedMessage { ParsedMessage::Blob(blob) => blob.epoch, } } + + /// `SyncEnvelope::epoch_seq`, as every message carries it. + pub fn epoch_seq(&self) -> u32 { + match self { + ParsedMessage::Hello(hello) => hello.epoch_seq, + ParsedMessage::HelloReply(reply) => reply.epoch_seq, + ParsedMessage::Sync(envelope) => envelope.epoch_seq, + ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq, + ParsedMessage::Blob(blob) => blob.epoch_seq, + } + } } #[derive(Debug, Clone)] @@ -223,6 +238,8 @@ pub struct HelloRequest { /// another swarm doesn't answer. pub swarm_id: Option, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec, /// The sender is ready (same meaning as `SyncEnvelope::ready`). @@ -236,6 +253,8 @@ pub struct HelloReply { pub swarm_id: String, pub peers: Vec, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec, /// The sender is ready. @@ -246,6 +265,8 @@ pub struct HelloReply { /// How long before this HELLO the sender had last heard from the recipient, if ever /// (in milliseconds): tells a restarted node how long it was away. pub away_ms: Option, + /// The sender's own run (as its HELLOs carry it): a new one means it restarted. + pub sender_run_id: RunId, } #[derive(Debug, Clone)] @@ -253,6 +274,10 @@ pub struct SyncEnvelope { pub swarm_id: String, /// Epoch of the document the message was generated from. pub epoch: Epoch, + /// 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. + pub epoch_seq: u32, pub message: Message, /// The sender was ready when it generated the message: its document is the swarm's, /// not one it is still filling (joining, rejoining). Only such a peer can tell a diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 82e6016..f519928 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -802,16 +802,20 @@ fn format_message( struct EpochShadow { generation: u64, id: ByteBuf, + /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`). + #[serde(default)] + seq: u32, } -fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { +fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow { EpochShadow { generation: epoch.generation, id: ByteBuf::from(epoch.id.to_vec()), + seq, } } -fn epoch_from_shadow(shadow: EpochShadow) -> Result { +fn epoch_from_shadow(shadow: &EpochShadow) -> Result { if shadow.generation == 0 && shadow.id.is_empty() { return Ok(Epoch::GENESIS); } @@ -850,7 +854,7 @@ fn keepalive_to_bytes(msg: &Keepalive) -> Result, rmp_serde::encode::Err rmp_serde::to_vec(&KeepaliveShadow { heads, digest, - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, syncs_with_you: msg.syncs_with_you, }) @@ -880,7 +884,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result { }; Ok(Keepalive { heads, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, ready: shadow.ready, syncs_with_you: shadow.syncs_with_you, }) @@ -908,6 +913,7 @@ pub struct HelloReplyShadow<'a> { ready: bool, run_id: ByteBuf, away_ms: Option, + sender_run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -955,7 +961,7 @@ mod automerge_msg_codec { pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result, rmp_serde::encode::Error> { rmp_serde::to_vec(&HelloRequestShadow { swarm_id: msg.swarm_id.clone(), - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), @@ -965,7 +971,8 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result Result, rmp_serde::enco let shadow = HelloReplyShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), away_ms: msg.away_ms, + sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()), }; rmp_serde::to_vec(&shadow) } @@ -997,11 +1005,13 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result Ok(HelloReply { swarm_id: shadow.swarm_id.into_owned(), peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, away_ms: shadow.away_ms, + sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?, }) } @@ -1010,7 +1020,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result, rmp_serde:: let shadow = SyncEnvelopeShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, }; rmp_serde::to_vec(&shadow) @@ -1019,7 +1029,8 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result Result, rmp_serde::encode::Error> BlobBody::Missing => BlobBodyShadow::Missing, }; rmp_serde::to_vec(&BlobShadow { - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), id, body, }) @@ -1123,7 +1134,8 @@ fn bytes_to_blob(bytes: &[u8]) -> Result { BlobBodyShadow::Missing => BlobBody::Missing, }; Ok(BlobMessage { - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, id, body, }) @@ -1181,6 +1193,7 @@ mod tests { generation: u64::MAX, id: [0xff; 32], }, + epoch_seq: u32::MAX, ready: true, syncs_with_you: true, }; @@ -1203,6 +1216,7 @@ mod tests { let id = BlobId::Change(ChangeHash([9; 32])); let chunk = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Chunk { index: 5, @@ -1215,12 +1229,14 @@ mod tests { let decoded = bytes_to_blob(&bytes).unwrap(); assert_eq!(decoded.id, id); assert_eq!(decoded.epoch, epoch); + assert_eq!(decoded.epoch_seq, 4); assert!( matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) ); let manifest = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Manifest { size: 100, -- cgit v1.3.1