aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/dashboard/dashboard.html25
-rw-r--r--src/api/dashboard/mod.rs11
-rw-r--r--src/swarm/links.rs39
-rw-r--r--src/transport/mod.rs4
-rw-r--r--src/transport/wire.rs3
5 files changed, 62 insertions, 20 deletions
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(),
}],
}),