From 8366f7c5de474753dbfbf88431917e79c15e5389 Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 05:21:25 +0100 Subject: replace eager links by trees --- docs/dashboard.md | 8 +--- docs/glossary.md | 9 ++-- docs/measurements.md | 4 +- docs/openapi.yaml | 13 ++---- docs/peers.md | 11 +---- src/api/dashboard/clocks.rs | 2 - src/api/dashboard/dashboard.html | 36 +-------------- src/api/dashboard/mod.rs | 2 - src/api/stats.rs | 11 ++--- src/api/stats/prometheus.rs | 18 ++------ src/document/sync.rs | 7 --- src/simtest/bench.rs | 10 ++--- src/simtest/identity.rs | 1 - src/simtest/relay.rs | 63 +------------------------- src/simtest/timing.rs | 1 - src/state/peer.rs | 10 ----- src/swarm/links.rs | 24 +++------- src/swarm/peer/keepalive.rs | 11 ++--- src/swarm/relay.rs | 97 +++++++--------------------------------- src/swarm/stats.rs | 21 +++------ src/swarm/sync.rs | 18 +++----- src/swarm/tree.rs | 1 + src/transport/libp2p/tests.rs | 1 - src/transport/mod.rs | 8 ---- src/transport/mycelium/mod.rs | 1 - src/transport/paths.rs | 1 - src/transport/sim.rs | 1 - src/transport/wire.rs | 10 +---- 28 files changed, 61 insertions(+), 339 deletions(-) diff --git a/docs/dashboard.md b/docs/dashboard.md index 70397a0..1fe51dc 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -99,9 +99,6 @@ else the fresher. Then: * **Dotted**: one end reports no links: what is drawn may be out of date. * **Narrow, over a red band**: either end holds the other dormant, or the path in use is down (no path to the peer answers; `/peers` shows its connection as `null`). -* **A chevron on the line in use**: the link is eager that way ([`peers.md`](peers.md)): - the end it points away from sends the other its changes as soon as it applies them, - rather than announcing them; one each way if both. Shown if either end says so. Messages this node exchanges run along its links as dots, from sender to receiver, all at one speed: over a longer link they arrive later. @@ -330,7 +327,7 @@ another version of it (the state's `page`). member (`swarm::topology`), as this node computes them. The measured one is every member's links as it reports them in reply to a LINKS (sent to every member, at most 512, while the producer runs). The LINKS subscribes: the member replies again whenever -its links change (a peer, a transport, relayed, dormant, eager, or a round trip moved by a +its links change (a peer, a transport, relayed, dormant, or a round trip moved by a fifth; it looks every 2 s), for 90 s, renewed every 30 s (each renewal answered), and ended when the producer stops; a node that is gone stops it by not renewing. A reply older than 90 s is left out. A member keeps 64 subscribers at most. With `links: every @@ -367,8 +364,7 @@ every way has failed, all are tried again. Every hop does the same for what it a The links a member reports: its neighbours and the peers it holds dormant (a neighbour that goes dormant is one no more, but stays listed until it is heard from again), by which transport, whether only through a relay, the round trip (as the paths layer -measures it), whether it holds the peer dormant or the path is down, which way the link -is eager, the peer's other +measures it), whether it holds the peer dormant or the path is down, the peer's other transports with a path up (each with its fastest path's round trip, and whether it goes only through a relay), and the peer's clock against the member's ([`measurements.md`](measurements.md), Per peer, `clock`). A member answers a LINKS from members only, at most every 5 s, and diff --git a/docs/glossary.md b/docs/glossary.md index 050a885..42c7aac 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -103,11 +103,10 @@ 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). +**writer's tree**: how a writer's changes pass on, the same on every node: from the +member list's graph and whom each member says it reaches, each member's parent is the +neighbour it reaches closest to the writer. Whoever applies a writer's change sends it +at once to its children in the writer's tree; announcements catch what the tree misses. **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/measurements.md b/docs/measurements.md index f93a79a..086cd17 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -83,9 +83,7 @@ or caught up over many small messages. | `relay.announceRounds` | `announce_rounds_total` | Announcement rounds: our new heads to the neighbours after applying their changes (partial mesh only). | | `relay.pulls` | `pulls_total` | Pulls of changes we lacked from a neighbour that announced them. | | `relay.pullsQueued` | `pulls_queued_total` | Announcements of heads we lacked that waited for the pull in progress. Many compared with `pulls`: changes arrive from several sides at once. | -| `relay.grafts` | `grafts_total` | Links made eager: a pull brought changes over them, which from then on the neighbour sends as it applies them ([`peers.md`](peers.md)). | -| `relay.prunes` | `prunes_total` | Links made lazy again: changes came over them that another path had brought first. Grafts and prunes growing together, long after start: links keep switching, as when writers in different places take turns. | -| `relay.eagerForwards` | `eager_forwards_total` | Changes forwarded over eager links: a sync message per neighbour, instead of an announcement and a pull. | +| `relay.treeForwards` | `tree_forwards_total` | Changes passed on along writers' trees ([`peers.md`](peers.md)): a sync message per child, at once, instead of an announcement and a pull. | | `relay.blobRequestsDropped` | `blob_requests_dropped_total` | Blob requests not answered because too many were being answered; the requester asks again or asks another peer. Also logged as a warning, at most once a minute with the count. | ## Topology diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 20d2ccc..b73827c 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -753,9 +753,8 @@ paths: those of the member list, and any a link reported goes to: `alias`, `self`, `dashboard`: its `--dashboard-url`), `intended` (by node ID: the neighbours the member list gives it, as this node computes them), `measured` (by node ID: `secsAgo` and - `links`, each with `peer`, `transport`, `relayed`, `down`, `rttMs`, `dormant`, - `eagerTo` and `eagerFrom` (the member sends the peer its changes as it applies - them, and the peer the member: eager links) and `others` (the peer's other transports: `transport`, `rttMs`, `relayed`), as the member + `links`, each with `peer`, `transport`, `relayed`, `down`, `rttMs`, `dormant` and + `others` (the peer's other transports: `transport`, `rttMs`, `relayed`), as the member answered a LINKS; ours always, the others' while fresh) and `live` (whether the members' links are subscribed to, `/admin/dashboard/links`). One producer per node makes it every 2 s while anyone looks (and for a minute after this @@ -1977,7 +1976,7 @@ components: type: integer relay: type: object - required: [announceRounds, pulls, pullsQueued, grafts, prunes, eagerForwards, blobRequestsDropped] + required: [announceRounds, pulls, pullsQueued, treeForwards, blobRequestsDropped] properties: announceRounds: type: integer @@ -1985,11 +1984,7 @@ components: type: integer pullsQueued: type: integer - grafts: - type: integer - prunes: - type: integer - eagerForwards: + treeForwards: type: integer blobRequestsDropped: type: integer diff --git a/docs/peers.md b/docs/peers.md index 9118fde..760412f 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -290,15 +290,8 @@ 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. 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 +keepalive intervals. Changes that pass on along a writer's tree (below) go at once, +without announcement or pull; the announcements catch what the trees miss. 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/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 { .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)>, /// 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 { 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 { .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) 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) }); } -/// 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, @@ -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> { + // Some((caught_up, heads changed)) 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 @@ -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 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, /// 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, 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 { syncs_with_you: true, leaving: false, announcing: false, - eager: false, reaches: None, }; let proof = IdentityProof { @@ -447,8 +446,6 @@ pub(crate) fn samples() -> Vec { 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 { -- cgit v1.3.1