slow round back off

5 files changed, 92 insertions, 5 deletions

diff --git a/src/node.rs b/src/node.rs
index b376baa..5bfce31 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -142,6 +142,7 @@ impl Node {
             pulling: Default::default(),
             announce_pending: Default::default(),
             announce_slow: Default::default(),
+            slow_rounds: Default::default(),
             reach: Default::default(),
             trees: Default::default(),
             rtt: Default::default(),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 53d8178..e127877 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -154,6 +154,8 @@ pub(crate) struct AppState {
     /// The slow round after changes passed on along writers' trees is scheduled
     /// (`swarm::relay`).
     pub(crate) announce_slow: Arc<std::sync::Mutex<Option<crate::swarm::relay::Pending>>>,
+    /// How long the next slow round waits.
+    pub(crate) slow_rounds: Arc<std::sync::Mutex<crate::swarm::relay::SlowRounds>>,
     /// Distances in writers' trees (`swarm::tree`).
     pub(crate) trees: Arc<std::sync::Mutex<crate::swarm::tree::Trees>>,
     /// Whom we reach, as told our neighbours (`swarm::reach`).
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 32b35fc..8f79886 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -153,6 +153,8 @@ pub(crate) struct PeerState {
     /// with its epoch: once they are ours, it has every change we would announce to it
     /// (`swarm::relay`).
     pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>,
+    /// When we last told it our heads in a slow round (`swarm::relay::SlowRounds`).
+    pub(crate) slow_announced_at: Option<Instant>,
     /// Whom it said it reaches (`swarm::reach`), by pair key, and when.
     pub(crate) reaches: Option<(Instant, Vec<u64>)>,
     /// Whose change the next sync message to it passes on along the writer's tree
@@ -286,6 +288,7 @@ impl PeerState {
             syncs_with_us: None,
             reported_heads: None,
             last_heads: None,
+            slow_announced_at: None,
             reaches: None,
             tree_writer: None,
             left: false,
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index cfafbd3..53f1aa3 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -190,6 +190,11 @@ pub(crate) async fn handle_message(
             }
         }
         transport::ParsedMessage::StatusReply(remote) => {
+            // An answer to our slow round's announcement: a tree missed.
+            let announced = peer_state.lock().await.slow_announced_at.take();
+            if let Some(at) = announced {
+                crate::swarm::relay::slow_round_answered(state, at.elapsed());
+            }
             on_status(state, &peer, &remote, reactivated).await;
         }
         transport::ParsedMessage::Links(request) => {
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);
+    }
+}