diff options
| author | tv <tv@krebsco.de> | 2026-11-07 22:35:10 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-11-07 22:35:10 +0100 |
| commit | 0e03a77bcb1d39f1fbda23d8e704ac40302afb93 (patch) | |
| tree | 79245416e9721cdccb1b6879b8f1cfc43629cc74 | |
| parent | c016fc783ff29e9069a446e17876046df64760a8 (diff) | |
dashboard: without LINKS; LINKS removed
The dashboard reads every member's links from the members themselves;
LINKS and LINKS_REPLY, swarm::links and their wire tags go (wire
version bumped). Nodes the browser can't reach are drawn from what the
reachable ones measure.
Auth plan M3.
| -rw-r--r-- | docs/architecture.md | 4 | ||||
| -rw-r--r-- | docs/dashboard.md | 104 | ||||
| -rw-r--r-- | docs/measurements.md | 13 | ||||
| -rw-r--r-- | docs/openapi.yaml | 38 | ||||
| -rw-r--r-- | docs/operations.md | 3 | ||||
| -rw-r--r-- | docs/peers.md | 5 | ||||
| -rw-r--r-- | src/api/dashboard/dashboard.html | 143 | ||||
| -rw-r--r-- | src/api/dashboard/links.rs | 216 | ||||
| -rw-r--r-- | src/api/dashboard/mod.rs | 132 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 1 | ||||
| -rw-r--r-- | src/node.rs | 4 | ||||
| -rw-r--r-- | src/simtest/bench.rs | 9 | ||||
| -rw-r--r-- | src/simtest/links.rs | 259 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 1 | ||||
| -rw-r--r-- | src/simtest/remakes.rs | 2 | ||||
| -rw-r--r-- | src/state/mod.rs | 2 | ||||
| -rw-r--r-- | src/swarm/links.rs | 1132 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 6 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 3 | ||||
| -rw-r--r-- | src/transport/mod.rs | 97 | ||||
| -rw-r--r-- | src/transport/paths.rs | 2 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 11 | ||||
| -rw-r--r-- | src/transport/wire.rs | 49 |
24 files changed, 332 insertions, 1905 deletions
diff --git a/docs/architecture.md b/docs/architecture.md index 0f0bad7..f4ce611 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -77,7 +77,7 @@ it switched to the right document without trusting anyone. ## Modules **[`transport/`](../src/transport/)** — [`mod.rs`](../src/transport/mod.rs) defines the message types (HELLO, HELLO_REPLY, sync -envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, LINKS/LINKS_REPLY, PROBE/PROBE_REPLY) and the `Transport` trait; +envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, PROBE/PROBE_REPLY) and the `Transport` trait; [`wire.rs`](../src/transport/wire.rs) is the wire format (a version byte, bumped with every change to the format, and a kind byte, then the message in MessagePack, derived from the types in [`mod.rs`](../src/transport/mod.rs) with codecs for hashes and IDs); [`mycelium/`](../src/transport/mycelium/) (with the `mycelium` feature) is the REST @@ -118,7 +118,6 @@ ignored 25-node soak test with a propagation bound (`cargo test soak -- --ignore | `tree` | writers' trees: passing each writer's changes on to the members after us in its tree, the same on every node | [`peers.md`](peers.md) | | `reach` | whom each neighbour reaches, as it says in its STATUS: the 2-hop neighbourhood trees are built on | [`peers.md`](peers.md) | | `relay` | announcing heads where trees don't reach (at once) and to everyone else (slow round), and pulling what's missing | module docs | -| `links` | answering LINKS with this node's links, for the dashboard's map | [`dashboard.md`](dashboard.md) | | [`topology/`](../src/swarm/topology/) | which members are neighbours: ring plus low-score links, stable members preferred | module docs | | `members` | the member registry: aliases, alive stamps, tiers, the electorate | module docs | | `stability` | a node's own connectivity record and its stable/transient tier | module docs | @@ -237,7 +236,6 @@ replies, admit (swarm key), mark the peer seen, record its epoch (ignoring late then by type: HELLO answered, HELLO_REPLY establishes membership and discovers peers, sync queued on its lane if on our epoch (held back from everyone but the source while catching up), STATUS answered with a STATUS_REPLY (an announcement only if the receiver lacks some of its heads) and compared for divergence, blobs served or collected, -LINKS answered with the node's links if the asker is a member, and again on each change while it subscribes, or with other members' links it asks for the asker, which can't reach them (`swarm::links`, for the dashboard); PING and PONG (the paths layer answered the PING already) only mark the peer seen, and after a PONG the peer's round-trip time is its path's again. diff --git a/docs/dashboard.md b/docs/dashboard.md index 5a66afd..05069d1 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -2,8 +2,8 @@ A page showing one node's view of the swarm, live: a map of the members and the links between them, the node's peers, its figures, and the document, which it can also edit. -This page is about using it; how it works underneath (events, LINKS, what is sent -when) is at the end, [Underneath](#underneath). +This page is about using it; how it works underneath (events, subscriptions, what is +sent when) is at the end, [Underneath](#underneath). ## Opening it @@ -67,10 +67,10 @@ are collapsed is kept in the browser too. Every member of the swarm, by alias (or a shortened node ID): * **This node**: filled blue, slightly larger. -* **Filled**: the member reports its links to this node. -* **Hollow**: it reports none: it doesn't answer this node's LINKS, or its last answer - is more than 90 s old (it is unreachable from here, stopped, or doesn't yet take this - node for a member). +* **Filled**: the page has the member's links: this node's own, and those of the nodes + it subscribes to (below). +* **Hollow**: it has none of its links: the page doesn't subscribe to it, or doesn't + reach it. * **Black centre, ring of this node's blue**: subscribed: the page reads its links from it, directly (below); the ring grey while the browser doesn't reach it (why, in its tooltip). @@ -94,10 +94,9 @@ Every member of the swarm, by alias (or a shortened node ID): Settled but not decided yet, the coordinator is hollow, with a golden question mark in it. Of the remakes in this node's document: every member's vote shows, as it reaches this node. * **A golden glow**, growing from it and fading, once: it just entered a new epoch (a - remake, [`remakes.md`](remakes.md)). For every member that reports its links (they - say its epoch), and this node's peers, as they change while the page is open: so - after a remake, a wave across the map as the members switch. With `links: every - 30 s`, up to that late. + remake, [`remakes.md`](remakes.md)). For this node's peers and the nodes subscribed + to (their links say their epoch), as they change while the page is open: so after a + remake, a wave across the map as the members switch. ### Links @@ -118,7 +117,7 @@ else the fresher. Then: * **Solid**: direct. * **Dashed**: through a relay. -* **Dotted**: one end reports no links: what is drawn may be out of date. +* **Dotted**: the page has one end's links only: drawn as that end reports it. * **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`). @@ -145,12 +144,7 @@ legend and the sliders greyed meanwhile); the choice is kept in the browser. A label on the map that changes, a round trip or a node's name, breathes once: it glows around its new text and fades (not where the system asks for reduced motion). -The map's controls are at its top right, over the map as the legend is (below): -`links: live` says the members tell this node of their links' -changes as they happen. Clicked, it's `links: every 30 s`: the members are only asked -every 30 s, which costs the swarm fewer messages and leaves the map up to 30 s behind -(this node's own links stay current). It's the node's choice, for every page open on -it, until clicked again or the node restarts. +The map's controls are at its top right, over the map as the legend is (below). The legend, over the top of the map, names the transports that appear; its contents are on an underlay of their own shape, as the map's labels, so they read over what is @@ -205,7 +199,6 @@ measured. they were (on a dynamic map they then move to their places among the other node's links, smoothly; new ones are placed among them), the view, the pins, what is in focus, the legend shown or hidden, what is open in the figures and in the document, - `links: live` or `every 30 s` (the other node's choice, for every page open on it), and this browser's settings for the page (sizes, collapsed boxes, the layout, the messages' size and speed, the console's filter), which are per host otherwise. They travel in the URL's fragment, which no request carries, and the page removes it @@ -263,7 +256,7 @@ The box is as tall as its rows, up to eight; then it scrolls. The figures of what is in focus, as a tree, its keys at the top level and how many beside the title: this node's `/admin/stats`, or a peer's `/peers` entry and, after it, -the links the member reports (`links`). +the links it reports (`links`), where the page has them. Branches stay as opened across updates, and ages count on between them. What each figure means: [`measurements.md`](measurements.md). Some values get a reading after them, muted, in parentheses: a node ID its alias @@ -316,7 +309,7 @@ miss the commit before the others switch, and rejoin). The answer shows below: t The messages this node exchanged with its peers, one by one: when (on the 24-hour clock, to the millisecond), which way (`→` sent, `←` received, `↛` refused: the transport wouldn't send it), with whom, the type (`HELLO`, `SYNC`, `STATUS_REPLY`, -`LINKS`, …, in its kind's colour, as on the map) and the size. The latest 500, the +`PING`, …, in its kind's colour, as on the map) and the size. The latest 500, the newest at the bottom; scrolled along while at the bottom, which `following…` beside the title says. `clear` empties it; what comes from then on fills it again. They arrive with every update (every 2 s). The PINGs the paths layer sends on its own, and the PONGs it answers @@ -338,7 +331,7 @@ the map or in the peers, goes back. Pointing at a message marks the others of its exchange (a lighter grey, and a bar on their left); while one is shown, its exchange stays marked. An exchange is a PING and its PONG (by their nonce, wherever they were listed), a HELLO and its replies (by the -run of the node that sent the HELLO), a LINKS and its replies (by their nonce), a blob's +run of the node that sent the HELLO), a blob's requests and answers, from whichever peers (by the blob), and a delta offered in a SYNC with the blob it is fetched as. A STATUS and its reply carry nothing that ties them, nor does an ordinary SYNC. @@ -379,15 +372,14 @@ again. * **The status stays at `connecting…`, or the page doesn't load**: the node isn't built with `dashboard` (404), or the user isn't in `--htpasswd` (401). -* **A hollow node**: no links report from it. If it is also dormant here, it is - probably down or unreachable; if not, it may not take this node for a member yet - (it hasn't registered it), or its answers don't arrive. A node that just joined - through others shows as soon as one of them reports a link to it, hollow until its - registration reaches this node and it is asked for its links (within 30 s). +* **A hollow node**: the page has none of its links: subscribe to it (double-click). If + the ring stays grey, the browser doesn't reach its dashboard (why, in its tooltip). + A node that just joined through this one shows at once, hollow; through others, once + its registration reaches this node (the graph is computed again within 30 s). * **Only dashed grey between two nodes**: the member list wants them as neighbours, but neither reports a link to the other: they haven't reached each other (yet). -* **A dotted link**: one end has stopped reporting; the link is as it was last - reported. +* **A dotted link**: the page has one end's links only (subscribe to the other to + see its side). * **A red band**: one end holds the other dormant. If the other end still reports the link, the trouble is likely in one direction only. @@ -408,42 +400,16 @@ them; the page reconnects once the node is back, and reloads if the node now ser another version of it (the state's `page`). **The map's graphs.** The intended one is the neighbours the member list gives every -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), with the epoch it is in. The LINKS subscribes: the -member replies again whenever its links change (a peer, a transport, relayed, dormant, -or a round trip moved by a fifth; it looks every 2 s) or it enters a new epoch, 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 -30 s` (`POST /admin/dashboard/links?live=false`; `true` back), the subscriptions end -and the members are only asked every 30 s, for every page open on this node, until -asked otherwise or the node restarts. - -A member this node can't reach (on none of its transports: one on Tor alone, from a -node on libp2p alone; or one that doesn't answer its HELLOs, as behind NAT) is asked -through a peer that can: an active member it reaches -that, by the member list, shares a transport with the member, else one on a transport -this node isn't on (the lowest pair score of several). The LINKS names the members to -ask (`of`): the peer asks them on this node's behalf, itself or again through a peer -of its own (three nodes at most), for as long as this node asked, and passes their -links on in its replies (`relayed`, each with its age), several together (64 members' -at most, and 256 KiB as encoded but for one member's alone): it holds what arrives a -while. The asker's window (250 ms, at most 5 s) is for the whole way: -each hop holds its share of what is left, the window divided among it and the hops -that may follow (at least 10 ms), and passes the rest on, so links reach the asker -within the window however long the chain. Such a subscription isn't ended by a message: every hop's runs -out unless renewed, so it ends within 90 s of the page closing or the node going away. -A hop renews its own as soon as it is renewed, for the same time, so the hops run in -step, each a message's delay behind the one before (it never runs out first), and a -renewal keeps its nonce, so what is on its way under the last one still counts. A -renewal lost on the way leaves the hops behind it to run out at the old time: with 90 s -renewed every 30 s, one lost costs nothing. - -A way of asking a member that brings none of its links within 20 s (the member doesn't -answer this node after all, or the peer relaying doesn't pass anything on) is left out -for that member for 5 minutes: the next renewal goes the next way (directly, then the -peer picked as above, then the next one), and the old way runs out by itself. Once -every way has failed, all are tried again. Every hop does the same for what it asks on. +member (`swarm::topology`), as this node computes them, every 30 s. The measured one is +this node's links, in the state, and those of the nodes the page subscribes to, which +it reads from each node itself: `GET /admin/dashboard/links/own/events` there, a +`links` event with its links at once and again whenever they change (a peer, a +transport, relayed, dormant, or a round trip moved by a fifth; it looks every 2 s) or +it enters a new epoch, with the epoch it is in. The page asks with a token from this +node that reads the member list only (`Authorization: Bearer`, [`access.md`](access.md), +Tokens), refreshed half way through its life; the other node allows the page's origin +by CORS where it is a member's dashboard (`--cors-origins` for others). A stream that +fails is opened again, waiting longer each time (up to a minute). 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 @@ -451,13 +417,11 @@ which transport, whether only through a relay, the round trip (as the paths laye 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 -asks for each peer that has it relay at most 512 members at once and 32 a second on -average (the rest of a request is dropped); every build answers, with or without the -feature (`swarm::links`). +Per peer, `clock`). -**Clocks.** The clocks in the links chain every member's clock to this node's, along -the narrowest bounds, and the map takes each against their median: a member more than +**Clocks.** The clocks in the links on the map (this node's and the subscribed nodes') +chain every member's clock to this node's, along the narrowest bounds, and the page +takes each against their median: a member more than `--clock-skew-warn` off it, beyond the chained bound, is ringed. With fewer than three members compared, none is. diff --git a/docs/measurements.md b/docs/measurements.md index a444862..9036031 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -2,8 +2,7 @@ What a node measures about itself and its peers, where to read it, and what the numbers mean. Everything here comes from traffic and work the node has anyway; it -sends no probes of its own and nothing is measured by anyone else, but for the -dashboard's map (LINKS, only while someone looks at it). +sends no probes of its own and nothing is measured by anyone else. Four places to read: @@ -179,7 +178,7 @@ clocks"); setting them is the operating system's job (NTP). Messages and payload bytes by direction (`sent`, `received`, `refused`) and kind (`hello`, `helloReply`, `sync`, `status`, `statusReply`, `blob`, `identity` for WHO and -ME, `paths` for PING and PONG, `links` for LINKS and their replies, `probe` for PROBE +ME, `paths` for PING and PONG, `probe` for PROBE and PROBE_REPLY, `plumtree` for PRUNE and GRAFT, `hyparview` for NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT). The layers below the node count WHO and ME, and the paths layer's own PINGs and PONGs, under the address they went to or came from; a peer @@ -302,13 +301,13 @@ outgrowing the full save (busy swarm, or a `--document-rewrite` set low). In builds with the `dashboard` cargo feature, `GET /admin/dashboard` is a page showing `/admin/stats`, `/peers` and a map of the swarm, live: [`dashboard.md`](dashboard.md), and how it works -underneath (events, LINKS) at its end. Its LINKS are the one thing a node sends only to -measure: the links every member reports, while someone looks. +underneath (events, subscriptions) at its end. Other members' links it reads from +them, over HTTP, from the nodes the page subscribes to: no messages in the swarm. ## What is not measured -* Nothing about other nodes but what they send us: no probes, no remote figures (but - the dashboard's LINKS). +* Nothing about other nodes but what they send us: no probes, no remote figures (the + dashboard's page reads those of the nodes it subscribes to from them). * Clock skew between nodes that aren't each other's peers: only peers' clocks are compared (`clock`, per peer). * Remake durations and outcomes: the remake reports (`GET /admin/remakes`) hold them. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 20cbf80..4b82ff7 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -992,20 +992,18 @@ paths: The dashboard's latest snapshot: `node` (`id`, `alias`), `page` (the page's version, a hash: a page showing another reloads), `stats` (as `/admin/stats`), `peers` (as `/peers`) and `graph`: `members` (by node ID, - those of the member list, and any a link reported goes to: `alias`, `self`, + those of the member list, and any of this node's links 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`, `epoch` - (the epoch the member said it is in: `generation`, `id`) and `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`), and `remakes` + it, as this node computes them) and `measured` (this node's own, by its node ID: + `secsAgo`, `epoch` (`generation`, `id`) and `links`, each with `peer`, `transport`, `relayed`, `down`, `rttMs`, `dormant`, `clock` and + `others` (the peer's other transports: `transport`, `rttMs`, `relayed`); other + nodes' the page reads from them, `/admin/dashboard/links/own/events`), and `remakes` (those of this node's epoch: `id`, `coordinator`, `phase` as in `/admin/remakes`, `voters` (node IDs), `fromGeneration`, the epoch's, and once committed `switchDelayMs` and `switchInMs`, this node's wait to switch and what is left of it). One producer per node makes it every 2 s while anyone looks (and for a minute after this - request), subscribed to the members' links (each tells of a change as it - happens; renewed every 30 s); every client gets the same. Only with the `dashboard` cargo feature. + request); every client gets the same. Only with the `dashboard` cargo feature. responses: '200': description: The snapshot. @@ -1076,30 +1074,11 @@ paths: summary: Dashboard refresh description: | The dashboard's producer builds the state now rather than at its next tick, and - asks the members for their links again; what changed comes as events. Only with + computes the member list's graph again; what changed comes as events. Only with the `dashboard` cargo feature. responses: '204': description: Done. - /admin/dashboard/links: - post: - tags: [Admin] - summary: Dashboard links, live or asked - description: | - Whether the dashboard's producer subscribes to the members' links (`live=true`, - the default: each member tells of a change as it happens) or only asks for them - every 30 s (`live=false`: fewer messages in the swarm, a map up to 30 s - behind). For every client of this node, until set again or the node restarts; - the state's `graph.live` says which. Only with the `dashboard` cargo feature. - parameters: - - name: live - in: query - required: true - schema: - type: boolean - responses: - '204': - description: Done. /admin/dashboard/tree: get: tags: [Admin] @@ -2075,9 +2054,6 @@ components: paths: $ref: '#/components/schemas/Counter' description: PING and PONG; the paths layer's own PINGs counted under the address (so only in the totals). - links: - $ref: '#/components/schemas/Counter' - description: LINKS and LINKS_REPLY, asked for and answered for a map of the swarm (the dashboard). probe: $ref: '#/components/schemas/Counter' description: PROBE and PROBE_REPLY, asking others whether they reach a member before recording a link to it as unreached. diff --git a/docs/operations.md b/docs/operations.md index 2fd8e7f..f8dcdef 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -249,7 +249,6 @@ And from the log, at warn or error: since its last rewrite (`--document-rewrite`); the state file (small); Tor's state directory. * **Network**: per neighbour, about two keepalives per `--keepalive` interval when - quiet ([`measurements.md`](measurements.md), Traffic), plus sync for every write, and LINKS while a - dashboard is open; a joining node downloads the document once (`savedBytes`, about). + quiet ([`measurements.md`](measurements.md), Traffic), plus sync for every write; a joining node downloads the document once (`savedBytes`, about). * **Processes and ports**: the HTTP API (`--bind-addr`); a supervised Mycelium daemon and its overlay port; libp2p's listen addresses. diff --git a/docs/peers.md b/docs/peers.md index 85d65a9..4aa1334 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -246,7 +246,7 @@ stateDiagram-v2 - Dormancy is only judged while the node is ready, so joining nodes keep fast retries. A peer's farewell (it is shutting down: a STATUS with `leaving`) makes it dormant at once, whatever our phase; it isn't counted as an incident (`dormancies`). What that - run still sends while it stops (sync messages, PINGs, a LINKS_REPLY) doesn't make it + run still sends while it stops (sync messages, PINGs) doesn't make it active again: only a HELLO or HELLO_REPLY from its next run does. - **A suspended machine** (a laptop asleep) counts its peers as silent for as long as it slept: the monotonic clock the timers run on stops meanwhile, so without that every @@ -376,8 +376,7 @@ what they know, contacting nobody: an active session that answers; a session tha doesn't, held dormant or unanswered HELLOs; or can't tell): if some don't and none does, the member is away and nothing is recorded (asked again after another hold); otherwise, or with no answer within 10 s, the link is recorded. A node that reaches nobody at all is the one away, and -records nothing. PROBEs are their own message, apart from the dashboard's -LINKS, which a node may leave unanswered. +records nothing. A writer's changes pass on along its tree ([`swarm/tree.rs`](../src/swarm/tree.rs)), the same on every node: built on the graph the member list gives (as if every member reached every other), less diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index 5b776cd..57a3a0f 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -197,11 +197,6 @@ /* After the sliders, as tall; the kinds and the sliders greyed while the dots are off. */ .legend button.toggle { float: none; margin-left: 0; line-height: 1em; } .legend.off .message, .legend.off .slider { opacity: .45; } - /* The LINKS kind, a toggle: shown (in the console, as dots, in the rates) or not. */ - .legend button.message::before { content: ""; display: inline-block; vertical-align: middle; - margin-right: 4px; } - .legend button.message.hidden-kind { text-decoration: line-through; } - #console.no-links > div.links-kind { display: none; } /* The map and, over its top, its controls (links, layout, legend) and the legend: they take no room from the map, which stays as it is, the legend shown or hidden. What is under them shows through; their contents are on an underlay of their own @@ -316,7 +311,6 @@ <div class="map-area"> <div class="map-overlay"> <div class="map-controls"> - <button id="links-toggle" class="toggle">links: live</button> <button id="rtt-toggle" class="toggle">layout: static</button> <button id="tree-toggle" class="toggle">tree: off</button> <button id="legend-toggle" class="toggle">hide legend</button> @@ -324,8 +318,8 @@ <div id="legends"> <div class="legend"> <span class="node self">this node</span> - <span class="node">reports its links</span> - <span class="node silent">reports none (no LINKS_REPLY)</span> + <span class="node">its links known</span> + <span class="node silent">its links unknown (not subscribed)</span> <span class="node dormant-node">dormant (to this node)</span> <span class="node clock-off">clock off the swarm's</span> <span class="node focus">in focus (click a node)</span> @@ -398,7 +392,7 @@ <div id="column-hint" class="note">Drag the space left of these boxes to the right to hide the column.</div> <section id="console-box"> - <h2>Console <label class="choice" title="Only the messages with the node or over the link in focus (clicked on the map or in the peers); all while nothing is"><input type="checkbox" id="console-filter">filter selected</label><span id="console-following" class="following" title="Scrolled to the bottom: new messages are scrolled to as they come; scroll up to stop">following…</span><button id="console-clear" class="toggle inline" title="Empty this page's console: what comes from now on fills it again">clear</button><span id="console-rates" class="rates" title="Messages and bytes per second sent and received, as load averages are: over the last 1, 5 and 15 minutes (exponentially weighted), since the page was opened; with the filter, those with the node or over the link in focus; without LINKS while their kind is off in the map's legend (they go only while a dashboard asks)"></span></h2> + <h2>Console <label class="choice" title="Only the messages with the node or over the link in focus (clicked on the map or in the peers); all while nothing is"><input type="checkbox" id="console-filter">filter selected</label><span id="console-following" class="following" title="Scrolled to the bottom: new messages are scrolled to as they come; scroll up to stop">following…</span><button id="console-clear" class="toggle inline" title="Empty this page's console: what comes from now on fills it again">clear</button><span id="console-rates" class="rates" title="Messages and bytes per second sent and received, as load averages are: over the last 1, 5 and 15 minutes (exponentially weighted), since the page was opened; with the filter, those with the node or over the link in focus"></span></h2> <div class="scroll" id="console-scroll"><div id="console" class="console"></div> <div id="console-none" class="note" hidden>No information available</div></div> </section> @@ -1953,7 +1947,6 @@ function render(ages = false) { const skew = state.stats.clock; document.getElementById("clock-warning").textContent = skew && skew.off ? "clock " + seconds(skew.offsetMs) + " off its peers'" : ""; - showLinksToggle(); held = false; const boxes = [["peers-box", renderPeers], ["figures-box", renderStats]]; // Its ages are in tooltips only: drawn on events, not every second. @@ -1994,21 +1987,6 @@ legendToggle.addEventListener("click", () => { }); // Hidden as on the page it was opened from (`handOver`). if (handOver && handOver.legend === false) showLegend(false); -// The members' links subscribed to (`live`: each tells of a change as it happens), or -// only asked for every 30 s; the node's choice, for every page (`POST -// /admin/dashboard/links`), shown as the state has it. -const linksToggle = document.getElementById("links-toggle"); -function showLinksToggle() { - const live = !state || state.graph.live !== false; - linksToggle.textContent = live ? "links: live" : "links: every 30 s"; - linksToggle.title = live - ? "The members tell of their links' changes as they happen. Click to only ask them every 30 s (fewer messages; for every page on this node)" - : "The members are asked for their links every 30 s. Click to have them tell of changes as they happen (for every page on this node)"; -} -linksToggle.addEventListener("click", () => { - const live = !state || state.graph.live !== false; - fetch("dashboard/links?live=" + !live, { method: "POST" }).catch(() => {}); -}); // A writer's tree, shown and hidden (`treeShown`). const treeToggle = document.getElementById("tree-toggle"); function showTreeToggle() { @@ -2089,21 +2067,10 @@ function handOff(id) { focused: focused === id ? null : focused, settings, legend: !document.getElementById("legends").hidden, - links: state ? state.graph.live !== false : undefined, figures: [...open], document: [...documentOpen], })); } -// Links live or every 30 s as on the page it was opened from (`handOver`), once its -// first state says otherwise. The node's choice: for every page open on it. -let linksHandedOver = false; -function handOverLinks() { - if (linksHandedOver || !handOver || typeof handOver.links !== "boolean") return; - linksHandedOver = true; - if ((state.graph.live !== false) !== handOver.links) { - fetch("dashboard/links?live=" + handOver.links, { method: "POST" }).catch(() => {}); - } -} // `[x, y]` turned by `angle`. function turned(x, y, angle) { const c = Math.cos(angle), s = Math.sin(angle); @@ -2744,8 +2711,7 @@ function messages(byKind) { // The colour of a kind of message (`traffic`'s kinds), on the map and in the console; // the replies as their requests. const KIND_COLOURS = [["sync", "#ea580c"], ["blob", "#a16207"], ["hello", "#16a34a"], - ["status", "#e11d48"], ["identity", "#0d9488"], ["paths", "var(--muted)"], - ["links", "#db2777"]]; + ["status", "#e11d48"], ["identity", "#0d9488"], ["paths", "var(--muted)"]]; const kindColour = kind => (KIND_COLOURS.find(([name]) => kind === name || kind === name + "Reply") || [null, "var(--fg)"])[1]; // How large the dots are drawn, times `dotRadius` (the slider after their legend, kept @@ -2759,39 +2725,6 @@ let flowSpeed = .3; try { flowSpeed = Number(localStorage.getItem("continuum-dot-speed")) || .3; } catch (_) {} flowSpeed = Math.min(.9, Math.max(.1, flowSpeed)); let dotSlider = null, speedSlider = null; -// LINKS and their replies (`traffic`'s `links`), and whether they are shown -// (`linksToggleInLegend`). -const isLinks = kind => kind === "links"; -let linksShown = true; -try { linksShown = localStorage.getItem("continuum-links-shown") !== "off"; } catch (_) {} -// The LINKS kind in the legend, a toggle: whether LINKS and their replies, which go only -// while a dashboard asks for the members' links, are shown: in the console, as dots on -// the map, and counted in the console's rates. Off, what is shown is the swarm's own -// traffic (kept in this browser). -function linksToggleInLegend(colour) { - const button = el("button", { class: "toggle message", style: "--colour: " + colour }, "links"); - const show = () => { - button.classList.toggle("hidden-kind", !linksShown); - document.getElementById("console").classList.toggle("no-links", !linksShown); - button.title = linksShown - ? "LINKS go only while a dashboard asks for the members' links. Click to leave them out: of the console, the dots and the rates" - : "Left out of the console, the dots and the rates. Click to show them again"; - }; - button.addEventListener("click", () => { - linksShown = !linksShown; - try { localStorage.setItem("continuum-links-shown", linksShown ? "on" : "off"); } catch (_) {} - if (!linksShown) { - for (let i = flows.length - 1; i >= 0; i--) if (isLinks(flows[i].kind)) flows.splice(i, 1); - } - show(); - showRates(); - filterConsole(); - const scroll = document.getElementById("console-scroll"); - scroll.scrollTop = scroll.scrollHeight; - }); - show(); - return button; -} function showMessagesLegend() { const legend = document.getElementById("messages-legend"); legend.classList.toggle("off", !flowsShown); @@ -2811,9 +2744,8 @@ function showMessagesLegend() { try { localStorage.setItem("continuum-dot-speed", String(flowSpeed)); } catch (_) {} }); // Before the toggle, which ends the line. - legend.prepend(...KIND_COLOURS.map(([kind, colour]) => kind === "links" - ? linksToggleInLegend(colour) - : el("span", { class: "message", style: "--colour: " + colour }, + legend.prepend(...KIND_COLOURS.map(([kind, colour]) => + el("span", { class: "message", style: "--colour: " + colour }, kind === "paths" ? "paths (pings)" : kind)), el("label", { class: "slider" }, "size", slider), el("label", { class: "slider" }, "speed", speed)); @@ -2876,7 +2808,7 @@ function addFlows(node, current, before, skip) { for (const [kind, count] of Object.entries(counts[way])) { const earlier = was[way][kind] || { messages: 0, bytes: 0 }; const grown = Math.max(0, count.messages - earlier.messages); - if (grown > 0 && (linksShown || !isLinks(kind))) { + if (grown > 0) { kinds[kind] = { messages: grown, bytes: Math.max(0, count.bytes - earlier.bytes) }; total += grown; } @@ -2923,7 +2855,6 @@ function remoteMessagesArrived(id, entry, list) { const now = performance.now(); const links = new Map(); for (const message of kept) { - if (!linksShown && isLinks(message.counted)) continue; const key = message.from + " " + message.to; const link = links.get(key) || { from: message.from, to: message.to, kinds: {}, total: 0 }; const kind = link.kinds[message.counted] || (link.kinds[message.counted] = { messages: 0, bytes: 0 }); @@ -2949,44 +2880,37 @@ function remoteMessagesArrived(id, entry, list) { // weighs in by how long ago it was, `e^(-age/window)`), from the counts each update // brings. The first sample is taken as it is, so they needn't climb from 0 first. --- const RATE_WINDOWS_S = [60, 300, 900]; -// By peer, and in all (`null`), with LINKS (`all`) and without (`withoutLinks`): the -// averages (`messages`, `bytes`: per window); and when the counts were last taken. +// By peer, and in all (`null`): the averages (`messages`, `bytes`: per window); and when +// the counts were last taken. const rates = new Map(); let ratesAt = null; function measureRates(current) { const now = performance.now(); if (lastTraffic !== null && ratesAt !== null && now > ratesAt) { const secs = (now - ratesAt) / 1000; - // Of `what` (`messages`, `bytes`), sent and received, by how much each grew; LINKS - // left out if `withoutLinks`. - const grown = (counts, before, what, withoutLinks) => ["sent", "received"].reduce((sum, way) => { - const total = of => Object.entries(of[way]).reduce((sum, [kind, count]) => - withoutLinks && isLinks(kind) ? sum : sum + count[what], 0); + // Of `what` (`messages`, `bytes`), sent and received, by how much each grew. + const grown = (counts, before, what) => ["sent", "received"].reduce((sum, way) => { + const total = of => Object.values(of[way]).reduce((sum, count) => sum + count[what], 0); return sum + Math.max(0, total(counts) - total(before)); }, 0); const average = (known, rate) => known ? known.map((average, i) => rate + (average - rate) * Math.exp(-secs / RATE_WINDOWS_S[i])) : RATE_WINDOWS_S.map(() => rate); - const sample = (key, variant, messages, bytes) => { - const known = rates.get(key) || {}; - const was = known[variant]; - known[variant] = { messages: average(was && was.messages, messages / secs), - bytes: average(was && was.bytes, bytes / secs) }; - rates.set(key, known); + const sample = (key, messages, bytes) => { + const was = rates.get(key); + rates.set(key, { messages: average(was && was.messages, messages / secs), + bytes: average(was && was.bytes, bytes / secs) }); }; - for (const variant of ["all", "withoutLinks"]) { - const withoutLinks = variant === "withoutLinks"; - let messagesInAll = 0, bytesInAll = 0; - for (const [id, counts] of current) { - const before = lastTraffic.get(id); - if (!before) continue; - const messages = grown(counts, before, "messages", withoutLinks); - const bytes = grown(counts, before, "bytes", withoutLinks); - sample(id, variant, messages, bytes); - messagesInAll += messages; - bytesInAll += bytes; - } - sample(null, variant, messagesInAll, bytesInAll); + let messagesInAll = 0, bytesInAll = 0; + for (const [id, counts] of current) { + const before = lastTraffic.get(id); + if (!before) continue; + const messages = grown(counts, before, "messages"); + const bytes = grown(counts, before, "bytes"); + sample(id, messages, bytes); + messagesInAll += messages; + bytesInAll += bytes; } + sample(null, messagesInAll, bytesInAll); // Peers gone from the list: their averages too. for (const key of [...rates.keys()]) if (key !== null && !current.has(key)) rates.delete(key); } @@ -2994,13 +2918,11 @@ function measureRates(current) { showRates(); } // The averages shown: in all, or with the filter, of the peer in focus; none (–) for a -// link between other nodes, a node that is no peer, or before two updates came. With -// LINKS or without (`linksShown`). +// link between other nodes, a node that is no peer, or before two updates came. function showRates() { const filtering = consoleFiltered && focused !== null; const peer = filtering ? peerOf(focused) : null; - const both = filtering && peer === null ? undefined : rates.get(peer); - const shown = both && both[linksShown ? "all" : "withoutLinks"]; + const shown = filtering && peer === null ? undefined : rates.get(peer); const format = rate => rate < 100 ? rate.toFixed(2) : String(Math.round(rate)); // Bytes in the unit the largest of the three needs, all three alike. const largest = shown ? Math.max(...shown.bytes) : 0; @@ -3109,8 +3031,7 @@ function filterConsole() { const filtering = consoleFiltered && focused !== null; log.classList.toggle("filtering", filtering); document.getElementById("console-none").hidden = - !filtering || log.querySelector(":scope > div.peer-focused" + - (linksShown ? "" : ":not(.links-kind)")) !== null; + !filtering || log.querySelector(":scope > div.peer-focused") !== null; const view = filtering ? focused : null; if (view !== consoleView) { consoleView = view; @@ -3128,7 +3049,6 @@ function consoleLine(message) { "data-peer": message.node ? "" : message.peer, "data-from": message.from || "", "data-to": message.to || "", class: (shownMessage && shownMessage.seq === message.seq ? "focused" : "") + - (isLinks(message.counted) ? " links-kind" : "") + (relatedMarked !== null && message.related === relatedMarked ? " related" : "") + peerClasses(message), title: message.way === "refused" ? "refused: the transport wouldn't send it" @@ -3273,7 +3193,6 @@ function connect() { status("live", true); pulse(); render(); trafficArrived(); epochsArrived(); remakesArrived(); if (treeShown && writerTree === null) loadTree(); - handOverLinks(); if (token === null && tokenTimer === null) obtainTokens(); for (const id of subscriptions) if (!subscribed.has(id)) follow(id); }); @@ -3451,8 +3370,8 @@ document.addEventListener("visibilitychange", () => { } } }); -// A click on the status: live, everything now (a tick on the node, the members asked -// again, what's open of the document read again); otherwise, reconnecting now. +// A click on the status: live, everything now (a tick on the node, what's open of the +// document read again); otherwise, reconnecting now. connection.title = "Click to refresh now"; connection.style.cursor = "pointer"; connection.addEventListener("click", () => { diff --git a/src/api/dashboard/links.rs b/src/api/dashboard/links.rs new file mode 100644 index 0000000..1f0b886 --- /dev/null +++ b/src/api/dashboard/links.rs @@ -0,0 +1,216 @@ +//! This node's links, as it measures them, for the dashboard's map (`own_links`): its +//! own in the state, and to any page that subscribes to them +//! (`GET /admin/dashboard/links/own/events`, also from another member's page). + +use crate::state::AppState; +use crate::swarm::topology; +use crate::transport::paths::{self, ClockOffset}; +use crate::transport::{Connection, Destination}; + +/// Links listed at most (a node has about `--neighbours`, more if others prefer it). +const MAX_LINKS: usize = 256; +/// Other transports in a link at most. +const MAX_OTHERS: usize = 8; +/// One of this node's links. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct Link { + /// The peer's node ID. + pub(crate) peer: String, + /// The transport its messages go over now (the scheme of the path in use), if any. + pub(crate) transport: Option<String>, + /// Only through a relay (libp2p). + pub(crate) relayed: bool, + /// The path in use is down: none to the peer answers (`paths`), whatever its + /// connection was. + pub(crate) down: bool, + /// The round trip on that path, in microseconds, if measured. + pub(crate) rtt_us: Option<u32>, + /// Silent for long (`peers.md`, dormancy). + pub(crate) dormant: bool, + /// The peer's other transports (`MAX_OTHERS` at most): where else it proved + /// itself, those paths up. + pub(crate) others: Vec<OtherPath>, + /// How far the peer's clock is ahead of this node's (`paths::ClockOffset`), if + /// measured: with other members', the page shows whose clock is off. + pub(crate) clock: Option<ClockOffset>, +} + +/// A transport of a link besides the one in use. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct OtherPath { + /// Its scheme (`mycelium`, `libp2p`, `tor`). + pub(crate) transport: String, + /// The round trip of its fastest path, in microseconds, if any is measured. + pub(crate) rtt_us: Option<u32>, + /// Only through a relay (libp2p): no direct connection at any of its paths. + pub(crate) relayed: bool, +} + +/// This node's links: its neighbours, and the peers it holds dormant, as it measures +/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its +/// link stays listed, dormant, until it is heard from again: left out, the link would +/// show only as that peer last reported it, from before. +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 = peer_state.lock().await.liveness.is_dormant(); + if !dormant && !topology::is_neighbour(state, &peer) { + continue; + } + let destination = state.transport.destination(&peer); + let transport = destination + .as_ref() + .and_then(|destination| scheme_of(&destination.address)); + links.push(Link { + others: others(destination.as_ref(), transport.as_deref()), + transport, + relayed: destination + .as_ref() + .is_some_and(|destination| destination.connection == Some(Connection::Relayed)), + // Ranked up before down: the first is down only if all are. + down: destination + .as_ref() + .and_then(|destination| destination.paths.first()) + .is_some_and(|path| path.down), + // The path in use's (`Transport::rtt`, the first of `destination.paths`). + rtt_us: state + .transport + .rtt(&peer) + .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), + dormant, + clock: state.transport.clock_offset(&peer), + peer, + }); + } + links.sort_by(|a, b| a.peer.cmp(&b.peer)); + links.truncate(MAX_LINKS); + links +} + +/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it +/// has one. +fn scheme_of(address: &str) -> Option<String> { + Some(paths::scheme_of(address)) + .filter(|scheme| !scheme.is_empty()) + .map(str::to_owned) +} + +/// The transports besides `used` the peer proved itself at, with a path up: each with +/// the round trip of its fastest path, if any is measured, and whether all of them go +/// through a relay. +fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> { + let mut others: Vec<OtherPath> = Vec::new(); + for path in destination.map_or(&[][..], |destination| &destination.paths[..]) { + let Some(transport) = scheme_of(&path.address) else { + continue; + }; + if path.down || used == Some(transport.as_str()) { + continue; + } + let rtt_us = path + .rtt_ms + .filter(|ms| ms.is_finite() && *ms >= 0.0) + .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32); + let relayed = path.connection == Some(Connection::Relayed); + match others.iter_mut().find(|other| other.transport == transport) { + Some(other) => { + other.rtt_us = match (other.rtt_us, rtt_us) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + }; + other.relayed &= relayed; + } + None => others.push(OtherPath { + transport, + rtt_us, + relayed, + }), + } + } + others.sort_by(|a, b| a.transport.cmp(&b.transport)); + others.truncate(MAX_OTHERS); + others +} + +/// Whether `new` says something `old` didn't: another peer, transport, relay or +/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock +/// measured or moved beyond both bounds. Round trips and clock samples wobble; small +/// moves aren't news. +pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool { + let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) { + (Some(a), Some(b)) => { + a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms) + } + (a, b) => a.is_some() != b.is_some(), + }; + let moved = |a: Option<u32>, b: Option<u32>| match (a, b) { + (Some(a), Some(b)) => { + let by = a.abs_diff(b); + by > 1000 && by > a.max(b) / 5 + } + (a, b) => a.is_some() != b.is_some(), + }; + old.len() != new.len() + || old.iter().zip(new).any(|(a, b)| { + a.peer != b.peer + || a.transport != b.transport + || a.relayed != b.relayed + || a.down != b.down + || a.dormant != b.dormant + || moved(a.rtt_us, b.rtt_us) + || clock_moved(a.clock, b.clock) + || a.others.len() != b.others.len() + || a.others.iter().zip(&b.others).any(|(x, y)| { + x.transport != y.transport + || x.relayed != y.relayed + || moved(x.rtt_us, y.rtt_us) + }) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn link(peer: &str, rtt_us: Option<u32>) -> Link { + Link { + peer: peer.into(), + transport: Some("libp2p".into()), + relayed: false, + down: false, + rtt_us, + dormant: false, + others: Vec::new(), + clock: None, + } + } + + #[test] + fn only_news_is_a_change() { + let old = [link("a", Some(10_000))]; + assert!(!changed(&old, &[link("a", Some(11_000))])); + assert!(changed(&old, &[link("a", Some(13_000))])); + assert!(changed(&old, &[link("a", None)])); + assert!(changed(&old, &[link("b", Some(10_000))])); + assert!(changed(&old, &[])); + let mut dormant = link("a", Some(10_000)); + dormant.dormant = true; + assert!(changed(&old, &[dormant])); + let mut down = link("a", Some(10_000)); + down.down = true; + assert!(changed(&old, &[down])); + // Small round trips: a millisecond's wobble isn't news. + assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))])); + // A clock: news when measured, and when it moved beyond both bounds. + let clocked = |offset_ms| Link { + clock: Some(ClockOffset { + offset_ms, + within_ms: 50, + }), + ..link("a", Some(10_000)) + }; + assert!(changed(&old, &[clocked(0)])); + assert!(!changed(&[clocked(0)], &[clocked(80)])); + assert!(changed(&[clocked(0)], &[clocked(5_000)])); + } +} diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index 78e8352..9cb4ea4 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -10,16 +10,14 @@ //! themselves: they alone changing is no change, and the page counts them on. //! //! The map: the graph the member list makes (`topology::intended_graph`, as this node -//! sees it), and the links every member measures: subscribed to with LINKS while anyone -//! looks (renewed every `LINKS_EVERY`, ended when nobody does; `swarm::links`), so a -//! member tells of a change as it happens; the same answers for all clients. Or, if -//! asked (`POST /admin/dashboard/links?live=false`, for every client, until asked -//! otherwise or the node restarts), only asked every `LINKS_EVERY`, without -//! subscribing: fewer messages in the swarm, a map up to `LINKS_EVERY` behind. +//! sees it), and the links this node measures (`own_links`). Other members' links the +//! page reads from them itself, from the nodes it subscribes to +//! (`GET /admin/dashboard/links/own/events`, there). + +mod links; use std::collections::{BTreeMap, BTreeSet}; use std::convert::Infallible; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; @@ -35,23 +33,16 @@ use tokio::time::Instant; use crate::locks::MutexExt; use crate::state::AppState; -use crate::swarm::{links, members, topology}; -use crate::transport::{Hex, Link}; +use crate::swarm::{members, topology}; +use crate::transport::Hex; +use links::Link; /// How often the snapshot is built and compared, while anyone looks. const TICK: Duration = Duration::from_secs(2); -/// How often the subscriptions to the members' links are renewed (each renewal is -/// answered: a reply at least this often), and the graph recomputed; changes come in -/// between, as they happen (`swarm::links`). -const LINKS_EVERY: Duration = Duration::from_secs(30); -/// How long a subscription lasts unless renewed: a node that is gone stops it. -const LINKS_LEASE: Duration = Duration::from_secs(90); -/// A member's links older than this are left out (it didn't answer since). -const LINKS_KEPT_FOR: Duration = Duration::from_secs(90); +/// How often the member list's graph is recomputed (it costs about `members²`). +const GRAPH_EVERY: Duration = Duration::from_secs(30); /// The producer runs on this long after the last request for the state. const LINGER: Duration = Duration::from_secs(60); -/// Members asked for their links at most, each round. -const MAX_ASKED: usize = 512; /// A snapshot or a patch as published: its number, and its JSON. A snapshot's text is /// made when first asked for (a client connecting, or one that fell behind), not every @@ -104,11 +95,8 @@ struct Hub { /// to the clients connected: a JSON list. messages: broadcast::Sender<Arc<String>>, control: Mutex<Control>, - /// Someone asked for it now (`refresh`): a tick at once, members asked again. + /// Someone asked for it now (`refresh`): a tick at once, the graph recomputed. poke: Notify, - /// The members' links subscribed to (true, at first), or only asked for every - /// `LINKS_EVERY` (`set_links`). - live: AtomicBool, /// True once the node stops: every client gets an `offline` event, and its stream /// ends (so the HTTP server needn't wait for clients to go). closing: watch::Receiver<bool>, @@ -133,7 +121,6 @@ impl Hub { documents, messages, poke: Notify::new(), - live: AtomicBool::new(true), control: Mutex::new(Control { running: false, wanted_until: Instant::now(), @@ -186,12 +173,11 @@ impl Hub { log.watch(true); let mut logged = 0u64; loop { - let live = self.live.load(Ordering::Relaxed); if Instant::now() >= graph_due { - graph.refresh(&self.state, live).await; - graph_due = Instant::now() + LINKS_EVERY; + graph.refresh(&self.state); + graph_due = Instant::now() + GRAPH_EVERY; } - let snapshot = build(&self.state, &graph, live).await; + let snapshot = build(&self.state, &graph).await; let stripped = without_ages(&snapshot); let changed = last.as_ref().is_none_or(|(_, old)| *old != stripped); if changed { @@ -236,11 +222,7 @@ impl Hub { return; } if unwanted() { - // Nobody looks: the members needn't tell us any more. Unless someone - // came meanwhile, then it goes on (and subscribes again). - for member in self.state.links.subscribed() { - links::unsubscribe(&self.state, &member).await; - } + // Nobody looks: it stops. Unless someone came meanwhile, then it goes on. let mut control = self.control.lock_or_recover(); if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until { log.watch(false); @@ -314,34 +296,20 @@ enum Next { Closing, } -/// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when -/// members are asked for their links too: subscribed to, if `live`. +/// The intended graph, recomputed every `GRAPH_EVERY`. #[derive(Default)] struct Graph { intended: BTreeMap<String, BTreeSet<String>>, } impl Graph { - async fn refresh(&mut self, state: &AppState, live: bool) { - let lease = if live { LINKS_LEASE } else { Duration::ZERO }; + fn refresh(&mut self, state: &AppState) { self.intended = topology::intended_graph(state).unwrap_or_default(); - let members: Vec<String> = { - let replica = state.doc.read().await; - members::members(&replica.doc) - }; - let members: Vec<String> = members - .into_iter() - .filter(|member| member != state.local_id()) - .take(MAX_ASKED) - .collect(); - // Those we can't reach, through peers that can (`links::ask_all`). - links::ask_all(state, members, lease).await; } } -/// The snapshot: the node, its figures, its peers, and the map (`live`: its links -/// subscribed to). -async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue { +/// The snapshot: the node, its figures, its peers, and the map. +async fn build(state: &AppState, graph: &Graph) -> JsonValue { // The peers once, for their list and the figures' totals over them. let list = super::peers::peer_list(state).await; let stats = serde_json::to_value(super::stats::collect(state, &list).await).unwrap_or_default(); @@ -369,43 +337,18 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue { .map(|(member, neighbours)| (member.clone(), json!(neighbours))) .collect(); let own = links::own_links(state).await; - let replies: Vec<_> = state - .links - .replies() - .into_iter() - .filter(|(_, age, _)| *age <= LINKS_KEPT_FOR) - .collect(); - // With each its epoch: the page shows who entered a new one. - let epochs = state.links.epochs(); - let epoch_json = |epoch: Option<crate::transport::Epoch>| { - epoch.map(|epoch| json!(crate::document::replica::EpochRecord::from(epoch))) - }; + // With its epoch: the page shows who entered a new one. let mut measured = Map::new(); measured.insert( local.to_owned(), json!({ "secsAgo": 0, "links": links_json(&own), - "epoch": epoch_json(Some(state.local_epoch())) }), + "epoch": crate::document::replica::EpochRecord::from(state.local_epoch()) }), ); - for (member, age, member_links) in &replies { - measured.insert( - member.clone(), - json!({ "secsAgo": age.as_secs(), "links": links_json(member_links), - "epoch": epoch_json(epochs.get(member).copied()) }), - ); - } // Everyone on the map: in the member list's graph, measured, or at the other end of - // a link measured. A node that joined through others is linked to them before its + // a link measured. A node that joined through this one is linked to it before its // registration reaches this node, and before the graph is computed again - // (`LINKS_EVERY`); it shows then, hollow (no links of its own reported). - let linked: BTreeSet<&String> = own - .iter() - .chain( - replies - .iter() - .flat_map(|(_, _, member_links)| member_links.iter()), - ) - .map(|link| &link.peer) - .collect(); + // (`GRAPH_EVERY`); it shows then, hollow (no links of its own reported). + let linked: BTreeSet<&String> = own.iter().map(|link| &link.peer).collect(); let mut members_json = Map::new(); for member in graph.intended.keys().chain(measured.keys()).chain(linked) { members_json.insert( @@ -423,7 +366,6 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue { "members": members_json, "intended": intended, "measured": measured, - "live": live, }, "remakes": remakes, }) @@ -522,7 +464,6 @@ pub(crate) fn router(state: AppState) -> Router<AppState> { .route("/admin/dashboard/state", get(current_state)) .route("/admin/dashboard/events", get(events)) .route("/admin/dashboard/refresh", post(refresh)) - .route("/admin/dashboard/links", post(set_links)) .route("/admin/dashboard/tree", get(writer_tree)) .route("/admin/dashboard/links/own", get(own_links_now)) .route("/admin/dashboard/links/own/events", get(own_links_events)) @@ -667,7 +608,7 @@ async fn page_html() -> Html<&'static str> { } /// `POST /admin/dashboard/refresh`: a tick now rather than at the next `TICK`, and the -/// members asked for their links again; what changed comes as events, as ever. +/// graph recomputed; what changed comes as events, as ever. async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode { hub.want(); hub.poke.notify_one(); @@ -675,29 +616,6 @@ async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode { } #[derive(serde::Deserialize)] -struct LinksQuery { - live: bool, -} - -/// `POST /admin/dashboard/links?live=…`: the members' links subscribed to (`true`), or -/// only asked for every `LINKS_EVERY` (`false`), for every client; the subscriptions -/// ended at once, and the members asked again now. -async fn set_links( - Extension(hub): Extension<Arc<Hub>>, - Query(query): Query<LinksQuery>, -) -> StatusCode { - hub.live.store(query.live, Ordering::Relaxed); - if !query.live { - for member in hub.state.links.subscribed() { - links::unsubscribe(&hub.state, &member).await; - } - } - hub.want(); - hub.poke.notify_one(); - StatusCode::NO_CONTENT -} - -#[derive(serde::Deserialize)] struct TreeQuery { writer: String, } diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index bcf4033..08928b4 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -190,7 +190,6 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou (direction, "blob", by.blob), (direction, "identity", by.identity), (direction, "paths", by.paths), - (direction, "links", by.links), (direction, "probe", by.probe), (direction, "plumtree", by.plumtree), (direction, "hyparview", by.hyparview), diff --git a/src/node.rs b/src/node.rs index 9b14447..23a881d 100644 --- a/src/node.rs +++ b/src/node.rs @@ -148,7 +148,6 @@ impl Node { clock_ahead_ms: Default::default(), clock_skew: Default::default(), blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)), - links: Default::default(), protocol_writers: Arc::new(config.protocol_writers), first_users: Arc::new(config.first_users), node_key: config.node_key, @@ -218,9 +217,6 @@ impl Node { }); trace!("Sync worker task spawned"); - // Subscribers to our links (`swarm::links`; any build has them). - tokio::spawn(crate::swarm::links::watch(state.clone())); - let peer_worker_state = state.clone(); tokio::spawn(async move { peer::run_peer_worker(peer_worker_state).await; diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs index f8d8123..04b2eb6 100644 --- a/src/simtest/bench.rs +++ b/src/simtest/bench.rs @@ -491,7 +491,7 @@ impl Totals { } /// The kinds, in the order of the report. -const KINDS: [&str; 12] = [ +const KINDS: [&str; 11] = [ "SYNC", "STATUS", "STATUS_REPLY", @@ -500,13 +500,12 @@ const KINDS: [&str; 12] = [ "HELLO_REPLY", "PING/PONG", "WHO/ME", - "LINKS", "PROBE", "PRUNE/GRAFT", "NEIGHBOUR", ]; -fn counters(by: &ByKind) -> [Counter; 12] { +fn counters(by: &ByKind) -> [Counter; 11] { [ by.sync, by.status, @@ -516,14 +515,13 @@ fn counters(by: &ByKind) -> [Counter; 12] { by.hello_reply, by.paths, by.identity, - by.links, by.probe, by.plumtree, by.hyparview, ] } -fn counters_mut(by: &mut ByKind) -> [&mut Counter; 12] { +fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] { [ &mut by.sync, &mut by.status, @@ -533,7 +531,6 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 12] { &mut by.hello_reply, &mut by.paths, &mut by.identity, - &mut by.links, &mut by.probe, &mut by.plumtree, &mut by.hyparview, diff --git a/src/simtest/links.rs b/src/simtest/links.rs deleted file mode 100644 index 652e3ed..0000000 --- a/src/simtest/links.rs +++ /dev/null @@ -1,259 +0,0 @@ -//! Links relayed (`swarm::links`): a member the asker can't reach, on a transport the -//! asker isn't on, is asked through a node on both, for as long as the asker renews. - -use std::time::Duration; - -use serde_json::json; - -use super::{Sim, read, write}; -use crate::state::AppState; -use crate::swarm::bootstrap::Start; -use crate::swarm::{links, members, topology}; - -const SOON: Duration = Duration::from_secs(20); - -/// Whether `node` has `member`'s links, listing `peer`. -fn has_links(node: &AppState, member: &str, peer: &str) -> bool { - node.links - .replies() - .iter() - .any(|(from, _, links)| from == member && links.iter().any(|link| link.peer == peer)) -} - -/// a on `sim` alone, c on `sim2` alone, b on both: a and c never reach each other, yet -/// a's writes reach c, and a gets c's links through b, while it renews; once it stops, -/// the subscription runs out at b and at c. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn links_are_relayed_to_members_out_of_reach() { - let sim = Sim::with_identity(61); - let a = sim.start_on("a", &["sim"], &[], Start::Genesis); - let b = sim.start_on("b", &["sim", "sim2"], &["a"], Start::Join); - let c = sim.start_on("c", &["sim2"], &["b"], Start::Join); - sim.wait_ready(&[&a, &b, &c], SOON).await; - let (a_id, b_id, c_id) = ( - a.local_id().to_owned(), - b.local_id().to_owned(), - c.local_id().to_owned(), - ); - - { - let (a, c_id) = (&a, c_id.clone()); - sim.wait_until("a knows c, out of its reach", SOON, move || { - let c_id = c_id.clone(); - async move { - let listed = members::members(&a.doc.read().await.doc).contains(&c_id); - // On `sim2` as a's view of the member list has it, which picks the relay. - let on = topology::transports_of(a, &c_id); - listed && on.contains("sim2") && !a.transport.can_reach(&c_id) - } - }) - .await; - } - - // Changes pass through b. - write(&a, "x", json!(1)).await; - { - let c = &c; - sim.wait_until("c has a's write", SOON, move || async move { - read(c, "x").await == Some(json!(1)) - }) - .await; - } - - // c's links reach a through b; b asks c for a. - let lease = Duration::from_secs(3); - links::ask_all(&a, vec![b_id.clone(), c_id.clone()], lease).await; - { - let (a, b_id, c_id) = (&a, b_id.clone(), c_id.clone()); - sim.wait_until("a has c's links, relayed by b", SOON, move || { - let (b_id, c_id) = (b_id.clone(), c_id.clone()); - async move { has_links(a, &c_id, &b_id) } - }) - .await; - } - assert!( - b.links - .relaying() - .contains(&(c_id.clone(), vec![a_id.clone()])), - "{:?}", - b.links.relaying() - ); - assert!(c.links.subscribers().contains(&b_id)); - assert!(!c.links.subscribers().contains(&a_id)); - - // Renewed: it lasts beyond its lease. - for _ in 0..3 { - tokio::time::sleep(lease / 2).await; - links::ask_all(&a, vec![c_id.clone()], lease).await; - } - assert!(!b.links.relaying().is_empty()); - assert!(c.links.subscribers().contains(&b_id)); - - // Not renewed: it runs out at every hop, without a message. - { - let (b, c, b_id) = (&b, &c, b_id.clone()); - sim.wait_until( - "the relayed subscription ran out at b and at c", - lease * 2, - move || { - let b_id = b_id.clone(); - async move { b.links.relaying().is_empty() && !c.links.subscribers().contains(&b_id) } - }, - ) - .await; - } -} - -/// a, b and c all on `sim`, but nothing passes between a and c (as behind NAT): once a -/// judges c unreachable, it asks c's links through b, though it has c's address. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn links_are_relayed_to_members_that_dont_answer() { - let sim = Sim::with_identity(64); - let cut = crate::transport::sim::LinkParams { - loss: 1.0, - ..Default::default() - }; - sim.net.set_link("a", "c", cut.clone()); - sim.net.set_link("c", "a", cut); - let b = sim.start_on("b", &["sim"], &[], Start::Genesis); - let a = sim.start_on("a", &["sim"], &["b"], Start::Join); - let c = sim.start_on("c", &["sim"], &["b"], Start::Join); - sim.wait_ready(&[&a, &b, &c], SOON).await; - let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned()); - { - let (a, c_id) = (&a, c_id.clone()); - sim.wait_until( - "a knows c, and that it doesn't answer", - SOON * 3, - move || { - let c_id = c_id.clone(); - async move { - let listed = members::members(&a.doc.read().await.doc).contains(&c_id); - listed && a.transport.can_reach(&c_id) && !topology::reaches(a, &c_id).await - } - }, - ) - .await; - } - links::ask_all( - &a, - vec![b_id.clone(), c_id.clone()], - Duration::from_secs(10), - ) - .await; - { - let (a, b_id, c_id) = (&a, b_id.clone(), c_id.clone()); - sim.wait_until("a has c's links, relayed by b", SOON, move || { - let (b_id, c_id) = (b_id.clone(), c_id.clone()); - async move { has_links(a, &c_id, &b_id) } - }) - .await; - } - assert!(!c.links.subscribers().contains(&a.local_id().to_owned())); -} - -/// a and c can't talk; b and d both could relay. Once the relay a asks through is cut -/// off from c (and from the other relay, so it can't pass the ask on), it brings -/// nothing: after `FAILED_AFTER` a's renewal goes through the other one, and the -/// subscription at the first runs out by itself. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn links_fail_over_to_another_relay() { - let sim = Sim::with_identity(67); - let cut = crate::transport::sim::LinkParams { - loss: 1.0, - ..Default::default() - }; - let cut_between = |x: &str, y: &str| { - sim.net.set_link(x, y, cut.clone()); - sim.net.set_link(y, x, cut.clone()); - }; - cut_between("a", "c"); - let b = sim.start_on("b", &["sim"], &[], Start::Genesis); - let d = sim.start_on("d", &["sim"], &["b"], Start::Join); - let a = sim.start_on("a", &["sim"], &["b"], Start::Join); - let c = sim.start_on("c", &["sim"], &["b"], Start::Join); - sim.wait_ready(&[&a, &b, &c, &d], SOON).await; - let (a_id, c_id) = (a.local_id().to_owned(), c.local_id().to_owned()); - { - let (a, c_id) = (&a, c_id.clone()); - sim.wait_until( - "a knows c, and that it doesn't answer", - SOON * 3, - move || { - let c_id = c_id.clone(); - async move { - let listed = members::members(&a.doc.read().await.doc).contains(&c_id); - listed && !topology::reaches(a, &c_id).await - } - }, - ) - .await; - } - - let lease = Duration::from_secs(3); - links::ask_all(&a, vec![c_id.clone()], lease).await; - { - let (b, d, c_id, a_id) = (&b, &d, c_id.clone(), a_id.clone()); - sim.wait_until("one of b and d relays c's links to a", SOON, move || { - let relayed = (c_id.clone(), vec![a_id.clone()]); - async move { - b.links.relaying().contains(&relayed) || d.links.relaying().contains(&relayed) - } - }) - .await; - } - let relayed = (c_id.clone(), vec![a_id.clone()]); - // With their names on the simulated net, which links are cut by. - let ((first, first_name), (other, other_name)) = if b.links.relaying().contains(&relayed) { - ((&b, "b"), (&d, "d")) - } else { - ((&d, "d"), (&b, "b")) - }; - - // The relay in use reaches c no more, nor the other relay; a still reaches it. - cut_between(first_name, "c"); - cut_between(first_name, other_name); - let cut_at = tokio::time::Instant::now(); - - // a renews as a dashboard does, and gets c's links through the other relay. - let renewing = { - let (a, c_id) = (a.clone(), c_id.clone()); - tokio::spawn(async move { - loop { - tokio::time::sleep(lease / 2).await; - links::ask_all(&a, vec![c_id.clone()], lease).await; - } - }) - }; - { - let (a, other, relayed, c_id) = (&a, other, relayed.clone(), c_id.clone()); - sim.wait_until( - "the other relay relays c's links to a, and they arrive", - SOON * 3, - move || { - let (relayed, c_id) = (relayed.clone(), c_id.clone()); - async move { - other.links.relaying().contains(&relayed) - && a.links - .replies() - .iter() - .any(|(from, age, _)| from == &c_id && *age < cut_at.elapsed()) - } - }, - ) - .await; - } - { - let (first, relayed) = (first, relayed.clone()); - sim.wait_until( - "the first relay's subscription ran out", - lease * 2, - move || { - let relayed = relayed.clone(); - async move { !first.links.relaying().contains(&relayed) } - }, - ) - .await; - } - renewing.abort(); -} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 118ec9d..a3001c8 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -13,7 +13,6 @@ mod bench; mod deltas; mod hyparview; mod identity; -mod links; mod reach; mod relay; mod remakes; diff --git a/src/simtest/remakes.rs b/src/simtest/remakes.rs index 95c5697..6fa4235 100644 --- a/src/simtest/remakes.rs +++ b/src/simtest/remakes.rs @@ -346,8 +346,6 @@ async fn coordinator_cut_off_after_committing_rejoins_the_others() { Who, Me, Ping, - Links, - LinksReply, ], ..LinkParams::default() } diff --git a/src/state/mod.rs b/src/state/mod.rs index 272855f..671aff7 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -138,8 +138,6 @@ pub(crate) struct AppState { pub(crate) clock_skew: Arc<std::sync::Mutex<crate::swarm::skew::Watch>>, /// Chunked transfers served and fetched (`swarm::blob`). pub(crate) blobs: Arc<crate::swarm::blob::Blobs>, - /// LINKS asked and answered, for a map of the swarm (`swarm::links`). - pub(crate) links: Arc<crate::swarm::links::Links>, /// HTTP API users who may write protocol data through the store API /// (`--protocol-writers`). pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>, diff --git a/src/swarm/links.rs b/src/swarm/links.rs deleted file mode 100644 index 2234783..0000000 --- a/src/swarm/links.rs +++ /dev/null @@ -1,1132 +0,0 @@ -//! 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 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 -//! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing -//! 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, 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. -//! -//! A member the asker can't reach (on none of its transports: one on Tor alone, asked -//! from one on libp2p alone; or one that doesn't answer it, as behind NAT: as choosing -//! neighbours judges it, `topology::reaches`) is asked through a peer that can -//! (`relay_for`): a LINKS -//! naming the members (`of`) asks the receiver to ask them on the sender's behalf, and -//! to pass their links on (`relayed`). The receiver asks each itself, or through a peer -//! of its own while hops are left (`hops`), for as long as the sender asked, and holds -//! what arrives a while to send it together: the window (`batch_ms`) is the asker's for -//! the whole way (at most `MAX_BATCH`), shared out among the hops that may be left, so -//! the links reach the asker within it however long the chain (every hop holds -//! `MIN_BATCH` at least). Such a -//! subscription isn't ended by a message: it runs out, at every hop, unless the asker -//! renews it, so one that is gone (or no longer looks) stops it within a lease. -//! -//! A way of asking a member that brings none of its links (`FAILED_AFTER`: we don't -//! reach it after all, or the relay doesn't) is left out a while for that member, and -//! the next ask (the renewal) goes the next way: so the old way runs out by itself, at -//! no cost but the wait. Every hop does so for what it asks on. - -use std::collections::{HashMap, HashSet}; -use std::sync::Mutex; -use std::time::Duration; - -use tokio::time::Instant; -use tracing::debug; - -use crate::locks::MutexExt; -use crate::state::AppState; -use crate::swarm::{members, topology}; -use crate::transport::paths::{self, ClockOffset}; -use crate::transport::{ - Connection, Destination, Epoch, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, - RelayedLinks, -}; - -/// Links one reply lists at most (a member has about `--neighbours`, more if others -/// prefer it). -pub(crate) const MAX_LINKS: usize = 256; -/// Other transports in a link at most. -pub(crate) const MAX_OTHERS: usize = 8; -/// A member answers each other one at most this often. -const ANSWER_EVERY: Duration = Duration::from_secs(5); -/// Replies kept from members at most (one each), and requests out. -const MAX_KEPT: usize = 4096; -/// Subscriptions a member keeps at most, and how long one lasts at most. -const MAX_SUBSCRIBERS: usize = 64; -const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300); -/// How often a member with subscribers looks whether its links changed. -const WATCH_EVERY: Duration = Duration::from_secs(2); -/// A reply to a LINKS that doesn't subscribe is taken for this long. -const REPLY_WITHIN: Duration = Duration::from_secs(30); -/// Members a relayed LINKS names at most, and that one LINKS_REPLY passes on at most. -const MAX_RELAYED: usize = 64; -/// Bytes of relayed links one LINKS_REPLY carries at most, as encoded (one member's -/// alone may be more): what a member reports is usually small (about 12 links of 150 -/// bytes), but may be up to `MAX_LINKS` links, and 64 such would make a message of -/// megabytes. -const MAX_RELAYED_BYTES: usize = 256 * 1024; -/// How long relayed links are held to be sent together, along the whole way: the -/// asker's window (at most `MAX_BATCH`), each hop holding at least `MIN_BATCH`. -pub(crate) const DEFAULT_BATCH: Duration = Duration::from_millis(250); -const MIN_BATCH: Duration = Duration::from_millis(10); -const MAX_BATCH: Duration = Duration::from_secs(5); -/// Members a peer may have us ask for it, at most: as many at once (`RELAY_BURST`), and -/// on average this many per second (`RELAY_RATE`): a dashboard asking every member -/// (`MAX_ASKED` there, 512) every 30 s needs less. The rest of a request is dropped. -const RELAY_BURST: f64 = 512.0; -const RELAY_RATE: f64 = 32.0; -/// Nodes a relayed LINKS goes through at most, the first included. -const MAX_HOPS: u8 = 3; -/// A way of asking a member (directly, or through a peer) that brought none of its -/// links for this long after we asked has failed it (`split`): the next ask goes -/// another way. More than a window (`MAX_BATCH`) along a few hops of slow transports. -const FAILED_AFTER: Duration = Duration::from_secs(20); -/// How long a way that failed a member is left out for that member. -const FAILED_KEPT: Duration = Duration::from_secs(300); - -/// The requests out and the replies in, and when each asker was last answered. -#[derive(Default)] -pub(crate) struct Links { - inner: Mutex<Inner>, - /// Someone subscribed (`answer`): `watch` looks again. - subscribed: tokio::sync::Notify, -} - -#[derive(Default)] -struct Inner { - /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its - /// subscription's end, or `REPLY_WITHIN`). - asked: HashMap<String, (u64, Instant)>, - /// The latest LINKS_REPLY of each member: when, and its links. - replies: HashMap<String, (Instant, Vec<Link>)>, - /// Each member's epoch, as its latest LINKS_REPLY said (`LinksReply::epoch`). - epochs: HashMap<String, Epoch>, - /// When we last answered each member. - answered: HashMap<String, Instant>, - /// Who subscribed to our links: the nonce to answer with, and until when. - subscribers: HashMap<String, (u64, Instant)>, - /// The links subscribers were last sent (`watch`). - /// And in which epoch: a new one is news too (a dashboard shows who entered it). - published: Option<(Epoch, Vec<Link>)>, - /// Our relayed LINKS out, by the peer asked to relay: its nonce, until when replies - /// to it are taken, and the members asked for through it. - via: HashMap<String, (u64, Instant, HashSet<String>)>, - /// What we ask on others' behalf, by member asked for: who asked (`Relaying`). - relaying: HashMap<String, HashMap<String, Relaying>>, - /// Relayed links due to an asker, by asker: the members whose links changed. - due: HashMap<String, HashSet<String>>, - /// How many members each peer may still have us ask for it (`RELAY_BURST`, - /// `RELAY_RATE`), as of when. - relay_budget: HashMap<String, (f64, Instant)>, - /// How we ask each member (`split`). - routes: HashMap<String, Route>, -} - -/// How we ask a member: the way we last did, since when, whether its links came since, -/// and the ways that failed it lately (`FAILED_AFTER`), each since when. -struct Route { - /// Through this peer; None: directly. - via: Option<String>, - asked: Instant, - arrived: bool, - failed: HashMap<Option<String>, Instant>, -} - -/// One asker's relayed subscription to a member. -#[derive(Clone, Copy)] -struct Relaying { - /// The nonce to answer with. - nonce: u64, - /// Until when, if still wanted. - until: Instant, - /// How long its links are held to be sent with others. - window: Duration, - /// Asked for once (no subscription): ends with the first links passed on. - once: bool, -} - -impl Links { - /// The latest links each member replied with, and how long ago. - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec<Link>)> { - let now = Instant::now(); - self.inner - .lock_or_recover() - .replies - .iter() - .map(|(member, (at, links))| { - ( - member.clone(), - now.saturating_duration_since(*at), - links.clone(), - ) - }) - .collect() - } - - /// The epoch each member was in, as its latest links said (`replies`). - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn epochs(&self) -> HashMap<String, Epoch> { - self.inner.lock_or_recover().epochs.clone() - } - - /// Who is subscribed to our links (not run out). - #[cfg(test)] - pub(crate) fn subscribers(&self) -> Vec<String> { - let now = Instant::now(); - self.inner - .lock_or_recover() - .subscribers - .iter() - .filter(|(_, (_, until))| now < *until) - .map(|(peer, _)| peer.clone()) - .collect() - } - - /// The members we ask on others' behalf (relayed, not run out), and who asked. - #[cfg(test)] - pub(crate) fn relaying(&self) -> Vec<(String, Vec<String>)> { - let now = Instant::now(); - self.inner - .lock_or_recover() - .relaying - .iter() - .map(|(member, askers)| { - let askers: Vec<String> = askers - .iter() - .filter(|(_, relaying)| now < relaying.until) - .map(|(asker, _)| asker.clone()) - .collect(); - (member.clone(), askers) - }) - .filter(|(_, askers)| !askers.is_empty()) - .collect() - } - - /// The members our subscriptions are with, and the peers relaying for us (not run - /// out). - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn subscribed(&self) -> Vec<String> { - let now = Instant::now(); - let inner = self.inner.lock_or_recover(); - inner - .asked - .iter() - .filter(|(_, (_, until))| now < *until) - .map(|(member, _)| member.clone()) - .chain( - inner - .via - .iter() - .filter(|(_, (_, until, _))| now < *until) - .map(|(relay, _)| relay.clone()), - ) - .collect() - } -} - -/// This node's links: its neighbours, and the peers it holds dormant, as it measures -/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its -/// link stays listed, dormant, until it is heard from again: left out, the link would -/// show only as that peer last reported it, from before. -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 = peer_state.lock().await.liveness.is_dormant(); - if !dormant && !topology::is_neighbour(state, &peer) { - continue; - } - let destination = state.transport.destination(&peer); - let transport = destination - .as_ref() - .and_then(|destination| scheme_of(&destination.address)); - links.push(Link { - others: others(destination.as_ref(), transport.as_deref()), - transport, - relayed: destination - .as_ref() - .is_some_and(|destination| destination.connection == Some(Connection::Relayed)), - // Ranked up before down: the first is down only if all are. - down: destination - .as_ref() - .and_then(|destination| destination.paths.first()) - .is_some_and(|path| path.down), - // The path in use's (`Transport::rtt`, the first of `destination.paths`). - rtt_us: state - .transport - .rtt(&peer) - .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), - dormant, - clock: state.transport.clock_offset(&peer), - peer, - }); - } - links.sort_by(|a, b| a.peer.cmp(&b.peer)); - links.truncate(MAX_LINKS); - links -} - -/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it -/// has one. -fn scheme_of(address: &str) -> Option<String> { - Some(paths::scheme_of(address)) - .filter(|scheme| !scheme.is_empty()) - .map(str::to_owned) -} - -/// The transports besides `used` the peer proved itself at, with a path up: each with -/// the round trip of its fastest path, if any is measured, and whether all of them go -/// through a relay. -fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> { - let mut others: Vec<OtherPath> = Vec::new(); - for path in destination.map_or(&[][..], |destination| &destination.paths[..]) { - let Some(transport) = scheme_of(&path.address) else { - continue; - }; - if path.down || used == Some(transport.as_str()) { - continue; - } - let rtt_us = path - .rtt_ms - .filter(|ms| ms.is_finite() && *ms >= 0.0) - .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32); - let relayed = path.connection == Some(Connection::Relayed); - match others.iter_mut().find(|other| other.transport == transport) { - Some(other) => { - other.rtt_us = match (other.rtt_us, rtt_us) { - (Some(a), Some(b)) => Some(a.min(b)), - (a, b) => a.or(b), - }; - other.relayed &= relayed; - } - None => others.push(OtherPath { - transport, - rtt_us, - relayed, - }), - } - } - others.sort_by(|a, b| a.transport.cmp(&b.transport)); - others.truncate(MAX_OTHERS); - others -} - -/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for -/// that long; its replies land in `AppState::links`. -#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) { - let nonce = { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) { - inner.asked.clear(); - } - let until = now + subscribe.max(REPLY_WITHIN); - // The same nonce while its replies are taken: a renewal doesn't drop a reply on - // its way under the last one (in a chain of relays, every hop renews at once, - // `relay`), as `ask_via` keeps one per relay. - match inner.asked.get_mut(member) { - Some((nonce, taken)) if now < *taken => { - *taken = (*taken).max(until); - *nonce - } - _ => { - let nonce = crate::transport::paths::random_nonce(); - inner.asked.insert(member.to_owned(), (nonce, until)); - nonce - } - } - }; - let (epoch, epoch_seq) = state.epoch_stamp(); - let request = ParsedMessage::Links(LinksRequest { - epoch, - epoch_seq, - nonce, - subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), - of: Vec::new(), - batch_ms: 0, - hops: 0, - }); - if let Err(error) = state.transport.send_message(member, &request).await { - debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); - } -} - -/// Asks `members` for their links (`ask`): those we can reach ourselves, the others -/// through peers that can (`relay_for`, `ask_via`), with our window for what they pass -/// on (`DEFAULT_BATCH`). -#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: Duration) { - let (direct, through) = split(state, members, None, true).await; - for member in direct { - let state = state.clone(); - tokio::spawn(async move { ask(&state, &member, subscribe).await }); - } - for (relay, members) in through { - ask_via( - state, - &relay, - members, - subscribe, - DEFAULT_BATCH, - MAX_HOPS - 1, - ) - .await; - } -} - -/// `members` we can reach ourselves, and the others by the peer to ask them through -/// (none for those no peer leads to, nor without `through_peers`); not `exclude` (who -/// asked us). A way that brought none of a member's links since we last asked it that -/// way (`FAILED_AFTER`) is left out for a while (`FAILED_KEPT`): directly, the next -/// relay, and so on; once every way failed, they are all tried again. -async fn split( - state: &AppState, - members: Vec<String>, - exclude: Option<&str>, - through_peers: bool, -) -> (Vec<String>, HashMap<String, Vec<String>>) { - let mut direct = Vec::new(); - let mut through: HashMap<String, Vec<String>> = HashMap::new(); - for member in members { - if member == state.local_id() || Some(member.as_str()) == exclude { - continue; - } - let failed = failed_ways(state, &member); - let mut via = way(state, &member, exclude, through_peers, &failed).await; - if via.is_none() && !failed.is_empty() { - forget_failed(state, &member); - via = way(state, &member, exclude, through_peers, &HashMap::new()).await; - } - let Some(via) = via else { - continue; - }; - asked_by(state, &member, via.clone()); - match via { - None => direct.push(member), - Some(relay) => through.entry(relay).or_default().push(member), - } - } - (direct, through) -} - -/// How to ask `member`, but the `failed` ways: Some(None) directly, Some(Some(peer)) -/// through `peer` (if `through_peers`); None if no way is left. -async fn way( - state: &AppState, - member: &str, - exclude: Option<&str>, - through_peers: bool, - failed: &HashMap<Option<String>, Instant>, -) -> Option<Option<String>> { - if !failed.contains_key(&None) && topology::reaches(state, member).await { - return Some(None); - } - if !through_peers { - return None; - } - relay_for(state, member, exclude, failed).await.map(Some) -} - -/// The ways that failed `member` lately; the one we last asked it by among them if it -/// brought nothing within `FAILED_AFTER`. -fn failed_ways(state: &AppState, member: &str) -> HashMap<Option<String>, Instant> { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - let Some(route) = inner.routes.get_mut(member) else { - return HashMap::new(); - }; - route - .failed - .retain(|_, since| now.saturating_duration_since(*since) < FAILED_KEPT); - if !route.arrived && now.saturating_duration_since(route.asked) >= FAILED_AFTER { - debug!(member = %member, via = ?route.via, "LINKS brought nothing that way; asking another"); - route.failed.insert(route.via.clone(), now); - } - route.failed.clone() -} - -/// Every way may be tried again for `member`. -fn forget_failed(state: &AppState, member: &str) { - if let Some(route) = state.links.inner.lock_or_recover().routes.get_mut(member) { - route.failed.clear(); - } -} - -/// We ask `member` `via` now: the wait for its links starts over, unless we asked it so -/// already and are still waiting. -fn asked_by(state: &AppState, member: &str, via: Option<String>) { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - if inner.routes.len() >= MAX_KEPT && !inner.routes.contains_key(member) { - inner.routes.clear(); - } - let route = inner - .routes - .entry(member.to_owned()) - .or_insert_with(|| Route { - via: via.clone(), - asked: now, - arrived: false, - failed: HashMap::new(), - }); - if route.via != via || route.arrived { - route.via = via; - route.asked = now; - route.arrived = false; - } -} - -/// The peer to ask `member` through, which we can't reach: an active member we reach -/// (that answers us) that, by the member list, is on a transport `member` is on; else one on a transport -/// we aren't on (it may know one that is). Of several, the one with the lowest pair score -/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`, nor one -/// that failed `member` (`failed`). -async fn relay_for( - state: &AppState, - member: &str, - exclude: Option<&str>, - failed: &HashMap<Option<String>, Instant>, -) -> Option<String> { - let theirs = topology::transports_of(state, member); - let ours = topology::transports_of(state, state.local_id()); - let mut best: Option<(bool, u64, String)> = None; - for (peer, peer_state) in state.peers_snapshot().await { - if peer == member - || Some(peer.as_str()) == exclude - || failed.contains_key(&Some(peer.clone())) - || !state.transport.can_reach(&peer) - { - continue; - } - { - let peer_state = peer_state.lock().await; - if peer_state.is_unreachable() - || peer_state.liveness.is_dormant() - || !matches!(peer_state.standing, crate::state::Standing::Member { .. }) - { - continue; - } - } - let on = topology::transports_of(state, &peer); - let shares = !on.is_disjoint(&theirs); - if !shares && on.is_subset(&ours) { - continue; - } - let score = topology::pair_score(&peer, member); - // Sharing a transport with the member first, then the lowest score. - let better = best.as_ref().is_none_or(|(best_shares, best_score, _)| { - (shares, std::cmp::Reverse(score)) > (*best_shares, std::cmp::Reverse(*best_score)) - }); - if better { - best = Some((shares, score, peer)); - } - } - best.map(|(_, _, peer)| peer) -} - -/// Asks `relay` to ask `members` on our behalf (a LINKS naming them), with `subscribe`, -/// passing their links on within `window`, through `hops` more nodes at most: in LINKS -/// of `MAX_RELAYED` members at most. One nonce per relay while its replies are taken, so -/// its replies for every member we asked it for are taken. -async fn ask_via( - state: &AppState, - relay: &str, - members: Vec<String>, - subscribe: Duration, - window: Duration, - hops: u8, -) { - let nonce = { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - if inner.via.len() >= MAX_KEPT && !inner.via.contains_key(relay) { - inner.via.retain(|_, (_, until, _)| now < *until); - } - let until = now + subscribe.max(REPLY_WITHIN); - let entry = inner.via.entry(relay.to_owned()).or_insert_with(|| { - ( - crate::transport::paths::random_nonce(), - until, - HashSet::new(), - ) - }); - if now >= entry.1 { - entry.0 = crate::transport::paths::random_nonce(); - entry.2.clear(); - } - entry.1 = entry.1.max(until); - entry.2.extend(members.iter().cloned()); - entry.0 - }; - let (epoch, epoch_seq) = state.epoch_stamp(); - for members in members.chunks(MAX_RELAYED) { - let request = ParsedMessage::Links(LinksRequest { - epoch, - epoch_seq, - nonce, - subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), - of: members.to_vec(), - batch_ms: u32::try_from(window.as_millis()).unwrap_or(u32::MAX), - hops, - }); - if let Err(error) = state.transport.send_message(relay, &request).await { - debug!(relay = %relay, error = %format!("{error:#}"), "Relayed LINKS failed"); - } - } -} - -/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any -/// more. A peer relaying for us isn't told: what it asks for us runs out by itself. -#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn unsubscribe(state: &AppState, member: &str) { - let nonce = { - let mut inner = state.links.inner.lock_or_recover(); - if inner.via.remove(member).is_some() && !inner.asked.contains_key(member) { - return; - } - match inner.asked.remove(member) { - Some((nonce, _)) => nonce, - None => return, - } - }; - let (epoch, epoch_seq) = state.epoch_stamp(); - let request = ParsedMessage::Links(LinksRequest { - epoch, - epoch_seq, - nonce, - subscribe_secs: 0, - of: Vec::new(), - batch_ms: 0, - hops: 0, - }); - if let Err(error) = state.transport.send_message(member, &request).await { - debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); - } -} - -/// A LINKS from `peer`: answered if it's a member, and not answered lately. -pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) { - // From the neighbours' view, kept as the member list changes: no document read per - // request. - let member = match crate::swarm::topology::listed(state, peer) { - Some(listed) => listed, - None => { - let replica = state.doc.read().await; - members::members(&replica.doc) - .iter() - .any(|member| member == peer) - } - }; - if !member { - debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member"); - return; - } - if !request.of.is_empty() { - relay(state, peer, request).await; - return; - } - { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - // A subscription is taken, renewed or ended even when the reply is due later. - if request.subscribe_secs == 0 { - inner.subscribers.remove(peer); - } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer) - { - let lasts = - Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION); - inner - .subscribers - .insert(peer.to_owned(), (request.nonce, now + lasts)); - state.links.subscribed.notify_one(); - } - if inner - .answered - .get(peer) - .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY) - { - return; - } - if inner.answered.len() >= MAX_KEPT { - inner - .answered - .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY); - } - inner.answered.insert(peer.to_owned(), now); - } - let (epoch, epoch_seq) = state.epoch_stamp(); - let reply = ParsedMessage::LinksReply(LinksReply { - epoch, - epoch_seq, - nonce: request.nonce, - links: own_links(state).await, - relayed: Vec::new(), - }); - if let Err(error) = state.transport.send_message(peer, &reply).await { - debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); - } -} - -/// A relayed LINKS from `asker` (`answer`): the members it names are asked on its -/// behalf, ourselves or through a peer of ours while hops are left, and what we know of -/// them already is passed on at once (within its window); the rest as it arrives. -async fn relay(state: &AppState, asker: &str, request: LinksRequest) { - let now = Instant::now(); - let lease = Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION); - let once = lease.is_zero(); - // What is left of the asker's window, shared out among this hop and those that may - // follow: ours is a share, the rest goes on. - let left = Duration::from_millis(u64::from(request.batch_ms)).min(MAX_BATCH); - let window = (left / (u32::from(request.hops) + 1)).max(MIN_BATCH); - let rest = left.saturating_sub(window); - let mut members: Vec<String> = request - .of - .into_iter() - .filter(|member| member != asker && member != state.local_id()) - .take(MAX_RELAYED) - .collect(); - let known: Vec<String> = { - let mut inner = state.links.inner.lock_or_recover(); - // Within the asker's budget: each member costs one. - if inner.relay_budget.len() >= MAX_KEPT && !inner.relay_budget.contains_key(asker) { - inner.relay_budget.retain(|_, (budget, at)| { - *budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE - < RELAY_BURST - }); - } - let (budget, at) = inner - .relay_budget - .entry(asker.to_owned()) - .or_insert((RELAY_BURST, now)); - *budget = (*budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE) - .min(RELAY_BURST); - *at = now; - let allowed = (*budget as usize).min(members.len()); - *budget -= allowed as f64; - if allowed < members.len() { - debug!(asker = %asker, asked = members.len(), allowed, "Relayed LINKS over its budget; asking for part of it"); - members.truncate(allowed); - } - for member in &members { - if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) { - inner - .relaying - .retain(|_, askers| askers.values().any(|relaying| now < relaying.until)); - } - let askers = inner.relaying.entry(member.clone()).or_default(); - askers.retain(|_, relaying| now < relaying.until); - if askers.len() >= MAX_SUBSCRIBERS && !askers.contains_key(asker) { - continue; - } - askers.insert( - asker.to_owned(), - Relaying { - nonce: request.nonce, - // Asked once: for as long as its reply is awaited. - until: now + if once { REPLY_WITHIN } else { lease }, - window, - once, - }, - ); - } - members - .iter() - .filter(|member| { - inner - .replies - .get(*member) - .is_some_and(|(at, _)| now.saturating_duration_since(*at) < REPLY_WITHIN) - }) - .cloned() - .collect() - }; - for member in known { - forward(state, &member); - } - let (direct, through) = split(state, members, Some(asker), request.hops > 0).await; - for member in direct { - // Asked once while we subscribe anyway (our own dashboard, or another asker's): - // what comes is passed on, and a LINKS with 0 would end that subscription. - let subscribed = once - && state - .links - .inner - .lock_or_recover() - .asked - .get(&member) - .is_some_and(|(_, until)| now + REPLY_WITHIN < *until); - if !subscribed { - ask(state, &member, lease).await; - } - } - for (next, members) in through { - ask_via(state, &next, members, lease, rest, request.hops - 1).await; - } -} - -/// `member`'s links changed (or arrived): due to those who asked us for them -/// (`relaying`), each sent with what else is due to it at the end of its window. -fn forward(state: &AppState, member: &str) { - let now = Instant::now(); - let mut starting = Vec::new(); - { - let mut inner = state.links.inner.lock_or_recover(); - let askers: Vec<(String, Duration)> = match inner.relaying.get_mut(member) { - Some(askers) => { - askers.retain(|_, relaying| now < relaying.until); - askers - .iter() - .map(|(asker, relaying)| (asker.clone(), relaying.window)) - .collect() - } - None => return, - }; - for (asker, window) in askers { - let due = inner.due.entry(asker.clone()).or_default(); - if due.is_empty() { - starting.push((asker, window)); - } - due.insert(member.to_owned()); - } - } - for (asker, window) in starting { - let state = state.clone(); - tokio::spawn(async move { - tokio::time::sleep(window).await; - pass_on(&state, &asker).await; - }); - } -} - -/// What is due to `asker`: the links of the members it asked for, as we have them, in -/// as few LINKS_REPLYs as `MAX_RELAYED` allows (one per nonce it asked with). -async fn pass_on(state: &AppState, asker: &str) { - let now = Instant::now(); - let mut by_nonce: HashMap<u64, Vec<RelayedLinks>> = HashMap::new(); - { - let mut inner = state.links.inner.lock_or_recover(); - let due = inner.due.remove(asker).unwrap_or_default(); - for member in due { - let Some(relaying) = inner - .relaying - .get(&member) - .and_then(|askers| askers.get(asker)) - .copied() - .filter(|relaying| now < relaying.until) - else { - continue; - }; - let Some((at, links)) = inner.replies.get(&member) else { - continue; - }; - let Some(epoch) = inner.epochs.get(&member).copied() else { - continue; - }; - let age_ms = - u32::try_from(now.saturating_duration_since(*at).as_millis()).unwrap_or(u32::MAX); - by_nonce - .entry(relaying.nonce) - .or_default() - .push(RelayedLinks { - member: member.clone(), - age_ms, - epoch, - links: links.clone(), - }); - if relaying.once - && let Some(askers) = inner.relaying.get_mut(&member) - { - askers.remove(asker); - } - } - } - let (epoch, epoch_seq) = state.epoch_stamp(); - for (nonce, relayed) in by_nonce { - for part in by_size(relayed) { - let reply = ParsedMessage::LinksReply(LinksReply { - epoch, - epoch_seq, - nonce, - links: Vec::new(), - relayed: part, - }); - if let Err(error) = state.transport.send_message(asker, &reply).await { - debug!(asker = %asker, error = %format!("{error:#}"), "Relayed LINKS_REPLY failed"); - } - } - } -} - -/// `relayed` in parts for a LINKS_REPLY each: `MAX_RELAYED` members' links at most, and -/// `MAX_RELAYED_BYTES` as encoded (but for one member's alone). -fn by_size(relayed: Vec<RelayedLinks>) -> Vec<Vec<RelayedLinks>> { - let mut parts: Vec<Vec<RelayedLinks>> = Vec::new(); - let mut bytes = 0; - for links in relayed { - let size = rmp_serde::encode::to_vec(&links).map_or(0, |encoded| encoded.len()); - let full = parts.last().is_none_or(|part| { - part.len() >= MAX_RELAYED || (!part.is_empty() && bytes + size > MAX_RELAYED_BYTES) - }); - if full { - parts.push(Vec::new()); - bytes = 0; - } - bytes += size; - parts.last_mut().expect("just made").push(links); - } - parts -} - -/// A LINKS_REPLY from `peer`: its own links kept if it answers our latest LINKS to it, -/// and those it relays if it answers our latest relayed one, while their replies are -/// taken (a subscription's go on); then passed on to whoever asked us for them. -pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) { - let now = Instant::now(); - let mut arrived = Vec::new(); - { - let mut inner = state.links.inner.lock_or_recover(); - let own = inner - .asked - .get(peer) - .is_some_and(|(nonce, until)| *nonce == reply.nonce && now < *until); - let via: Option<HashSet<String>> = inner - .via - .get(peer) - .filter(|(nonce, until, _)| *nonce == reply.nonce && now < *until) - .map(|(_, _, members)| members.clone()); - let mut keep = |member: String, age: Duration, epoch: Epoch, mut links: Vec<Link>| { - links.truncate(MAX_LINKS); - for link in &mut links { - link.others.truncate(MAX_OTHERS); - } - if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(&member) { - inner.replies.clear(); - inner.epochs.clear(); - } - let at = now.checked_sub(age).unwrap_or(now); - // Arrived, if older news: the way works. - if let Some(route) = inner.routes.get_mut(&member) { - route.arrived = true; - } - // Not older news over newer. - let known = inner.replies.get(&member); - if known.is_some_and(|(known, _)| *known > at) { - return; - } - // The same links as we have, lately: no news, so not passed on. A copy that - // went round between relays (each asking the other for the member) comes - // back a little newer each time, as its age counts from when each sent it: - // passed on, it would go round for good. - let news = known.is_none_or(|(known_at, known_links)| { - *known_links != links || now.saturating_duration_since(*known_at) >= REPLY_WITHIN - }) || inner.epochs.get(&member) != Some(&epoch); - inner.replies.insert(member.clone(), (at, links)); - inner.epochs.insert(member.clone(), epoch); - if news { - arrived.push(member); - } - }; - if own { - keep(peer.to_owned(), Duration::ZERO, reply.epoch, reply.links); - } - if let Some(members) = via { - for relayed in reply.relayed.into_iter().take(MAX_RELAYED) { - if members.contains(&relayed.member) { - keep( - relayed.member, - Duration::from_millis(u64::from(relayed.age_ms)), - relayed.epoch, - relayed.links, - ); - } - } - } - } - for member in arrived { - forward(state, &member); - } -} - -/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's -/// links; sent to each of them when they changed since they were last sent. Without -/// any, it waits for one. Runs for the node's life. -pub(crate) async fn watch(state: AppState) { - let mut every = tokio::time::interval(WATCH_EVERY); - every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - loop { - every.tick().await; - let subscribers: Vec<(String, u64)> = { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - inner.subscribers.retain(|_, (_, until)| now < *until); - if inner.subscribers.is_empty() { - inner.published = None; - } - inner - .subscribers - .iter() - .map(|(peer, (nonce, _))| (peer.clone(), *nonce)) - .collect() - }; - if subscribers.is_empty() { - // A new subscriber was just answered (`answer`, or within `ANSWER_EVERY`): - // the next look is due an interval later. - state.links.subscribed.notified().await; - every.reset(); - continue; - } - let links = own_links(&state).await; - let (epoch, epoch_seq) = state.epoch_stamp(); - { - let mut inner = state.links.inner.lock_or_recover(); - if inner - .published - .as_ref() - .is_some_and(|(published_in, published)| { - *published_in == epoch && !changed(published, &links) - }) - { - continue; - } - inner.published = Some((epoch, links.clone())); - } - for (peer, nonce) in subscribers { - let reply = ParsedMessage::LinksReply(LinksReply { - epoch, - epoch_seq, - nonce, - links: links.clone(), - relayed: Vec::new(), - }); - if let Err(error) = state.transport.send_message(&peer, &reply).await { - debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); - } - } - } -} - -/// Whether `new` says something `old` didn't: another peer, transport, relay or -/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock -/// measured or moved beyond both bounds. Round trips and clock samples wobble; small -/// moves aren't news. -pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool { - let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) { - (Some(a), Some(b)) => { - a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms) - } - (a, b) => a.is_some() != b.is_some(), - }; - let moved = |a: Option<u32>, b: Option<u32>| match (a, b) { - (Some(a), Some(b)) => { - let by = a.abs_diff(b); - by > 1000 && by > a.max(b) / 5 - } - (a, b) => a.is_some() != b.is_some(), - }; - old.len() != new.len() - || old.iter().zip(new).any(|(a, b)| { - a.peer != b.peer - || a.transport != b.transport - || a.relayed != b.relayed - || a.down != b.down - || a.dormant != b.dormant - || moved(a.rtt_us, b.rtt_us) - || clock_moved(a.clock, b.clock) - || a.others.len() != b.others.len() - || a.others.iter().zip(&b.others).any(|(x, y)| { - x.transport != y.transport - || x.relayed != y.relayed - || moved(x.rtt_us, y.rtt_us) - }) - }) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn relayed_links_are_sent_in_parts_by_count_and_size() { - let relayed = |links: usize| RelayedLinks { - member: "m".into(), - age_ms: 0, - epoch: Epoch { - generation: 1, - id: [0; 32], - }, - links: (0..links) - .map(|i| link(&format!("peer-{i:0>70}"), Some(1500))) - .collect(), - }; - // Small: by count. - let parts = by_size((0..130).map(|_| relayed(2)).collect()); - assert_eq!(parts.iter().map(Vec::len).collect::<Vec<_>>(), [64, 64, 2]); - // Large: by size, each part within the bytes but for one alone. - let big = rmp_serde::encode::to_vec(&relayed(MAX_LINKS)) - .unwrap() - .len(); - assert!(20 * big > MAX_RELAYED_BYTES, "{big}"); - let parts = by_size((0..20).map(|_| relayed(MAX_LINKS)).collect()); - assert_eq!(parts.iter().map(Vec::len).sum::<usize>(), 20); - for part in &parts { - let bytes: usize = part - .iter() - .map(|links| rmp_serde::encode::to_vec(links).unwrap().len()) - .sum(); - assert!(part.len() == 1 || bytes <= MAX_RELAYED_BYTES, "{bytes}"); - } - assert!(parts.len() > 1); - } - - fn link(peer: &str, rtt_us: Option<u32>) -> Link { - Link { - peer: peer.into(), - transport: Some("libp2p".into()), - relayed: false, - down: false, - rtt_us, - dormant: false, - others: Vec::new(), - clock: None, - } - } - - #[test] - fn only_news_is_a_change() { - let old = [link("a", Some(10_000))]; - assert!(!changed(&old, &[link("a", Some(11_000))])); - assert!(changed(&old, &[link("a", Some(13_000))])); - assert!(changed(&old, &[link("a", None)])); - assert!(changed(&old, &[link("b", Some(10_000))])); - assert!(changed(&old, &[])); - let mut dormant = link("a", Some(10_000)); - dormant.dormant = true; - assert!(changed(&old, &[dormant])); - let mut down = link("a", Some(10_000)); - down.down = true; - assert!(changed(&old, &[down])); - // Small round trips: a millisecond's wobble isn't news. - assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))])); - // A clock: news when measured, and when it moved beyond both bounds. - let clocked = |offset_ms| Link { - clock: Some(ClockOffset { - offset_ms, - within_ms: 50, - }), - ..link("a", Some(10_000)) - }; - assert!(changed(&old, &[clocked(0)])); - assert!(!changed(&[clocked(0)], &[clocked(80)])); - assert!(changed(&[clocked(0)], &[clocked(5_000)])); - } -} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index ccffaa0..7b5b09c 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -10,7 +10,6 @@ pub(crate) mod document_file; pub(crate) mod flood; pub(crate) mod hop; pub(crate) mod hyparview; -pub(crate) mod links; pub(crate) mod members; pub(crate) mod pair; pub(crate) mod peer; diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 0c41466..d25c7de 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -242,12 +242,6 @@ pub(crate) async fn handle_message( heard_heads(state, &peer, &remote); on_status(state, &peer, &remote, reactivated).await; } - transport::ParsedMessage::Links(request) => { - crate::swarm::links::answer(state, &peer, request).await; - } - transport::ParsedMessage::LinksReply(reply) => { - crate::swarm::links::received(state, &peer, reply); - } transport::ParsedMessage::Probe(request) => { crate::swarm::working_links::answer(state, &peer, request).await; } diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index c23d2b2..2d812e5 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -530,7 +530,8 @@ pub(crate) async fn reaches(state: &AppState, peer: &str) -> bool { } /// The transports `member` is on, by the member list (the schemes of its addresses); -/// ours as we are on them. None known: empty. +/// ours as we are on them. None known: empty. For the tests. +#[cfg(test)] pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> { let view = state.neighbours.read_or_recover(); match view.as_ref() { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 8cbf412..f5b2db1 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -238,11 +238,6 @@ pub(crate) enum ParsedMessage { nonce: u64, at: crate::clock::UnixMillis, }, - /// Asks a member how it is linked to its peers, for a map of the swarm - /// (`swarm::links`). - Links(LinksRequest), - /// Answers a LINKS. - LinksReply(LinksReply), /// Asks whether the recipient reaches a member (`swarm::working_links`). Probe(ProbeRequest), /// Answers a PROBE. @@ -397,8 +392,6 @@ pub(crate) enum MessageKind { Me, Ping, Pong, - Links, - LinksReply, Probe, ProbeReply, Prune, @@ -421,8 +414,6 @@ impl ParsedMessage { ParsedMessage::Me(_) => MessageKind::Me, ParsedMessage::Ping(_) => MessageKind::Ping, ParsedMessage::Pong { .. } => MessageKind::Pong, - ParsedMessage::Links(_) => MessageKind::Links, - ParsedMessage::LinksReply(_) => MessageKind::LinksReply, ParsedMessage::Probe(_) => MessageKind::Probe, ParsedMessage::ProbeReply(_) => MessageKind::ProbeReply, ParsedMessage::Prune(_) => MessageKind::Prune, @@ -446,8 +437,6 @@ impl ParsedMessage { Some((status.epoch, status.epoch_seq)) } ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)), - ParsedMessage::Links(request) => Some((request.epoch, request.epoch_seq)), - ParsedMessage::LinksReply(reply) => Some((reply.epoch, reply.epoch_seq)), ParsedMessage::Probe(request) => Some((request.epoch, request.epoch_seq)), ParsedMessage::ProbeReply(reply) => Some((reply.epoch, reply.epoch_seq)), ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => { @@ -530,92 +519,6 @@ pub(crate) struct ProbeReply { pub(crate) reached: Option<bool>, } -/// LINKS: what a member is asked for a map of the swarm (`swarm::links`). -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct LinksRequest { - #[serde(with = "crate::transport::wire::codec::epoch")] - pub(crate) epoch: Epoch, - /// See `SyncEnvelope::epoch_seq`. - pub(crate) epoch_seq: u32, - /// Repeated in the reply: which request it answers. - pub(crate) nonce: u64, - /// For this long (seconds), a LINKS_REPLY again whenever the links change (a - /// subscription; the next LINKS renews it). 0: this reply only, and a subscription - /// ends (a relayed one runs out instead). - pub(crate) subscribe_secs: u32, - /// Members the receiver is to ask on the sender's behalf, which the sender can't - /// reach, passing their links on (relayed, `swarm::links`); none: the receiver's own. - pub(crate) of: Vec<String>, - /// How long relayed links may be held to send several in one LINKS_REPLY, by the - /// receiver and the hops after it together (milliseconds; `swarm::links`). - pub(crate) batch_ms: u32, - /// How many more nodes a relayed request may go through, after the receiver. - pub(crate) hops: u8, -} - -/// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct LinksReply { - #[serde(with = "crate::transport::wire::codec::epoch")] - pub(crate) epoch: Epoch, - /// See `SyncEnvelope::epoch_seq`. - pub(crate) epoch_seq: u32, - /// The request's. - pub(crate) nonce: u64, - /// The sender's own, if asked for; at most `swarm::links::MAX_LINKS`. - pub(crate) links: Vec<Link>, - /// Other members' links, passed on (`LinksRequest::of`). - pub(crate) relayed: Vec<RelayedLinks>, -} - -/// A member's links as a LINKS_REPLY passes them on. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct RelayedLinks { - pub(crate) member: String, - /// How long ago the member reported them (milliseconds). - pub(crate) age_ms: u32, - /// The member's epoch, as its LINKS_REPLY said (`LinksReply::epoch`): a dashboard - /// shows who entered a new one, swarm-wide. - #[serde(with = "crate::transport::wire::codec::epoch")] - pub(crate) epoch: Epoch, - pub(crate) links: Vec<Link>, -} - -/// One of a member's links, in a LINKS_REPLY. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub(crate) struct Link { - /// The peer's node ID. - pub(crate) peer: String, - /// The transport its messages go over now (the scheme of the path in use), if any. - pub(crate) transport: Option<String>, - /// Only through a relay (libp2p). - pub(crate) relayed: bool, - /// The path in use is down: none to the peer answers (`paths`), whatever its - /// connection was. - pub(crate) down: bool, - /// The round trip on that path, in microseconds, if measured. - pub(crate) rtt_us: Option<u32>, - /// Silent for long (`peers.md`, dormancy). - pub(crate) dormant: 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>, - /// How far the peer's clock is ahead of the member's (`paths::ClockOffset`), if - /// measured: with every member's, a map of the swarm shows whose clock is off. - pub(crate) clock: Option<paths::ClockOffset>, -} - -/// A transport of a link besides the one in use, in a LINKS_REPLY. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub(crate) struct OtherPath { - /// Its scheme (`mycelium`, `libp2p`, `tor`). - pub(crate) transport: String, - /// The round trip of its fastest path, in microseconds, if any is measured. - pub(crate) rtt_us: Option<u32>, - /// Only through a relay (libp2p): no direct connection at any of its paths. - pub(crate) relayed: bool, -} - #[derive(Clone, Serialize, Deserialize)] pub(crate) struct HelloRequest { /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in diff --git a/src/transport/paths.rs b/src/transport/paths.rs index aca0684..844a897 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -209,7 +209,7 @@ impl Table { /// `usable`, with each one's place in the configured order, best first (as /// `Paths::rank` says). With `sending`, for a message about to go: they count as /// used (`ACTIVE_FOR`), and the one in use may change for a faster one. Without, a - /// look at the choice as it stands, changing nothing (`/peers`, figures, LINKS). + /// look at the choice as it stands, changing nothing (`/peers`, figures, the dashboard's links). fn rank( &mut self, mut usable: Vec<(String, usize)>, diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 8220041..13ca9fa 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -51,8 +51,6 @@ pub(crate) struct ByKind { pub(crate) identity: Counter, /// PING and PONG (`transport::paths`), counted by the paths layer under the address. pub(crate) paths: Counter, - /// LINKS and LINKS_REPLY (`swarm::links`). - pub(crate) links: Counter, /// PROBE and PROBE_REPLY (`swarm::working_links`). pub(crate) probe: Counter, /// PRUNE and GRAFT (`swarm::plumtree`). @@ -72,7 +70,6 @@ impl ByKind { MessageKind::Blob, MessageKind::Who, MessageKind::Ping, - MessageKind::Links, MessageKind::Probe, MessageKind::Prune, MessageKind::Neighbour, @@ -92,7 +89,6 @@ impl ByKind { MessageKind::Blob => &self.blob, MessageKind::Who | MessageKind::Me => &self.identity, MessageKind::Ping | MessageKind::Pong => &self.paths, - MessageKind::Links | MessageKind::LinksReply => &self.links, MessageKind::Probe | MessageKind::ProbeReply => &self.probe, MessageKind::Prune | MessageKind::Graft => &self.plumtree, MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => { @@ -111,7 +107,6 @@ impl ByKind { MessageKind::Blob => &mut self.blob, MessageKind::Who | MessageKind::Me => &mut self.identity, MessageKind::Ping | MessageKind::Pong => &mut self.paths, - MessageKind::Links | MessageKind::LinksReply => &mut self.links, MessageKind::Probe | MessageKind::ProbeReply => &mut self.probe, MessageKind::Prune | MessageKind::Graft => &mut self.plumtree, MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => { @@ -199,7 +194,7 @@ pub(crate) struct Logged { /// What ties `message`, with `peer`, to other messages: a PING and its PONG (the /// nonce alone: a PING may go to an address and its PONG come back up from the node -/// there, `paths`), a LINKS and its replies (the nonce), a HELLO and its replies (the run of the +/// there, `paths`), a HELLO and its replies (the run of the /// node that sent the HELLO, which the reply echoes), a blob's requests and answers, from /// any peer (its ID), and a delta offered in a SYNC with the blob it is fetched as. None /// for the rest (a STATUS and its reply carry nothing of the kind). @@ -209,8 +204,6 @@ fn related(peer: &str, message: &ParsedMessage) -> Option<String> { ParsedMessage::Ping(nonce) | ParsedMessage::Pong { nonce, .. } => { format!("ping {nonce}") } - ParsedMessage::Links(request) => format!("links {peer} {}", request.nonce), - ParsedMessage::LinksReply(reply) => format!("links {peer} {}", reply.nonce), ParsedMessage::Probe(request) => format!("probe {peer} {}", request.nonce), ParsedMessage::ProbeReply(reply) => format!("probe {peer} {}", reply.nonce), ParsedMessage::Hello(hello) => format!("hello {peer} {}", Hex(&hello.run_id)), @@ -321,8 +314,6 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) { MessageKind::Me => ("ME", "identity"), MessageKind::Ping => ("PING", "paths"), MessageKind::Pong => ("PONG", "paths"), - MessageKind::Links => ("LINKS", "links"), - MessageKind::LinksReply => ("LINKS_REPLY", "links"), MessageKind::Probe => ("PROBE", "probe"), MessageKind::ProbeReply => ("PROBE_REPLY", "probe"), MessageKind::Prune => ("PRUNE", "plumtree"), diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 0733d9c..8ead9b0 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 = 38; +pub(crate) const VERSION: u8 = 39; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -29,8 +29,6 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::Me => 7, MessageKind::Ping => 8, MessageKind::Pong => 9, - MessageKind::Links => 10, - MessageKind::LinksReply => 11, MessageKind::Probe => 12, MessageKind::ProbeReply => 13, MessageKind::Prune => 14, @@ -90,8 +88,6 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> { } ParsedMessage::Ping(nonce) => rmp_serde::encode::write(&mut payload, nonce)?, ParsedMessage::Pong { nonce, at } => rmp_serde::encode::write(&mut payload, &(nonce, at))?, - ParsedMessage::Links(request) => rmp_serde::encode::write(&mut payload, request)?, - ParsedMessage::LinksReply(reply) => rmp_serde::encode::write(&mut payload, reply)?, ParsedMessage::Probe(request) => rmp_serde::encode::write(&mut payload, request)?, ParsedMessage::ProbeReply(reply) => rmp_serde::encode::write(&mut payload, reply)?, ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => { @@ -166,10 +162,6 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> { } else if is(MessageKind::Pong) { let (nonce, at) = rmp_serde::from_slice(body)?; ParsedMessage::Pong { nonce, at } - } else if is(MessageKind::Links) { - ParsedMessage::Links(rmp_serde::from_slice(body)?) - } else if is(MessageKind::LinksReply) { - ParsedMessage::LinksReply(rmp_serde::from_slice(body)?) } else if is(MessageKind::Probe) { ParsedMessage::Probe(rmp_serde::from_slice(body)?) } else if is(MessageKind::ProbeReply) { @@ -504,43 +496,6 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { nonce: u64::MAX, at: crate::clock::UnixMillis(5), }, - ParsedMessage::Links(crate::transport::LinksRequest { - epoch, - epoch_seq: 1, - nonce: 3, - subscribe_secs: 60, - of: vec!["q".into()], - batch_ms: 250, - hops: 2, - }), - ParsedMessage::LinksReply(crate::transport::LinksReply { - epoch, - epoch_seq: 1, - nonce: 3, - links: vec![crate::transport::Link { - peer: "p".into(), - transport: Some("libp2p".into()), - relayed: true, - down: false, - rtt_us: Some(1500), - dormant: false, - others: vec![crate::transport::OtherPath { - transport: "mycelium".into(), - rtt_us: None, - relayed: false, - }], - clock: Some(crate::transport::paths::ClockOffset { - offset_ms: -2_500, - within_ms: 40, - }), - }], - relayed: vec![crate::transport::RelayedLinks { - member: "q".into(), - age_ms: 1200, - epoch, - links: Vec::new(), - }], - }), ParsedMessage::Probe(crate::transport::ProbeRequest { epoch, epoch_seq: 1, @@ -610,7 +565,7 @@ mod tests { #[test] fn every_kind_round_trips() { let samples = samples(); - assert_eq!(samples.len(), 19, "a kind is missing from `samples`"); + assert_eq!(samples.len(), 17, "a kind is missing from `samples`"); for message in samples { let bytes = encode(&message).unwrap(); let decoded = decode(&bytes).unwrap(); |
