aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/auth.rs200
-rw-r--r--src/api/dashboard/dashboard.html497
-rw-r--r--src/api/dashboard/mod.rs42
-rw-r--r--src/api/listen.rs220
-rw-r--r--src/api/mod.rs83
-rw-r--r--src/api/peers.rs4
-rw-r--r--src/api/store.rs27
-rw-r--r--src/config/mod.rs97
-rw-r--r--src/main.rs46
-rw-r--r--src/node.rs6
-rw-r--r--src/simtest/links.rs105
-rw-r--r--src/simtest/mod.rs78
-rw-r--r--src/simtest/relay.rs66
-rw-r--r--src/state/mod.rs6
-rw-r--r--src/swarm/links.rs585
-rw-r--r--src/swarm/members.rs65
-rw-r--r--src/swarm/peer/worker.rs10
-rw-r--r--src/swarm/relay.rs56
-rw-r--r--src/swarm/topology/mod.rs59
-rw-r--r--src/transport/mod.rs23
-rw-r--r--src/transport/traffic.rs52
-rw-r--r--src/transport/wire.rs10
22 files changed, 2150 insertions, 187 deletions
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(),
+ }],
}),
]
}