diff options
| -rw-r--r-- | docs/api.md | 10 | ||||
| -rw-r--r-- | docs/architecture.md | 5 | ||||
| -rw-r--r-- | docs/dashboard.md | 99 | ||||
| -rw-r--r-- | docs/development.md | 5 | ||||
| -rw-r--r-- | docs/example.toml | 33 | ||||
| -rw-r--r-- | docs/measurements.md | 6 | ||||
| -rw-r--r-- | docs/openapi.yaml | 11 | ||||
| -rw-r--r-- | docs/operations.md | 68 | ||||
| -rw-r--r-- | docs/peers.md | 5 | ||||
| -rw-r--r-- | docs/trust.md | 7 | ||||
| -rw-r--r-- | src/api/auth.rs | 200 | ||||
| -rw-r--r-- | src/api/dashboard/dashboard.html | 497 | ||||
| -rw-r--r-- | src/api/dashboard/mod.rs | 42 | ||||
| -rw-r--r-- | src/api/listen.rs | 220 | ||||
| -rw-r--r-- | src/api/mod.rs | 83 | ||||
| -rw-r--r-- | src/api/peers.rs | 4 | ||||
| -rw-r--r-- | src/api/store.rs | 27 | ||||
| -rw-r--r-- | src/config/mod.rs | 97 | ||||
| -rw-r--r-- | src/main.rs | 46 | ||||
| -rw-r--r-- | src/node.rs | 6 | ||||
| -rw-r--r-- | src/simtest/links.rs | 105 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 78 | ||||
| -rw-r--r-- | src/simtest/relay.rs | 66 | ||||
| -rw-r--r-- | src/state/mod.rs | 6 | ||||
| -rw-r--r-- | src/swarm/links.rs | 585 | ||||
| -rw-r--r-- | src/swarm/members.rs | 65 | ||||
| -rw-r--r-- | src/swarm/peer/worker.rs | 10 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 56 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 59 | ||||
| -rw-r--r-- | src/transport/mod.rs | 23 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 52 | ||||
| -rw-r--r-- | src/transport/wire.rs | 10 |
32 files changed, 2364 insertions, 222 deletions
diff --git a/docs/api.md b/docs/api.md index 5b4dac4..cdc722e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -31,8 +31,9 @@ curl -u alice:secret -X POST 'localhost:3000/store/stats/visits?increment=1' # * Bodies are JSON (`application/json`, the default), text (`text/plain`, stored as a string), or a value of a type of its own (Types); answers are JSON, or as `Accept` asks. Numbers in JSON are stored as integers where they fit, as floats otherwise. -* Every request needs a user from the node's `--htpasswd` (basic authentication). Users - are all alike: each can read and write everything but protocol data (below). +* Every request needs a user from the node's `--htpasswd` (basic authentication), or + named by a reverse proxy in front that authenticated it (`--user-header`; + [`operations.md`](operations.md), Behind a reverse proxy). Users are all alike: each can read and write everything but protocol data (below). * A key with `/` in it, or an empty key, can be written inside a value at its parent, but not named in a path: read it through the parent. @@ -193,6 +194,9 @@ node or by sync: {"key":"settings/theme","deleted":true} ``` +While nothing changes, a blank line every 30 s (`--stream-heartbeat`) keeps proxies in +front from taking the stream for dead; skip blank lines. + An event carries the value as a read would return it then. Changes in quick succession may come as one event with the latest value: a stream tells the value, not every write. A change that leaves the value as it was sends nothing. The connection stays open; after @@ -222,7 +226,7 @@ Load balancers can send requests to any ready node: `/ready` is the check. Every change records the node that committed it (as its Automerge author) and, for a write through the API, the user who wrote it (in its message, as a trailer like Git's `Signed-off-by:`: `User: alice`), with when (by the writer's clock). User names are -each node's own (`--htpasswd`), so who wrote it is that user at that node: `alice` on +each node's own (`--htpasswd`, or its proxy's), so who wrote it is that user at that node: `alice` on two nodes is one person only where their operators make it so, which nothing checks. `GET /admin/changes` lists them, newest first, with the paths each touched; `GET /admin/changes/settings` those that diff --git a/docs/architecture.md b/docs/architecture.md index e0e2c94..9b9db71 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -225,14 +225,15 @@ catch-up source. The phases, what each serves and how the files come in: [`lifec the write lock, stamped with the clock) → `publish_change` (bumps the version, emits `DocumentChanged`) → the sync worker generates a message per neighbour → each neighbour applies it, relays its new heads to *its* neighbours, which pull what they -lack. In a full mesh (up to `k + 1` members) the writer reaches everyone directly. +lack. In a full mesh (up to `k + 1` members) the writer reaches everyone directly, but +those on none of its transports, to whom the nodes between announce. **An inbound message.** Consumer → `handle_message`: drop our own and stale-run replies, admit (swarm key), mark the peer seen, record its epoch (ignoring late ones), 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 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 (`swarm::links`, for the dashboard); +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 b7e3069..25bcfe8 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -14,10 +14,10 @@ cargo build --release --features mycelium,dashboard ``` It is at `/admin/dashboard` on the node's HTTP API, by default -`http://127.0.0.1:3000/admin/dashboard`, and asks for a user from `--htpasswd`, as the -rest of the API does. The API listens on localhost unless `--bind-addr` says otherwise; +`http://127.0.0.1:3000/admin/dashboard`, and asks for a user from `--htpasswd` (or one +a reverse proxy names, `--user-header`), as the rest of the API does. The API listens on localhost unless `--bind-addr` says otherwise; to look at a node on another machine, forward the port (`ssh -L 3000:localhost:3000 -<machine>`) or put a TLS proxy in front ([`trust.md`](trust.md)). +<machine>`) or put a TLS proxy in front ([`operations.md`](operations.md), Behind a reverse proxy). Every node has its own dashboard, and shows the swarm as that node sees it: what it knows of the others, what they report to it, and its own figures. @@ -100,16 +100,19 @@ else the fresher. Then: * **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`). -Messages this node exchanges run along its links as dots, from sender to receiver. +Messages this node exchanges run along its links as dots, from sender to receiver, all +at one speed: over a longer link they arrive later. They come from the per-peer message counts, which arrive every 2 s: what was sent and received since is spread over the next 2 s, at most 6 dots per link and direction, shared out by message kind, each kind in its colour (named in the legend; the console colours them alike). A dot is as large as its kind's messages were on average -(by the size's logarithm: a ping small, a large sync message large); the slider right -after their legend makes all dots larger or smaller, kept in the browser. A picture +(by the size's logarithm: a ping small, a large sync message large); the `size` slider +right after their legend makes all dots larger or smaller, and the `speed` slider after +it makes them run faster or slower (0.1 to 0.9 map units per millisecond, 0.3 at +first: a link of 200 in about 0.7 s); both are kept in the browser. A picture of the traffic, not a record of it; links between other members show none. -`messages: on` after the slider hides them (`messages: off` shows them again; the -legend and the slider greyed meanwhile); the choice is kept in the browser. +`messages: on` after the sliders hides them (`messages: off` shows them again; the +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). @@ -120,8 +123,10 @@ every 30 s, which costs the swarm fewer messages and leaves the map up to 30 s b (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 legend above the map names the transports that appear; `hide legend` beside the -box's title gives its room to the map (`show legend` brings it back). +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 +under them, and the map is dragged through it as anywhere else. `hide legend` beside +the box's title hides it (`show legend` brings it back); the map stays as it is. `layout: static`, beside it, is the default: nodes move only when the map is laid out (below). A click switches to `layout: dynamic`: each measured link is as @@ -146,10 +151,23 @@ measured. * **Drag a node**: move it. On a dynamic map, the others follow at once, and once let go of, it moves back among them; on a static map it stays where it is put until the map is laid out again. -* **Double-click a node**: pin it where it is (a dot in its middle): the layout leaves - it there. A double-click whose second press drags the node pins it where it's - dropped. Double-clicked again, it's let go, and moves to its place among the others - (on a static map, only it moves). +* **Double-click a node**: open its dashboard, in this page, where the node publishes + one (`--dashboard-url`, in the member list; e.g. behind a reverse proxy, + [`operations.md`](operations.md)). The other dashboard asks for its own credentials + where its proxy or node does. It opens as this one was left: the map's nodes where + 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 + at once. Links are drawn as the other node sees them. Where it publishes none, its + halo flashes red and "no dashboard" rises from it, fading out. +* **Middle-click a node**: pin it where it is (a dot in its middle): the layout leaves + it there. Dragged with the middle button, it is pinned where it's dropped. + Middle-clicked again, it's let go, and moves to its place among the others (on a + static map, only it moves). The map is laid out when the page opens, and again when the window or the box is resized. A member that comes later is placed among the others, which stay where they @@ -224,7 +242,8 @@ The messages this node exchanged with its peers, one by one: when (on the 24-hou 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 -newest at the bottom; scrolled along while at the bottom. They arrive with every update +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 PINGs with, are listed too: under the node at the address they went to, where it is known, else under the address. Not listed: WHO and ME, which stay in the layers @@ -242,6 +261,29 @@ requests and answers, from whichever peers (by the blob), and a delta offered in with the blob it is fetched as. A STATUS and its reply carry nothing that ties them, nor does an ordinary SYNC. +The map marks the message's two nodes and the link between them in the grey its line +has in the console, each in its own shape: a circle around each node, and the link's +lines as they bow. The message shown and the one pointed at are both marked, each in +its line's grey; where they mark the same node or link, the shown one's wins. + +The other way round, the messages exchanged with a node, or over one of this node's +links, are tinted in the colour of its halo on the map: lightly while it is pointed at +there, stronger while it is in focus (clicked there or in the peers); where both tint a +message, in focus wins. + +`filter selected`, beside the console's title, shows only the messages with the node or +over the link in focus (clicked on the map or in the peers; pointing at one doesn't +count), and all of them while nothing is. Where there are none to show (a link between +two other nodes, a node this one exchanges nothing with), it says so. Whenever what it +shows changes (turned on or off, another focus, or none), the console goes to the newest +message and follows from there. The choice is kept in the browser. + +At the right of the console's title, `msg/s` gives the messages per second sent and received, as load averages +do: over the last 1, 5 and 15 minutes (exponentially weighted), since the page was +opened, from the message counts each update brings; below it, the bytes per second +alike, in `B/s`, `KiB/s` or `MiB/s`, whichever the largest of the three needs. With the filter on, they are those +with the node or over the link in focus; `–` where there are none. + ## While you read Updates don't pull things from under you: while text is selected in a box, that box @@ -254,7 +296,9 @@ are done. The map isn't redrawn under the pointer: what you point at stays put. 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. + (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). * **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 @@ -290,6 +334,25 @@ older than 90 s is left out. A member keeps 64 subscribers at most. With `links: 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) 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. + The links a member reports: its neighbours and the peers it holds dormant (a neighbour that goes dormant is one no more, but stays listed until it is heard from again), by which transport, whether only through a relay, the round trip (as the paths layer @@ -297,7 +360,9 @@ measures it), whether it holds the peer dormant or the path is down, the peer's 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 -every build answers, with or without the feature (`swarm::links`). +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`). **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 diff --git a/docs/development.md b/docs/development.md index 93726b1..d7f6068 100644 --- a/docs/development.md +++ b/docs/development.md @@ -50,3 +50,8 @@ take as long as the protocol's timeouts make them, however fast the machine, and busy machine some fail now and then (a node's timer fires before the work it waits for is done). One failing alone (`cargo sim <name>`) and passing then is that, not a regression; one failing again and again is a regression. + +A node on several transports is a node on several simulated nets, one per scheme +(`Sim::start_on`, under the identity layer): nodes on different nets can't reach each +other, as one on Tor alone and one on libp2p alone can't, so a swarm whose transports +don't all meet is simulated with the simulated transport alone. diff --git a/docs/example.toml b/docs/example.toml index 95b6f52..00bfed3 100644 --- a/docs/example.toml +++ b/docs/example.toml @@ -37,6 +37,12 @@ node-key-file = "/var/lib/continuum/node.key" # member list; not unique, never an identity. # alias = "node-${env:HOSTNAME}" +# Where people reach this node's dashboard (http:// or https://, e.g. behind a reverse +# proxy). Published in the member list: the other nodes' dashboards open it on a +# double-click on this node. A build without the dashboard feature warns and publishes +# none (removing one published before). +# dashboard-url = "https://node-1.example.com/admin/dashboard" + # Only relay (libp2p circuit relay), without being a node: no document, no swarm, no # node key. Needs libp2p alone (libp2p's enable, no other), its listen and # key-file; options about a swarm or a document (node-key-file, seeds, genesis, @@ -47,14 +53,33 @@ node-key-file = "/var/lib/continuum/node.key" # --- The HTTP API ------------------------------------------------------------------------ -# Address of the HTTP API. Plain HTTP: to expose it, put a TLS proxy in front. -# bind-addr = "127.0.0.1:3000" +# Where the HTTP API listens: TCP addresses, and Unix sockets (unix:PATH, made readable +# and writable by the node's user and group only), as many as given. Plain HTTP: to +# expose it, put a TLS proxy in front. +# bind-addr = ["127.0.0.1:3000", "unix:/run/continuum/api.sock"] # Users of the HTTP API (an htpasswd file with Argon2 hashes). Required (but with -# relay-only: without it, no HTTP API). +# user-header, or relay-only: without either, no HTTP API). htpasswd = "/etc/continuum/htpasswd" -# HTTP API users (from htpasswd) who may write protocol data under /store/_continuum +# A request header naming the user, set by a reverse proxy in front of the HTTP API that +# authenticated the request itself (e.g. X-Remote-User, from nginx's $remote_user). +# Believed only from trusted-proxies; other requests need htpasswd's basic +# authentication. The proxy must set or clear it on every request. +# user-header = "X-Remote-User" + +# How often GET /stream sends a blank line while nothing changes, so proxies in front +# keep it open: below their read timeout (nginx's proxy_read_timeout, 60 s by default). +# 0: never. +# stream-heartbeat = "30s" + +# Where the reverse proxies whose user-header is believed connect from: IP addresses (or +# ranges, as ADDRESS/BITS), and unix for the API's Unix sockets (bind-addr). On a host +# with other users, leave out the addresses and have the proxy use a Unix socket: others +# can connect over loopback too. +# trusted-proxies = ["127.0.0.1", "::1", "unix"] + +# HTTP API users (from htpasswd or user-header) who may write protocol data under /store/_continuum # (remake records, members, forget records), e.g. to repair a stuck remake. Such writes # act on the whole swarm and are logged. Epoch metadata and the swarm ID are never # writable. diff --git a/docs/measurements.md b/docs/measurements.md index 2adc0f5..fb01de0 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -168,9 +168,9 @@ 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, counted under the address they went to or came from, so only in the totals, and -`paths` for PING and PONG: per peer, but the paths layer's own PINGs under the address, -so only in the totals). Payload bytes are what a transport's message carries (the wire +ME, and `paths` for PING and PONG). 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 +(`/peers`) has those of its addresses too, where it is known to be at them. Payload bytes are what a transport's message carries (the wire encoding): not the transport's framing (for Mycelium, its topic and the base64 of its HTTP API). `refused` is what the transports wouldn't accept to send (no path to the peer), not counted as sent. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 78912fa..701cdf9 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -7,7 +7,8 @@ info: as JSON by path, that syncs over Mycelium, libp2p or Tor. Every endpoint requires HTTP basic authentication (users from `--htpasswd`, - Argon2 hashes), except `/health`, `/ready`, the Swagger UI (`/swagger-ui`) and + Argon2 hashes), or a user named by a trusted reverse proxy in front + (`--user-header`), except `/health`, `/ready`, the Swagger UI (`/swagger-ui`) and this description (`/openapi.yaml`). `/debug/*` exists only with `--debug-api`. @@ -328,7 +329,8 @@ paths: description: | Newline-delimited JSON: one event with the current value, then one whenever the resolved value changes (by a local write or by sync). The connection stays - open. + open; while nothing changes, a blank line every `--stream-heartbeat` (30 s) + keeps proxies from closing it: skip blank lines. responses: '200': description: Event stream. @@ -747,8 +749,9 @@ paths: description: | 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: - `alias`, `self`), `intended` (by node ID: the neighbours the member list gives + `/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`, + `dashboard`: its `--dashboard-url`), `intended` (by node ID: the neighbours the member list gives it, as this node computes them), `measured` (by node ID: `secsAgo` and `links`, each with `peer`, `transport`, `relayed`, `down`, `rttMs`, `dormant` and `others` (the peer's other transports: `transport`, `rttMs`, `relayed`), as the member diff --git a/docs/operations.md b/docs/operations.md index fe09537..fae3b6c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -103,6 +103,74 @@ One node at a time, restarting each with its files. What to know: switching ([`remakes.md`](remakes.md), Limits). Don't remake while the swarm runs two versions. * **The dashboard** reloads by itself once its node serves another version of it. +## Behind a reverse proxy + +The HTTP API speaks plain HTTP on `127.0.0.1:3000`; a reverse proxy in front adds TLS, +and can authenticate users itself (single sign-on, client certificates, its own +password file). It then names the user in a header, which the node believes from where +the proxy connects alone (`--user-header`, `--trusted-proxies`); requests from +elsewhere still need `--htpasswd`, which becomes optional. The proxy must set or clear +that header on every request, or a client could set it itself. + +Where the proxy connects decides who can pose as it. Every user of a host connects +from its loopback addresses, so trusting those (as `--trusted-proxies` does by +default) trusts everyone on the host. A Unix socket is better: `--bind-addr` takes +`unix:PATH` too, as many places as given, and makes the socket readable and writable +by the node's user and group only, so only members of that group (the proxy's user, +added to it) can connect at all. Trust it alone with `--trusted-proxies unix`, and keep +a TCP address for the rest (a command-line tool, with `--htpasswd`): + +```sh +continuum --bind-addr 127.0.0.1:3000,unix:/run/continuum/api.sock \ + --user-header X-Remote-User --trusted-proxies unix --htpasswd htpasswd ... +``` + +```nginx +location / { + auth_basic "continuum"; + auth_basic_user_file /etc/nginx/continuum.htpasswd; + proxy_set_header X-Remote-User $remote_user; + proxy_pass http://unix:/run/continuum/api.sock; +} +``` + +The directory of the socket must exist and let the proxy's user through (with systemd, +`RuntimeDirectory=continuum` and `RuntimeDirectoryMode=0750`, the proxy's user in the +node's group). A socket left behind by an earlier run is replaced, and the node removes +its own when it stops. With a proxy on another host, trust its address instead, and +have the node listen where only that host reaches it. + +Where a socket won't do, the host's firewall can keep everyone but the proxy's user +off the API's port, as Linux filters local connections by the user that opens them. +With nftables, and nginx running as `www-data`: + +``` +table inet continuum { + chain output { + type filter hook output priority 0; + oifname "lo" tcp dport 3000 meta skuid != "www-data" reject + } +} +``` + +(with iptables: `-A OUTPUT -o lo -p tcp --dport 3000 -m owner ! --uid-owner www-data -j +REJECT`). The loopback addresses can then stay trusted; root can still get past it, as it +can past a socket's permissions. Every address of the API on loopback is trusted alike, +so a command-line tool of another user goes through the proxy then. + +The names the proxy gives are the users of `--protocol-writers` and of who wrote what +([`api.md`](api.md)). + +What streams (`/stream`, the dashboard's events) says `X-Accel-Buffering: no`, so nginx +passes it on as it comes, and says it's alive while nothing changes: the dashboard's +events every 15 s, `/stream` every 30 s (a blank line, `--stream-heartbeat`), within +nginx's `proxy_read_timeout` (60 s). With a proxy that gives up sooner, lower +`--stream-heartbeat` below its timeout. + +The API may also be served under a path of its own (`location /continuum/ { +proxy_pass http://unix:/run/continuum/api.sock:/; }`): its redirects (`Location`), the dashboard's +requests and the Swagger UI's are relative, so they stay under it. + ## What to alert on From the metrics (`GET /admin/stats` as Prometheus, [`measurements.md`](measurements.md)): diff --git a/docs/peers.md b/docs/peers.md index 95b4dab..c63cec1 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -284,7 +284,10 @@ the current pull is answered, if they still have something it lacks. The regular STATUS (every third keepalive) triggers the same pull, so gaps close within a few keepalive intervals. A node that just caught up announces its heads once. In a full mesh nothing is announced: the -writer reaches everyone (while a node doesn't know its neighbours yet, it announces). +writer reaches everyone (while a node doesn't know its neighbours yet, it announces); +but to the neighbours the writer can't reach, those on none of its transports (by the +member list, the writer's neighbours as this node computes them): a node on Tor and +libp2p announces to one on Tor alone what came from one on libp2p alone. ## Keepalive and self-healing diff --git a/docs/trust.md b/docs/trust.md index 521a175..165fc9b 100644 --- a/docs/trust.md +++ b/docs/trust.md @@ -34,7 +34,12 @@ secret like the swarm key: readable by its owner only (a node warns otherwise; a it creates is). **The HTTP API.** Every endpoint but the API description needs a user from -`--htpasswd`. It listens on `127.0.0.1:3000` by default (`--bind-addr`) and speaks +`--htpasswd`, or one a reverse proxy names (`--user-header`): the node believes that +header from `--trusted-proxies` alone, but from those blindly, so whoever can send +requests from there can be any user. With the default loopback addresses, that is every +user of the host; with only the API's Unix socket (`--trusted-proxies unix`), those its +file's permissions let in: the node's user and group ([`operations.md`](operations.md), Behind a +reverse proxy). It listens on `127.0.0.1:3000` by default (`--bind-addr`) and speaks plain HTTP: to expose it, put a TLS proxy in front. All users may read and write user data and use the admin endpoints (remakes, forgetting peers, exports, log filters). There are no read-only users: whoever may open the dashboard may also edit and delete diff --git a/src/api/auth.rs b/src/api/auth.rs index d58640e..400feed 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -1,10 +1,12 @@ use anyhow::Context; use argon2::{Argon2, password_hash::PasswordVerifier}; -use axum::extract::State; +use axum::extract::{ConnectInfo, State}; + +use super::listen::Peer; use axum::{ body::Body, http::{ - HeaderMap, Request, StatusCode, + HeaderMap, HeaderName, Request, StatusCode, header::{AUTHORIZATION, WWW_AUTHENTICATE}, }, response::{IntoResponse, Response}, @@ -16,6 +18,7 @@ use std::{ collections::HashMap, fs::File, io::{BufRead, BufReader}, + net::IpAddr, sync::Arc, time::{Duration, Instant}, }; @@ -49,26 +52,136 @@ fn load_users(path: &str) -> anyhow::Result<HashMap<String, String>> { Ok(users) } +/// Who may use the HTTP API: the users of an htpasswd file (basic authentication), +/// and the user a reverse proxy in front names (`--user-header`), if it is one of +/// `--trusted-proxies`. #[derive(Clone)] pub(crate) struct AuthState { idle_timeout: Duration, - users: Arc<HashMap<String, String>>, // user -> PHC hash string + /// None without an htpasswd file: then only the proxy's users. + users: Option<Arc<HashMap<String, String>>>, // user -> PHC hash string + proxy: Option<Arc<ProxyAuth>>, cache: Arc<DashMap<[u8; 32], Instant>>, /// Bounds concurrent Argon2 verifications (CPU and ~19 MiB each). verify_slots: Arc<Semaphore>, } -impl AuthState { - pub(crate) fn new(path: &str) -> anyhow::Result<Self> { - let users = load_users(path)?; +/// A reverse proxy authenticates, and names the user in `header`. +struct ProxyAuth { + header: HeaderName, + trusted: Vec<Trusted>, +} + +/// Where a proxy is believed from (`--trusted-proxies`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Trusted { + Addresses(Range), + /// The API's Unix sockets (`unix`). + Unix, +} - Ok(Self { +impl Trusted { + fn parse(text: &str) -> anyhow::Result<Self> { + if text.trim() == "unix" { + return Ok(Self::Unix); + } + Range::parse(text).map(Self::Addresses) + } + + fn holds(&self, peer: Peer) -> bool { + match (self, peer) { + (Self::Addresses(range), Peer::Tcp(address)) => range.contains(address.ip()), + (Self::Unix, Peer::Unix) => true, + _ => false, + } + } +} + +/// An address, or the addresses sharing its first `bits` bits. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Range { + address: IpAddr, + bits: u32, +} + +impl Range { + fn parse(text: &str) -> anyhow::Result<Self> { + let (address, bits) = match text.trim().split_once('/') { + Some((address, bits)) => (address, Some(bits)), + None => (text.trim(), None), + }; + let address: IpAddr = address.parse().with_context(|| { + format!("trusted-proxies: {text:?} is neither an IP address nor unix") + })?; + let all = if address.is_ipv4() { 32 } else { 128 }; + let bits = match bits { + None => all, + Some(bits) => bits + .parse() + .ok() + .filter(|bits| *bits <= all) + .with_context(|| format!("trusted-proxies: {text:?}: not 0 to {all} bits"))?, + }; + Ok(Self { address, bits }) + } + + fn contains(&self, address: IpAddr) -> bool { + // An IPv4 address as IPv6 (`::ffff:a.b.c.d`) is that IPv4 address. + let address = match address { + IpAddr::V6(v6) => v6.to_ipv4_mapped().map_or(address, IpAddr::V4), + v4 => v4, + }; + let leading = + |a: u128, b: u128, width: u32| self.bits == 0 || (a ^ b) >> (width - self.bits) == 0; + match (self.address, address) { + (IpAddr::V4(range), IpAddr::V4(address)) => { + leading(u32::from(range).into(), u32::from(address).into(), 32) + } + (IpAddr::V6(range), IpAddr::V6(address)) => { + leading(u128::from(range), u128::from(address), 128) + } + _ => false, + } + } +} + +impl AuthState { + /// From the options; None if there is neither an htpasswd file nor a user header + /// (no users: no HTTP API, for a relay). + pub(crate) fn from_settings( + settings: &crate::config::Settings, + ) -> anyhow::Result<Option<Self>> { + let users = settings + .htpasswd + .as_deref() + .map(load_users) + .transpose()? + .map(Arc::new); + let proxy = settings + .user_header + .as_deref() + .map(|header| { + let header = HeaderName::try_from(header.trim()) + .with_context(|| format!("user-header: {header:?} is no header name"))?; + let trusted = settings + .trusted_proxies + .iter() + .map(|trusted| Trusted::parse(trusted)) + .collect::<anyhow::Result<Vec<_>>>()?; + Ok::<_, anyhow::Error>(Arc::new(ProxyAuth { header, trusted })) + }) + .transpose()?; + if users.is_none() && proxy.is_none() { + return Ok(None); + } + Ok(Some(Self { idle_timeout: Duration::from_secs(300), cache: Arc::new(DashMap::new()), - users: Arc::new(users), + users, + proxy, verify_slots: Arc::new(Semaphore::new(2)), - }) + })) } } @@ -87,10 +200,25 @@ fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> { #[derive(Clone, Debug)] pub(crate) struct AuthenticatedUser(pub(crate) String); +/// The user the trusted proxy the request came from names, if it does. +fn proxied_user(state: &AuthState, headers: &HeaderMap, from: Option<Peer>) -> Option<String> { + let proxy = state.proxy.as_ref()?; + let user = headers.get(&proxy.header)?.to_str().ok()?.trim(); + if user.is_empty() { + return None; + } + let trusted = from.is_some_and(|from| proxy.trusted.iter().any(|trusted| trusted.holds(from))); + if !trusted { + tracing::debug!(from = ?from, header = %proxy.header, "User header from an address not in trusted-proxies; ignored"); + return None; + } + Some(user.to_owned()) +} + /// The user, if the credentials are valid. async fn authenticate(state: &AuthState, headers: &HeaderMap) -> Option<String> { let (user, pass) = parse_basic(headers)?; - let hash = state.users.get(&user).cloned()?; + let hash = state.users.as_ref()?.get(&user).cloned()?; verify(state, &user, pass, hash).await.then_some(user) } @@ -136,22 +264,60 @@ async fn verify(state: &AuthState, user: &str, pass: String, hash: String) -> bo verified } +/// The user of a request: as the trusted proxy it came from names it, else by basic +/// authentication; `401` if neither. pub(crate) async fn basic_auth( State(state): State<AuthState>, headers: HeaderMap, mut request: Request<Body>, next: axum::middleware::Next, ) -> Response { - let Some(user) = authenticate(&state, &headers).await else { + let from = request + .extensions() + .get::<ConnectInfo<Peer>>() + .map(|ConnectInfo(peer)| *peer); + let user = match proxied_user(&state, &headers, from) { + Some(user) => Some(user), + None => authenticate(&state, &headers).await, + }; + let Some(user) = user else { sleep(Duration::from_millis(1500)).await; - return ( - StatusCode::UNAUTHORIZED, - [(WWW_AUTHENTICATE, r#"Basic realm="Restricted""#)], - "Unauthorized", - ) - .into_response(); + // A challenge only where basic authentication can answer it. + return match state.users { + Some(_) => ( + StatusCode::UNAUTHORIZED, + [(WWW_AUTHENTICATE, r#"Basic realm="Restricted""#)], + "Unauthorized", + ) + .into_response(), + None => (StatusCode::UNAUTHORIZED, "Unauthorized").into_response(), + }; }; request.extensions_mut().insert(AuthenticatedUser(user)); next.run(request).await } + +#[cfg(test)] +mod tests { + use super::Range; + + #[test] + fn ranges_hold_their_addresses() { + let range = |text: &str| Range::parse(text).unwrap(); + let ip = |text: &str| text.parse().unwrap(); + assert!(range("127.0.0.1").contains(ip("127.0.0.1"))); + assert!(!range("127.0.0.1").contains(ip("127.0.0.2"))); + assert!(range("10.0.0.0/8").contains(ip("10.1.2.3"))); + assert!(!range("10.0.0.0/8").contains(ip("11.0.0.1"))); + assert!(range("::1").contains(ip("::1"))); + assert!(range("fd00::/8").contains(ip("fd12::1"))); + assert!(range("0.0.0.0/0").contains(ip("192.0.2.1"))); + // IPv4 as IPv6, as a dual-stack listener sees it. + assert!(range("127.0.0.1").contains(ip("::ffff:127.0.0.1"))); + assert!(!range("127.0.0.1").contains(ip("::1"))); + for bad in ["localhost", "10.0.0.0/33", "::/129", "10.0.0.0/x"] { + assert!(Range::parse(bad).is_err(), "{bad}"); + } + } +} diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index ba2b893..25837c8 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -8,6 +8,12 @@ :root { --bg: #fafaf9; --fg: #1c1917; --muted: #78716c; --line: #e7e5e4; --panel: #ffffff; --ok: #15803d; --warn: #c2410c; --bad: #b91c1c; --intended: #a8a29e; --self: #1d4ed8; + /* How strong the grey under a console message is, pointed at or shown, and under + its nodes and link on the map. */ + --pointed: .12; --shown: .22; + /* How strong the halo of a node or link is, pointed at or in focus, and the tint of + the console messages exchanged with it. */ + --halo-hovered: .15; --halo-focused: .35; } @media (prefers-color-scheme: dark) { :root { @@ -68,6 +74,28 @@ section.collapsed { align-self: start; } section.collapsed h2 { margin: 0; } section.collapsed h2 .toggle { display: none; } + /* A choice beside a box's title (the console's filter). */ + section h2 label.choice { font-size: 12px; text-transform: none; letter-spacing: 0; + margin-left: 1em; cursor: pointer; } + section h2 label.choice input { vertical-align: -2px; margin: 0 3px 0 0; } + section.collapsed h2 label.choice { display: none; } + /* The console's rates, at the right of its title: messages and bytes per second, a + row each, small enough for both to fit the title's line, centred on it. Columns as + wide as their widest content will be, so the numbers stay put as they change. */ + section h2 .rates { float: right; display: grid; + grid-template-columns: repeat(3, minmax(5ch, auto)) minmax(5.5ch, auto); + column-gap: 5px; height: 18px; align-content: center; + font-size: 9px; line-height: 9px; text-transform: none; letter-spacing: 0; + margin-left: 1em; font-variant-numeric: tabular-nums; } + section h2 .rates span { text-align: right; } + section h2 .rates span.unit { text-align: left; } + section.collapsed h2 .rates { display: none; } + /* After the console's filter: that it follows new messages (scrolled to the bottom). */ + section h2 .following { font-size: 12px; text-transform: none; letter-spacing: 0; + margin-left: 1em; font-style: italic; } + section.collapsed h2 .following { display: none; } + /* A button within a title's line, where it comes (not at its end, as toggles are). */ + section h2 button.inline { float: none; margin-left: 1em; line-height: 1em; } /* The right column dragged off to the side: the map takes the width. */ @media (min-width: 901px) { main.right-hidden > section:not(#map) { display: none; } @@ -86,6 +114,9 @@ #map { grid-area: map; } #peers-box { grid-area: peers; } #figures-box { grid-area: figures; } + /* Its title on one line: what doesn't fit ends in an ellipsis (the whole of it its + tooltip). */ + #figures-box h2 { white-space: nowrap; overflow: hidden; text-overflow: ellipsis; } #document-box { grid-area: document; } #console-box { grid-area: console; } #console-box .scroll { flex: 1; } @@ -95,9 +126,17 @@ .console > div { cursor: pointer; border-left: 3px solid transparent; padding-left: 4px; } /* Of one exchange with the message pointed at or shown (`related`): a lighter grey, and the bar. */ + /* Exchanged with the node or over the link pointed at on the map, or in focus: the + colour of its halo; in focus wins. */ + .console > div.peer-hovered { + background: color-mix(in srgb, var(--self) calc(var(--halo-hovered) * 100%), transparent); } + .console > div.peer-focused { + background: color-mix(in srgb, var(--self) calc(var(--halo-focused) * 100%), transparent); } + /* Filtered (`filter selected`): only the messages with what is in focus. */ + .console.filtering > div:not(.peer-focused) { display: none; } .console > div.related { background: rgba(127, 127, 127, .08); border-left-color: var(--self); } - .console > div:hover { background: var(--hover, rgba(127, 127, 127, .12)); } - .console > div.focused { background: rgba(127, 127, 127, .22); } + .console > div:hover { background: rgb(127 127 127 / var(--pointed)); } + .console > div.focused { background: rgb(127 127 127 / var(--shown)); } /* The message takes the box's width and scrolls with it, not in a box of its own. */ #stats pre.message { margin: 4px 0 0; font-size: 12px; white-space: pre-wrap; max-height: none; overflow: visible; } @@ -126,14 +165,23 @@ /* A kind of message: a dot in its colour, as on the map. */ .legend .message::before { width: 8px; height: 8px; border: none; border-radius: 50%; background: var(--colour); } - /* The dots' size, right after their legend, no taller than its text. */ - .legend .dot-size { display: inline-flex; align-items: center; gap: 4px; } - .legend .dot-size::before { display: none; } - .legend .dot-size input { width: 6em; height: 1em; margin: 0; } - /* After the slider, as tall; the kinds and the slider greyed while the dots are off. */ + /* The dots' size and speed, right after their legend, no taller than its text. */ + .legend .slider { display: inline-flex; align-items: center; gap: 4px; } + .legend .slider::before { display: none; } + .legend .slider input { width: 6em; height: 1em; margin: 0; } + /* 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 .dot-size { opacity: .45; } - #legends { flex: none; } + .legend.off .message, .legend.off .slider { opacity: .45; } + /* The map and, over its top, the legend: shown or hidden, it takes no room from the + map, which stays as it is. What is under it shows through; its contents are on an + underlay of their own shape (as the map's labels), and only its controls take the + pointer: the map under the rest is dragged as anywhere else. */ + .map-area { position: relative; flex: 1; min-height: 0; display: flex; } + #legends { position: absolute; top: 0; left: 0; right: 0; pointer-events: none; } + #legends .legend > * { filter: drop-shadow(2px 0 0 var(--panel)) + drop-shadow(-2px 0 0 var(--panel)) drop-shadow(0 2px 0 var(--panel)) + drop-shadow(0 -2px 0 var(--panel)); } + #legends button, #legends input, #legends label { pointer-events: auto; } /* Hides the legend, and shows it again. */ button.toggle { font: inherit; font-size: 12px; text-transform: none; letter-spacing: 0; float: right; margin-left: 6px; padding: 0 6px; color: var(--muted); background: none; @@ -205,6 +253,7 @@ <h2>Swarm <button id="legend-toggle" class="toggle">hide legend</button> <button id="rtt-toggle" class="toggle">layout: static</button> <button id="links-toggle" class="toggle">links: live</button></h2> + <div class="map-area"> <div id="legends"> <div class="legend"> <span class="node self">this node</span> @@ -219,10 +268,12 @@ class="toggle">messages: on</button></div> <div class="legend"> <span class="hint">drag to pan, wheel or pinch to zoom, turn with two fingers; - drag a node to move it, double-click it to pin it there or let it go</span> + drag a node to move it, middle-click it to pin it there or let it go, + double-click it for its dashboard</span> </div> </div> - <svg id="graph"><g id="view"><g id="flows" pointer-events="none"></g></g></svg> + <svg id="graph"><g id="view"><g id="flows" pointer-events="none"></g><g id="flashes" pointer-events="none"></g></g></svg> + </div> </section> <section id="peers-box"> <h2>Peers</h2> @@ -239,8 +290,9 @@ <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</h2> - <div class="scroll" id="console-scroll"><div id="console" class="console"></div></div> + <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> </main> <script> @@ -252,6 +304,31 @@ const PAGE_VERSION = "__PAGE_VERSION__"; let state = null; let stateAt = 0; +// --- Hand-off: a double click on a node opens its dashboard, on another host, as this +// page was left (`handOff`): the map where it was, the view, the pins, the focus and the +// settings this browser keeps for the page (another host's are its own). It comes in +// the URL's fragment, which no request carries: taken here, before anything reads the +// settings, and gone from the URL at once. --- +const HAND_OFF = "#continuum="; +const handOver = (() => { + if (!location.hash.startsWith(HAND_OFF)) return null; + try { + const handed = JSON.parse(decodeURIComponent(location.hash.slice(HAND_OFF.length))); + for (const [key, value] of Object.entries(handed.settings || {})) { + if (key.startsWith("continuum-") && typeof value === "string") { + try { localStorage.setItem(key, value); } catch (_) {} + } + } + return handed; + } catch (_) { + return null; + } finally { + try { + history.replaceState(null, "", location.pathname + location.search); + } catch (_) { /* not where history works */ } + } +})(); + // Fields that grow by themselves: the node's list (`GROWING` and `is_age` in its // dashboard), filled in when it serves the page. const GROWING = new Set(__GROWING__); @@ -309,7 +386,7 @@ function age(secs) { // What is in focus, its figures shown: a node's ID, or a link's pair key (`pairKey`, // two IDs). null: this node's (`/admin/stats`). -let focused = (() => { +let focused = handOver ? handOver.focused || null : (() => { try { return decodeURIComponent(location.hash.slice(1)) || null; } catch { return null; } })(); const pairKey = (a, b) => a < b ? a + " " + b : b + " " + a; @@ -327,6 +404,7 @@ function focus(id) { : "#" + encodeURIComponent(focused)); } catch { /* not where history works */ } render(); + markPeerMessages(); } // --- Peers --- @@ -370,6 +448,11 @@ function renderPeers() { // --- Figures: a tree, its open branches kept across renders --- const open = new Set(["stats", "peer", "links", "from", "to", "paths", "paths.destination"]); +// As the page it was opened from had them (`handOver`). +if (handOver && Array.isArray(handOver.figures)) { + open.clear(); + for (const path of handOver.figures) if (typeof path === "string") open.add(path); +} function tree(key, value, path) { // Empty: nothing to open. if (value !== null && typeof value === "object" && Object.keys(value).length === 0) { @@ -417,6 +500,7 @@ function renderStats() { const field = (key, value) => el("div", { class: "leaf" }, el("span", { class: "key" }, key + ": "), value); title.replaceChildren("Message: ", what); + title.title = title.textContent; document.getElementById("stats").replaceChildren( field("at", clockTime(new Date(message.at))), field(message.way === "received" ? "from" : "to", message.peer), @@ -455,6 +539,7 @@ function renderStats() { destination: peer.destination, addresses: peer.addresses }, "paths")); } title.replaceChildren("Link: ", what); + title.title = title.textContent; document.getElementById("stats").replaceChildren(...shown); return; } @@ -474,6 +559,7 @@ function renderStats() { if (report) shown.push(tree("links", report, "links")); } title.replaceChildren(focused === null ? "Figures: " : "Peer: ", what); + title.title = title.textContent; document.getElementById("stats").replaceChildren(...shown); } @@ -666,6 +752,7 @@ function hover(id) { if (hovered === id) return; hovered = id; if (state) renderGraph(); + markPeerMessages(); } function hoverable(element, id) { element.addEventListener("mouseenter", () => hover(id)); @@ -800,11 +887,35 @@ function renderGraph() { settle(); } const children = []; + // Under everything: the nodes and the link of the message shown in the console, and + // of the one pointed at, each in the grey its line has there (the shown one's where + // both mark the same), in its own shape (the lines of a link as they bow, a node's + // circle). A group per grey, so where its shapes overlap it is no darker. + const marking = new Map(); + for (const [message, how] of [[pointedMessage, "pointed"], [shownMessage, "shown"]]) { + if (message === null) continue; + for (const key of [state.node.id, message.peer, pairKey(state.node.id, message.peer)]) { + marking.set(key, how); + } + } + const marks = Object.fromEntries(["pointed", "shown"].map(how => [how, + drawing("marks " + how, "g", { "pointer-events": "none", + style: "opacity: var(--" + how + ")" })])); + const marked = { pointed: [], shown: [] }; + // A shape under `key` (a node's ID, a link's pair key), if it is marked. + const mark = (key, tag, attrs) => { + const how = marking.get(key); + if (how) marked[how].push(svg(tag, attrs)); + }; + const markLine = (key, d, width) => mark(key, "path", { d, fill: "none", + stroke: "rgb(127 127 127)", "stroke-width": width + 8, "stroke-linecap": "round" }); + children.push(marks.pointed, marks.shown); // Behind a link: a halo while it's hovered or in focus (`width`: of its lines). const halo = (key, p, q, width) => { if (key !== hovered && key !== focused) return; children.push(drawing("halo " + key, "line", { x1: p.x, y1: p.y, x2: q.x, y2: q.y, - stroke: "var(--self)", "stroke-opacity": key === focused ? .35 : .15, + stroke: "var(--self)", + style: "stroke-opacity: var(" + (key === focused ? "--halo-focused" : "--halo-hovered") + ")", "stroke-width": width + 8, "stroke-linecap": "round" })); }; // Over a link: an invisible, wider line to point at and click. @@ -825,6 +936,7 @@ function renderGraph() { if (measured.has(key)) continue; const p = positions.get(a), q = positions.get(b); halo(key, p, q, 1); + markLine(key, "M " + p.x + " " + p.y + " L " + q.x + " " + q.y, 1); children.push(drawing("intended " + key, "line", { x1: p.x, y1: p.y, x2: q.x, y2: q.y, stroke: "var(--intended)", "stroke-dasharray": "4 4", "stroke-width": 1 })); tooltip(hit(key, p, q, 1), name(a) + " – " + name(b) + ": intended (member list), not measured"); @@ -874,6 +986,7 @@ function renderGraph() { } else if (line.used ? link.relayed : line.relayed) { attrs["stroke-dasharray"] = "7 4"; } + markLine(key, attrs.d, attrs["stroke-width"]); const drawnLine = drawing("path " + key + " " + i, "path", attrs); const description = name(a) + " – " + name(b) + (line.transport ? " via " + line.transport : "") + (line.used ? " (in use)" : "") + @@ -916,7 +1029,8 @@ function renderGraph() { // A halo as links have: light under the pointer, stronger in focus. if (shown || id === hovered) { children.push(drawing("ring " + id, "circle", { cx: p.x, cy: p.y, r: 13, - fill: "var(--self)", "fill-opacity": shown ? .35 : .15, stroke: "none" })); + fill: "var(--self)", stroke: "none", + style: "fill-opacity: var(" + (shown ? "--halo-focused" : "--halo-hovered") + ")" })); } const pressed = element => { element.addEventListener("pointerdown", event => { event.stopPropagation(); drag(event, id); }); @@ -929,7 +1043,9 @@ function renderGraph() { } // Dormant as this node sees it (`/peers`): a red outline. const dormant = (state.peers.peers[id] || {}).status === "dormant"; - const circle = drawing("node " + id, "circle", { cx: p.x, cy: p.y, r: member.self ? 8 : 6, + const radius = member.self ? 8 : 6; + mark(id, "circle", { cx: p.x, cy: p.y, r: radius + 6, fill: "rgb(127 127 127)" }); + const circle = drawing("node " + id, "circle", { cx: p.x, cy: p.y, r: radius, fill: member.self ? "var(--self)" : reported ? "var(--fg)" : "var(--panel)", stroke: member.self ? "var(--self)" : dormant ? "var(--bad)" : "var(--fg)", "stroke-width": dormant ? 2.5 : 1.5 }, pressed); @@ -939,7 +1055,7 @@ function renderGraph() { (dormant ? " (dormant)" : "") + (clock && clock.off ? " (clock " + seconds(clock.offsetMs) + " off the swarm's, ±" + clock.withinMs + " ms)" : "")); - // Pinned (double-clicked): a dot in the middle. + // Pinned (middle-clicked): a dot in the middle. const pin = pinned.has(id) ? [drawing("pin " + id, "circle", { cx: p.x, cy: p.y, r: 2, fill: member.self || reported ? "var(--panel)" : "var(--fg)", "pointer-events": "none" })] : []; @@ -948,6 +1064,8 @@ function renderGraph() { label(nameLabel, name(id)); children.push(circle, ...pin, nameLabel); } + marks.pointed.replaceChildren(...marked.pointed); + marks.shown.replaceChildren(...marked.shown); show(document.getElementById("view"), children); } @@ -955,6 +1073,11 @@ function renderGraph() { // read again where the node says it changed (`document` events) --- // What is open, by path (as JSON): the first level to begin with. const documentOpen = new Set([JSON.stringify([])]); +// As the page it was opened from had them (`handOver`): the same document, so the same +// paths, opened as the tree is drawn. +if (handOver && Array.isArray(handOver.document)) { + for (const path of handOver.document) if (typeof path === "string") documentOpen.add(path); +} // The open branches read, by path (as JSON): to read again on a change. const documentBranches = new Map(); function storeUrl(path, shallow) { @@ -1378,14 +1501,18 @@ function release() { } document.addEventListener("selectionchange", release); -// The legend, hidden and shown again; the map takes the room. +// The legend, hidden and shown again, over the map: the map stays as it is. const legendToggle = document.getElementById("legend-toggle"); -legendToggle.addEventListener("click", () => { +function showLegend(shown) { const legends = document.getElementById("legends"); - legends.hidden = !legends.hidden; - legendToggle.textContent = legends.hidden ? "show legend" : "hide legend"; - if (state) renderGraph(); + legends.hidden = !shown; + legendToggle.textContent = shown ? "hide legend" : "show legend"; +} +legendToggle.addEventListener("click", () => { + showLegend(document.getElementById("legends").hidden); }); +// 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. @@ -1429,6 +1556,55 @@ function moveView() { "translate(" + view.x + " " + view.y + ") rotate(" + degrees(view.turn) + ") scale(" + view.k + ")"); } +// The map as the page it was opened from left it (`handOver`): its nodes where they +// were (new ones are placed among them; on a dynamic map all then move to their places +// smoothly, `settle`), the view and the pins. +if (handOver) { + for (const [id, at] of Object.entries(handOver.positions || {})) { + const [x, y] = Array.isArray(at) ? at : []; + if (Number.isFinite(x) && Number.isFinite(y)) positions.set(id, { x, y }); + } + for (const id of handOver.pinned || []) if (positions.has(id)) pinned.add(id); + if (positions.size > 0) relayout = false; + const handed = handOver.view || {}; + for (const key of ["x", "y", "k", "turn"]) { + if (Number.isFinite(handed[key])) view[key] = handed[key]; + } + moveView(); +} +// The fragment that opens another node's dashboard (`id`'s) as this page is now +// (`handOver`). What is in focus stays, but `id` itself: its page is its own. +function handOff(id) { + const settings = {}; + try { + for (let i = 0; i < localStorage.length; i++) { + const key = localStorage.key(i); + if (key && key.startsWith("continuum-")) settings[key] = localStorage.getItem(key); + } + } catch (_) {} + const round = value => Math.round(value * 10) / 10; + return HAND_OFF + encodeURIComponent(JSON.stringify({ + positions: Object.fromEntries([...positions].map(([node, p]) => [node, [round(p.x), round(p.y)]])), + pinned: [...pinned], + view: { ...view }, + focused: focused === id ? null : focused, + settings, + legend: !document.getElementById("legends").hidden, + links: state.graph.live !== false, + 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); @@ -1543,21 +1719,23 @@ graph.addEventListener("pointerdown", event => { // Not the browser's drag and drop of the picture. graph.addEventListener("dragstart", event => event.preventDefault()); // A node: dragged, it moves, held where the pointer is (`grabbed`) until let go; a click, -// it is in focus; a second one soon after pins it where it is (`pinned`: the layout -// leaves it there), or lets it go if pinned (`unpin`), and so does a second press soon -// after that drags it (pinned where it's dropped). Told apart here: the map holds the -// pointer while pressed (`pressing`), so the browser's clicks and double-clicks go to -// the map, not the node. +// it is in focus; a second one soon after opens its dashboard (`openDashboard`). A +// middle click pins it where it is (`pinned`: the layout leaves it +// there), or lets it go if pinned (`unpin`); dragged with the middle button, it is +// pinned where it's dropped. Told apart here: the map holds the pointer while pressed +// (`pressing`), so the browser's clicks go to the map, not the node. let lastNodeClick = { id: null, at: 0 }; const DOUBLE_CLICK_MS = 400; let grabbed = null; -// Of the press under way on a node: whether it's the second of a double-click, and -// whether it moved the node. -let secondPress = false, dragMoved = false; +// Of the press under way on a node: whether it's the middle button's, and whether it +// moved the node. +let middlePress = false, dragMoved = false; function drag(event, id) { const p = positions.get(id), from = { x: p.x, y: p.y }; grabbed = id; - secondPress = lastNodeClick.id === id && performance.now() - lastNodeClick.at < DOUBLE_CLICK_MS; + middlePress = event.button === 1; + // Not the browser's scrolling by the middle button. + if (middlePress) event.preventDefault(); dragMoved = false; pressing(event, (dx, dy) => { dragMoved = true; @@ -1569,25 +1747,70 @@ function drag(event, id) { // Dynamic: the others follow it at once. if (byRtt) settle(); }, () => { + if (middlePress) { + if (pinned.has(id)) unpin(id); else pin(id); + return; + } const now = performance.now(); if (lastNodeClick.id === id && now - lastNodeClick.at < DOUBLE_CLICK_MS) { - lastNodeClick = { id: null, at: now }; - if (pinned.has(id)) unpin(id); else pin(id); + lastNodeClick = { id: null, at: 0 }; + openDashboard(id); } else { lastNodeClick = { id, at: now }; focus(id); } }); } +// A node's dashboard, where it publishes one (`--dashboard-url`, in the member list), +// opened in this page; else its halo flashes red, larger than usual, and "no dashboard" +// rises from it in the same red, both fading out. This node's own: here already. +const FLASH_MS = 1200; +function openDashboard(id) { + const member = state.graph.members[id]; + if (!member || member.self) return; + if (member.dashboard) { + location.assign(member.dashboard.split("#")[0] + handOff(id)); + return; + } + const layer = document.getElementById("flashes"); + const halo = svg("circle", { r: 17, fill: "var(--bad)", "fill-opacity": .5 }); + const words = svg("text", { "text-anchor": "middle", "font-size": 12, "font-weight": 600, + fill: "var(--bad)", "fill-opacity": .5 }); + words.textContent = "no dashboard"; + layer.append(halo, words); + const started = performance.now(); + const step = now => { + const p = positions.get(id); + const t = Math.min(1, Math.max(0, (now - started) / FLASH_MS)); + if (!p || t >= 1) { + halo.remove(); + words.remove(); + return; + } + const opacity = String(1 - t); + halo.setAttribute("cx", p.x); + halo.setAttribute("cy", p.y); + halo.setAttribute("opacity", opacity); + // Up on the screen, however the map is turned or zoomed: from above the node, slowly. + const [x, y] = turned(0, -(22 + (calm.matches ? 0 : 30 * t)), -view.turn); + const attrs = upright({ x: p.x + x / view.k, y: p.y + y / view.k }); + words.setAttribute("x", attrs.x); + words.setAttribute("y", attrs.y); + if (attrs.transform) words.setAttribute("transform", attrs.transform); + else words.removeAttribute("transform"); + words.setAttribute("opacity", opacity); + requestAnimationFrame(step); + }; + requestAnimationFrame(step); +} // A dragged node let go of: dynamic, it moves to its place among the others again; -// dragged by a double-click's second press, its pin toggled. +// dragged with the middle button, it is pinned there. function dropped() { if (grabbed === null) return; const id = grabbed; grabbed = null; - if (dragMoved && secondPress) { - lastNodeClick = { id: null, at: 0 }; - if (pinned.has(id)) unpin(id); else pin(id); + if (dragMoved && middlePress) { + pin(id); return; } if (byRtt) settle(); @@ -1613,9 +1836,9 @@ graph.addEventListener("mouseleave", () => hover(null)); // as it grew since the last update, drawn as dots running along the link the way the // messages went. Counts come with each update (every 2 s), so the dots are spread // over the next 2 s, a few at most per link and way: a picture of the traffic, not a -// record of it. Kept above everything else on the map (`flows`, after what `show` -// places). --- -const FLOW_MS = 700, FLOW_SPREAD_MS = 2000, FLOWS_AT_MOST = 6; +// record of it. All at one speed (`flowSpeed`), so over a longer link they arrive +// later. Kept above everything else on the map (`flows`, after what `show` places). --- +const FLOW_SPREAD_MS = 2000, FLOWS_AT_MOST = 6; // Shown or not (`flows-toggle`, kept in this browser; shown at first). let flowsShown = true; try { flowsShown = localStorage.getItem("continuum-flows") !== "0"; } catch (_) {} @@ -1656,7 +1879,12 @@ const kindColour = kind => (KIND_COLOURS.find(([name]) => kind === name || let dotScale = 2; try { dotScale = Number(localStorage.getItem("continuum-dot-scale")) || 2; } catch (_) {} dotScale = Math.min(4, Math.max(1.25, dotScale)); -let dotSlider = null; +// How fast the dots run, in map units per ms (the slider after the size's, kept in this +// browser): .3 crosses a link of 200 in about 0.7 s. +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; function showMessagesLegend() { const legend = document.getElementById("messages-legend"); legend.classList.toggle("off", !flowsShown); @@ -1668,13 +1896,21 @@ function showMessagesLegend() { dotScale = Number(slider.value); try { localStorage.setItem("continuum-dot-scale", String(dotScale)); } catch (_) {} }); + const speed = speedSlider = el("input", { type: "range", min: ".1", max: ".9", step: ".1", + title: "How fast the dots run" }); + speed.value = String(flowSpeed); + speed.addEventListener("input", () => { + flowSpeed = Number(speed.value); + try { localStorage.setItem("continuum-dot-speed", String(flowSpeed)); } catch (_) {} + }); // Before the toggle, which ends the line. legend.prepend(...KIND_COLOURS.map(([kind, colour]) => el("span", { class: "message", style: "--colour: " + colour }, kind === "paths" ? "paths (pings)" : kind)), - el("label", { class: "dot-size" }, "size", slider)); + el("label", { class: "slider" }, "size", slider), + el("label", { class: "slider" }, "speed", speed)); } - dotSlider.disabled = !flowsShown; + dotSlider.disabled = speedSlider.disabled = !flowsShown; } showMessagesLegend(); // A dot's radius for a message of `bytes`: grows with the size's logarithm, from 1.5 @@ -1734,10 +1970,73 @@ function trafficArrived() { } } } + measureRates(current); lastTraffic = current; if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows); } +// --- The console's rates: messages and bytes per second sent and received, per peer +// and in all, averaged as load averages are, over 1, 5 and 15 minutes (each sample +// 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`): 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. + 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, messages, bytes) => { + const known = rates.get(key); + rates.set(key, { messages: average(known && known.messages, messages / secs), + bytes: average(known && known.bytes, bytes / secs) }); + }; + 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"), 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); + } + ratesAt = now; + 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. +function showRates() { + const filtering = consoleFiltered && focused !== null; + const peer = filtering ? peerOf(focused) : null; + 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; + const [unit, per] = largest < 1024 ? ["B/s", 1] : largest < 1048576 ? ["KiB/s", 1024] + : ["MiB/s", 1048576]; + const row = (values, unit) => [...(values ? values.map(format) : ["–", "", ""]), unit]; + const cells = [...row(shown && shown.messages, "msg/s"), + ...row(shown && shown.bytes.map(bytes => bytes / per), unit)]; + const box = document.getElementById("console-rates"); + if (box.childElementCount !== cells.length) { + box.replaceChildren(...cells.map((_, i) => el("span", { class: i % 4 === 3 ? "unit" : "" }))); + } + cells.forEach((cell, i) => text(box.children[i], cell)); +} + // --- Console: the messages this node exchanged with its peers, one by one, as the // node logs them (`messages` events, `traffic::MessageLog`), the latest at the bottom, // `CONSOLE_LINES` at most. Scrolled along while at the bottom. A click on one shows it @@ -1758,6 +2057,14 @@ const WAYS = { sent: "→", received: "←", refused: "↛" }; // The exchange marked in the console (a message's `related`): the one of the message // pointed at, else of the one shown; null: none. let relatedMarked = null; +// The message pointed at in the console, if any: its nodes and link are marked on the +// map (`renderGraph`), else the shown one's. +let pointedMessage = null; +function pointAt(message) { + pointedMessage = message; + markRelated(message ? message.related : shownRelated()); + if (state) renderGraph(); +} function markRelated(key) { relatedMarked = key || null; for (const line of document.getElementById("console").children) { @@ -1765,12 +2072,68 @@ function markRelated(key) { } } const shownRelated = () => shownMessage && shownMessage.related; +// The peer whose messages `key` (a node, or a link's pair key) stands for: the node, +// or the other end of one of this node's links; null: none (this node itself, whose +// every message it would be, or a link between others). +function peerOf(key) { + if (key === null || !state) return null; + const self = state.node.id; + if (!isPair(key)) return key === self ? null : key; + const [a, b] = key.split(" "); + return a === self ? b : b === self ? a : null; +} +// A console line's tint for the node or link pointed at on the map, or in focus. +function peerClasses(peer) { + return (peer === peerOf(focused) ? " peer-focused" : "") + + (peer === peerOf(hovered) ? " peer-hovered" : ""); +} +function markPeerMessages() { + const focusedPeer = peerOf(focused), hoveredPeer = peerOf(hovered); + for (const line of document.getElementById("console").children) { + line.classList.toggle("peer-focused", focusedPeer !== null && line.dataset.peer === focusedPeer); + line.classList.toggle("peer-hovered", hoveredPeer !== null && line.dataset.peer === hoveredPeer); + } + filterConsole(); +} +// `filter selected`: only the messages with the node or over the link in focus (those +// tinted for it, `peer-focused`), all while nothing is; where there are none (a link +// between other nodes, a node this one exchanges nothing with), a note. Pointing at +// one doesn't count. Kept in this browser. +let consoleFiltered = false; +try { consoleFiltered = localStorage.getItem("continuum-console-filter") === "1"; } catch (_) {} +const consoleFilter = document.getElementById("console-filter"); +consoleFilter.checked = consoleFiltered; +consoleFilter.addEventListener("change", () => { + consoleFiltered = consoleFilter.checked; + try { localStorage.setItem("continuum-console-filter", consoleFiltered ? "1" : "0"); } catch (_) {} + filterConsole(); +}); +// What the console is filtered by (what is in focus), null: nothing. When it changes +// (filtered or not any more, or another focus), the console goes to the bottom, and +// follows from there. +let consoleView = null; +function filterConsole() { + showRates(); + const log = document.getElementById("console"); + const filtering = consoleFiltered && focused !== null; + log.classList.toggle("filtering", filtering); + document.getElementById("console-none").hidden = + !filtering || log.querySelector(":scope > div.peer-focused") !== null; + const view = filtering ? focused : null; + if (view !== consoleView) { + consoleView = view; + const scroll = document.getElementById("console-scroll"); + scroll.scrollTop = scroll.scrollHeight; + } +} // A line: when, which way, with whom, the type (in its kind's colour) and the size. function consoleLine(message) { const line = el("div", { "data-seq": String(message.seq), "data-related": message.related || "", + "data-peer": message.peer, class: (shownMessage && shownMessage.seq === message.seq ? "focused" : "") + - (relatedMarked !== null && message.related === relatedMarked ? " related" : ""), + (relatedMarked !== null && message.related === relatedMarked ? " related" : "") + + peerClasses(message.peer), title: message.way === "refused" ? "refused: the transport wouldn't send it" : "" }, el("span", { class: "when" }, clockTime(new Date(message.at)) + " "), WAYS[message.way] + " " + name(message.peer) + " ", @@ -1782,8 +2145,8 @@ function consoleLine(message) { showMessage(shownMessage && shownMessage.seq === message.seq ? null : message); }); // The messages of its exchange marked while pointed at; then the shown one's again. - line.addEventListener("pointerenter", () => markRelated(message.related)); - line.addEventListener("pointerleave", () => markRelated(shownRelated())); + line.addEventListener("pointerenter", () => pointAt(message)); + line.addEventListener("pointerleave", () => pointAt(null)); return line; } function showMessage(message) { @@ -1804,25 +2167,56 @@ function messagesArrived(messages) { const following = scroll.scrollTop + scroll.clientHeight >= scroll.scrollHeight - 4; log.append(...messages.slice(-CONSOLE_LINES).map(consoleLine)); while (log.childElementCount > CONSOLE_LINES) log.firstElementChild.remove(); + filterConsole(); if (following) scroll.scrollTop = scroll.scrollHeight; + showFollowing(); } +// `following…` while the console is at its bottom: it scrolls to new messages then +// (`messagesArrived`); scrolled up, it stays where it is. +const consoleScroll = document.getElementById("console-scroll"); +function showFollowing() { + const scroll = consoleScroll; + document.getElementById("console-following").hidden = + scroll.scrollTop + scroll.clientHeight < scroll.scrollHeight - 4; +} +consoleScroll.addEventListener("scroll", showFollowing, { passive: true }); +// `clear`: the console emptied; it fills again with what comes from now on. +document.getElementById("console-clear").addEventListener("click", () => { + consoleMessages.length = 0; + pointedMessage = null; + if (shownMessage !== null) showMessage(null); + markRelated(null); + document.getElementById("console").replaceChildren(); + filterConsole(); + showFollowing(); +}); function showConsole() { document.getElementById("console").replaceChildren(...consoleMessages.map(consoleLine)); + filterConsole(); const scroll = document.getElementById("console-scroll"); scroll.scrollTop = scroll.scrollHeight; + showFollowing(); } function drawFlows() { flowFrame = null; const layer = document.getElementById("flows"); const now = performance.now(); + // How far along its link each is (0 to 1), as long as the link is now: arrived, or + // its link gone, it goes. + const along = flow => { + const p = positions.get(flow.from), q = positions.get(flow.to); + if (!p || !q) return null; + const length = Math.max(Math.hypot(q.x - p.x, q.y - p.y), 1); + return { p, q, t: (now - flow.start) * flowSpeed / length }; + }; for (let i = flows.length - 1; i >= 0; i--) { - if (now > flows[i].start + FLOW_MS) flows.splice(i, 1); + const at = along(flows[i]); + if (at === null || at.t >= 1) flows.splice(i, 1); } const dots = []; for (const flow of flows) { - const p = positions.get(flow.from), q = positions.get(flow.to); - if (now < flow.start || !p || !q) continue; - const t = (now - flow.start) / FLOW_MS; + if (now < flow.start) continue; + const { p, q, t } = along(flow); dots.push({ x: p.x + (q.x - p.x) * t, y: p.y + (q.y - p.y) * t, r: flow.r, kind: flow.kind, opacity: Math.sin(Math.PI * t) }); } @@ -1874,6 +2268,7 @@ function connect() { if (state !== null) documentChanged({ all: true }); state = snapshot; stateAt = Date.now(); live = true; status("live", true); pulse(); render(); trafficArrived(); + handOverLinks(); }); events.addEventListener("patch", event => { if (!state) return; diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index 2557906..8d954c0 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -302,14 +302,13 @@ impl Graph { let replica = state.doc.read().await; members::members(&replica.doc) }; - for member in members + let members: Vec<String> = members .into_iter() .filter(|member| member != state.local_id()) .take(MAX_ASKED) - { - let state = state.clone(); - tokio::spawn(async move { links::ask(&state, &member, lease).await }); - } + .collect(); + // Those we can't reach, through peers that can (`links::ask_all`). + links::ask_all(state, members, lease).await; } } @@ -320,9 +319,13 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue { let list = super::peers::peer_list(state).await; let stats = serde_json::to_value(super::stats::collect(state, &list).await).unwrap_or_default(); let peers = serde_json::to_value(&list).unwrap_or_default(); - let aliases = { + let (aliases, dashboards) = { let replica = state.doc.read().await; - members::aliases(&replica.doc) + let dashboards: std::collections::HashMap<String, String> = members::members(&replica.doc) + .into_iter() + .filter_map(|pk| members::dashboard(&replica.doc, &pk).map(|url| (pk, url))) + .collect(); + (members::aliases(&replica.doc), dashboards) }; let local = state.local_id(); let intended: Map<String, JsonValue> = graph @@ -357,12 +360,25 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue { ), u64::try_from(state.clock_skew_warn.as_millis()).unwrap_or(u64::MAX), ); - // Everyone on the map: in the member list's graph, or measured. + // 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 + // 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(); let mut members_json = Map::new(); - for member in graph.intended.keys().chain(measured.keys()) { + for member in graph.intended.keys().chain(measured.keys()).chain(linked) { members_json.insert( member.clone(), - json!({ "alias": aliases.get(member), "self": member == local }), + json!({ "alias": aliases.get(member), "self": member == local, + "dashboard": dashboards.get(member) }), ); } json!({ @@ -619,7 +635,11 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse { } } }; - Sse::new(stream).keep_alive(KeepAlive::default()) + ( + // Through nginx too as it comes, not buffered (`operations.md`). + [(super::UNBUFFERED.0, super::UNBUFFERED.1)], + Sse::new(stream).keep_alive(KeepAlive::default()), + ) } /// The node stops: the client may reconnect once it is back. diff --git a/src/api/listen.rs b/src/api/listen.rs new file mode 100644 index 0000000..df2df22 --- /dev/null +++ b/src/api/listen.rs @@ -0,0 +1,220 @@ +//! Where the HTTP API listens (`--bind-addr`): TCP addresses, and Unix sockets +//! (`unix:PATH`), any number of them at once, all serving the same API. A request +//! carries where it came from (`Peer`), which decides whether a reverse proxy's user +//! header is believed (`api::auth`, `--trusted-proxies`): on a Unix socket, its file's +//! permissions decide who may connect at all, which on one host is what an address +//! can't. + +use std::future::Future; +use std::net::SocketAddr; +use std::path::PathBuf; +use std::str::FromStr; + +use anyhow::{Context, Result}; +use axum::Router; +use axum::extract::connect_info::Connected; +use axum::serve::IncomingStream; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use tracing::{info, warn}; + +/// One place the API listens. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Bind { + Tcp(SocketAddr), + /// A Unix socket at this path, made by the node (readable and writable by its user + /// and group only), and removed when it stops. + Unix(PathBuf), +} + +const UNIX: &str = "unix:"; + +impl FromStr for Bind { + type Err = anyhow::Error; + + fn from_str(text: &str) -> Result<Self> { + let text = text.trim(); + match text.strip_prefix(UNIX) { + Some(path) if !path.is_empty() => Ok(Self::Unix(PathBuf::from(path))), + Some(_) => { + anyhow::bail!("bind-addr: unix: needs a path (unix:/run/continuum/api.sock)") + } + None => text.parse().map(Self::Tcp).with_context(|| { + format!("bind-addr: {text:?} is neither ADDRESS:PORT nor unix:PATH") + }), + } + } +} + +impl std::fmt::Display for Bind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Tcp(address) => write!(f, "{address}"), + Self::Unix(path) => write!(f, "{UNIX}{}", path.display()), + } + } +} + +impl Serialize for Bind { + fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> { + serializer.collect_str(self) + } +} + +impl<'de> Deserialize<'de> for Bind { + fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> { + let text = String::deserialize(deserializer)?; + text.parse() + .map_err(|error: anyhow::Error| serde::de::Error::custom(format!("{error:#}"))) + } +} + +/// `--bind-addr`: one place (as before it took more), or several. +pub(crate) fn one_or_more<'de, D: Deserializer<'de>>( + deserializer: D, +) -> Result<Vec<Bind>, D::Error> { + #[derive(Deserialize)] + #[serde(untagged)] + enum OneOrMore { + One(Bind), + More(Vec<Bind>), + } + Ok(match OneOrMore::deserialize(deserializer)? { + OneOrMore::One(bind) => vec![bind], + OneOrMore::More(binds) => binds, + }) +} + +/// Where a request came from (`axum::extract::ConnectInfo`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Peer { + Tcp(SocketAddr), + /// Over one of the API's Unix sockets. + Unix, +} + +impl Connected<IncomingStream<'_, tokio::net::TcpListener>> for Peer { + fn connect_info(stream: IncomingStream<'_, tokio::net::TcpListener>) -> Self { + Self::Tcp(*stream.remote_addr()) + } +} + +#[cfg(unix)] +impl Connected<IncomingStream<'_, tokio::net::UnixListener>> for Peer { + fn connect_info(_stream: IncomingStream<'_, tokio::net::UnixListener>) -> Self { + Self::Unix + } +} + +/// Serves `app` at every one of `binds` until `stop` resolves (each server then drains +/// its connections); resolves once all have stopped, with the first error if any did. +/// Binds them all before it returns: one that can't be bound is an error at once. +pub(crate) async fn serve<F>( + binds: &[Bind], + app: Router, + stop: impl Fn() -> F, +) -> Result<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> +where + F: Future<Output = ()> + Send + 'static, +{ + let mut servers = tokio::task::JoinSet::new(); + for bind in binds { + let service = app.clone().into_make_service_with_connect_info::<Peer>(); + let stopped = stop(); + match bind { + Bind::Tcp(address) => { + let listener = tokio::net::TcpListener::bind(address) + .await + .with_context(|| format!("Binding the HTTP API to {address}"))?; + info!(address = %listener.local_addr()?, "HTTP API listening"); + servers.spawn(async move { + axum::serve(listener, service) + .with_graceful_shutdown(stopped) + .await + .map_err(anyhow::Error::from) + }); + } + #[cfg(unix)] + Bind::Unix(path) => { + let listener = bind_unix(path)?; + info!(socket = %path.display(), "HTTP API listening"); + let path = path.clone(); + servers.spawn(async move { + let served = axum::serve(listener, service) + .with_graceful_shutdown(stopped) + .await; + if let Err(error) = std::fs::remove_file(&path) { + warn!(socket = %path.display(), error = %error, "Couldn't remove the HTTP API's socket"); + } + served.map_err(anyhow::Error::from) + }); + } + #[cfg(not(unix))] + Bind::Unix(_) => anyhow::bail!("bind-addr: Unix sockets need a Unix"), + } + } + Ok(Box::pin(async move { + let mut first = Ok(()); + while let Some(served) = servers.join_next().await { + let served = served + .map_err(anyhow::Error::from) + .and_then(|served| served); + if let Err(error) = served + && first.is_ok() + { + first = Err(error); + } + } + first + })) +} + +/// A Unix socket at `path`: one left there by an earlier run is replaced, anything else +/// there is an error. Its user and group may connect, nobody else. +#[cfg(unix)] +fn bind_unix(path: &std::path::Path) -> Result<tokio::net::UnixListener> { + use std::os::unix::fs::{FileTypeExt, PermissionsExt}; + match std::fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_socket() => std::fs::remove_file(path) + .with_context(|| format!("Removing the old socket {}", path.display()))?, + Ok(_) => anyhow::bail!( + "bind-addr: {} exists and is no socket; not replacing it", + path.display() + ), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error).with_context(|| format!("Looking at {}", path.display())); + } + } + let listener = tokio::net::UnixListener::bind(path) + .with_context(|| format!("Binding the HTTP API to {}", path.display()))?; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660)) + .with_context(|| format!("Setting who may use {}", path.display()))?; + Ok(listener) +} + +#[cfg(test)] +mod tests { + use super::Bind; + + #[test] + fn binds_parse() { + assert_eq!( + "127.0.0.1:3000".parse::<Bind>().unwrap(), + Bind::Tcp("127.0.0.1:3000".parse().unwrap()) + ); + assert_eq!( + "unix:/run/continuum/api.sock".parse::<Bind>().unwrap(), + Bind::Unix("/run/continuum/api.sock".into()) + ); + assert_eq!( + "unix:/run/continuum/api.sock" + .parse::<Bind>() + .unwrap() + .to_string(), + "unix:/run/continuum/api.sock" + ); + for bad in ["unix:", "localhost", "3000"] { + assert!(bad.parse::<Bind>().is_err(), "{bad}"); + } + } +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 9bb2bed..e4ccee6 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -9,6 +9,7 @@ mod dashboard; mod debug; mod document; mod health; +pub(crate) mod listen; mod log; mod peers; mod stats; @@ -77,13 +78,17 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> api.layer(middleware::from_fn_with_state(auth_state, basic_auth)) // The hand-written spec (`docs/openapi.yaml`), and the Swagger UI showing it. .merge( - SwaggerUi::new("/swagger-ui").config(utoipa_swagger_ui::Config::from("/openapi.yaml")), + // Relative to the UI's page (`/swagger-ui/`), so it is found behind a proxy that + // serves the API under a path of its own too. + SwaggerUi::new("/swagger-ui") + .config(utoipa_swagger_ui::Config::from("../openapi.yaml")), ) .route("/openapi.yaml", get(openapi_spec)) .route("/health", get(health::node_health)) .route("/ready", get(health::ready)) // Outermost, so authentication failures get the error body too. .layer(middleware::from_fn(json_errors)) + .layer(middleware::from_fn(relative_redirects)) .with_state(state) } @@ -107,9 +112,17 @@ pub(crate) fn relay_only_router( .layer(middleware::from_fn_with_state(auth_state, basic_auth)) .route("/health", get(health::health)) .layer(middleware::from_fn(json_errors)) + .layer(middleware::from_fn(relative_redirects)) .with_state(relay) } +/// On a response that streams: a reverse proxy passes it on as it comes, not once it +/// has buffered enough (nginx's `X-Accel-Buffering`). +pub(crate) const UNBUFFERED: (axum::http::HeaderName, &str) = ( + axum::http::HeaderName::from_static("x-accel-buffering"), + "no", +); + /// An error answer: `{"error": message}` with `details`' fields added (`docs/openapi.yaml`, /// `Error`). pub(crate) fn error_response( @@ -124,6 +137,56 @@ pub(crate) fn error_response( (status, Json(body)).into_response() } +/// Redirects to a path of this API (`Location: /store/…`) made relative to the request's +/// (`3`, `swagger-ui/`): they then lead to the same place behind a reverse proxy +/// that serves the API under a path of its own (`/continuum/store/…`). +async fn relative_redirects(request: Request, next: Next) -> Response { + let from = request.uri().path().to_owned(); + let mut response = next.run(request).await; + let relative = response + .headers() + .get(axum::http::header::LOCATION) + .and_then(|location| location.to_str().ok()) + .filter(|location| location.starts_with('/') && !location.starts_with("//")) + .map(|location| relative_to(&from, location)); + if let Some(location) = relative.and_then(|location| HeaderValue::from_str(&location).ok()) { + response + .headers_mut() + .insert(axum::http::header::LOCATION, location); + } + response +} + +/// `to` (a path, with its query if any) as a reference relative to the page at `from` +/// (a path). +fn relative_to(from: &str, to: &str) -> String { + let (path, query) = match to.split_once('?') { + Some((path, query)) => (path, Some(query)), + None => (to, None), + }; + // The page's directory, and the target's segments (the last its name). + let from: Vec<&str> = from.split('/').collect(); + let directory = &from[..from.len() - 1]; + let target: Vec<&str> = path.split('/').collect(); + let common = directory + .iter() + .zip(&target[..target.len() - 1]) + .take_while(|(a, b)| a == b) + .count(); + let mut relative = "../".repeat(directory.len() - common); + relative.push_str(&target[common..].join("/")); + // Not empty, and not taken for a scheme (`a:b`). + let first = relative.split('/').next().unwrap_or(""); + if relative.is_empty() || first.contains(':') { + relative.insert_str(0, "./"); + } + if let Some(query) = query { + relative.push('?'); + relative.push_str(query); + } + relative +} + /// Gives every error answer the same body (`error_response`). Handlers that fail with /// a bare status or a text message, and axum's own rejections (a malformed query or /// body), get it here: the text becomes the message, or the status's reason if there @@ -229,3 +292,21 @@ pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError { pub(crate) fn media_type(value: &str) -> &str { value.split(';').next().unwrap_or("").trim() } + +#[cfg(test)] +mod redirect_tests { + use super::relative_to; + + #[test] + fn redirects_are_relative_to_the_request() { + assert_eq!(relative_to("/store/queue/-1", "/store/queue/3"), "3"); + assert_eq!( + relative_to("/store/a/-1/b", "/store/a/0/b?x=1"), + "../0/b?x=1" + ); + assert_eq!(relative_to("/store/-1", "/store/2"), "2"); + assert_eq!(relative_to("/swagger-ui", "/swagger-ui/"), "swagger-ui/"); + assert_eq!(relative_to("/store/a/b", "/store/a/"), "./"); + assert_eq!(relative_to("/store/x", "/store/a:b"), "./a:b"); + } +} diff --git a/src/api/peers.rs b/src/api/peers.rs index 3fe4c39..537ad98 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -121,7 +121,9 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { for (peer, peer_state_mutex) in snapshot { let neighbour = crate::swarm::topology::is_neighbour(&state, &peer); - let traffic = state.traffic.peer(&peer).unwrap_or_default(); + let traffic = state + .traffic + .node(&peer, &state.transport.identities_of(&peer)); let sync_throughput = state .throughput .lock_or_recover() diff --git a/src/api/store.rs b/src/api/store.rs index 2b21c83..d419d37 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -861,6 +861,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo let stream_state = state.clone(); let stream_key = key.clone(); + let heartbeat = state.stream_heartbeat; let output = async_stream::stream! { let (mut last_value, mut conflicted) = initial; @@ -872,9 +873,27 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo return; } } + // When the last line went out: a blank one follows after `heartbeat` without + // another, so proxies in front don't take the stream for dead. + let mut sent = tokio::time::Instant::now(); loop { - let relevant = match changes.recv().await { + let beat = async { + match sent.checked_add(heartbeat).filter(|_| !heartbeat.is_zero()) { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending::<()>().await, + } + }; + let received = tokio::select! { + received = changes.recv() => Some(received), + () = beat => None, + }; + let Some(received) = received else { + sent = tokio::time::Instant::now(); + yield Ok(Bytes::from_static(b"\n")); + continue; + }; + let relevant = match received { // Conflicted paths can change without a patch on their path; re-read always. Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)), Err(broadcast::error::RecvError::Lagged(skipped)) => { @@ -910,6 +929,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo return; } } + sent = tokio::time::Instant::now(); last_value = next_value; } }; @@ -923,6 +943,11 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo axum::http::header::CACHE_CONTROL, HeaderValue::from_static("no-cache, no-transform"), ); + // Through nginx too as it comes, not buffered (`operations.md`). + response.headers_mut().insert( + super::UNBUFFERED.0, + HeaderValue::from_static(super::UNBUFFERED.1), + ); Ok(response) } diff --git a/src/config/mod.rs b/src/config/mod.rs index c7f96a3..a6f70aa 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -90,24 +90,38 @@ pub(crate) const FORCE: &[&str] = &["swarmId", "badDocumentFile", "replaceDocume pub(crate) const OPTIONS: &[OptionSpec] = &[ option( "bind-addr", - Kind::Text, - "HOSTNAME:PORT", + Kind::List, + "ADDRESS:PORT|unix:PATH,[...]", Some("127.0.0.1:3000"), - "Address of the HTTP API.", + "Where the HTTP API listens: TCP addresses, and Unix sockets (unix:PATH, made readable and writable by the node's user and group only), as many as given.", ), option( "htpasswd", Kind::Text, "PATH", None, - "Users of the HTTP API (htpasswd file with Argon2 hashes). Required, but with relay-only (then without it, no HTTP API).", + "Users of the HTTP API (htpasswd file with Argon2 hashes). Required, but with user-header or relay-only (then without either, no HTTP API).", + ), + option( + "user-header", + Kind::Text, + "NAME", + None, + "A request header naming the user, set by a reverse proxy in front of the HTTP API that authenticated the request itself (e.g. X-Remote-User, from nginx's $remote_user). Believed only from trusted-proxies; other requests need htpasswd's basic authentication. The proxy must set or clear it on every request.", + ), + option( + "trusted-proxies", + Kind::List, + "ADDRESS,[ADDRESS...]", + Some("127.0.0.1,::1,unix"), + "Where the reverse proxies whose user-header is believed connect from: IP addresses (or ranges, as ADDRESS/BITS), and unix for the API's Unix sockets (bind-addr). On a host with other users, leave out the addresses and have the proxy use a Unix socket: others can connect over loopback too.", ), option( "protocol-writers", Kind::List, "USER,[USER...]", None, - "HTTP API users (from htpasswd) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata and the swarm ID are never writable.", + "HTTP API users (from htpasswd or user-header) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata and the swarm ID are never writable.", ), option( "log-format", @@ -229,6 +243,20 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "A name for this node, shown next to its node ID (/peers, metrics). Published in the swarm's member list; not unique, never an identity.", ), option( + "dashboard-url", + Kind::Text, + "URL", + None, + "Where people reach this node's dashboard (http:// or https://, e.g. behind a reverse proxy). Published in the swarm's member list: the other nodes' dashboards open it on a double-click on this node. A build without the dashboard feature warns and publishes none (removing one published before).", + ), + option( + "stream-heartbeat", + Kind::Duration, + "DURATION", + Some("30s"), + "How often GET /stream sends a blank line while nothing changes, so proxies in front keep it open: below their read timeout (nginx's proxy_read_timeout, 60 s by default). 0: never.", + ), + option( "keepalive", Kind::Duration, "DURATION", @@ -480,10 +508,16 @@ pub(crate) fn spec(name: &str) -> Option<&'static OptionSpec> { #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(rename_all = "kebab-case", deny_unknown_fields)] pub(crate) struct Settings { - #[serde(default = "default_bind_addr")] - pub(crate) bind_addr: SocketAddr, - /// Required (`check`), but with `relay_only`. + #[serde( + default = "default_bind_addr", + deserialize_with = "crate::api::listen::one_or_more" + )] + pub(crate) bind_addr: Vec<crate::api::listen::Bind>, + /// Required (`check`), but with `user_header` or `relay_only`. pub(crate) htpasswd: Option<String>, + pub(crate) user_header: Option<String>, + #[serde(default = "default_trusted_proxies")] + pub(crate) trusted_proxies: Vec<String>, #[serde(default)] pub(crate) protocol_writers: Vec<String>, #[serde(default)] @@ -520,9 +554,15 @@ pub(crate) struct Settings { #[serde(default)] pub(crate) swarm_key_create: bool, pub(crate) alias: Option<String>, + pub(crate) dashboard_url: Option<String>, #[serde(default = "default_keepalive", with = "crate::units::duration_option")] pub(crate) keepalive: Duration, #[serde( + default = "default_stream_heartbeat", + with = "crate::units::duration_option" + )] + pub(crate) stream_heartbeat: Duration, + #[serde( default, deserialize_with = "crate::units::optional_duration", serialize_with = "crate::units::serialize_optional_duration" @@ -618,8 +658,14 @@ pub(crate) struct Settings { pub(crate) debug_epoch: Option<u64>, } -fn default_bind_addr() -> SocketAddr { - SocketAddr::from(([127, 0, 0, 1], 3000)) +fn default_bind_addr() -> Vec<crate::api::listen::Bind> { + vec![crate::api::listen::Bind::Tcp(SocketAddr::from(( + [127, 0, 0, 1], + 3000, + )))] +} +fn default_trusted_proxies() -> Vec<String> { + vec!["127.0.0.1".to_owned(), "::1".to_owned(), "unix".to_owned()] } fn default_document_append() -> Duration { Duration::from_secs(5) @@ -630,6 +676,9 @@ fn default_document_rewrite() -> Duration { fn default_keepalive() -> Duration { Duration::from_secs(60) } +fn default_stream_heartbeat() -> Duration { + Duration::from_secs(30) +} fn default_neighbours() -> usize { 12 } @@ -696,8 +745,10 @@ impl Settings { if self.relay_only { self.check_relay_only()?; } else { - if self.htpasswd.is_none() { - bail!("htpasswd is required: the users of the HTTP API"); + if self.htpasswd.is_none() && self.user_header.is_none() { + bail!( + "htpasswd is required: the users of the HTTP API (or user-header, with a reverse proxy that authenticates them)" + ); } if self.node_key_file.is_none() { bail!( @@ -766,6 +817,17 @@ impl Settings { if let Some(alias) = &self.alias { check_alias(alias)?; } + if let Some(url) = &self.dashboard_url { + let scheme = url.starts_with("http://") || url.starts_with("https://"); + if !scheme + || url.len() > 2048 + || url.chars().any(|c| c.is_whitespace() || c.is_control()) + { + bail!( + "dashboard-url: {url:?} is no http:// or https:// URL (of at most 2048 characters)" + ); + } + } Ok(()) } } @@ -820,6 +882,7 @@ impl Settings { ("swarm-key-create", self.swarm_key_create), ("protocol-writers", !self.protocol_writers.is_empty()), ("alias", self.alias.is_some()), + ("dashboard-url", self.dashboard_url.is_some()), ("debug-api", self.debug_api), ]; let set: Vec<&str> = swarm_options @@ -1496,6 +1559,16 @@ fn resolve(merged: Layer, files: Vec<PathBuf>) -> Result<Loaded> { settings.alias = settings.alias.map(|alias| alias.trim().to_owned()); settings.transports = settings.ordered_transports(); settings.check().context("invalid configuration")?; + // No dashboard to be reached there: none is published (and one published before is + // removed, `swarm::members`). + if !cfg!(feature = "dashboard") + && let Some(url) = settings.dashboard_url.take() + { + warn!( + url = %url, + "dashboard-url is set, but this build has no dashboard (the dashboard feature); not publishing it" + ); + } Ok(Loaded { settings, origins: merged diff --git a/src/main.rs b/src/main.rs index a10eb73..c3d7458 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,7 +4,6 @@ //! main.rs only wires things together: arguments, the starting document, the node //! (`node`), bootstrap and the HTTP server. -use std::future::IntoFuture; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -94,10 +93,11 @@ async fn main() -> Result<()> { // the spot, and startup (starting transports, bootstrap) can take minutes. let stop = watch_for_shutdown().context("Installing signal handlers")?; // A broken htpasswd will abort startup. - let auth_state = AuthState::new(args.htpasswd.as_deref().context("htpasswd is required")?)?; + let auth_state = + AuthState::from_settings(&args)?.context("htpasswd or user-header is required")?; info!( - bind_addr = args.bind_addr.to_string(), + bind_addr = ?args.bind_addr.iter().map(ToString::to_string).collect::<Vec<_>>(), transports = ?args.transports, "Starting replication service" ); @@ -219,8 +219,10 @@ async fn main() -> Result<()> { .with_probe_max(args.dormant_probe_max), neighbours_k: args.neighbours, alias: args.alias.clone(), + dashboard_url: args.dashboard_url.clone(), electorate_window: args.electorate_window, clock_skew_warn: args.clock_skew_warn, + stream_heartbeat: args.stream_heartbeat, superseded_writes: args.superseded_writes, // At most 4MiB (`Settings::check`). blob_chunk_bytes: args.blob_chunk as u32, @@ -260,20 +262,16 @@ async fn main() -> Result<()> { let joined = AtomicBool::new(false); let app = api::router(state.clone(), auth_state, args.debug_api); - let listener = tokio::net::TcpListener::bind(args.bind_addr) - .await - .with_context(|| format!("Binding local API to {}", args.bind_addr))?; - - info!(address = %listener.local_addr()?, "HTTP API listening"); - - // The node shutting down (`AppState::closing`): the server drains. + // The node shutting down (`AppState::closing`): the servers drain. let closing = state.closing.subscribe(); - let server_builder = axum::serve(listener, app).with_graceful_shutdown(async move { - stopped(closing).await; - info!("Shutdown signal received. Draining API connections...") - }); - - let server_fut = server_builder.into_future(); + let server_fut = api::listen::serve(&args.bind_addr, app, move || { + let closing = closing.clone(); + async move { + stopped(closing).await; + info!("Shutdown signal received. Draining API connections...") + } + }) + .await?; tokio::pin!(server_fut); // Joins, then runs until stopped; returns only if joining fails. @@ -336,7 +334,7 @@ async fn main() -> Result<()> { #[cfg(feature = "libp2p")] async fn relay_only(args: &config::Settings) -> Result<()> { let stop = watch_for_shutdown().context("Installing signal handlers")?; - let auth_state = args.htpasswd.as_deref().map(AuthState::new).transpose()?; + let auth_state = AuthState::from_settings(args)?; let transport: Arc<dyn Transport> = Arc::new(transport::libp2p::Libp2pTransport::start( transport::libp2p::Libp2pConfig::from_settings(args, true)?, )?); @@ -345,19 +343,13 @@ async fn relay_only(args: &config::Settings) -> Result<()> { let server = match auth_state { Some(auth_state) => { let app = api::relay_only_router(Arc::clone(&transport), auth_state); - let listener = tokio::net::TcpListener::bind(args.bind_addr) - .await - .with_context(|| format!("Binding local API to {}", args.bind_addr))?; - info!(address = %listener.local_addr()?, "HTTP API listening"); let stop = stop.clone(); - Some(tokio::spawn(async move { - axum::serve(listener, app) - .with_graceful_shutdown(stopped(stop)) - .await - })) + let served = + api::listen::serve(&args.bind_addr, app, move || stopped(stop.clone())).await?; + Some(tokio::spawn(served)) } None => { - info!("No htpasswd: no HTTP API"); + info!("No htpasswd or user-header: no HTTP API"); None } }; diff --git a/src/node.rs b/src/node.rs index 4597e66..8b18cd4 100644 --- a/src/node.rs +++ b/src/node.rs @@ -45,8 +45,12 @@ pub(crate) struct NodeConfig { pub(crate) timing: Timing, pub(crate) neighbours_k: usize, pub(crate) alias: Option<String>, + /// `--dashboard-url`. + pub(crate) dashboard_url: Option<String>, pub(crate) electorate_window: Duration, pub(crate) clock_skew_warn: Duration, + /// `--stream-heartbeat`. + pub(crate) stream_heartbeat: Duration, /// `--superseded-writes`. pub(crate) superseded_writes: crate::remake::Superseded, pub(crate) blob_chunk_bytes: u32, @@ -116,8 +120,10 @@ impl Node { handshakes: Default::default(), neighbours_k: config.neighbours_k, alias: config.alias, + dashboard_url: config.dashboard_url, electorate_window: config.electorate_window, clock_skew_warn: config.clock_skew_warn, + stream_heartbeat: config.stream_heartbeat, superseded_writes: config.superseded_writes, clock_ahead_ms: Default::default(), clock_skew: Default::default(), diff --git a/src/simtest/links.rs b/src/simtest/links.rs new file mode 100644 index 0000000..44836bd --- /dev/null +++ b/src/simtest/links.rs @@ -0,0 +1,105 @@ +//! 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; + } +} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 61c81d1..267768f 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -11,6 +11,8 @@ mod deltas; mod identity; +mod links; +mod relay; mod remakes; mod soak; mod timing; @@ -47,7 +49,14 @@ pub(crate) const SOON: Duration = Duration::from_secs(10); /// A simulated swarm: the net and the nodes started on it. pub(crate) struct Sim { + /// The net of the transport `sim`, every node's but those started on others + /// (`start_on`). pub(crate) net: SimNet, + /// Further nets, by the scheme of their transport: nodes on different nets can't + /// reach each other, as nodes on different transports (one on Tor alone, one on + /// libp2p alone) can't. Made when first used. + nets: std::sync::Mutex<std::collections::HashMap<&'static str, SimNet>>, + seed: u64, /// Nodes under the identity layer (`transport::identity`), named by node IDs; else /// by their names on the net. identity: bool, @@ -60,6 +69,8 @@ impl Sim { init_logging(); Self { net: SimNet::new(seed), + nets: Default::default(), + seed, identity: false, nodes: Default::default(), } @@ -99,8 +110,71 @@ impl Sim { start: Start, adjust: impl FnOnce(&mut NodeConfig), ) -> AppState { + self.start_node(id, self.net.transport(id), seeds, start, adjust) + } + + /// [`start`](Self::start) under the identity layer, on the nets of `schemes` + /// instead of `sim`'s alone (the first preferred): it reaches only nodes on one of + /// them. Seeds by name are resolved on the first. + pub(crate) fn start_on( + &self, + id: &str, + schemes: &[&'static str], + seeds: &[&str], + start: Start, + ) -> AppState { + self.start_on_with(id, schemes, seeds, start, |_| {}) + } + + /// [`start_on`](Self::start_on), with the configuration adjusted first. + pub(crate) fn start_on_with( + &self, + id: &str, + schemes: &[&'static str], + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> AppState { + assert!( + self.identity, + "transports by scheme need the identity layer" + ); + let mut transports = schemes.iter().map(|scheme| { + let transport: Arc<dyn crate::transport::Transport> = + self.net_for(scheme).transport(id); + (*scheme, transport) + }); + let (scheme, first) = transports.next().expect("a transport at least"); + let more: Vec<_> = transports.collect(); + let key = Arc::new(sim_key(id)); + self.start_node(id, first, seeds, start, move |config| { + config.identity = Some(IdentityConfig { key, scheme, more }); + adjust(config); + }) + } + + /// The net of the transport `scheme`. + pub(crate) fn net_for(&self, scheme: &'static str) -> SimNet { + if scheme == "sim" { + return self.net.clone(); + } + let mut nets = self.nets.lock().unwrap(); + let index = nets.len() as u64 + 1; + nets.entry(scheme) + .or_insert_with(|| SimNet::new(self.seed.wrapping_add(index))) + .clone() + } + + fn start_node( + &self, + id: &str, + transport: Arc<dyn crate::transport::Transport>, + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> AppState { let mut config = NodeConfig { - transport: self.net.transport(id), + transport, identity: self.identity.then(|| IdentityConfig { key: Arc::new(sim_key(id)), scheme: "sim", @@ -112,8 +186,10 @@ impl Sim { timing: Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT), neighbours_k: 8, alias: Some(id.to_owned()), + dashboard_url: None, electorate_window: Duration::from_secs(30 * 24 * 3600), clock_skew_warn: Duration::from_secs(2), + stream_heartbeat: Duration::from_secs(30), superseded_writes: Default::default(), blob_chunk_bytes: 64 * 1024, swarm_key: None, diff --git a/src/simtest/relay.rs b/src/simtest/relay.rs new file mode 100644 index 0000000..9b016b8 --- /dev/null +++ b/src/simtest/relay.rs @@ -0,0 +1,66 @@ +//! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't +//! reach: those on none of its transports. + +use std::time::Duration; + +use serde_json::json; + +use super::{SYNC_REPLY_TIMEOUT, Sim, read, write}; +use crate::state::Timing; +use crate::swarm::bootstrap::Start; +use crate::swarm::topology; +use crate::transport::sim::DELIVERY_LIMIT; + +const SOON: Duration = Duration::from_secs(30); + +/// a on `sim` alone, c on `sim2` alone, b on both; a full mesh at each (every member +/// it reaches is its neighbour), so a pushes its write to b alone, and b, which only +/// received it, must announce it to c. With a long keepalive, the regular STATUS (every +/// third keepalive, 15 s) can't be what brings it: c has it within a second (an +/// announcement waits 200 ms, then c pulls). Without the announcement, a STATUS that +/// happens to fall into that second would pass it too, once in about 15 runs. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn changes_reach_members_the_writer_cant_reach_at_once() { + let keepalive = Duration::from_secs(5); + let slow = move |config: &mut crate::node::NodeConfig| { + config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT); + }; + let sim = Sim::with_identity(62); + let a = sim.start_on_with("a", &["sim"], &[], Start::Genesis, slow); + let b = sim.start_on_with("b", &["sim", "sim2"], &["a"], Start::Join, slow); + let c = sim.start_on_with("c", &["sim2"], &["b"], Start::Join, slow); + sim.wait_ready(&[&a, &b, &c], SOON).await; + let c_id = c.local_id().to_owned(); + { + let (a, b, c_id) = (&a, &b, c_id.clone()); + sim.wait_until( + "a and b know their neighbours, a full mesh, c beyond a's reach", + SOON, + move || { + let c_id = c_id.clone(); + async move { + topology::view_known(a) + && topology::view_known(b) + && topology::is_full_mesh(a) + && topology::is_full_mesh(b) + && topology::is_neighbour(b, &c_id) + && topology::transports_of(a, &c_id).contains("sim2") + && !a.transport.can_reach(&c_id) + } + }, + ) + .await; + } + sim.wait_converged(&[&a, &b, &c], SOON).await; + + write(&a, "x", json!(1)).await; + { + let c = &c; + sim.wait_until( + "c has a's write within a second", + Duration::from_secs(1), + move || async move { read(c, "x").await == Some(json!(1)) }, + ) + .await; + } +} diff --git a/src/state/mod.rs b/src/state/mod.rs index aebc59a..fef6448 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -107,11 +107,17 @@ pub(crate) struct AppState { pub(crate) neighbours_k: usize, /// Our alias (`--alias`), published in the member list (`swarm::members`). pub(crate) alias: Option<String>, + /// Where people reach our dashboard (`--dashboard-url`), published in the member list + /// (`swarm::members`). + pub(crate) dashboard_url: Option<String>, /// Members alive within this window vote on remakes (`--electorate-window`). pub(crate) electorate_window: Duration, /// This node's clock is off when it differs from most of its peers' by more than /// this (`--clock-skew-warn`, `swarm::skew`); 0: never. pub(crate) clock_skew_warn: Duration, + /// How often `GET /stream` says it's alive while nothing changes + /// (`--stream-heartbeat`); 0: never. + pub(crate) stream_heartbeat: Duration, /// What a replay does with writes a newer one superseded (`--superseded-writes`). pub(crate) superseded_writes: crate::remake::Superseded, /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as diff --git a/src/swarm/links.rs b/src/swarm/links.rs index b84f57a..7f8f0b0 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -15,8 +15,20 @@ //! 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) 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. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Mutex; use std::time::Duration; @@ -28,7 +40,7 @@ use crate::state::AppState; use crate::swarm::{members, topology}; use crate::transport::paths::{self, ClockOffset}; use crate::transport::{ - Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, + Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, RelayedLinks, }; /// Links one reply lists at most (a member has about `--neighbours`, more if others @@ -47,6 +59,25 @@ const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300); 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; /// The requests out and the replies in, and when each asker was last answered. #[derive(Default)] @@ -69,6 +100,29 @@ struct Inner { subscribers: HashMap<String, (u64, Instant)>, /// The links subscribers were last sent (`watch`). published: Option<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)>, +} + +/// 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 { @@ -90,16 +144,57 @@ impl Links { .collect() } - /// The members our subscriptions are with (not run out). - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn subscribed(&self) -> Vec<String> { + /// 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() } } @@ -194,33 +289,185 @@ fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPat /// 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 = crate::transport::paths::random_nonce(); - { + 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 = Instant::now() + subscribe.max(REPLY_WITHIN); - inner.asked.insert(member.to_owned(), (nonce, until)); - } + 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).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); not `exclude` (who asked us). +async fn split( + state: &AppState, + members: Vec<String>, + exclude: Option<&str>, +) -> (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; + } + if state.transport.can_reach(&member) { + direct.push(member); + } else if let Some(relay) = relay_for(state, &member, exclude).await { + through.entry(relay).or_default().push(member); + } + } + (direct, through) +} + +/// The peer to ask `member` through, which we can't reach: an active member we reach +/// 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`. +async fn relay_for(state: &AppState, member: &str, exclude: Option<&str>) -> 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 || !state.transport.can_reach(&peer) { + continue; + } + { + let peer_state = peer_state.lock().await; + if 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. +/// 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, @@ -232,6 +479,9 @@ pub(crate) async fn unsubscribe(state: &AppState, member: &str) { 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"); @@ -250,6 +500,10 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) 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(); @@ -285,33 +539,276 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) 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 LINKS_REPLY from `peer`: kept if it answers our latest LINKS to it, while its -/// replies are taken (a subscription's go on). -pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) { - let mut inner = state.links.inner.lock_or_recover(); - let taken = inner - .asked - .get(peer) - .is_some_and(|(nonce, until)| *nonce == reply.nonce && Instant::now() < *until); - if !taken { - return; +/// 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); } - reply.links.truncate(MAX_LINKS); - for link in &mut reply.links { - link.others.truncate(MAX_OTHERS); + let (direct, through) = split(state, members, Some(asker)).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; + } } - if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) { - inner.replies.clear(); + if request.hops > 0 { + 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 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, + 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, 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(); + } + let at = now.checked_sub(age).unwrap_or(now); + // Not older news over newer. + if inner + .replies + .get(&member) + .is_some_and(|(known, _)| *known > at) + { + return; + } + inner.replies.insert(member.clone(), (at, links)); + arrived.push(member); + }; + if own { + keep(peer.to_owned(), Duration::ZERO, 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.links, + ); + } + } + } + } + for member in arrived { + forward(state, &member); } - inner - .replies - .insert(peer.to_owned(), (Instant::now(), reply.links)); } /// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's @@ -361,6 +858,7 @@ pub(crate) async fn watch(state: AppState) { 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"); @@ -409,6 +907,35 @@ fn changed(old: &[Link], new: &[Link]) -> bool { 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, + 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(), diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 05c6501..a534c04 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -10,7 +10,8 @@ //! //! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name //! for people reading /peers and metrics (not unique and never an identity: the public -//! key stays that), where it can be reached (`addresses`, as its transport gives them: +//! key stays that), where people reach its dashboard (`--dashboard-url`, `dashboard`), +//! where it can be reached (`addresses`, as its transport gives them: //! hints for the others' transports), and an `alive` stamp the node renews now and //! then: the members renewed within `--electorate-window` are the electorate of a //! remake. @@ -114,6 +115,34 @@ fn set_alias(doc: &mut impl Transactable, pk: &str, alias: Option<&str>) -> Resu } } +fn dashboard_path(pk: &str) -> Vec<String> { + let mut path = member_path(pk); + path.push("dashboard".to_owned()); + path +} + +/// Where people reach `pk`'s dashboard, if it says. +pub(crate) fn dashboard(doc: &Document, pk: &str) -> Option<String> { + match store::read_path(doc, &dashboard_path(pk)) { + Ok(Some(JsonValue::String(url))) => Some(url), + _ => None, + } +} + +fn set_dashboard( + doc: &mut impl Transactable, + pk: &str, + url: Option<&str>, +) -> Result<(), StoreError> { + match url { + Some(url) => store::write_path(doc, &dashboard_path(pk), &json!(url)), + None => match store::delete_path(doc, &dashboard_path(pk), false) { + Err(StoreError::NotFound) => Ok(()), + result => result, + }, + } +} + fn addresses_path(pk: &str) -> Vec<String> { let mut path = member_path(pk); path.push("addresses".to_owned()); @@ -260,8 +289,9 @@ const RENEWAL_MOST_OFTEN: Duration = Duration::from_secs(3600); pub(crate) const MIN_ELECTORATE_WINDOW: Duration = Duration::from_secs(7 * RENEWAL_MOST_OFTEN.as_secs()); -/// Registers this node if it is ready and its entry is missing, keeps its alias as -/// configured (set, changed or removed after a restart with another `--alias`) and its +/// Registers this node if it is ready and its entry is missing, keeps its alias and its +/// dashboard's URL as configured (set, changed or removed after a restart with another +/// `--alias` or `--dashboard-url`) and its /// addresses as the transport gives them, and renews its `alive` stamp now and then /// (see `electorate`). pub(crate) async fn ensure_registered(state: &AppState) { @@ -271,7 +301,15 @@ pub(crate) async fn ensure_registered(state: &AppState) { let local = state.local_id().to_owned(); let own = member_path(&local); let window = state.electorate_window.as_millis() as u64; - let (registered, recorded_alias, recorded_addresses, alive, recorded_window, shortest) = { + let ( + registered, + recorded_alias, + recorded_dashboard, + recorded_addresses, + alive, + recorded_window, + shortest, + ) = { let replica = state.doc.read().await; let registered = store::read_path(&replica.doc, &own) .ok() @@ -280,6 +318,7 @@ pub(crate) async fn ensure_registered(state: &AppState) { ( registered, alias(&replica.doc, &local), + dashboard(&replica.doc, &local), addresses(&replica.doc, &local), millis_at(&replica.doc, &alive_path(&local)), millis_at(&replica.doc, &window_path(&local)), @@ -301,6 +340,9 @@ pub(crate) async fn ensure_registered(state: &AppState) { if let Some(alias) = &wanted { entry["alias"] = json!(alias); } + if let Some(url) = &state.dashboard_url { + entry["dashboard"] = json!(url); + } match state .mutate_doc(move |doc| store::write_path(doc, &own, &entry)) .await @@ -352,6 +394,21 @@ pub(crate) async fn ensure_registered(state: &AppState) { } } + if recorded_dashboard != state.dashboard_url { + let url = state.dashboard_url.clone(); + let local = local.clone(); + match state + .mutate_doc(move |doc| set_dashboard(doc, &local, url.as_deref())) + .await + { + Ok(()) => info!( + url = state.dashboard_url.as_deref(), + "Dashboard URL updated in the member list" + ), + Err(error) => warn!(error = %error, "Failed to record our dashboard URL"), + } + } + if recorded_alias == wanted { return; } diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index e8e6f8b..6196571 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -170,7 +170,15 @@ pub(crate) async fn run_peer_worker(state: AppState) { crate::swarm::state_file::update(&state, false).await; { let known = state.peer_states.read().await; - state.traffic.retain(|peer| known.contains_key(peer)); + // And their addresses, where the layers below the node count + // (`Traffic::node`). + let addresses: std::collections::HashSet<String> = known + .keys() + .flat_map(|peer| state.transport.identities_of(peer)) + .collect(); + state + .traffic + .retain(|peer| known.contains_key(peer) || addresses.contains(peer)); state .throughput .lock_or_recover() diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 1736cec..5ab93f2 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -16,7 +16,10 @@ //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! -//! In a full mesh the writer reaches everyone itself, so nothing is announced. +//! In a full mesh the writer reaches everyone itself, so nothing is announced, but to +//! the neighbours the writer can't reach: those on none of its transports (a member on +//! Tor alone, reached through one on Tor and libp2p, from a writer on libp2p alone). +//! Whom it reaches is what the member list makes its neighbours (`topology::neighbours_of`). use std::collections::{HashMap, HashSet}; use std::sync::atomic::Ordering; @@ -44,7 +47,11 @@ pub(crate) async fn announce(state: &AppState, from: &str) { // While the view is unknown everyone counts as a neighbour, but that isn't a full // mesh: the others may not hear from the writer. if !state.initial_sync_done.load(Ordering::Acquire) - || (topology::view_known(state) && topology::is_full_mesh(state)) + || (topology::view_known(state) + && topology::is_full_mesh(state) + && unreached(state, &HashSet::from([from.to_owned()])) + .await + .is_empty()) { return; } @@ -80,26 +87,55 @@ pub(crate) async fn announce(state: &AppState, from: &str) { /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. async fn announce_now(state: &AppState, senders: &HashSet<String>) { + let neighbours = unreached(state, senders).await; + if neighbours.is_empty() { + return; + } state.stats.announce_round(); + for peer in neighbours { + if let Err(error) = send_status(state, &peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); + } + } +} + +/// The neighbours to tell of the changes `senders` sent us. The peer whose changes we +/// announce has them already, so with a single sender it is skipped; with several, each +/// lacks the others' changes. In a full mesh, also those every sender is a neighbour of: +/// the writers reached them themselves. +async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> { let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, }; - let neighbours: Vec<String> = state + let full_mesh = topology::view_known(state) && topology::is_full_mesh(state); + // Unknown (a sender not in the member list): nobody counts as reached by it. + let reached: Vec<Option<std::collections::BTreeSet<String>>> = if full_mesh { + senders + .iter() + .map(|sender| topology::neighbours_of(state, sender)) + .collect() + } else { + Vec::new() + }; + let reached_by_all = |peer: &str| { + full_mesh + && reached.iter().zip(senders).all(|(neighbours, sender)| { + sender == peer || neighbours.as_ref().is_some_and(|n| n.contains(peer)) + }) + }; + state .peer_states .read() .await .keys() .filter(|peer| { - skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer) + skip.as_deref() != Some(peer.as_str()) + && topology::is_neighbour(state, peer) + && !reached_by_all(peer) }) .cloned() - .collect(); - for peer in neighbours { - if let Err(error) = send_status(state, &peer).await { - debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); - } - } + .collect() } /// Pulls in progress: one neighbour at a time, so each change arrives about once. The diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 30fa56e..df75f46 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -133,31 +133,54 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool { pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> { let view = state.neighbours.read_or_recover(); let view = view.as_ref()?; - let unreachable = |member: &str| view.unreachable.contains(member); - let stable = |member: &str| view.stable.contains(member); let graph = view .members .iter() - .map(|member| { - let neighbours = if member == state.local_id() { - view.neighbours.clone() - } else { - let own = view.transports.get(member).cloned().unwrap_or_default(); - neighbours_across( - &view.transports, - member, - &own, - view.k, - &unreachable, - &stable, - ) - }; - (member.clone(), neighbours) - }) + .map(|member| (member.clone(), view.neighbours_of(state.local_id(), member))) .collect(); Some(graph) } +/// `member`'s neighbours as the member list makes them, as this node sees it (as in +/// `intended_graph`); None until the view is known, or for one the list doesn't have. +pub(crate) fn neighbours_of(state: &AppState, member: &str) -> Option<BTreeSet<String>> { + let view = state.neighbours.read_or_recover(); + let view = view.as_ref()?; + view.members + .iter() + .any(|listed| listed == member) + .then(|| view.neighbours_of(state.local_id(), member)) +} + +/// The transports `member` is on, by the member list (the schemes of its addresses); +/// ours as we are on them. None known: empty. +pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> { + let view = state.neighbours.read_or_recover(); + match view.as_ref() { + Some(view) if member == state.local_id() => view.local_transports.clone(), + Some(view) => view.transports.get(member).cloned().unwrap_or_default(), + None if member == state.local_id() => state.transport.transports().into_iter().collect(), + None => BTreeSet::new(), + } +} + +impl NeighbourView { + fn neighbours_of(&self, local: &str, member: &str) -> BTreeSet<String> { + if member == local { + return self.neighbours.clone(); + } + let own = self.transports.get(member).cloned().unwrap_or_default(); + neighbours_across( + &self.transports, + member, + &own, + self.k, + &|member| self.unreachable.contains(member), + &|member| self.stable.contains(member), + ) + } +} + /// How the neighbours changed. pub(crate) struct NeighbourChange { pub(crate) added: Vec<String>, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 96f7651..b8b57af 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -409,8 +409,16 @@ pub(crate) struct LinksRequest { 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. + /// 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. @@ -422,7 +430,18 @@ pub(crate) struct LinksReply { pub(crate) epoch_seq: u32, /// The request's. pub(crate) nonce: u64, - /// At most `swarm::links::MAX_LINKS`. + /// 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, pub(crate) links: Vec<Link>, } diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index bf9438f..b7907fb 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -30,6 +30,11 @@ impl Counter { self.messages += 1; self.bytes += bytes as u64; } + + fn absorb(&mut self, other: Counter) { + self.messages += other.messages; + self.bytes += other.bytes; + } } #[derive(Debug, Default, Clone, Copy, Serialize)] @@ -51,6 +56,37 @@ pub(crate) struct ByKind { } impl ByKind { + fn absorb(&mut self, other: &ByKind) { + for kind in [ + MessageKind::Hello, + MessageKind::HelloReply, + MessageKind::Sync, + MessageKind::Status, + MessageKind::StatusReply, + MessageKind::Blob, + MessageKind::Who, + MessageKind::Ping, + MessageKind::Links, + ] { + let theirs = *other.get(kind); + self.get_mut(kind).absorb(theirs); + } + } + + fn get(&self, kind: MessageKind) -> &Counter { + match kind { + MessageKind::Hello => &self.hello, + MessageKind::HelloReply => &self.hello_reply, + MessageKind::Sync => &self.sync, + MessageKind::Status => &self.status, + MessageKind::StatusReply => &self.status_reply, + MessageKind::Blob => &self.blob, + MessageKind::Who | MessageKind::Me => &self.identity, + MessageKind::Ping | MessageKind::Pong => &self.paths, + MessageKind::Links | MessageKind::LinksReply => &self.links, + } + } + fn get_mut(&mut self, kind: MessageKind) -> &mut Counter { match kind { MessageKind::Hello => &mut self.hello, @@ -282,6 +318,22 @@ impl Traffic { self.lock().peers.get(peer).copied() } + /// A node's traffic: counted under it, and under `addresses`, where it is (the + /// layers below the node count what they send and receive on their own there: WHO + /// and ME, the paths layer's PINGs and PONGs). + pub(crate) fn node(&self, node: &str, addresses: &[String]) -> PeerTraffic { + let table = self.lock(); + let mut traffic = table.peers.get(node).copied().unwrap_or_default(); + for address in addresses.iter().filter(|address| address.as_str() != node) { + if let Some(there) = table.peers.get(address) { + traffic.sent.absorb(&there.sent); + traffic.received.absorb(&there.received); + traffic.refused.absorb(&there.refused); + } + } + traffic + } + pub(crate) fn total(&self) -> PeerTraffic { self.lock().total } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 7717200..0769e9a 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 = 14; +pub(crate) const VERSION: u8 = 15; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -428,6 +428,9 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { epoch_seq: 1, nonce: 3, subscribe_secs: 60, + of: vec!["q".into()], + batch_ms: 250, + hops: 2, }), ParsedMessage::LinksReply(crate::transport::LinksReply { epoch, @@ -450,6 +453,11 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { within_ms: 40, }), }], + relayed: vec![crate::transport::RelayedLinks { + member: "q".into(), + age_ms: 1200, + links: Vec::new(), + }], }), ] } |