allow links subscriptions

1 files changed, 187 insertions, 8 deletions

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))]));
+    }
+}