aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/openapi.yaml4
-rw-r--r--src/api/dashboard/mod.rs38
-rw-r--r--src/node.rs3
-rw-r--r--src/swarm/links.rs195
-rw-r--r--src/transport/mod.rs4
-rw-r--r--src/transport/wire.rs15
6 files changed, 232 insertions, 27 deletions
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 664579a..b04c98a 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -548,8 +548,8 @@ paths:
`others` (the peer's other transports: `transport`, `rttMs`, `relayed`), as the member
answered a LINKS; ours always, the others' while fresh). One producer
per node makes it every 2 s while anyone looks (and for a minute after this
- request), asking members for their links every 10 s; every client gets the
- same. Only with the `dashboard` cargo feature.
+ request), subscribed to the members' links (each tells of a change as it
+ happens; renewed every 30 s); every client gets the same. Only with the `dashboard` cargo feature.
responses:
'200':
description: The snapshot.
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index dd454ec..f0cbbfd 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -10,8 +10,9 @@
//! themselves: they alone changing is no change, and the page counts them on.
//!
//! The map: the graph the member list makes (`topology::intended_graph`, as this node
-//! sees it), and the links every member measures, asked for with LINKS every
-//! `LINKS_EVERY` while anyone looks (`swarm::links`), the same answers for all clients.
+//! sees it), and the links every member measures: subscribed to with LINKS while anyone
+//! looks (renewed every `LINKS_EVERY`, ended when nobody does; `swarm::links`), so a
+//! member tells of a change as it happens; the same answers for all clients.
use std::collections::{BTreeMap, BTreeSet};
use std::convert::Infallible;
@@ -34,10 +35,14 @@ use crate::transport::Link;
/// How often the snapshot is built and compared, while anyone looks.
const TICK: Duration = Duration::from_secs(2);
-/// How often members are asked for their links, and the graph recomputed.
-const LINKS_EVERY: Duration = Duration::from_secs(10);
+/// How often the subscriptions to the members' links are renewed (each renewal is
+/// answered: a reply at least this often), and the graph recomputed; changes come in
+/// between, as they happen (`swarm::links`).
+const LINKS_EVERY: Duration = Duration::from_secs(30);
+/// How long a subscription lasts unless renewed: a node that is gone stops it.
+const LINKS_LEASE: Duration = Duration::from_secs(90);
/// A member's links older than this are left out (it didn't answer since).
-const LINKS_KEPT_FOR: Duration = Duration::from_secs(30);
+const LINKS_KEPT_FOR: Duration = Duration::from_secs(90);
/// The producer runs on this long after the last request for the state.
const LINGER: Duration = Duration::from_secs(60);
/// Members asked for their links at most, each round.
@@ -164,14 +169,27 @@ impl Hub {
if let Some(changed) = document_changes(&mut changes) {
let _ = self.documents.send(Arc::new(changed.to_string()));
}
- {
+ let unwanted = || {
+ self.patches.receiver_count() == 0
+ && Instant::now() >= self.control.lock_or_recover().wanted_until
+ };
+ if *self.closing.borrow() {
+ self.control.lock_or_recover().running = false;
+ return;
+ }
+ if unwanted() {
+ // Nobody looks: the members needn't tell us any more. Unless someone
+ // came meanwhile, then it goes on (and subscribes again).
+ for member in self.state.links.subscribed() {
+ links::unsubscribe(&self.state, &member).await;
+ }
let mut control = self.control.lock_or_recover();
- let unwanted =
- self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until;
- if unwanted || *self.closing.borrow() {
+ if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
control.running = false;
return;
}
+ drop(control);
+ graph_due = Instant::now();
}
}
}
@@ -232,7 +250,7 @@ impl Graph {
.take(MAX_ASKED)
{
let state = state.clone();
- tokio::spawn(async move { links::ask(&state, &member).await });
+ tokio::spawn(async move { links::ask(&state, &member, LINKS_LEASE).await });
}
}
}
diff --git a/src/node.rs b/src/node.rs
index be722e6..0b8f663 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -171,6 +171,9 @@ impl Node {
});
trace!("Sync worker task spawned");
+ // Subscribers to our links (`swarm::links`; any build has them).
+ tokio::spawn(crate::swarm::links::watch(state.clone()));
+
let peer_worker_state = state.clone();
tokio::spawn(async move {
peer::run_peer_worker(peer_worker_state).await;
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index 04f2e2d..333fc8f 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -8,6 +8,12 @@
//! 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.
+//!
+//! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
+//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
+//! round trip that moved by a fifth), looked at every `WATCH_EVERY`, until the
+//! subscription runs out (the asker renews it with its next LINKS, so one that is gone
+//! stops it by not renewing) or a LINKS with 0 ends it.
use std::collections::HashMap;
use std::sync::Mutex;
@@ -32,6 +38,13 @@ pub(crate) const MAX_OTHERS: usize = 8;
const ANSWER_EVERY: Duration = Duration::from_secs(5);
/// Replies kept from members at most (one each), and requests out.
const MAX_KEPT: usize = 4096;
+/// Subscriptions a member keeps at most, and how long one lasts at most.
+const MAX_SUBSCRIBERS: usize = 64;
+const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300);
+/// How often a member with subscribers looks whether its links changed.
+const WATCH_EVERY: Duration = Duration::from_secs(2);
+/// A reply to a LINKS that doesn't subscribe is taken for this long.
+const REPLY_WITHIN: Duration = Duration::from_secs(30);
/// The requests out and the replies in, and when each asker was last answered.
#[derive(Default)]
@@ -41,12 +54,17 @@ pub(crate) struct Links {
#[derive(Default)]
struct Inner {
- /// Our LINKS out, by member: its nonce.
- asked: HashMap<String, u64>,
+ /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its
+ /// subscription's end, or `REPLY_WITHIN`).
+ asked: HashMap<String, (u64, Instant)>,
/// 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>,
+ /// 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>>,
}
impl Links {
@@ -67,6 +85,19 @@ impl Links {
})
.collect()
}
+
+ /// The members our subscriptions are with (not run out).
+ #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+ pub(crate) fn subscribed(&self) -> Vec<String> {
+ let now = Instant::now();
+ self.inner
+ .lock_or_recover()
+ .asked
+ .iter()
+ .filter(|(_, (_, until))| now < *until)
+ .map(|(member, _)| member.clone())
+ .collect()
+ }
}
/// This node's links: its neighbours, as it measures them.
@@ -151,22 +182,48 @@ fn rtt_of(destination: Option<&Destination>) -> Option<Duration> {
.map(|ms| Duration::from_secs_f64(ms / 1000.0))
}
-/// Asks `member` for its links (a LINKS); its reply lands in `AppState::links`.
+/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for
+/// that long; its replies land in `AppState::links`.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-pub(crate) async fn ask(state: &AppState, member: &str) {
+pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) {
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 until = Instant::now() + subscribe.max(REPLY_WITHIN);
+ inner.asked.insert(member.to_owned(), (nonce, until));
}
let (epoch, epoch_seq) = state.epoch_stamp();
let request = ParsedMessage::Links(LinksRequest {
epoch,
epoch_seq,
nonce,
+ subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX),
+ });
+ if let Err(error) = state.transport.send_message(member, &request).await {
+ debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
+ }
+}
+
+/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any
+/// more.
+#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+pub(crate) async fn unsubscribe(state: &AppState, member: &str) {
+ let nonce = {
+ let mut inner = state.links.inner.lock_or_recover();
+ match inner.asked.remove(member) {
+ Some((nonce, _)) => nonce,
+ None => return,
+ }
+ };
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ let request = ParsedMessage::Links(LinksRequest {
+ epoch,
+ epoch_seq,
+ nonce,
+ subscribe_secs: 0,
});
if let Err(error) = state.transport.send_message(member, &request).await {
debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
@@ -188,6 +245,17 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest)
{
let now = Instant::now();
let mut inner = state.links.inner.lock_or_recover();
+ // A subscription is taken, renewed or ended even when the reply is due later.
+ if request.subscribe_secs == 0 {
+ inner.subscribers.remove(peer);
+ } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer)
+ {
+ let lasts =
+ Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION);
+ inner
+ .subscribers
+ .insert(peer.to_owned(), (request.nonce, now + lasts));
+ }
if inner
.answered
.get(peer)
@@ -214,13 +282,17 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest)
}
}
-/// A LINKS_REPLY from `peer`: kept if it answers our LINKS to it.
+/// A LINKS_REPLY from `peer`: kept if it answers our latest LINKS to it, while its
+/// replies are taken (a subscription's go on).
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) {
+ let taken = inner
+ .asked
+ .get(peer)
+ .is_some_and(|(nonce, until)| *nonce == reply.nonce && Instant::now() < *until);
+ if !taken {
return;
}
- inner.asked.remove(peer);
reply.links.truncate(MAX_LINKS);
for link in &mut reply.links {
link.others.truncate(MAX_OTHERS);
@@ -232,3 +304,110 @@ pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) {
.replies
.insert(peer.to_owned(), (Instant::now(), reply.links));
}
+
+/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's
+/// links; sent to each of them when they changed since they were last sent. Runs for
+/// the node's life.
+pub(crate) async fn watch(state: AppState) {
+ let mut every = tokio::time::interval(WATCH_EVERY);
+ every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+ loop {
+ every.tick().await;
+ let subscribers: Vec<(String, u64)> = {
+ let now = Instant::now();
+ let mut inner = state.links.inner.lock_or_recover();
+ inner.subscribers.retain(|_, (_, until)| now < *until);
+ if inner.subscribers.is_empty() {
+ inner.published = None;
+ continue;
+ }
+ inner
+ .subscribers
+ .iter()
+ .map(|(peer, (nonce, _))| (peer.clone(), *nonce))
+ .collect()
+ };
+ let links = own_links(&state).await;
+ {
+ let mut inner = state.links.inner.lock_or_recover();
+ if inner
+ .published
+ .as_ref()
+ .is_some_and(|published| !changed(published, &links))
+ {
+ continue;
+ }
+ inner.published = Some(links.clone());
+ }
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ for (peer, nonce) in subscribers {
+ let reply = ParsedMessage::LinksReply(LinksReply {
+ epoch,
+ epoch_seq,
+ nonce,
+ links: links.clone(),
+ });
+ if let Err(error) = state.transport.send_message(&peer, &reply).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
+ }
+ }
+ }
+}
+
+/// Whether `new` says something `old` didn't: another peer, transport, relay or
+/// dormancy, or a round trip measured or moved by more than a fifth (and 1 ms). Round
+/// trips wobble; small moves aren't news.
+fn changed(old: &[Link], new: &[Link]) -> bool {
+ let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
+ (Some(a), Some(b)) => {
+ let by = a.abs_diff(b);
+ by > 1000 && by > a.max(b) / 5
+ }
+ (a, b) => a.is_some() != b.is_some(),
+ };
+ old.len() != new.len()
+ || old.iter().zip(new).any(|(a, b)| {
+ a.peer != b.peer
+ || a.transport != b.transport
+ || a.relayed != b.relayed
+ || a.dormant != b.dormant
+ || moved(a.rtt_us, b.rtt_us)
+ || a.others.len() != b.others.len()
+ || a.others.iter().zip(&b.others).any(|(x, y)| {
+ x.transport != y.transport
+ || x.relayed != y.relayed
+ || moved(x.rtt_us, y.rtt_us)
+ })
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn link(peer: &str, rtt_us: Option<u32>) -> Link {
+ Link {
+ peer: peer.into(),
+ transport: Some("libp2p".into()),
+ relayed: false,
+ rtt_us,
+ dormant: false,
+ others: Vec::new(),
+ }
+ }
+
+ #[test]
+ fn only_news_is_a_change() {
+ let old = [link("a", Some(10_000))];
+ assert!(!changed(&old, &[link("a", Some(11_000))]));
+ assert!(changed(&old, &[link("a", Some(13_000))]));
+ assert!(changed(&old, &[link("a", None)]));
+ assert!(changed(&old, &[link("b", Some(10_000))]));
+ assert!(changed(&old, &[]));
+ let mut dormant = link("a", Some(10_000));
+ dormant.dormant = true;
+ assert!(changed(&old, &[dormant]));
+ // Small round trips: a millisecond's wobble isn't news.
+ assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
+ }
+}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 7fbbd73..903dd07 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -403,6 +403,10 @@ pub(crate) struct LinksRequest {
pub(crate) epoch_seq: u32,
/// Repeated in the reply: which request it answers.
pub(crate) nonce: u64,
+ /// For this long (seconds), a LINKS_REPLY again whenever the links change (a
+ /// subscription; the next LINKS renews it). 0: this reply only, and a subscription
+ /// ends.
+ pub(crate) subscribe_secs: u32,
}
/// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them.
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index dc94ab3..40e32a2 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 = 10;
+pub(crate) const VERSION: u8 = 11;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {
@@ -409,6 +409,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
epoch,
epoch_seq: 1,
nonce: 3,
+ subscribe_secs: 60,
}),
ParsedMessage::LinksReply(crate::transport::LinksReply {
epoch,
@@ -599,7 +600,7 @@ mod tests {
});
assert_eq!(
hex(&encode(&hello).unwrap()),
- format!("0a0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+ format!("0b0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
);
let reply = ParsedMessage::HelloReply(HelloReply {
@@ -619,7 +620,7 @@ mod tests {
assert_eq!(
hex(&encode(&reply).unwrap()),
format!(
- "0a0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
+ "0b0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
run("04"),
run("06")
)
@@ -635,7 +636,7 @@ mod tests {
});
assert_eq!(
hex(&encode(&status).unwrap()),
- format!("0a03969290c400{epoch_hex}01c3c2c2")
+ format!("0b03969290c400{epoch_hex}01c3c2c2")
);
let who = ParsedMessage::Who(IdentityProof {
@@ -644,7 +645,7 @@ mod tests {
});
assert_eq!(
hex(&encode(&who).unwrap()),
- format!("0a0692c420{}c440{}", "08".repeat(32), "09".repeat(64))
+ format!("0b0692c420{}c440{}", "08".repeat(32), "09".repeat(64))
);
let ParsedMessage::Me(proof) = decode(
&encode(&ParsedMessage::Me(IdentityProof {
@@ -658,10 +659,10 @@ mod tests {
};
assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64]));
- assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0a0807");
+ assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0b0807");
assert_eq!(
hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()),
- "0a09cfffffffffffffffff"
+ "0b09cfffffffffffffffff"
);
let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else {
panic!("not a PONG");