consolidate backoff code

1 files changed, 25 insertions, 20 deletions

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