diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/peer/inbound.rs | 24 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 19 |
2 files changed, 34 insertions, 9 deletions
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); |