From 2b4359f23d20b14bae1896512f678d02b7f3c34c Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 03:25:00 +0100 Subject: introduce eager links --- docs/glossary.md | 6 +++ docs/peers.md | 10 ++++- src/document/sync.rs | 7 ++++ src/simtest/identity.rs | 1 + src/simtest/relay.rs | 63 ++++++++++++++++++++++++++++- src/simtest/timing.rs | 1 + src/state/peer.rs | 10 +++++ src/swarm/peer/keepalive.rs | 11 ++++-- src/swarm/relay.rs | 92 +++++++++++++++++++++++++++++++++++++------ src/swarm/sync.rs | 18 ++++++--- src/transport/libp2p/tests.rs | 1 + src/transport/mod.rs | 4 ++ src/transport/mycelium/mod.rs | 1 + src/transport/paths.rs | 1 + src/transport/sim.rs | 1 + src/transport/wire.rs | 8 +++- 16 files changed, 210 insertions(+), 25 deletions(-) diff --git a/docs/glossary.md b/docs/glossary.md index 71d3014..050a885 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -103,6 +103,12 @@ closest by round-trip time first. **relay**: in a partial mesh, a node announces new heads to its neighbours, which pull the changes they lack. +**eager link** and **lazy link**: over a lazy link, changes are announced and pulled +(the relay); over an eager one, a neighbour gets them as sync messages as soon as the +other end applies them. A pull that brought changes makes a link eager in that +direction, changes that came first another way make it lazy again (Plumtree's GRAFT and +PRUNE). + **blob**: bytes too large for one message, fetched in chunks from several peers at once: an epoch's snapshot change (for a rejoining node) or a remake's content (for voters). diff --git a/docs/peers.md b/docs/peers.md index a56ee62..c0e574d 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -290,7 +290,15 @@ for what's missing). It pulls from one neighbour at a time (15 s timeout); neigh 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 STATUS (every third keepalive) triggers the same pull, so gaps close within a few -keepalive intervals. A node +keepalive intervals. Links start lazy like this, announced over and pulled from; a +pull that brought changes this node lacked makes the link eager in that direction +(Plumtree's GRAFT): its STATUSes ask the neighbour for its changes as soon as the +neighbour applies them, and the neighbour sends them at once as sync messages, without +announcing or waiting. Changes this node had already, over an eager link, make it lazy +again (Plumtree's PRUNE): another path brings them first. Eager links so form where +changes actually flow, over links that deliver; announcements over the others catch +what they miss. Every STATUS restates the link's mode, and a link starts lazy again +whenever its sync does (a new neighbour, dormancy, a new run of the peer). A node that just caught up announces its heads once. In a full mesh too: the member list makes every member the writer's neighbour, but the writer may reach fewer of them (one behind NAT may reach a single one, one on libp2p alone none on Tor alone), and no node can tell diff --git a/src/document/sync.rs b/src/document/sync.rs index 44ad833..8780bf2 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -30,6 +30,13 @@ pub(crate) fn decode(payload: &SyncPayload) -> anyhow::Result { .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}")) } +impl Received { + /// Whether it carries changes (it may carry only heads and what the peer needs). + pub(crate) fn carries_changes(&self) -> bool { + !self.0.changes.is_empty() + } +} + /// A sync message being handed to our transport (for Mycelium: the HTTP request is in /// progress). #[derive(Clone)] diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index 04dbed3..7379048 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -111,6 +111,7 @@ fn status(node: &AppState) -> ParsedMessage { syncs_with_you: false, leaving: false, announcing: false, + eager: false, }) } diff --git a/src/simtest/relay.rs b/src/simtest/relay.rs index dadeed3..88ece74 100644 --- a/src/simtest/relay.rs +++ b/src/simtest/relay.rs @@ -1,5 +1,6 @@ //! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't -//! reach: those on none of its transports. +//! reach: those on none of its transports, or behind a link that fails; and the links +//! a pull makes eager. use std::time::Duration; @@ -117,3 +118,63 @@ async fn changes_reach_members_the_writer_fails_to_reach_at_once() { .await; } } + +/// Whether `node` asked `peer` for its changes eagerly, and whether `peer` heard so. +async fn eager(node: &crate::state::AppState, peer: &crate::state::AppState) -> bool { + let asked = match node.get_peer_state(peer.local_id()).await { + Some(state) => state.lock().await.eager_from, + None => false, + }; + let heard = match peer.get_peer_state(node.local_id()).await { + Some(state) => state.lock().await.eager_to, + None => false, + }; + asked && heard +} + +/// As above, x reaching b alone: c pulls x's first write from b, which makes the link +/// from b to c eager; x's next write b forwards to c as it applies it, without +/// announcing it. The link from x's side stays lazy: nothing was pulled over it. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pulls_make_links_eager() { + let keepalive = Duration::from_secs(5); + let slow = move |config: &mut crate::node::NodeConfig| { + config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT); + }; + let sim = Sim::with_identity(65); + let cut = crate::transport::sim::LinkParams { + loss: 1.0, + ..Default::default() + }; + sim.net.set_link("x", "c", cut.clone()); + sim.net.set_link("c", "x", cut); + let b = sim.start_on_with("b", &["sim"], &[], Start::Genesis, slow); + let c = sim.start_on_with("c", &["sim"], &["b"], Start::Join, slow); + let x = sim.start_on_with("x", &["sim"], &["b"], Start::Join, slow); + sim.wait_ready(&[&b, &c, &x], SOON).await; + // c may have pulled already: what x writes when it registers comes that way too. + sim.wait_converged(&[&b, &c, &x], SOON).await; + + write(&x, "y", json!(1)).await; + { + let (b, c) = (&b, &c); + sim.wait_until( + "c pulled x's write from b, and the link from b is eager", + SOON, + move || async move { read(c, "y").await == Some(json!(1)) && eager(c, b).await }, + ) + .await; + } + assert!(!eager(&b, &x).await, "nothing was pulled from x"); + + write(&x, "y", json!(2)).await; + { + let c = &c; + sim.wait_until( + "c has x's next write within a second", + Duration::from_secs(1), + move || async move { read(c, "y").await == Some(json!(2)) }, + ) + .await; + } +} diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs index 7879ac7..ea59689 100644 --- a/src/simtest/timing.rs +++ b/src/simtest/timing.rs @@ -336,6 +336,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() { syncs_with_you: false, leaving: false, announcing: false, + eager: false, }) }; let a = &a; diff --git a/src/state/peer.rs b/src/state/peer.rs index c9fa675..c6558d7 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -146,6 +146,11 @@ pub(crate) struct PeerState { /// 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)>, + /// The link's modes (`swarm::relay`): whether the peer wants our changes as we apply + /// them (its STATUS says so: we forward them to it), and whether we want its (ours + /// says so). Both lazy (announced only) until a pull makes ours eager. + pub(crate) eager_to: bool, + pub(crate) eager_from: bool, /// 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`). @@ -237,6 +242,8 @@ impl PeerState { syncs_with_us: None, reported_heads: None, last_heads: None, + eager_to: false, + eager_from: false, left: false, hello_attempts: 0, unreachable_since: None, @@ -252,6 +259,9 @@ impl PeerState { let mut reply_backoff = self.sync.reply_backoff; reply_backoff.reset(); self.sync = SyncSession::new(reply_backoff); + // The link starts lazy again: a pull makes it eager when it proves itself. + self.eager_to = false; + self.eager_from = false; } /// An overdue sync answer from an established, active peer (see diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index c426f99..37fcba8 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -176,9 +176,12 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta stamp.epoch_seq, ) }; - let syncs_with_you = match state.get_peer_state(peer).await { - Some(peer_state) => peer_state.lock().await.syncs_on(epoch), - None => false, + let (syncs_with_you, eager) = match state.get_peer_state(peer).await { + Some(peer_state) => { + let peer_state = peer_state.lock().await; + (peer_state.syncs_on(epoch), peer_state.eager_from) + } + None => (false, false), }; transport::Status { heads, @@ -188,6 +191,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta syncs_with_you, leaving: state.is_closing(), announcing: false, + eager, } } @@ -224,6 +228,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport 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())); + peer_state.eager_to = remote.eager; 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 26719cd..fcca9c7 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -19,6 +19,16 @@ //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! +//! Links start *lazy*: changes are announced over them and pulled. A pull that brought +//! changes we lacked makes the link *eager* in that direction (Plumtree's GRAFT): our +//! STATUS asks the neighbour for its changes as soon as it applies them (`eager`), and +//! it sends them right away as sync messages, without announcing or waiting +//! (`forward_eagerly`). Changes we had already over an eager link make it lazy again +//! (Plumtree's PRUNE): another path brought them first. So eager links form where +//! changes actually flow, over links that deliver, and the announcements on the others +//! are what catches whatever they miss. A link starts lazy again whenever its sync does +//! (`PeerState::reset_sync`: a new neighbour, dormancy, a new run of the peer). +//! //! In a full mesh too: the member list makes every member the writer's neighbour, but //! that says nothing about whether the writer reaches them. One behind NAT may reach a //! single one of them, one on another transport (on Tor alone) none, and no node can @@ -55,6 +65,7 @@ pub(crate) async fn announce(state: &AppState, from: &str) { if !state.initial_sync_done.load(Ordering::Acquire) { return; } + forward_eagerly(state, from).await; let schedule = { let mut pending = state.announce_pending.lock_or_recover(); match pending.as_mut() { @@ -87,6 +98,19 @@ pub(crate) async fn announce(state: &AppState, from: &str) { }); } +/// What `from` sent us, at once to the neighbours that asked for our changes as we apply +/// them (eager links): a sync message each, with what it lacks (the sync worker's). +async fn forward_eagerly(state: &AppState, from: &str) { + for (peer, peer_state) in state.peers_snapshot().await { + if peer == from || !topology::is_neighbour(state, &peer) { + continue; + } + if peer_state.lock().await.eager_to { + state.request_sync(&peer); + } + } +} + /// One announcement round. The peer whose changes we announce has them already, so a /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. @@ -106,7 +130,8 @@ 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. 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. +/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and +/// one we forward to eagerly, which was sent the changes already. async fn unreached(state: &AppState, senders: &HashSet) -> Vec { let skip = match senders.len() { 1 => senders.iter().next().cloned(), @@ -121,13 +146,15 @@ async fn unreached(state: &AppState, senders: &HashSet) -> Vec { if skip.as_deref() == Some(peer.as_str()) || !topology::is_neighbour(state, &peer) { continue; } + let peer_state = peer_state.lock().await; + // Sent what we have as we applied it (`forward_eagerly`). + let eager = peer_state.eager_to; 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 { + drop(peer_state); + if !eager && !has_ours { unreached.push(peer); } } @@ -211,21 +238,60 @@ pub(crate) async fn pull_if_behind( true } -/// A sync message from `peer` was applied: a pull from it is answered, and the next -/// waiting neighbour's turn comes. -pub(crate) fn pulled(state: &AppState, peer: &str) { - { +/// A sync message from `peer` was applied (`changed`: it brought changes we lacked): a +/// pull from it is answered, and the next waiting neighbour's turn comes. A pull that +/// brought changes makes the link eager (`graft`). +pub(crate) async fn pulled(state: &AppState, peer: &str, changed: bool) { + let next = { let mut pulling = state.pulling.lock_or_recover(); if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) { return; } pulling.current = None; - if pulling.waiting.is_empty() { - return; - } + !pulling.waiting.is_empty() + }; + if changed { + graft(state, peer).await; } - let state = state.clone(); - tokio::spawn(async move { pull_next(&state).await }); + if next { + let state = state.clone(); + tokio::spawn(async move { pull_next(&state).await }); + } +} + +/// `peer` brought changes we lacked when we pulled: from now on we want them as it +/// applies them, and our STATUS says so (Plumtree's GRAFT). +async fn graft(state: &AppState, peer: &str) { + if !set_eager_from(state, peer, true).await { + return; + } + debug!(peer = %peer, "Link eager: we want its changes as it applies them"); + if let Err(error) = announce_heads(state, peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is eager failed"); + } +} + +/// Changes we had already came from `peer` over an eager link: another path brings them +/// first, so announcements will do; our STATUS says so (Plumtree's PRUNE). +pub(crate) async fn prune(state: &AppState, peer: &str) { + if !set_eager_from(state, peer, false).await { + return; + } + debug!(peer = %peer, "Link lazy: its changes reach us first another way"); + if let Err(error) = announce_heads(state, peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed"); + } +} + +/// Sets whether we want `peer`'s changes eagerly; whether that changed it. +async fn set_eager_from(state: &AppState, peer: &str, eager: bool) -> bool { + let Some(peer_state) = state.get_peer_state(peer).await else { + return false; + }; + let mut peer_state = peer_state.lock().await; + let changed = peer_state.eager_from != eager; + peer_state.eager_from = eager; + changed } /// Pulls from the first waiting neighbour that still has heads we lack. diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 3bfbf7d..37c3b0a 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -372,11 +372,12 @@ pub(crate) async fn handle_sync( let clock = Arc::clone(&state.clock); let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed); - // Some((caught_up, heads changed)) when applied, None when the message belongs - // to another epoch. - tokio::task::spawn_blocking(move || -> Result> { + // Some((caught_up, heads changed, a duplicate)) when applied, None when the + // message belongs to another epoch. + tokio::task::spawn_blocking(move || -> Result> { // Before the lock: decoding a large message takes a while. let message = crate::document::sync::decode(&message)?; + let carries_changes = message.carries_changes(); let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -406,6 +407,9 @@ pub(crate) async fn handle_sync( let heads_after = replica.heads().to_vec(); stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; + // Changes we had already, over a link we asked to be eager: the link isn't + // needed (`relay::prune`). + let duplicate = carries_changes && !changed && peer_state.eager_from; if changed && measure_propagation { // Now by the peers' clocks (their median): this node's own clock skew // doesn't count; the writer's still does. @@ -426,13 +430,14 @@ pub(crate) async fn handle_sync( // before it registered itself). check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc), changed, + duplicate, ))) }) .await .context("Sync task panicked")?? }; - let Some((caught_up, changed)) = applied else { + let Some((caught_up, changed, duplicate)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; @@ -451,7 +456,10 @@ pub(crate) async fn handle_sync( state.notify_store_changed(); - crate::swarm::relay::pulled(state, peer); + crate::swarm::relay::pulled(state, peer, changed).await; + if duplicate { + crate::swarm::relay::prune(state, peer).await; + } // Caught up: our other neighbours may lack what the source had. if changed || just_caught_up { crate::swarm::relay::announce(state, peer).await; diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs index 1d7c6ac..8f77bf7 100644 --- a/src/transport/libp2p/tests.rs +++ b/src/transport/libp2p/tests.rs @@ -60,6 +60,7 @@ fn status() -> ParsedMessage { syncs_with_you: false, leaving: false, announcing: false, + eager: false, }) } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 25f9b7c..7f1aa98 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -206,6 +206,10 @@ pub(crate) struct Status { /// STATUS answered only if the recipient's heads differ. Most announcements reach a /// neighbour that has the changes already, which has nothing to tell. pub(crate) announcing: bool, + /// The sender wants the recipient's changes as soon as the recipient applies them, + /// sent as sync messages (an eager link, `swarm::relay`), not only announced. Every + /// STATUS says so, or not, again: one lost changes nothing for long. + pub(crate) eager: bool, } /// Cheap to clone: large parts are shared (`bytes::Bytes`). diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 9850ef0..0562e72 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -718,6 +718,7 @@ mod tests { syncs_with_you: true, leaving: false, announcing: false, + eager: false, }; let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap(); // Mycelium's message data: a length byte, the topic, then our payload. diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 29b3c23..1cd21e8 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -1514,6 +1514,7 @@ mod tests { syncs_with_you: false, leaving: false, announcing: false, + eager: false, }) } diff --git a/src/transport/sim.rs b/src/transport/sim.rs index 7a679fc..af262aa 100644 --- a/src/transport/sim.rs +++ b/src/transport/sim.rs @@ -388,6 +388,7 @@ mod tests { syncs_with_you: false, leaving: false, announcing: false, + eager: false, }) } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index d3bf39d..40976ee 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 = 16; +pub(crate) const VERSION: u8 = 17; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -370,6 +370,7 @@ pub(crate) fn samples() -> Vec { syncs_with_you: true, leaving: false, announcing: false, + eager: false, }; let proof = IdentityProof { key: [8; 32], @@ -551,6 +552,7 @@ mod tests { syncs_with_you: false, leaving: false, announcing: false, + eager: false, }; let decoded = decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap(); @@ -575,6 +577,7 @@ mod tests { syncs_with_you: true, leaving: false, announcing: false, + eager: false, }; assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err()); } @@ -673,10 +676,11 @@ mod tests { syncs_with_you: false, leaving: false, announcing: true, + eager: false, }); assert_eq!( hex(&encode(&status).unwrap()), - format!("{v}03979290c400{epoch_hex}01c3c2c2c3") + format!("{v}03989290c400{epoch_hex}01c3c2c2c3c2") ); let who = ParsedMessage::Who(IdentityProof { -- cgit v1.3.1