add budget for relayed LINKS

1 files changed, 29 insertions, 1 deletions

diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index cdbae67..4b89e94 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -66,6 +66,11 @@ const MAX_RELAYED: usize = 64;
 pub(crate) const DEFAULT_BATCH: Duration = Duration::from_millis(250);
 const MIN_BATCH: Duration = Duration::from_millis(10);
 const MAX_BATCH: Duration = Duration::from_secs(5);
+/// Members a peer may have us ask for it, at most: as many at once (`RELAY_BURST`), and
+/// on average this many per second (`RELAY_RATE`): a dashboard asking every member
+/// (`MAX_ASKED` there, 512) every 30 s needs less. The rest of a request is dropped.
+const RELAY_BURST: f64 = 512.0;
+const RELAY_RATE: f64 = 32.0;
 /// Nodes a relayed LINKS goes through at most, the first included.
 const MAX_HOPS: u8 = 3;
 
@@ -97,6 +102,9 @@ struct Inner {
     relaying: HashMap<String, HashMap<String, Relaying>>,
     /// Relayed links due to an asker, by asker: the members whose links changed.
     due: HashMap<String, HashSet<String>>,
+    /// How many members each peer may still have us ask for it (`RELAY_BURST`,
+    /// `RELAY_RATE`), as of when.
+    relay_budget: HashMap<String, (f64, Instant)>,
 }
 
 /// One asker's relayed subscription to a member.
@@ -544,7 +552,7 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
     let left = Duration::from_millis(u64::from(request.batch_ms)).min(MAX_BATCH);
     let window = (left / (u32::from(request.hops) + 1)).max(MIN_BATCH);
     let rest = left.saturating_sub(window);
-    let members: Vec<String> = request
+    let mut members: Vec<String> = request
         .of
         .into_iter()
         .filter(|member| member != asker && member != state.local_id())
@@ -552,6 +560,26 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
         .collect();
     let known: Vec<String> = {
         let mut inner = state.links.inner.lock_or_recover();
+        // Within the asker's budget: each member costs one.
+        if inner.relay_budget.len() >= MAX_KEPT && !inner.relay_budget.contains_key(asker) {
+            inner.relay_budget.retain(|_, (budget, at)| {
+                *budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE
+                    < RELAY_BURST
+            });
+        }
+        let (budget, at) = inner
+            .relay_budget
+            .entry(asker.to_owned())
+            .or_insert((RELAY_BURST, now));
+        *budget = (*budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE)
+            .min(RELAY_BURST);
+        *at = now;
+        let allowed = (*budget as usize).min(members.len());
+        *budget -= allowed as f64;
+        if allowed < members.len() {
+            debug!(asker = %asker, asked = members.len(), allowed, "Relayed LINKS over its budget; asking for part of it");
+            members.truncate(allowed);
+        }
         for member in &members {
             if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) {
                 inner