consolidate backoff code

4 files changed, 99 insertions, 49 deletions

diff --git a/src/backoff.rs b/src/backoff.rs
new file mode 100644
index 0000000..1f87b8d
--- /dev/null
+++ b/src/backoff.rs
@@ -0,0 +1,58 @@
+//! Exponential backoff: a delay that doubles, up to a limit, while something keeps
+//! failing, and starts over when it succeeds. Only the delay; each user keeps its own
+//! point in time (next attempt, deadline).
+
+use std::time::Duration;
+
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct Backoff {
+    initial: Duration,
+    max: Duration,
+    current: Duration,
+}
+
+impl Backoff {
+    pub(crate) const fn new(initial: Duration, max: Duration) -> Self {
+        Self {
+            initial,
+            max,
+            current: initial,
+        }
+    }
+
+    /// The delay to use now.
+    pub(crate) fn current(&self) -> Duration {
+        self.current
+    }
+
+    /// Doubles the delay, up to the limit.
+    pub(crate) fn grow(&mut self) {
+        self.current = self.current.saturating_mul(2).min(self.max);
+    }
+
+    /// The delay to use now; the next call gives twice as much (up to the limit).
+    pub(crate) fn next(&mut self) -> Duration {
+        let delay = self.current;
+        self.grow();
+        delay
+    }
+
+    /// Back to the initial delay.
+    pub(crate) fn reset(&mut self) {
+        self.current = self.initial;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn doubles_up_to_the_limit_and_resets() {
+        let mut backoff = Backoff::new(Duration::from_secs(1), Duration::from_secs(5));
+        let delays: Vec<_> = (0..5).map(|_| backoff.next().as_secs()).collect();
+        assert_eq!(delays, [1, 2, 4, 5, 5]);
+        backoff.reset();
+        assert_eq!(backoff.current(), Duration::from_secs(1));
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index eac3c4c..d5043a8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -19,6 +19,7 @@ use tracing::{debug, error, info, trace, warn};
 use url::Url;
 
 mod api;
+mod backoff;
 mod document;
 mod logging;
 mod remake;
diff --git a/src/state.rs b/src/state.rs
index 49bc1a8..bbda4e4 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -12,6 +12,7 @@ use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
+use crate::backoff::Backoff;
 use crate::document::replica::{Epoch, Replica};
 use crate::remake::{Host, PeerSummary, RemakeState};
 use crate::swarm::sync::{ChangeSet, SyncEvent};
@@ -66,13 +67,11 @@ pub(crate) enum Liveness {
 pub(crate) enum HelloState {
     #[default]
     NotSent,
-    /// A send is in flight. `next_delay` is how long to wait after it before retrying.
-    Sending { next_delay: Duration },
-    /// HELLO was sent (or the send failed); no HELLO_REPLY yet.
-    AwaitingReply {
-        next_attempt: Instant,
-        next_delay: Duration,
-    },
+    /// A send is in flight.
+    Sending,
+    /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at
+    /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant).
+    AwaitingReply { next_attempt: Instant },
     /// HELLO_REPLY received. No more retries.
     Established,
 }
@@ -87,6 +86,11 @@ pub(crate) struct Sending {
     pub(crate) regenerate: bool,
 }
 
+/// First wait for a HELLO_REPLY before sending the HELLO again; doubles while none
+/// comes.
+const HELLO_RETRY: Duration = Duration::from_secs(1);
+const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
+
 /// First wait before generating again after our Mycelium refused a message; doubles
 /// while it keeps refusing.
 const SEND_RETRY: Duration = Duration::from_millis(500);
@@ -100,6 +104,8 @@ pub(crate) struct PeerState {
 
     pub(crate) standing: Standing,
     pub(crate) hello_state: HelloState,
+    /// Wait between unanswered HELLOs; reset whenever a new handshake starts.
+    pub(crate) hello_backoff: Backoff,
 
     /// Everything about the current Automerge sync with this peer; a reset replaces it.
     pub(crate) sync: SyncSession,
@@ -146,12 +152,12 @@ pub(crate) struct SyncSession {
     /// not kept; the next one is generated afresh and carries everything pending.
     pub(crate) retry_at: Option<Instant>,
     /// The wait after the next refusal.
-    pub(crate) retry_delay: Duration,
+    pub(crate) send_backoff: Backoff,
 
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
-    pub(crate) reply_timeout: Duration,
+    pub(crate) reply_backoff: Backoff,
 }
 
 /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
@@ -176,9 +182,9 @@ impl SyncSession {
             state: sync::State::new(),
             sending: None,
             retry_at: None,
-            retry_delay: SEND_RETRY,
+            send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
             reply: None,
-            reply_timeout: SYNC_REPLY_TIMEOUT,
+            reply_backoff: Backoff::new(SYNC_REPLY_TIMEOUT, SYNC_REPLY_TIMEOUT_MAX),
         }
     }
 
@@ -197,7 +203,7 @@ impl SyncSession {
     /// Our Mycelium took the message.
     pub(crate) fn send_succeeded(&mut self) -> bool {
         self.retry_at = None;
-        self.retry_delay = SEND_RETRY;
+        self.send_backoff.reset();
         self.sending
             .take()
             .is_some_and(|sending| sending.regenerate)
@@ -216,9 +222,8 @@ impl SyncSession {
             wait.sent.retain(|hash| !sending.sent.contains(hash));
         }
         self.state.in_flight = false;
-        let delay = self.retry_delay;
+        let delay = self.send_backoff.next();
         self.retry_at = Some(now + delay);
-        self.retry_delay = (delay * 2).min(SEND_RETRY_MAX);
         Some(delay)
     }
 
@@ -229,7 +234,7 @@ impl SyncSession {
             Some(wait) => wait.sent.extend(sent),
             None => {
                 self.reply = Some(SyncReplyWait {
-                    deadline: now + self.reply_timeout,
+                    deadline: now + self.reply_backoff.current(),
                     sent,
                 });
             }
@@ -240,7 +245,7 @@ impl SyncSession {
     /// are the same.
     pub(crate) fn reply_settled(&mut self) {
         self.reply = None;
-        self.reply_timeout = SYNC_REPLY_TIMEOUT;
+        self.reply_backoff.reset();
     }
 
     /// If the answer is overdue: forgets that the unanswered changes were sent and that
@@ -255,7 +260,7 @@ impl SyncSession {
             self.state.sent_hashes.remove(hash);
         }
         self.state.in_flight = false;
-        self.reply_timeout = (self.reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX);
+        self.reply_backoff.grow();
         Some(wait.sent.len())
     }
 }
@@ -266,6 +271,7 @@ impl PeerState {
             leads,
             standing: Standing::default(),
             hello_state: HelloState::default(),
+            hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
             sync: SyncSession::new(),
             liveness: Liveness::Active,
             added_at: Instant::now(),
@@ -307,7 +313,7 @@ impl PeerState {
         if matches!(self.liveness, Liveness::Dormant { .. }) {
             self.liveness = Liveness::Active;
             // Force a fresh handshake; its reply resets sync on both sides.
-            if !matches!(self.hello_state, HelloState::Sending { .. }) {
+            if self.hello_state != HelloState::Sending {
                 self.hello_state = HelloState::NotSent;
             }
             return true;
@@ -324,10 +330,9 @@ impl PeerState {
         self.liveness = Liveness::Dormant { since: now };
         self.reset_sync();
 
-        if !matches!(self.hello_state, HelloState::Sending { .. }) {
+        if self.hello_state != HelloState::Sending {
             self.hello_state = HelloState::AwaitingReply {
                 next_attempt: now + timing.dormant_probe,
-                next_delay: timing.dormant_probe,
             };
         }
         true
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index faa6f91..c8c86ac 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -17,9 +17,6 @@ use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
 use crate::swarm::sync::{SyncEvent, handle_sync};
 use crate::transport::{self, ParsedMessage};
 
-const HELLO_RETRY_INITIAL: Duration = Duration::from_secs(1);
-const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
-
 const PEER_TICK: Duration = Duration::from_millis(500);
 
 const META_ROOT: &str = "_continuum";
@@ -218,9 +215,9 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
     let claimed = {
         let mut peer_state = peer_state_mutex.lock().await;
         if peer_state.hello_state == HelloState::NotSent {
-            peer_state.hello_state = HelloState::Sending {
-                next_delay: HELLO_RETRY_INITIAL,
-            };
+            // A new handshake: retries start over.
+            peer_state.hello_state = HelloState::Sending;
+            peer_state.hello_backoff.reset();
             true
         } else {
             trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending");
@@ -233,7 +230,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
     }
 
     debug!(to = %peer, "Sending HELLO");
-    transmit_hello(state, peer, HELLO_RETRY_INITIAL).await
+    transmit_hello(state, peer).await
 }
 
 async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
@@ -241,32 +238,24 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
         return Ok(());
     };
 
-    let delay = {
+    {
         let mut peer_state = peer_state_mutex.lock().await;
         match peer_state.hello_state {
-            HelloState::AwaitingReply {
-                next_attempt,
-                next_delay,
-            } if Instant::now() >= next_attempt => {
-                peer_state.hello_state = HelloState::Sending { next_delay };
-                next_delay
+            HelloState::AwaitingReply { next_attempt } if Instant::now() >= next_attempt => {
+                peer_state.hello_state = HelloState::Sending;
             }
             _ => return Ok(()),
         }
-    };
+    }
 
-    debug!(
-        peer = %peer,
-        next_retry_ms = delay.as_millis(),
-        "No HELLO_REPLY yet; re-sending HELLO"
-    );
-    transmit_hello(state, peer, delay).await
+    debug!(peer = %peer, "No HELLO_REPLY yet; re-sending HELLO");
+    transmit_hello(state, peer).await
 }
 
 /// Sends a HELLO without holding any lock, then schedules the next retry.
 /// Success and failure both end in AwaitingReply: Mycelium accepting the message
 /// says nothing about whether the peer received it.
-async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> {
+async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
     let swarm_id = state.swarm_id().await;
     let auth = state
         .swarm_key
@@ -289,16 +278,13 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result
     if let Some(peer_state_mutex) = state.get_peer_state(peer).await {
         let mut peer_state = peer_state_mutex.lock().await;
 
-        if matches!(peer_state.hello_state, HelloState::Sending { .. }) {
-            let (wait, next_delay) = match peer_state.liveness {
-                Liveness::Dormant { .. } => {
-                    (state.timing.dormant_probe, state.timing.dormant_probe)
-                }
-                Liveness::Active => (delay, delay.saturating_mul(2).min(HELLO_RETRY_MAX)),
+        if peer_state.hello_state == HelloState::Sending {
+            let wait = match peer_state.liveness {
+                Liveness::Dormant { .. } => state.timing.dormant_probe,
+                Liveness::Active => peer_state.hello_backoff.next(),
             };
             peer_state.hello_state = HelloState::AwaitingReply {
                 next_attempt: Instant::now() + wait,
-                next_delay,
             };
         }
     }