slow round back off

1 files changed, 81 insertions, 5 deletions

diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 98d4ca0..58196d3 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -26,9 +26,9 @@
 //! go at once to the receiver's children in the writer's tree (`swarm::tree`). Then the
 //! tree carries them, and the round after `ANNOUNCE_DELAY` goes only to the neighbours
 //! the tree leaves out (`Plan::uncovered`: no member they reach comes before them). The
-//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`): in case their parent missed
-//! the change (it went away, or the nodes' views of the tree differed), coalescing all
-//! that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that
+//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`, longer while nobody lacks
+//! anything when told: `SlowRounds`): in case their parent missed the change (it went
+//! away, or the nodes' views of the tree differed), coalescing all that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that
 //! has no part in the writer's tree) are announced to all, as before.
 //!
 //! In a full mesh too: the member list makes every member the writer's neighbour, but
@@ -69,8 +69,63 @@ const ANNOUNCE_JITTER: Duration = Duration::from_millis(200);
 
 /// How long the slow round after changes passed on along writers' trees waits: what a
 /// tree misses arrives this much later, and while changes keep coming, every neighbour
-/// is told at most this often.
+/// is told at most this often. Doubles with every slow round that nobody lacked
+/// anything of (`SlowRounds`), up to `SLOW_ANNOUNCE_MAX`.
 const SLOW_ANNOUNCE_DELAY: Duration = Duration::from_secs(2);
+const SLOW_ANNOUNCE_MAX: Duration = Duration::from_secs(16);
+
+/// A STATUS_REPLY this soon after our slow round's announcement to the peer answers it.
+const SLOW_ANSWER_WITHIN: Duration = Duration::from_secs(5);
+
+/// How long the slow round waits. Since a neighbour answers an announcement only if it
+/// lacks something (`swarm::peer::inbound`), an answer to the slow round means a tree
+/// missed: then it is back to `SLOW_ANNOUNCE_DELAY`. While nobody answers, the trees
+/// work, and the round waits twice as long each time, up to `SLOW_ANNOUNCE_MAX`: it is
+/// a safety net, and most of its announcements go to neighbours that have everything.
+/// (A lost message along a tree is noticed sooner by the next one, which its receiver
+/// can't apply and asks after; the slow round is for the last of a run of writes, a
+/// parent gone away, views of the tree that differ.)
+pub(crate) struct SlowRounds {
+    delay: Duration,
+    /// An answer came since the last slow round was scheduled.
+    answered: bool,
+}
+
+impl Default for SlowRounds {
+    fn default() -> Self {
+        // The first round is the short one.
+        Self {
+            delay: SLOW_ANNOUNCE_DELAY,
+            answered: true,
+        }
+    }
+}
+
+impl SlowRounds {
+    /// The wait for the slow round being scheduled now: as long as the last if it was
+    /// answered (or the first), else twice that (up to the most).
+    fn next(&mut self) -> Duration {
+        self.delay = if std::mem::take(&mut self.answered) {
+            SLOW_ANNOUNCE_DELAY
+        } else {
+            (self.delay * 2).min(SLOW_ANNOUNCE_MAX)
+        };
+        self.delay
+    }
+}
+
+/// A STATUS_REPLY from `peer`, which we told of our heads in a slow round
+/// `announced_ago`: it lacked something, so a tree missed; the next slow round waits
+/// `SLOW_ANNOUNCE_DELAY` again.
+pub(crate) fn slow_round_answered(state: &AppState, announced_ago: Duration) {
+    if announced_ago < SLOW_ANSWER_WITHIN {
+        let mut rounds = state.slow_rounds.lock_or_recover();
+        if !rounds.answered {
+            debug!("A slow round's announcement was answered; next one in 2 s");
+        }
+        rounds.answered = true;
+    }
+}
 
 /// An announcement round scheduled: for changes from these peers, and the neighbours
 /// that were sent them along a writer's tree meanwhile (`swarm::tree`).
@@ -139,7 +194,8 @@ pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unau
         schedule
     };
     if schedule {
-        round(state, &state.announce_slow, SLOW_ANNOUNCE_DELAY, true);
+        let delay = state.slow_rounds.lock_or_recover().next();
+        round(state, &state.announce_slow, delay, true);
     }
 }
 
@@ -177,6 +233,9 @@ async fn announce_now(state: &AppState, pending: &Pending, slow: bool) {
     }
     state.stats.announce_round(pending.all, slow);
     for peer in neighbours {
+        if slow && let Some(peer_state) = state.get_peer_state(&peer).await {
+            peer_state.lock().await.slow_announced_at = Some(Instant::now());
+        }
         if let Err(error) = announce_heads(state, &peer, false).await {
             debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
         }
@@ -383,3 +442,20 @@ async fn pull_next(state: &AppState) {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The slow round waits twice as long after each one nobody answered, up to the
+    /// most, and the short time again after an answer.
+    #[test]
+    fn slow_rounds_back_off_while_unanswered() {
+        let mut rounds = SlowRounds::default();
+        let waits: Vec<u64> = (0..6).map(|_| rounds.next().as_secs()).collect();
+        assert_eq!(waits, [2, 4, 8, 16, 16, 16]);
+        rounds.answered = true;
+        assert_eq!(rounds.next(), SLOW_ANNOUNCE_DELAY);
+        assert_eq!(rounds.next(), SLOW_ANNOUNCE_DELAY * 2);
+    }
+}