aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/document/sync.rs7
-rw-r--r--src/simtest/identity.rs1
-rw-r--r--src/simtest/relay.rs63
-rw-r--r--src/simtest/timing.rs1
-rw-r--r--src/state/peer.rs10
-rw-r--r--src/swarm/peer/keepalive.rs11
-rw-r--r--src/swarm/relay.rs92
-rw-r--r--src/swarm/sync.rs18
-rw-r--r--src/transport/libp2p/tests.rs1
-rw-r--r--src/transport/mod.rs4
-rw-r--r--src/transport/mycelium/mod.rs1
-rw-r--r--src/transport/paths.rs1
-rw-r--r--src/transport/sim.rs1
-rw-r--r--src/transport/wire.rs8
14 files changed, 195 insertions, 24 deletions
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<Received> {
.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<String>) {
/// 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<String>) -> Vec<String> {
let skip = match senders.len() {
1 => senders.iter().next().cloned(),
@@ -121,13 +146,15 @@ async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> {
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<Option<(bool, bool)>> {
+ // Some((caught_up, heads changed, a duplicate)) when applied, None when the
+ // message belongs to another epoch.
+ tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> {
// 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<ParsedMessage> {
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 {