add links messages

Introduce two new wire messages, LINKS and LINKS_REPLY (swarm::links).
A member answers with its links: for each peer, the transport, whether
it’s relayed, the RTT, and whether it’s dormant.  It answers members
only, at most every 5 s.  This bumps wire::VERSION to 7.  Every build
answers, with or without the feature, so every node needs this patch
before the map shows anything beyond its own node.

9 files changed, 274 insertions, 8 deletions

diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 690effa..aef0616 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -179,6 +179,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
             (direction, "blob", by.blob),
             (direction, "identity", by.identity),
             (direction, "paths", by.paths),
+            (direction, "links", by.links),
         ]
     };
     let mut series = Vec::new();
diff --git a/src/node.rs b/src/node.rs
index 8aab601..be722e6 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -114,6 +114,7 @@ impl Node {
             alias: config.alias,
             electorate_window: config.electorate_window,
             blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)),
+            links: Default::default(),
             protocol_writers: Arc::new(config.protocol_writers),
             state_file: config.state_file,
             document_file: config.document_file,
diff --git a/src/state/mod.rs b/src/state/mod.rs
index fec164d..d7c80c2 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -112,6 +112,8 @@ pub(crate) struct AppState {
     pub(crate) electorate_window: Duration,
     /// Chunked transfers served and fetched (`swarm::blob`).
     pub(crate) blobs: Arc<crate::swarm::blob::Blobs>,
+    /// LINKS asked and answered, for a map of the swarm (`swarm::links`).
+    pub(crate) links: Arc<crate::swarm::links::Links>,
     /// HTTP API users who may write protocol data through the store API
     /// (`--protocol-writers`).
     pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>,
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
new file mode 100644
index 0000000..a12a1de
--- /dev/null
+++ b/src/swarm/links.rs
@@ -0,0 +1,177 @@
+//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a
+//! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its
+//! neighbours, each with the transport its messages go over, whether only through a
+//! relay, the round trip and whether the neighbour is dormant: as the member measures
+//! them, which no other node can.
+//!
+//! Every node answers, whatever its build: one with a dashboard asks any member, which
+//! may have none. Only members (in the member list) are answered, as the links name
+//! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing
+//! is written to the document: asked for while someone looks, forgotten after.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::{members, topology};
+use crate::transport::{Connection, Link, LinksReply, LinksRequest, ParsedMessage};
+
+/// Links one reply lists at most (a member has about `--neighbours`, more if others
+/// prefer it).
+pub(crate) const MAX_LINKS: usize = 256;
+/// A member answers each other one at most this often.
+const ANSWER_EVERY: Duration = Duration::from_secs(5);
+/// Replies kept from members at most (one each), and requests out.
+const MAX_KEPT: usize = 4096;
+
+/// The requests out and the replies in, and when each asker was last answered.
+#[derive(Default)]
+pub(crate) struct Links {
+    inner: Mutex<Inner>,
+}
+
+#[derive(Default)]
+struct Inner {
+    /// Our LINKS out, by member: its nonce.
+    asked: HashMap<String, u64>,
+    /// The latest LINKS_REPLY of each member: when, and its links.
+    replies: HashMap<String, (Instant, Vec<Link>)>,
+    /// When we last answered each member.
+    answered: HashMap<String, Instant>,
+}
+
+impl Links {
+    /// The latest links each member replied with, and how long ago.
+    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+    pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec<Link>)> {
+        let now = Instant::now();
+        self.inner
+            .lock_or_recover()
+            .replies
+            .iter()
+            .map(|(member, (at, links))| {
+                (
+                    member.clone(),
+                    now.saturating_duration_since(*at),
+                    links.clone(),
+                )
+            })
+            .collect()
+    }
+}
+
+/// This node's links: its neighbours, as it measures them.
+pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
+    let mut links = Vec::new();
+    for (peer, peer_state) in state.peers_snapshot().await {
+        if !topology::is_neighbour(state, &peer) {
+            continue;
+        }
+        let dormant = peer_state.lock().await.liveness.is_dormant();
+        let destination = state.transport.destination(&peer);
+        links.push(Link {
+            transport: destination.as_ref().and_then(|destination| {
+                destination
+                    .address
+                    .split_once(':')
+                    .map(|(scheme, _)| scheme.to_owned())
+            }),
+            relayed: destination
+                .as_ref()
+                .is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
+            rtt_us: state
+                .transport
+                .rtt(&peer)
+                .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
+            dormant,
+            peer,
+        });
+    }
+    links.sort_by(|a, b| a.peer.cmp(&b.peer));
+    links.truncate(MAX_LINKS);
+    links
+}
+
+/// 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) {
+    let nonce = crate::transport::paths::random_nonce();
+    {
+        let mut inner = state.links.inner.lock_or_recover();
+        if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) {
+            inner.asked.clear();
+        }
+        inner.asked.insert(member.to_owned(), nonce);
+    }
+    let (epoch, epoch_seq) = state.epoch_stamp();
+    let request = ParsedMessage::Links(LinksRequest {
+        epoch,
+        epoch_seq,
+        nonce,
+    });
+    if let Err(error) = state.transport.send_message(member, &request).await {
+        debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
+    }
+}
+
+/// A LINKS from `peer`: answered if it's a member, and not answered lately.
+pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) {
+    let member = {
+        let replica = state.doc.read().await;
+        members::members(&replica.doc)
+            .iter()
+            .any(|member| member == peer)
+    };
+    if !member {
+        debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member");
+        return;
+    }
+    {
+        let now = Instant::now();
+        let mut inner = state.links.inner.lock_or_recover();
+        if inner
+            .answered
+            .get(peer)
+            .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY)
+        {
+            return;
+        }
+        if inner.answered.len() >= MAX_KEPT {
+            inner
+                .answered
+                .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY);
+        }
+        inner.answered.insert(peer.to_owned(), now);
+    }
+    let (epoch, epoch_seq) = state.epoch_stamp();
+    let reply = ParsedMessage::LinksReply(LinksReply {
+        epoch,
+        epoch_seq,
+        nonce: request.nonce,
+        links: own_links(state).await,
+    });
+    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 LINKS to it.
+pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) {
+    let mut inner = state.links.inner.lock_or_recover();
+    if inner.asked.get(peer) != Some(&reply.nonce) {
+        return;
+    }
+    inner.asked.remove(peer);
+    reply.links.truncate(MAX_LINKS);
+    if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) {
+        inner.replies.clear();
+    }
+    inner
+        .replies
+        .insert(peer.to_owned(), (Instant::now(), reply.links));
+}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index d424529..7fb8e16 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -4,6 +4,7 @@
 pub(crate) mod auth;
 pub(crate) mod blob;
 pub(crate) mod bootstrap;
+pub(crate) mod links;
 pub(crate) mod members;
 pub(crate) mod pair;
 pub(crate) mod peer;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 6a149a2..d4df6c7 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -160,6 +160,12 @@ pub(crate) async fn handle_message(
         transport::ParsedMessage::StatusReply(remote) => {
             on_status(state, &peer, &remote, reactivated).await;
         }
+        transport::ParsedMessage::Links(request) => {
+            crate::swarm::links::answer(state, &peer, request).await;
+        }
+        transport::ParsedMessage::LinksReply(reply) => {
+            crate::swarm::links::received(state, &peer, reply);
+        }
         transport::ParsedMessage::Who(_)
         | transport::ParsedMessage::Me(_)
         | transport::ParsedMessage::Ping(_)
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index fd3651b..a15a1f7 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -218,6 +218,11 @@ pub(crate) enum ParsedMessage {
     Ping(u64),
     /// Answers a PING.
     Pong(u64),
+    /// Asks a member how it is linked to its peers, for a map of the swarm
+    /// (`swarm::links`).
+    Links(LinksRequest),
+    /// Answers a LINKS.
+    LinksReply(LinksReply),
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -343,6 +348,8 @@ pub(crate) enum MessageKind {
     Me,
     Ping,
     Pong,
+    Links,
+    LinksReply,
 }
 
 impl ParsedMessage {
@@ -358,6 +365,8 @@ impl ParsedMessage {
             ParsedMessage::Me(_) => MessageKind::Me,
             ParsedMessage::Ping(_) => MessageKind::Ping,
             ParsedMessage::Pong(_) => MessageKind::Pong,
+            ParsedMessage::Links(_) => MessageKind::Links,
+            ParsedMessage::LinksReply(_) => MessageKind::LinksReply,
         }
     }
 
@@ -372,6 +381,8 @@ impl ParsedMessage {
                 Some((status.epoch, status.epoch_seq))
             }
             ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
+            ParsedMessage::Links(request) => Some((request.epoch, request.epoch_seq)),
+            ParsedMessage::LinksReply(reply) => Some((reply.epoch, reply.epoch_seq)),
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
@@ -380,6 +391,45 @@ impl ParsedMessage {
     }
 }
 
+/// LINKS: what a member is asked for a map of the swarm (`swarm::links`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct LinksRequest {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// Repeated in the reply: which request it answers.
+    pub(crate) nonce: u64,
+}
+
+/// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct LinksReply {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// The request's.
+    pub(crate) nonce: u64,
+    /// At most `swarm::links::MAX_LINKS`.
+    pub(crate) links: Vec<Link>,
+}
+
+/// One of a member's links, in a LINKS_REPLY.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct Link {
+    /// The peer's node ID.
+    pub(crate) peer: String,
+    /// The transport its messages go over now (the scheme of the path in use), if any.
+    pub(crate) transport: Option<String>,
+    /// Only through a relay (libp2p).
+    pub(crate) relayed: bool,
+    /// The 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,
+}
+
 #[derive(Clone, Serialize, Deserialize)]
 pub(crate) struct HelloRequest {
     /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index b237db2..f8ca586 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -45,6 +45,8 @@ pub(crate) struct ByKind {
     pub(crate) identity: Counter,
     /// PING and PONG (`transport::paths`), counted by the paths layer under the address.
     pub(crate) paths: Counter,
+    /// LINKS and LINKS_REPLY (`swarm::links`).
+    pub(crate) links: Counter,
 }
 
 impl ByKind {
@@ -58,6 +60,7 @@ impl ByKind {
             MessageKind::Blob => &mut self.blob,
             MessageKind::Who | MessageKind::Me => &mut self.identity,
             MessageKind::Ping | MessageKind::Pong => &mut self.paths,
+            MessageKind::Links | MessageKind::LinksReply => &mut self.links,
         }
     }
 }
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 53fd460..3e1ab10 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 = 6;
+pub(crate) const VERSION: u8 = 7;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -29,6 +29,8 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::Me => 7,
         MessageKind::Ping => 8,
         MessageKind::Pong => 9,
+        MessageKind::Links => 10,
+        MessageKind::LinksReply => 11,
     }
 }
 
@@ -67,6 +69,8 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
         ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => {
             rmp_serde::encode::write(&mut payload, nonce)?
         }
+        ParsedMessage::Links(request) => rmp_serde::encode::write(&mut payload, request)?,
+        ParsedMessage::LinksReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
     }
     Ok(payload)
 }
@@ -99,6 +103,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::Ping(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::Pong) {
         ParsedMessage::Pong(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Links) {
+        ParsedMessage::Links(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::LinksReply) {
+        ParsedMessage::LinksReply(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -396,6 +404,23 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
         ParsedMessage::Me(proof),
         ParsedMessage::Ping(7),
         ParsedMessage::Pong(u64::MAX),
+        ParsedMessage::Links(crate::transport::LinksRequest {
+            epoch,
+            epoch_seq: 1,
+            nonce: 3,
+        }),
+        ParsedMessage::LinksReply(crate::transport::LinksReply {
+            epoch,
+            epoch_seq: 1,
+            nonce: 3,
+            links: vec![crate::transport::Link {
+                peer: "p".into(),
+                transport: Some("libp2p".into()),
+                relayed: true,
+                rtt_us: Some(1500),
+                dormant: false,
+            }],
+        }),
     ]
 }
 
@@ -413,7 +438,7 @@ mod tests {
     #[test]
     fn every_kind_round_trips() {
         let samples = samples();
-        assert_eq!(samples.len(), 10, "a kind is missing from `samples`");
+        assert_eq!(samples.len(), 12, "a kind is missing from `samples`");
         for message in samples {
             let bytes = encode(&message).unwrap();
             let decoded = decode(&bytes).unwrap();
@@ -566,7 +591,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&hello).unwrap()),
-            format!("060097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+            format!("070097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
         );
 
         let reply = ParsedMessage::HelloReply(HelloReply {
@@ -586,7 +611,7 @@ mod tests {
         assert_eq!(
             hex(&encode(&reply).unwrap()),
             format!(
-                "060199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
+                "070199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
                 run("04"),
                 run("06")
             )
@@ -601,7 +626,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&status).unwrap()),
-            format!("0603959290c400{epoch_hex}01c3c2")
+            format!("0703959290c400{epoch_hex}01c3c2")
         );
 
         let who = ParsedMessage::Who(IdentityProof {
@@ -610,7 +635,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&who).unwrap()),
-            format!("060692c420{}c440{}", "08".repeat(32), "09".repeat(64))
+            format!("070692c420{}c440{}", "08".repeat(32), "09".repeat(64))
         );
         let ParsedMessage::Me(proof) = decode(
             &encode(&ParsedMessage::Me(IdentityProof {
@@ -624,10 +649,10 @@ mod tests {
         };
         assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64]));
 
-        assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "060807");
+        assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "070807");
         assert_eq!(
             hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()),
-            "0609cfffffffffffffffff"
+            "0709cfffffffffffffffff"
         );
         let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else {
             panic!("not a PONG");