dashboard: swarm-wide epoch shine

1 files changed, 31 insertions, 8 deletions

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(),