aboutsummaryrefslogtreecommitdiffstats
path: root/src/api/dashboard/links.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/dashboard/links.rs')
-rw-r--r--src/api/dashboard/links.rs216
1 files changed, 216 insertions, 0 deletions
diff --git a/src/api/dashboard/links.rs b/src/api/dashboard/links.rs
new file mode 100644
index 0000000..1f0b886
--- /dev/null
+++ b/src/api/dashboard/links.rs
@@ -0,0 +1,216 @@
+//! This node's links, as it measures them, for the dashboard's map (`own_links`): its
+//! own in the state, and to any page that subscribes to them
+//! (`GET /admin/dashboard/links/own/events`, also from another member's page).
+
+use crate::state::AppState;
+use crate::swarm::topology;
+use crate::transport::paths::{self, ClockOffset};
+use crate::transport::{Connection, Destination};
+
+/// Links listed at most (a node has about `--neighbours`, more if others prefer it).
+const MAX_LINKS: usize = 256;
+/// Other transports in a link at most.
+const MAX_OTHERS: usize = 8;
+/// One of this node's links.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct Link {
+ /// The peer's node ID.
+ pub(crate) peer: String,
+ /// The transport its messages go over now (the scheme of the path in use), if any.
+ pub(crate) transport: Option<String>,
+ /// Only through a relay (libp2p).
+ pub(crate) relayed: bool,
+ /// The path in use is down: none to the peer answers (`paths`), whatever its
+ /// connection was.
+ pub(crate) down: bool,
+ /// The round trip on that path, in microseconds, if measured.
+ pub(crate) rtt_us: Option<u32>,
+ /// Silent for long (`peers.md`, dormancy).
+ pub(crate) dormant: bool,
+ /// The peer's other transports (`MAX_OTHERS` at most): where else it proved
+ /// itself, those paths up.
+ pub(crate) others: Vec<OtherPath>,
+ /// How far the peer's clock is ahead of this node's (`paths::ClockOffset`), if
+ /// measured: with other members', the page shows whose clock is off.
+ pub(crate) clock: Option<ClockOffset>,
+}
+
+/// A transport of a link besides the one in use.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct OtherPath {
+ /// Its scheme (`mycelium`, `libp2p`, `tor`).
+ pub(crate) transport: String,
+ /// The round trip of its fastest path, in microseconds, if any is measured.
+ pub(crate) rtt_us: Option<u32>,
+ /// Only through a relay (libp2p): no direct connection at any of its paths.
+ pub(crate) relayed: bool,
+}
+
+/// This node's links: its neighbours, and the peers it holds dormant, as it measures
+/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its
+/// link stays listed, dormant, until it is heard from again: left out, the link would
+/// show only as that peer last reported it, from before.
+pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
+ let mut links = Vec::new();
+ for (peer, peer_state) in state.peers_snapshot().await {
+ let dormant = peer_state.lock().await.liveness.is_dormant();
+ if !dormant && !topology::is_neighbour(state, &peer) {
+ continue;
+ }
+ let destination = state.transport.destination(&peer);
+ let transport = destination
+ .as_ref()
+ .and_then(|destination| scheme_of(&destination.address));
+ links.push(Link {
+ others: others(destination.as_ref(), transport.as_deref()),
+ transport,
+ relayed: destination
+ .as_ref()
+ .is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
+ // Ranked up before down: the first is down only if all are.
+ down: destination
+ .as_ref()
+ .and_then(|destination| destination.paths.first())
+ .is_some_and(|path| path.down),
+ // The path in use's (`Transport::rtt`, the first of `destination.paths`).
+ rtt_us: state
+ .transport
+ .rtt(&peer)
+ .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
+ dormant,
+ clock: state.transport.clock_offset(&peer),
+ peer,
+ });
+ }
+ links.sort_by(|a, b| a.peer.cmp(&b.peer));
+ links.truncate(MAX_LINKS);
+ links
+}
+
+/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it
+/// has one.
+fn scheme_of(address: &str) -> Option<String> {
+ Some(paths::scheme_of(address))
+ .filter(|scheme| !scheme.is_empty())
+ .map(str::to_owned)
+}
+
+/// The transports besides `used` the peer proved itself at, with a path up: each with
+/// the round trip of its fastest path, if any is measured, and whether all of them go
+/// through a relay.
+fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> {
+ let mut others: Vec<OtherPath> = Vec::new();
+ for path in destination.map_or(&[][..], |destination| &destination.paths[..]) {
+ let Some(transport) = scheme_of(&path.address) else {
+ continue;
+ };
+ if path.down || used == Some(transport.as_str()) {
+ continue;
+ }
+ let rtt_us = path
+ .rtt_ms
+ .filter(|ms| ms.is_finite() && *ms >= 0.0)
+ .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32);
+ let relayed = path.connection == Some(Connection::Relayed);
+ match others.iter_mut().find(|other| other.transport == transport) {
+ Some(other) => {
+ other.rtt_us = match (other.rtt_us, rtt_us) {
+ (Some(a), Some(b)) => Some(a.min(b)),
+ (a, b) => a.or(b),
+ };
+ other.relayed &= relayed;
+ }
+ None => others.push(OtherPath {
+ transport,
+ rtt_us,
+ relayed,
+ }),
+ }
+ }
+ others.sort_by(|a, b| a.transport.cmp(&b.transport));
+ others.truncate(MAX_OTHERS);
+ others
+}
+
+/// Whether `new` says something `old` didn't: another peer, transport, relay or
+/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock
+/// measured or moved beyond both bounds. Round trips and clock samples wobble; small
+/// moves aren't news.
+pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool {
+ let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) {
+ (Some(a), Some(b)) => {
+ a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms)
+ }
+ (a, b) => a.is_some() != b.is_some(),
+ };
+ let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
+ (Some(a), Some(b)) => {
+ let by = a.abs_diff(b);
+ by > 1000 && by > a.max(b) / 5
+ }
+ (a, b) => a.is_some() != b.is_some(),
+ };
+ old.len() != new.len()
+ || old.iter().zip(new).any(|(a, b)| {
+ a.peer != b.peer
+ || a.transport != b.transport
+ || a.relayed != b.relayed
+ || a.down != b.down
+ || a.dormant != b.dormant
+ || moved(a.rtt_us, b.rtt_us)
+ || clock_moved(a.clock, b.clock)
+ || a.others.len() != b.others.len()
+ || a.others.iter().zip(&b.others).any(|(x, y)| {
+ x.transport != y.transport
+ || x.relayed != y.relayed
+ || moved(x.rtt_us, y.rtt_us)
+ })
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn link(peer: &str, rtt_us: Option<u32>) -> Link {
+ Link {
+ peer: peer.into(),
+ transport: Some("libp2p".into()),
+ relayed: false,
+ down: false,
+ rtt_us,
+ dormant: false,
+ others: Vec::new(),
+ clock: None,
+ }
+ }
+
+ #[test]
+ fn only_news_is_a_change() {
+ let old = [link("a", Some(10_000))];
+ assert!(!changed(&old, &[link("a", Some(11_000))]));
+ assert!(changed(&old, &[link("a", Some(13_000))]));
+ assert!(changed(&old, &[link("a", None)]));
+ assert!(changed(&old, &[link("b", Some(10_000))]));
+ assert!(changed(&old, &[]));
+ let mut dormant = link("a", Some(10_000));
+ dormant.dormant = true;
+ assert!(changed(&old, &[dormant]));
+ let mut down = link("a", Some(10_000));
+ down.down = true;
+ assert!(changed(&old, &[down]));
+ // Small round trips: a millisecond's wobble isn't news.
+ assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
+ // A clock: news when measured, and when it moved beyond both bounds.
+ let clocked = |offset_ms| Link {
+ clock: Some(ClockOffset {
+ offset_ms,
+ within_ms: 50,
+ }),
+ ..link("a", Some(10_000))
+ };
+ assert!(changed(&old, &[clocked(0)]));
+ assert!(!changed(&[clocked(0)], &[clocked(80)]));
+ assert!(changed(&[clocked(0)], &[clocked(5_000)]));
+ }
+}