From 8f0b588831ac06913dd01a3ebd8bb523292d992e Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 02:43:43 +0100 Subject: skip neighbours with our heads --- docs/measurements.md | 2 +- docs/peers.md | 5 ++++- src/state/peer.rs | 5 +++++ src/swarm/peer/keepalive.rs | 1 + src/swarm/relay.rs | 51 ++++++++++++++++++++++++++++++++------------- 5 files changed, 47 insertions(+), 17 deletions(-) diff --git a/docs/measurements.md b/docs/measurements.md index fb01de0..131adca 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -137,7 +137,7 @@ growing is the sign, and that writer's own `clock.off` ([`trust.md`](trust.md), clocks"). Writers aren't known by node (Automerge's actors are per run), so their skew can't be taken off too. With sane clocks: in a full mesh most changes land in the first buckets; in a partial mesh each hop adds about a round trip -plus the 200 ms announcement delay. Mass in the minutes buckets means changes arrive by +plus the announcement delay (200 to 400 ms). Mass in the minutes buckets means changes arrive by keepalive-triggered pulls rather than relay, i.e. announcements are being lost. ## Clocks diff --git a/docs/peers.md b/docs/peers.md index 9868c0e..a56ee62 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -282,7 +282,10 @@ Changes travel through a partial mesh by announcement and pull ([`swarm/relay.rs pushes its own writes to its neighbours; after applying changes from one neighbour it sends its other neighbours a STATUS with its new heads (marked as an announcement: no keepalive, and answered only by a neighbour whose heads differ, so one that has the -changes already sends nothing back), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks +changes already sends nothing back; nor is a neighbour told whose latest heads, as its +own announcement or STATUS brought them, are this node's: announcements wait 200 to +400 ms, at random, so of two neighbours with the same news one is usually first, and +the other skips it), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks for what's missing). It pulls from one neighbour at a time (15 s timeout); neighbours that announce heads it lacks meanwhile wait their turn and are pulled from as soon as the current pull is answered, if they still have something it lacks. The regular diff --git a/src/state/peer.rs b/src/state/peer.rs index 24f9188..c9fa675 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -142,6 +142,10 @@ pub(crate) struct PeerState { /// How many heads its most recent keepalive listed (None if it sent a digest). pub(crate) reported_heads: Option, + /// The heads its most recent STATUS or STATUS_REPLY carried (an announcement's too), + /// with its epoch: once they are ours, it has every change we would announce to it + /// (`swarm::relay`). + pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>, /// It said farewell (`say_farewell`): what else that run of it still sends while it /// stops doesn't make it active again; a message from another run does (`returned`). @@ -232,6 +236,7 @@ impl PeerState { reports_ready: None, syncs_with_us: None, reported_heads: None, + last_heads: None, left: false, hello_attempts: 0, unreachable_since: None, diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 196b8b6..c426f99 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -223,6 +223,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport let became_ready = remote.ready && peer_state.reports_ready != Some(true); peer_state.reports_ready = Some(remote.ready); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); + peer_state.last_heads = Some((remote.epoch, remote.heads.clone())); drop(peer_state); if left { info!(peer = %peer, "Peer said farewell; dormant until it is heard from again"); diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index a4a4034..26719cd 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -13,7 +13,9 @@ //! Announcements are coalesced: the first applied change schedules one announcement //! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A //! round costs one STATUS per neighbour (plus the pulls and their replies), whatever the -//! write rate; the delay is the price per hop. Without coalescing, every applied sync +//! write rate; the delay is the price per hop. A neighbour that announced the same heads +//! to us first is skipped, so between two neighbours with the same news usually only one +//! STATUS goes. Without coalescing, every applied sync //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! @@ -39,9 +41,13 @@ use crate::transport::{self, HeadsSummary}; /// A pull that brought nothing for this long no longer blocks pulling from another peer. const PULL_TIMEOUT: Duration = Duration::from_secs(15); -/// How long an announcement waits for further changes to ride along. Added once per -/// hop to propagation; well below a round trip over the transport. +/// How long an announcement waits for further changes to ride along, and a random part +/// on top (`ANNOUNCE_JITTER`), so that of two neighbours with the same news one usually +/// announces first and the other, seeing the first's heads are its own, skips it +/// (`unreached`). Added once per hop to propagation; well below a round trip over the +/// transport. const ANNOUNCE_DELAY: Duration = Duration::from_millis(200); +const ANNOUNCE_JITTER: Duration = Duration::from_millis(200); /// After changes from `from` were applied and our heads changed: tell our other /// neighbours, in the next announcement round. @@ -67,7 +73,11 @@ pub(crate) async fn announce(state: &AppState, from: &str) { } let state = state.clone(); tokio::spawn(async move { - tokio::time::sleep(ANNOUNCE_DELAY).await; + let jitter = crate::transport::paths::random_nonce() + % u64::try_from(ANNOUNCE_JITTER.as_millis()) + .unwrap_or(1) + .max(1); + tokio::time::sleep(ANNOUNCE_DELAY + Duration::from_millis(jitter)).await; let senders = state .announce_pending .lock_or_recover() @@ -95,22 +105,33 @@ async fn announce_now(state: &AppState, senders: &HashSet) { /// The neighbours to tell of the changes `senders` sent us. The peer whose changes we /// announce has them already, so with a single sender it is skipped; with several, each -/// lacks the others' changes. +/// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an +/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have. async fn unreached(state: &AppState, senders: &HashSet) -> Vec { let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, }; - state - .peer_states - .read() - .await - .keys() - .filter(|peer| { - skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer) - }) - .cloned() - .collect() + let ours = { + let stamp = state.stamp.borrow(); + (stamp.epoch, HeadsSummary::of(stamp.heads.to_vec())) + }; + let mut unreached = Vec::new(); + for (peer, peer_state) in state.peers_snapshot().await { + if skip.as_deref() == Some(peer.as_str()) || !topology::is_neighbour(state, &peer) { + continue; + } + let has_ours = peer_state + .lock() + .await + .last_heads + .as_ref() + .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1); + if !has_ours { + unreached.push(peer); + } + } + unreached } /// Pulls in progress: one neighbour at a time, so each change arrives about once. The -- cgit v1.3.1