aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-11-03 05:21:25 +0100
committertv <tv@krebsco.de>2026-11-03 05:21:25 +0100
commit8366f7c5de474753dbfbf88431917e79c15e5389 (patch)
tree8c3ed8cca74760ff98b58264843765a3d8e65d3b /src
parent96d7744dd29a3c783b43a7a6697af7d5d5719a10 (diff)
replace eager links by trees
Diffstat (limited to 'src')
-rw-r--r--src/api/dashboard/clocks.rs2
-rw-r--r--src/api/dashboard/dashboard.html36
-rw-r--r--src/api/dashboard/mod.rs2
-rw-r--r--src/api/stats.rs11
-rw-r--r--src/api/stats/prometheus.rs18
-rw-r--r--src/document/sync.rs7
-rw-r--r--src/simtest/bench.rs10
-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/links.rs24
-rw-r--r--src/swarm/peer/keepalive.rs11
-rw-r--r--src/swarm/relay.rs97
-rw-r--r--src/swarm/stats.rs21
-rw-r--r--src/swarm/sync.rs18
-rw-r--r--src/swarm/tree.rs1
-rw-r--r--src/transport/libp2p/tests.rs1
-rw-r--r--src/transport/mod.rs8
-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.rs10
23 files changed, 48 insertions, 307 deletions
diff --git a/src/api/dashboard/clocks.rs b/src/api/dashboard/clocks.rs
index ac58676..9f069be 100644
--- a/src/api/dashboard/clocks.rs
+++ b/src/api/dashboard/clocks.rs
@@ -99,8 +99,6 @@ mod tests {
down: false,
rtt_us: None,
dormant: false,
- eager_to: false,
- eager_from: false,
others: Vec::new(),
clock: Some(ClockOffset {
offset_ms,
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 1b442ae..9132275 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -161,9 +161,6 @@
.legend .dormant::before { border-top-width: 1.5px; --colour: var(--muted);
box-shadow: 0 0 0 2.5px color-mix(in srgb, var(--bad) 50%, transparent); }
.legend .silent::before { border-top-style: dotted; --colour: var(--muted); }
- /* A chevron on the line, as on the map. */
- .legend .eager::before { content: "\203A"; width: auto; height: auto; border: none;
- font-weight: bold; font-size: 15px; line-height: 0; }
.legend .hint::before { display: none; }
/* A kind of message: a dot in its colour, as on the map. */
.legend .message::before { width: 8px; height: 8px; border: none; border-radius: 50%;
@@ -610,7 +607,6 @@ function renderLegend(measured) {
el("span", { class: "direct" }, "direct"),
el("span", { class: "relayed" }, "relayed"),
el("span", { class: "dormant" }, "dormant or down"),
- el("span", { class: "eager" }, "eager (changes sent on at once, this way)"),
el("span", { class: "silent" }, "an end reports no links"));
document.getElementById("legend").replaceChildren(...items);
}
@@ -641,18 +637,6 @@ function edges() {
}
}
}
- // Eager, which way, if either end says so: "a>b" where a sends b its changes as it
- // applies them (a's eagerTo, or b's eagerFrom).
- const eager = new Set();
- for (const [a, report] of Object.entries(state.graph.measured)) {
- for (const link of report.links) {
- if (link.eagerTo) eager.add(a + ">" + link.peer);
- if (link.eagerFrom) eager.add(link.peer + ">" + a);
- }
- }
- for (const edge of measured.values()) {
- edge.eager = [eager.has(edge.a + ">" + edge.b), eager.has(edge.b + ">" + edge.a)];
- }
// Dormant, or down, if either end says so, whichever report is drawn.
for (const edge of measured.values()) {
if ((edge.dormant && !edge.link.dormant) || (edge.down && !edge.link.down)) {
@@ -970,7 +954,7 @@ function renderGraph() {
tooltip(hit(key, p, q, 1), name(a) + " – " + name(b) + ": intended (member list), not measured");
}
const ms = rtt => rtt < 10 ? rtt.toFixed(1) : String(Math.round(rtt));
- for (const [key, { a, b, link, secsAgo, eager }] of measured) {
+ for (const [key, { a, b, link, secsAgo }] of measured) {
const p = positions.get(a), q = positions.get(b);
if (!p || !q) continue;
// The transport in use thick, in the middle; the others thin, beside it.
@@ -1020,8 +1004,6 @@ function renderGraph() {
(line.transport ? " via " + line.transport : "") + (line.used ? " (in use)" : "") +
((line.used ? link.relayed : line.relayed) ? ", relayed" : "") + (link.dormant ? ", dormant" : "") +
(line.used && link.down ? ", down" : "") +
- (line.used && eager[0] ? ", eager " + name(a) + " → " + name(b) : "") +
- (line.used && eager[1] ? ", eager " + name(b) + " → " + name(a) : "") +
(silent ? ", one end reports no links" : "") +
(typeof line.rttMs === "number" ? ", " + line.rttMs.toFixed(1) + " ms" : "") +
", measured " + age(secsAgo + elapsed()) + " ago";
@@ -1045,22 +1027,6 @@ function renderGraph() {
}
});
children.push(...drawnLines.reverse());
- // Eager, each way it is: a chevron on the line in use pointing that way, past the
- // middle towards the end it goes to (both ways: one each side of it).
- const ux = (q.x - p.x) / length, uy = (q.y - p.y) / length;
- [[eager[0], 1], [eager[1], -1]].forEach(([on, way]) => {
- if (!on) return;
- const t = eager[0] && eager[1] ? .5 + way * .06 : .5;
- const mx = p.x + (q.x - p.x) * t, my = p.y + (q.y - p.y) * t;
- const dx = ux * way * 5, dy = uy * way * 5;
- const d = "M " + (mx - dx + nx * 5) + " " + (my - dy + ny * 5) + " L " + (mx + dx) +
- " " + (my + dy) + " L " + (mx - dx - nx * 5) + " " + (my - dy - ny * 5);
- const shape = { d, fill: "none", "stroke-linecap": "round", "stroke-linejoin": "round" };
- children.push(drawing("eager under " + key + " " + way, "path",
- { ...shape, stroke: "var(--panel)", "stroke-width": 5 }));
- children.push(drawing("eager " + key + " " + way, "path",
- { ...shape, stroke: colourOf(link.transport), "stroke-width": 2 }));
- });
tooltip(covering, descriptions.join("\n"));
}
// Above every line, so none crosses them.
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 14464d4..8d954c0 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -407,8 +407,6 @@ fn links_json(links: &[Link]) -> JsonValue {
"down": link.down,
"rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0),
"dormant": link.dormant,
- "eagerTo": link.eager_to,
- "eagerFrom": link.eager_from,
"clock": link.clock,
"others": link.others.iter().map(|other| json!({
"transport": other.transport,
diff --git a/src/api/stats.rs b/src/api/stats.rs
index 9ac2c94..96e8ad3 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -140,11 +140,8 @@ struct RelayStats {
pulls: u64,
/// Announcements of heads we lacked that waited for the pull in progress.
pulls_queued: u64,
- /// Links made eager, and lazy again (`swarm::relay`).
- grafts: u64,
- prunes: u64,
- /// Changes forwarded over eager links, per neighbour.
- eager_forwards: u64,
+ /// Changes passed on along writers' trees, per child (`swarm::tree`).
+ tree_forwards: u64,
/// Blob requests not answered: too many were being answered.
blob_requests_dropped: u64,
}
@@ -345,9 +342,7 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR
announce_rounds: node.relay.announce_rounds,
pulls: node.relay.pulls,
pulls_queued: node.relay.pulls_queued,
- grafts: node.relay.grafts,
- prunes: node.relay.prunes,
- eager_forwards: node.relay.eager_forwards,
+ tree_forwards: node.relay.tree_forwards,
blob_requests_dropped: node.blob_requests_dropped,
},
topology: TopologyStats {
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 3d4bcc6..03643ee 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -651,22 +651,10 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
relay.pulls_queued as f64,
);
out.single(
- "grafts_total",
+ "tree_forwards_total",
"counter",
- "Links made eager: a pull brought changes over them.",
- relay.grafts as f64,
- );
- out.single(
- "prunes_total",
- "counter",
- "Links made lazy again: changes came over them that another path brought first.",
- relay.prunes as f64,
- );
- out.single(
- "eager_forwards_total",
- "counter",
- "Changes forwarded over eager links, per neighbour.",
- relay.eager_forwards as f64,
+ "Changes passed on along writers' trees, per child.",
+ relay.tree_forwards as f64,
);
out.single(
"blob_requests_dropped_total",
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 8780bf2..44ad833 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -30,13 +30,6 @@ 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/bench.rs b/src/simtest/bench.rs
index 8c24d5c..d80190d 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -173,9 +173,7 @@ impl Totals {
relay.announce_rounds += stats.relay.announce_rounds;
relay.pulls += stats.relay.pulls;
relay.pulls_queued += stats.relay.pulls_queued;
- relay.grafts += stats.relay.grafts;
- relay.prunes += stats.relay.prunes;
- relay.eager_forwards += stats.relay.eager_forwards;
+ relay.tree_forwards += stats.relay.tree_forwards;
let propagation = &mut totals.propagation;
for (sum, count) in propagation
.buckets
@@ -360,13 +358,11 @@ fn report(workload: &str, quiet: (Totals, Totals), during: (Totals, Totals), con
);
let (r0, r1) = (during.0.relay, during.1.relay);
println!(
- " relay per write: {:.1} announcement rounds, {:.1} pulls ({:.1} queued), {:.2} grafts, {:.2} prunes, {:.1} eager forwards",
+ " relay per write: {:.1} announcement rounds, {:.1} pulls ({:.1} queued), {:.1} along writers' trees",
(r1.announce_rounds - r0.announce_rounds) as f64 / writes,
(r1.pulls - r0.pulls) as f64 / writes,
(r1.pulls_queued - r0.pulls_queued) as f64 / writes,
- (r1.grafts - r0.grafts) as f64 / writes,
- (r1.prunes - r0.prunes) as f64 / writes,
- (r1.eager_forwards - r0.eager_forwards) as f64 / writes,
+ (r1.tree_forwards - r0.tree_forwards) as f64 / writes,
);
let mut propagation = Propagation::default();
let (p0, p1) = (during.0.propagation, during.1.propagation);
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index 154c72e..9f77836 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -111,7 +111,6 @@ fn status(node: &AppState) -> ParsedMessage {
syncs_with_you: false,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
})
}
diff --git a/src/simtest/relay.rs b/src/simtest/relay.rs
index 88ece74..da1ce0a 100644
--- a/src/simtest/relay.rs
+++ b/src/simtest/relay.rs
@@ -1,6 +1,5 @@
//! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't
-//! reach: those on none of its transports, or behind a link that fails; and the links
-//! a pull makes eager.
+//! reach: those on none of its transports, or behind a link that fails.
use std::time::Duration;
@@ -118,63 +117,3 @@ 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 d06e992..602a7bf 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -336,7 +336,6 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
syncs_with_you: false,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
})
};
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 06269e9..ca721b5 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -146,11 +146,6 @@ 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,
/// Whom it said it reaches (`swarm::reach`), by pair key, and when.
pub(crate) reaches: Option<(Instant, Vec<u64>)>,
/// Whose change the next sync message to it passes on along the writer's tree
@@ -247,8 +242,6 @@ impl PeerState {
syncs_with_us: None,
reported_heads: None,
last_heads: None,
- eager_to: false,
- eager_from: false,
reaches: None,
tree_writer: None,
left: false,
@@ -266,9 +259,6 @@ 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;
// Said over the session that ended.
self.reaches = None;
}
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index e0d483d..c430d58 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -1,9 +1,8 @@
//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a
//! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its
//! neighbours and the peers it holds dormant (`own_links`), each with the transport its
-//! messages go over, whether only through a relay, the round trip, whether it is
-//! dormant, and which way the link is eager (`swarm::relay`): as the member measures
-//! them, which no other node can.
+//! messages go over, whether only through a relay, the round trip and whether it is
+//! dormant: as the member measures them, which no other node can.
//!
//! Every node answers, whatever its build: one with a dashboard asks any member, which
//! may have none. Only members (in the member list) are answered, as the links name
@@ -11,8 +10,8 @@
//! is written to the document: asked for while someone looks, forgotten after.
//!
//! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
-//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, eager,
-//! or a round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
+//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
+//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
//! subscribes (round trips change without telling), until the subscription runs out
//! (the asker renews it with its next LINKS, so one that is gone stops it by not
//! renewing) or a LINKS with 0 ends it.
@@ -232,14 +231,7 @@ impl Links {
pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
let mut links = Vec::new();
for (peer, peer_state) in state.peers_snapshot().await {
- let (dormant, eager_to, eager_from) = {
- let peer_state = peer_state.lock().await;
- (
- peer_state.liveness.is_dormant(),
- peer_state.eager_to,
- peer_state.eager_from,
- )
- };
+ let dormant = peer_state.lock().await.liveness.is_dormant();
if !dormant && !topology::is_neighbour(state, &peer) {
continue;
}
@@ -264,8 +256,6 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
.rtt(&peer)
.map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
dormant,
- eager_to,
- eager_from,
clock: state.transport.clock_offset(&peer),
peer,
});
@@ -1020,8 +1010,6 @@ fn changed(old: &[Link], new: &[Link]) -> bool {
|| a.relayed != b.relayed
|| a.down != b.down
|| a.dormant != b.dormant
- || a.eager_to != b.eager_to
- || a.eager_from != b.eager_from
|| moved(a.rtt_us, b.rtt_us)
|| clock_moved(a.clock, b.clock)
|| a.others.len() != b.others.len()
@@ -1074,8 +1062,6 @@ mod tests {
down: false,
rtt_us,
dormant: false,
- eager_to: false,
- eager_from: false,
others: Vec::new(),
clock: None,
}
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index c04b1ad..8362d73 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -179,12 +179,9 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
stamp.epoch_seq,
)
};
- 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),
+ let syncs_with_you = match state.get_peer_state(peer).await {
+ Some(peer_state) => peer_state.lock().await.syncs_on(epoch),
+ None => false,
};
transport::Status {
heads,
@@ -194,7 +191,6 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
syncs_with_you,
leaving: state.is_closing(),
announcing: false,
- eager,
reaches: Some(crate::swarm::reach::published(state)),
}
}
@@ -232,7 +228,6 @@ 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;
if let Some(reaches) = &remote.reaches {
peer_state.reaches = Some((Instant::now(), reaches.clone()));
}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 7dbc5b0..f4eaf5a 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -19,15 +19,9 @@
//! 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).
+//! Changes that say whose they are (the writer's own, and what passes on along its tree)
+//! go at once to the receiver's children in the writer's tree (`swarm::tree`), which
+//! the announcement round then leaves out.
//!
//! 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
@@ -74,7 +68,6 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>)
if !state.initial_sync_done.load(Ordering::Acquire) {
return;
}
- forward_eagerly(state, from).await;
let sent = match writer {
Some(writer) => crate::swarm::tree::forward(state, writer, from).await,
None => Vec::new(),
@@ -106,20 +99,6 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>)
});
}
-/// 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.stats.eager_forward();
- 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.
@@ -140,8 +119,7 @@ async fn announce_now(state: &AppState, pending: &Pending) {
/// 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; and
-/// one we forward to eagerly, or `sent` the changes along a writer's tree, which has
-/// them already.
+/// one `sent` the changes along a writer's tree, which has them already.
async fn unreached(
state: &AppState,
senders: &HashSet<String>,
@@ -163,15 +141,13 @@ async fn unreached(
{
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);
- drop(peer_state);
- if !eager && !has_ours {
+ if !has_ours {
unreached.push(peer);
}
}
@@ -255,62 +231,21 @@ pub(crate) async fn pull_if_behind(
true
}
-/// 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 = {
+/// 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) {
+ {
let mut pulling = state.pulling.lock_or_recover();
if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
return;
}
pulling.current = None;
- !pulling.waiting.is_empty()
- };
- if changed {
- graft(state, peer).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;
- }
- state.stats.link_grafted();
- debug!(peer = %peer, "Link eager: we want its changes as it applies them");
- if let Err(error) = announce_heads(state, peer, false).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;
- }
- state.stats.link_pruned();
- debug!(peer = %peer, "Link lazy: its changes reach us first another way");
- if let Err(error) = announce_heads(state, peer, false).await {
- debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed");
+ if pulling.waiting.is_empty() {
+ return;
+ }
}
-}
-
-/// 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
+ let state = state.clone();
+ tokio::spawn(async move { pull_next(&state).await });
}
/// Pulls from the first waiting neighbour that still has heads we lack.
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index dd56bb0..8a00629 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -102,12 +102,9 @@ pub(crate) struct Relay {
pub(crate) pulls: u64,
/// Announcements of heads we lacked that had to wait for the pull in progress.
pub(crate) pulls_queued: u64,
- /// Links made eager (a pull brought changes over them) and lazy again (changes came
- /// over them that another path brought first).
- pub(crate) grafts: u64,
- pub(crate) prunes: u64,
- /// Changes forwarded over eager links: a sync message asked for per neighbour.
- pub(crate) eager_forwards: u64,
+ /// Changes passed on along writers' trees (`swarm::tree`): a sync message asked for
+ /// per child.
+ pub(crate) tree_forwards: u64,
}
/// When the initial sync finished, and what it took.
@@ -190,16 +187,8 @@ impl NodeStats {
self.lock().relay.pulls_queued += 1;
}
- pub(crate) fn link_grafted(&self) {
- self.lock().relay.grafts += 1;
- }
-
- pub(crate) fn link_pruned(&self) {
- self.lock().relay.prunes += 1;
- }
-
- pub(crate) fn eager_forward(&self) {
- self.lock().relay.eager_forwards += 1;
+ pub(crate) fn tree_forward(&self) {
+ self.lock().relay.tree_forwards += 1;
}
/// A sync message of `bytes` arrived.
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 403e0c1..187dacc 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -376,12 +376,11 @@ 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, a duplicate)) when applied, None when the
- // message belongs to another epoch.
- tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> {
+ // Some((caught_up, heads changed)) when applied, None when the message belongs
+ // to another epoch.
+ tokio::task::spawn_blocking(move || -> Result<Option<(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
@@ -411,9 +410,6 @@ 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.
@@ -434,14 +430,13 @@ 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, duplicate)) = applied else {
+ let Some((caught_up, changed)) = applied else {
debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
return Ok(());
};
@@ -460,10 +455,7 @@ pub(crate) async fn handle_sync(
state.notify_store_changed();
- crate::swarm::relay::pulled(state, peer, changed).await;
- if duplicate {
- crate::swarm::relay::prune(state, peer).await;
- }
+ crate::swarm::relay::pulled(state, peer);
// Caught up: our other neighbours may lack what the source had.
if changed || just_caught_up {
crate::swarm::relay::announce(state, peer, writer.filter(|_| changed)).await;
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index 489bc22..1dff92e 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -146,6 +146,7 @@ pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<St
if let Some(peer_state) = state.get_peer_state(child).await {
peer_state.lock().await.tree_writer = Some(writer);
}
+ state.stats.tree_forward();
state.request_sync(child);
}
if !children.is_empty() {
diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs
index 7bedeaa..9f013c2 100644
--- a/src/transport/libp2p/tests.rs
+++ b/src/transport/libp2p/tests.rs
@@ -60,7 +60,6 @@ fn status() -> ParsedMessage {
syncs_with_you: false,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
})
}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 899f4b8..fb198ec 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -206,10 +206,6 @@ 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,
/// Whom the sender reaches (`swarm::reach`): the pair keys (`topology::pair_key`)
/// of the members it has an active session with, sorted. In keepalive STATUSes and
/// their replies, and when it changed; not in announcements of heads (None).
@@ -473,10 +469,6 @@ pub(crate) struct Link {
pub(crate) rtt_us: Option<u32>,
/// Silent for long (`peers.md`, dormancy).
pub(crate) dormant: bool,
- /// Eager links (`swarm::relay`): the member sends the peer its changes as it applies
- /// them (the peer asked), and the peer the member (the member asked).
- pub(crate) eager_to: bool,
- pub(crate) eager_from: bool,
/// The peer's other transports (`swarm::links::MAX_OTHERS` at most): where else
/// it proved itself, those paths up.
pub(crate) others: Vec<OtherPath>,
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 76ef8cd..3e33a9f 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -718,7 +718,6 @@ mod tests {
syncs_with_you: true,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
};
let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap();
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 3e595ab..6f0a892 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -1514,7 +1514,6 @@ mod tests {
syncs_with_you: false,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
})
}
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index c9e45fb..906d60a 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -388,7 +388,6 @@ mod tests {
syncs_with_you: false,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
})
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 8e3baa6..67f6de0 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 = 22;
+pub(crate) const VERSION: u8 = 23;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {
@@ -370,7 +370,6 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
syncs_with_you: true,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
};
let proof = IdentityProof {
@@ -447,8 +446,6 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
down: false,
rtt_us: Some(1500),
dormant: false,
- eager_to: true,
- eager_from: false,
others: vec![crate::transport::OtherPath {
transport: "mycelium".into(),
rtt_us: None,
@@ -556,7 +553,6 @@ mod tests {
syncs_with_you: false,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
};
let decoded =
@@ -582,7 +578,6 @@ mod tests {
syncs_with_you: true,
leaving: false,
announcing: false,
- eager: false,
reaches: None,
};
assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
@@ -682,12 +677,11 @@ mod tests {
syncs_with_you: false,
leaving: false,
announcing: true,
- eager: false,
reaches: None,
});
assert_eq!(
hex(&encode(&status).unwrap()),
- format!("{v}03999290c400{epoch_hex}01c3c2c2c3c2c0")
+ format!("{v}03989290c400{epoch_hex}01c3c2c2c3c0")
);
let who = ParsedMessage::Who(IdentityProof {