dashboard: swarm-wide epoch shine

7 files changed, 71 insertions, 27 deletions

diff --git a/docs/dashboard.md b/docs/dashboard.md
index 8ce8374..b7247ab 100644
--- a/docs/dashboard.md
+++ b/docs/dashboard.md
@@ -77,8 +77,10 @@ Every member of the swarm, by alias (or a shortened node ID):
   allow. The members report their peers' clocks with their links, so this works for
   members this node doesn't reach itself; it needs three members compared.
 * **A golden glow**, growing from it and fading, once: it just entered a new epoch (a
-  remake, [`remakes.md`](remakes.md)). For this node and its peers, whose epochs this
-  node knows, as they change while the page is open.
+  remake, [`remakes.md`](remakes.md)). For every member that reports its links (they
+  say its epoch), and this node's peers, as they change while the page is open: so
+  after a remake, a wave across the map as the members switch. With `links: every
+  30 s`, up to that late.
 
 ### Links
 
@@ -345,9 +347,9 @@ another version of it (the state's `page`).
 **The map's graphs.** The intended one is the neighbours the member list gives every
 member (`swarm::topology`), as this node computes them. The measured one is every
 member's links as it reports them in reply to a LINKS (sent to every member, at most
-512, while the producer runs). The LINKS subscribes: the member replies again whenever
-its links change (a peer, a transport, relayed, dormant, or a round trip moved by a
-fifth; it looks every 2 s), for 90 s, renewed every 30 s (each renewal answered), and
+512, while the producer runs), with the epoch it is in. The LINKS subscribes: the
+member replies again whenever its links change (a peer, a transport, relayed, dormant,
+or a round trip moved by a fifth; it looks every 2 s) or it enters a new epoch, for 90 s, renewed every 30 s (each renewal answered), and
 ended when the producer stops; a node that is gone stops it by not renewing. A reply
 older than 90 s is left out. A member keeps 64 subscribers at most. With `links: every
 30 s` (`POST /admin/dashboard/links?live=false`; `true` back), the subscriptions end
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index f45fae7..677af66 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -752,8 +752,8 @@ paths:
         `/admin/stats`), `peers` (as `/peers`) and `graph`: `members` (by node ID,
         those of the member list, and any a link reported goes to: `alias`, `self`,
         `dashboard`: its `--dashboard-url`), `intended` (by node ID: the neighbours the member list gives
-        it, as this node computes them), `measured` (by node ID: `secsAgo` and
-        `links`, each with `peer`, `transport`, `relayed`, `down`, `rttMs`, `dormant` and
+        it, as this node computes them), `measured` (by node ID: `secsAgo`, `epoch`
+        (the epoch the member said it is in: `generation`, `id`) and `links`, each with `peer`, `transport`, `relayed`, `down`, `rttMs`, `dormant` and
         `others` (the peer's other transports: `transport`, `rttMs`, `relayed`), as the member
         answered a LINKS; ours always, the others' while fresh) and `live` (whether the
         members' links are subscribed to, `/admin/dashboard/links`). One producer
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 1bd2a19..1f27544 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -2017,21 +2017,28 @@ function openDashboard(id) {
   requestAnimationFrame(step);
 }
 // Nodes that entered a new epoch (a remake: `remakes.md`) shine: a golden halo grows
-// from the node and fades, as a breath, larger. Known for this node and its peers
-// (`/peers`: `localEpoch`, each peer's `epoch`); others' epochs the page doesn't see.
-const SHINE_MS = 1600;
+// from the node and fades, as a breath, larger. Their epochs come from `/peers` (this
+// node's, `localEpoch`, and each peer's) and from the links members report
+// (`graph.measured`, each with the epoch its member said it is in), so for every member
+// that reports. Only a later generation than the page knew shines, and only from a
+// report of the last two minutes: a late or older report doesn't make it shine again.
+const SHINE_MS = 1600, SHINE_FRESH_SECS = 120;
 const knownEpochs = new Map();
 function epochsArrived() {
   if (!state) return;
-  const current = new Map();
-  if (state.peers.localEpoch) current.set(state.node.id, state.peers.localEpoch.id);
+  const seen = [];
+  if (state.peers.localEpoch) seen.push([state.node.id, state.peers.localEpoch, 0]);
   for (const [id, peer] of Object.entries(state.peers.peers)) {
-    if (peer.epoch) current.set(id, peer.epoch.id);
+    if (peer.epoch) seen.push([id, peer.epoch, 0]);
+  }
+  for (const [id, report] of Object.entries(state.graph.measured)) {
+    if (report.epoch) seen.push([id, report.epoch, report.secsAgo || 0]);
   }
-  for (const [id, epoch] of current) {
+  for (const [id, epoch, secsAgo] of seen) {
     const was = knownEpochs.get(id);
-    if (was !== undefined && was !== epoch) shine(id);
-    knownEpochs.set(id, epoch);
+    if (was !== undefined && epoch.generation <= was) continue;
+    if (was !== undefined && secsAgo <= SHINE_FRESH_SECS) shine(id);
+    knownEpochs.set(id, epoch.generation);
   }
 }
 function shine(id) {
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 6682cbd..c82ce76 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -340,15 +340,22 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue {
         .into_iter()
         .filter(|(_, age, _)| *age <= LINKS_KEPT_FOR)
         .collect();
+    // With each its epoch: the page shows who entered a new one.
+    let epochs = state.links.epochs();
+    let epoch_json = |epoch: Option<crate::transport::Epoch>| {
+        epoch.map(|epoch| json!(crate::document::replica::EpochRecord::from(epoch)))
+    };
     let mut measured = Map::new();
     measured.insert(
         local.to_owned(),
-        json!({ "secsAgo": 0, "links": links_json(&own) }),
+        json!({ "secsAgo": 0, "links": links_json(&own),
+            "epoch": epoch_json(Some(state.local_epoch())) }),
     );
     for (member, age, member_links) in &replies {
         measured.insert(
             member.clone(),
-            json!({ "secsAgo": age.as_secs(), "links": links_json(member_links) }),
+            json!({ "secsAgo": age.as_secs(), "links": links_json(member_links),
+                "epoch": epoch_json(epochs.get(member).copied()) }),
         );
     }
     let clocks = clocks::swarm(
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index 54f18c6..2d9071b 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -47,7 +47,8 @@ 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, RelayedLinks,
+    Connection, Destination, Epoch, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage,
+    RelayedLinks,
 };
 
 /// Links one reply lists at most (a member has about `--neighbours`, more if others
@@ -107,12 +108,15 @@ struct Inner {
     asked: HashMap<String, (u64, Instant)>,
     /// The latest LINKS_REPLY of each member: when, and its links.
     replies: HashMap<String, (Instant, Vec<Link>)>,
+    /// Each member's epoch, as its latest LINKS_REPLY said (`LinksReply::epoch`).
+    epochs: HashMap<String, Epoch>,
     /// When we last answered each member.
     answered: HashMap<String, Instant>,
     /// Who subscribed to our links: the nonce to answer with, and until when.
     subscribers: HashMap<String, (u64, Instant)>,
     /// The links subscribers were last sent (`watch`).
-    published: Option<Vec<Link>>,
+    /// And in which epoch: a new one is news too (a dashboard shows who entered it).
+    published: Option<(Epoch, 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>)>,
@@ -169,6 +173,12 @@ impl Links {
             .collect()
     }
 
+    /// The epoch each member was in, as its latest links said (`replies`).
+    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+    pub(crate) fn epochs(&self) -> HashMap<String, Epoch> {
+        self.inner.lock_or_recover().epochs.clone()
+    }
+
     /// Who is subscribed to our links (not run out).
     #[cfg(test)]
     pub(crate) fn subscribers(&self) -> Vec<String> {
@@ -814,6 +824,9 @@ async fn pass_on(state: &AppState, asker: &str) {
             let Some((at, links)) = inner.replies.get(&member) else {
                 continue;
             };
+            let Some(epoch) = inner.epochs.get(&member).copied() else {
+                continue;
+            };
             let age_ms =
                 u32::try_from(now.saturating_duration_since(*at).as_millis()).unwrap_or(u32::MAX);
             by_nonce
@@ -822,6 +835,7 @@ async fn pass_on(state: &AppState, asker: &str) {
                 .push(RelayedLinks {
                     member: member.clone(),
                     age_ms,
+                    epoch,
                     links: links.clone(),
                 });
             if relaying.once
@@ -885,13 +899,14 @@ pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
             .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>| {
+        let mut keep = |member: String, age: Duration, epoch: Epoch, 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();
+                inner.epochs.clear();
             }
             let at = now.checked_sub(age).unwrap_or(now);
             // Arrived, if older news: the way works.
@@ -909,14 +924,15 @@ pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
             // passed on, it would go round for good.
             let news = known.is_none_or(|(known_at, known_links)| {
                 *known_links != links || now.saturating_duration_since(*known_at) >= REPLY_WITHIN
-            });
+            }) || inner.epochs.get(&member) != Some(&epoch);
             inner.replies.insert(member.clone(), (at, links));
+            inner.epochs.insert(member.clone(), epoch);
             if news {
                 arrived.push(member);
             }
         };
         if own {
-            keep(peer.to_owned(), Duration::ZERO, reply.links);
+            keep(peer.to_owned(), Duration::ZERO, reply.epoch, reply.links);
         }
         if let Some(members) = via {
             for relayed in reply.relayed.into_iter().take(MAX_RELAYED) {
@@ -924,6 +940,7 @@ pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
                     keep(
                         relayed.member,
                         Duration::from_millis(u64::from(relayed.age_ms)),
+                        relayed.epoch,
                         relayed.links,
                     );
                 }
@@ -964,18 +981,20 @@ pub(crate) async fn watch(state: AppState) {
             continue;
         }
         let links = own_links(&state).await;
+        let (epoch, epoch_seq) = state.epoch_stamp();
         {
             let mut inner = state.links.inner.lock_or_recover();
             if inner
                 .published
                 .as_ref()
-                .is_some_and(|published| !changed(published, &links))
+                .is_some_and(|(published_in, published)| {
+                    *published_in == epoch && !changed(published, &links)
+                })
             {
                 continue;
             }
-            inner.published = Some(links.clone());
+            inner.published = Some((epoch, links.clone()));
         }
-        let (epoch, epoch_seq) = state.epoch_stamp();
         for (peer, nonce) in subscribers {
             let reply = ParsedMessage::LinksReply(LinksReply {
                 epoch,
@@ -1036,6 +1055,10 @@ mod tests {
         let relayed = |links: usize| RelayedLinks {
             member: "m".into(),
             age_ms: 0,
+            epoch: Epoch {
+                generation: 1,
+                id: [0; 32],
+            },
             links: (0..links)
                 .map(|i| link(&format!("peer-{i:0>70}"), Some(1500)))
                 .collect(),
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 0f009cb..04e0506 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -453,6 +453,10 @@ pub(crate) struct RelayedLinks {
     pub(crate) member: String,
     /// How long ago the member reported them (milliseconds).
     pub(crate) age_ms: u32,
+    /// The member's epoch, as its LINKS_REPLY said (`LinksReply::epoch`): a dashboard
+    /// shows who entered a new one, swarm-wide.
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
     pub(crate) links: Vec<Link>,
 }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index b3a8ee9..475c7f2 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 = 27;
+pub(crate) const VERSION: u8 = 28;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -462,6 +462,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             relayed: vec![crate::transport::RelayedLinks {
                 member: "q".into(),
                 age_ms: 1200,
+                epoch,
                 links: Vec::new(),
             }],
         }),