diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/links.rs | 39 |
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(), |