diff options
| -rw-r--r-- | src/backoff.rs | 58 | ||||
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 45 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 44 |
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, }; } } |