diff options
| -rw-r--r-- | docs/openapi.yaml | 7 | ||||
| -rw-r--r-- | src/api/dashboard/dashboard.html | 81 | ||||
| -rw-r--r-- | src/api/dashboard/mod.rs | 8 | ||||
| -rw-r--r-- | src/swarm/links.rs | 70 | ||||
| -rw-r--r-- | src/transport/mod.rs | 12 | ||||
| -rw-r--r-- | src/transport/wire.rs | 18 |
6 files changed, 148 insertions, 48 deletions
diff --git a/docs/openapi.yaml b/docs/openapi.yaml index ad3b3c8..4ba5e25 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -494,10 +494,11 @@ paths: `/admin/stats`), `peers` (as `/peers`) and `graph`: `members` (by node ID: `alias`, `self`), `intended` (by node ID: the neighbours the member list gives it, as this node computes them) and `measured` (by node ID: `secsAgo` and - `links`, each with `peer`, `transport`, `relayed`, `rttMs`, `dormant`, as the - member answered a LINKS; ours always, the others' while fresh). One producer + `links`, each with `peer`, `transport`, `relayed`, `rttMs`, `dormant` and + `others` (the peer's other transports: `transport`, `rttMs`), as the member + answered a LINKS; ours always, the others' while fresh). One producer per node makes it every 2 s while anyone looks (and for a minute after this - request), asking members for their links every 30 s; every client gets the + request), asking members for their links every 10 s; every client gets the same. Only with the `dashboard` cargo feature. responses: '200': diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index c87341f..a8b697e 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -53,6 +53,8 @@ .legend span::before { content: ""; display: inline-block; width: 18px; height: 0; vertical-align: middle; margin-right: 4px; border-top: 2px solid var(--colour, currentColor); } .legend .intended::before { border-top-style: dashed; --colour: var(--intended); } + .legend .used::before { border-top-width: 3px; --colour: var(--muted); } + .legend .other::before { border-top-width: 1px; --colour: var(--muted); } .legend .direct::before { --colour: var(--muted); } .legend .relayed::before { border-top-style: dashed; --colour: var(--muted); } .legend .dormant::before { border-top-style: dotted; --colour: var(--bad); } @@ -242,14 +244,19 @@ function colourOf(transport) { function renderLegend(measured) { const transports = new Set(); - for (const { link } of measured.values()) transports.add(link.transport || ""); + for (const { link } of measured.values()) { + transports.add(link.transport || ""); + for (const other of link.others || []) transports.add(other.transport); + } const items = [el("span", { class: "intended" }, "intended (member list)")]; for (const transport of [...transports].sort()) { const item = el("span", {}, transport || "unknown transport"); item.style.setProperty("--colour", colourOf(transport || null)); items.push(item); } - items.push(el("span", { class: "direct" }, "direct"), + items.push(el("span", { class: "used" }, "in use"), + el("span", { class: "other" }, "other transport"), + el("span", { class: "direct" }, "direct"), el("span", { class: "relayed" }, "relayed"), el("span", { class: "dormant" }, "dormant")); document.getElementById("legend").replaceChildren(...items); @@ -263,13 +270,16 @@ function edges() { intended.set(key, [a, b]); } } - // One measured edge per pair: the fresher report wins. + // One measured edge per pair: a report with an RTT, then the fresher. const measured = new Map(); for (const [a, report] of Object.entries(state.graph.measured)) { for (const link of report.links) { const key = a < link.peer ? a + " " + link.peer : link.peer + " " + a; const known = measured.get(key); - if (!known || report.secsAgo < known.secsAgo) { + const timed = typeof link.rttMs === "number"; + const knownTimed = known && typeof known.link.rttMs === "number"; + if (!known || (timed && !knownTimed) || + (timed === knownTimed && report.secsAgo < known.secsAgo)) { measured.set(key, { a, b: link.peer, link, secsAgo: report.secsAgo }); } } @@ -335,31 +345,50 @@ function renderGraph() { children.push(svg("line", { x1: p.x, y1: p.y, x2: q.x, y2: q.y, stroke: "var(--intended)", "stroke-dasharray": "4 4", "stroke-width": 1 })); } + const ms = rtt => rtt < 10 ? rtt.toFixed(1) : String(Math.round(rtt)); for (const { a, b, link, secsAgo } of measured.values()) { const p = positions.get(a), q = positions.get(b); if (!p || !q) continue; - const line = svg("line", { x1: p.x, y1: p.y, x2: q.x, y2: q.y, - stroke: link.dormant ? "var(--bad)" : colourOf(link.transport), "stroke-width": 2 }); - // Dotted where dormant, dashed where relayed, solid where direct. - if (link.dormant) { - line.setAttribute("stroke-dasharray", "1 4"); - line.setAttribute("stroke-linecap", "round"); - } else if (link.relayed) { - line.setAttribute("stroke-dasharray", "7 4"); - } - const title = svg("title"); - title.textContent = name(a) + " – " + name(b) + (link.transport ? " via " + link.transport : "") + - (link.relayed ? ", relayed" : "") + (link.dormant ? ", dormant" : "") + - (typeof link.rttMs === "number" ? ", " + link.rttMs.toFixed(1) + " ms" : "") + - ", measured " + age(secsAgo + elapsed()) + " ago"; - line.append(title); - children.push(line); - if (typeof link.rttMs === "number") { - const label = svg("text", { x: (p.x + q.x) / 2, y: (p.y + q.y) / 2 - 3, - "text-anchor": "middle", "font-size": 10, fill: "var(--muted)" }); - label.textContent = link.rttMs < 10 ? link.rttMs.toFixed(1) : Math.round(link.rttMs); - children.push(label); - } + // The transport in use thick, in the middle; the others thin, beside it. + const lines = [{ transport: link.transport, rttMs: link.rttMs, used: true }]; + for (const other of link.others || []) lines.push({ ...other, used: false }); + const length = Math.max(Math.hypot(q.x - p.x, q.y - p.y), 1); + const nx = -(q.y - p.y) / length, ny = (q.x - p.x) / length; + lines.forEach((line, i) => { + // 0, then +1, -1, +2, -2 … steps of 6 px to the side. + const side = i === 0 ? 0 : (i % 2 ? 1 : -1) * Math.ceil(i / 2) * 6; + const ox = nx * side, oy = ny * side; + const drawn = svg("line", { x1: p.x + ox, y1: p.y + oy, x2: q.x + ox, y2: q.y + oy, + stroke: link.dormant ? "var(--bad)" : colourOf(line.transport), + "stroke-width": line.used ? 3 : 1.2 }); + // Dotted where dormant, dashed where relayed, solid where direct. + if (link.dormant) { + drawn.setAttribute("stroke-dasharray", line.used ? "1 5" : "1 3"); + drawn.setAttribute("stroke-linecap", "round"); + } else if (line.used && link.relayed) { + drawn.setAttribute("stroke-dasharray", "7 4"); + } + const title = svg("title"); + title.textContent = name(a) + " – " + name(b) + + (line.transport ? " via " + line.transport : "") + (line.used ? " (in use)" : "") + + (line.used && link.relayed ? ", relayed" : "") + (link.dormant ? ", dormant" : "") + + (typeof line.rttMs === "number" ? ", " + line.rttMs.toFixed(1) + " ms" : "") + + ", measured " + age(secsAgo + elapsed()) + " ago"; + drawn.append(title); + children.push(drawn); + if (typeof line.rttMs === "number") { + // The one in use at the middle, larger; the others further along, beside theirs. + const t = line.used ? .5 : .3; + const out = line.used ? -4 : Math.sign(side) * 9; + const label = svg("text", { + x: p.x + (q.x - p.x) * t + ox + nx * out, + y: p.y + (q.y - p.y) * t + oy + ny * out + (line.used ? 0 : 3), + "text-anchor": "middle", "font-size": line.used ? 11 : 9, + fill: line.used ? "var(--fg)" : "var(--muted)" }); + label.textContent = ms(line.rttMs); + children.push(label); + } + }); } for (const id of nodes) { const p = positions.get(id), member = state.graph.members[id]; diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index adaed10..6c7f1c5 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -35,9 +35,9 @@ use crate::transport::Link; /// How often the snapshot is built and compared, while anyone looks. const TICK: Duration = Duration::from_secs(2); /// How often members are asked for their links, and the graph recomputed. -const LINKS_EVERY: Duration = Duration::from_secs(30); +const LINKS_EVERY: Duration = Duration::from_secs(10); /// A member's links older than this are left out (it didn't answer since). -const LINKS_KEPT_FOR: Duration = Duration::from_secs(90); +const LINKS_KEPT_FOR: Duration = Duration::from_secs(30); /// The producer runs on this long after the last request for the state. const LINGER: Duration = Duration::from_secs(60); /// Members asked for their links at most, each round. @@ -230,6 +230,10 @@ fn links_json(links: &[Link]) -> JsonValue { "relayed": link.relayed, "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0), "dormant": link.dormant, + "others": link.others.iter().map(|other| json!({ + "transport": other.transport, + "rttMs": other.rtt_us.map(|us| f64::from(us) / 1000.0), + })).collect::<Vec<_>>(), }) }) .collect() diff --git a/src/swarm/links.rs b/src/swarm/links.rs index a12a1de..d7df0d5 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -19,11 +19,15 @@ use tracing::debug; use crate::locks::MutexExt; use crate::state::AppState; use crate::swarm::{members, topology}; -use crate::transport::{Connection, Link, LinksReply, LinksRequest, ParsedMessage}; +use crate::transport::{ + Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, +}; /// Links one reply lists at most (a member has about `--neighbours`, more if others /// prefer it). pub(crate) const MAX_LINKS: usize = 256; +/// Other transports in a link at most. +pub(crate) const MAX_OTHERS: usize = 8; /// A member answers each other one at most this often. const ANSWER_EVERY: Duration = Duration::from_secs(5); /// Replies kept from members at most (one each), and requests out. @@ -74,19 +78,17 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { } let dormant = peer_state.lock().await.liveness.is_dormant(); let destination = state.transport.destination(&peer); + let transport = destination + .as_ref() + .and_then(|destination| scheme_of(&destination.address)); links.push(Link { - transport: destination.as_ref().and_then(|destination| { - destination - .address - .split_once(':') - .map(|(scheme, _)| scheme.to_owned()) - }), + others: others(destination.as_ref(), transport.as_deref()), + transport, relayed: destination .as_ref() .is_some_and(|destination| destination.connection == Some(Connection::Relayed)), - rtt_us: state - .transport - .rtt(&peer) + rtt_us: rtt_of(destination.as_ref()) + .or_else(|| state.transport.rtt(&peer)) .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), dormant, peer, @@ -97,6 +99,51 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { links } +/// The scheme of `address` (`mycelium:…`): its transport. +fn scheme_of(address: &str) -> Option<String> { + address.split_once(':').map(|(scheme, _)| scheme.to_owned()) +} + +/// The transports besides `used` the peer proved itself at, with a path up: each with +/// the round trip of its fastest path, if any is measured. +fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> { + let mut others: Vec<OtherPath> = Vec::new(); + for path in destination.map_or(&[][..], |destination| &destination.paths[..]) { + let Some(transport) = scheme_of(&path.address) else { + continue; + }; + if path.down || used == Some(transport.as_str()) { + continue; + } + let rtt_us = path + .rtt_ms + .filter(|ms| ms.is_finite() && *ms >= 0.0) + .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32); + match others.iter_mut().find(|other| other.transport == transport) { + Some(other) => { + other.rtt_us = match (other.rtt_us, rtt_us) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + }; + } + None => others.push(OtherPath { transport, rtt_us }), + } + } + others.sort_by(|a, b| a.transport.cmp(&b.transport)); + others.truncate(MAX_OTHERS); + others +} + +/// The round-trip time of the path messages to the peer go over (`destination`'s +/// first path, best first), if measured: `Transport::rtt` takes the first address the +/// peer proved itself at, which needn't be the one in use. +fn rtt_of(destination: Option<&Destination>) -> Option<Duration> { + let path = destination?.paths.first()?; + path.rtt_ms + .filter(|ms| ms.is_finite() && *ms >= 0.0) + .map(|ms| Duration::from_secs_f64(ms / 1000.0)) +} + /// Asks `member` for its links (a LINKS); its reply lands in `AppState::links`. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) async fn ask(state: &AppState, member: &str) { @@ -168,6 +215,9 @@ pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) { } inner.asked.remove(peer); reply.links.truncate(MAX_LINKS); + for link in &mut reply.links { + link.others.truncate(MAX_OTHERS); + } if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) { inner.replies.clear(); } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index a15a1f7..9d28212 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -428,6 +428,18 @@ pub(crate) struct Link { pub(crate) rtt_us: Option<u32>, /// Silent for long (`peers.md`, dormancy). pub(crate) dormant: bool, + /// The peer's other transports (`swarm::links::MAX_OTHERS` at most): where else + /// it proved itself, those paths up. + pub(crate) others: Vec<OtherPath>, +} + +/// A transport of a link besides the one in use, in a LINKS_REPLY. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct OtherPath { + /// Its scheme (`mycelium`, `libp2p`, `tor`). + pub(crate) transport: String, + /// The round trip of its fastest path, in microseconds, if any is measured. + pub(crate) rtt_us: Option<u32>, } #[derive(Clone, Serialize, Deserialize)] diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 3e1ab10..87063d2 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{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 = 7; +pub(crate) const VERSION: u8 = 8; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -419,6 +419,10 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { relayed: true, rtt_us: Some(1500), dormant: false, + others: vec![crate::transport::OtherPath { + transport: "mycelium".into(), + rtt_us: None, + }], }], }), ] @@ -591,7 +595,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("070097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("080097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -611,7 +615,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "070199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "080199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -626,7 +630,7 @@ mod tests { }); assert_eq!( hex(&encode(&status).unwrap()), - format!("0703959290c400{epoch_hex}01c3c2") + format!("0803959290c400{epoch_hex}01c3c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -635,7 +639,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("070692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("080692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -649,10 +653,10 @@ mod tests { }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); - assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "070807"); + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "080807"); assert_eq!( hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()), - "0709cfffffffffffffffff" + "0809cfffffffffffffffff" ); let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else { panic!("not a PONG"); |