summaryrefslogtreecommitdiffstats
path: root/src/state.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/state.rs')
-rw-r--r--src/state.rs41
1 files changed, 32 insertions, 9 deletions
diff --git a/src/state.rs b/src/state.rs
index ee7fa7a..e6054ad 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -221,18 +221,20 @@ pub(crate) struct SyncReplyWait {
/// First wait for a sync answer; generous, since Mycelium delivers large messages in
/// many small packets.
const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
-/// Upper bound (Mycelium gives up on a message after five minutes).
-const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
+/// Upper bound, as a multiple of the first wait: 5 minutes (Mycelium gives up on a
+/// message after five minutes).
+const SYNC_REPLY_TIMEOUT_MAX_FACTOR: u32 = 30;
impl SyncSession {
- pub(crate) fn new() -> Self {
+ /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`).
+ pub(crate) fn new(reply_backoff: Backoff) -> Self {
Self {
state: sync::State::new(),
sending: None,
retry_at: None,
send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
reply: None,
- reply_backoff: Backoff::new(SYNC_REPLY_TIMEOUT, SYNC_REPLY_TIMEOUT_MAX),
+ reply_backoff,
}
}
@@ -321,13 +323,13 @@ impl SyncSession {
}
impl PeerState {
- pub(crate) fn new(leads: bool) -> Self {
+ pub(crate) fn new(leads: bool, timing: &Timing) -> Self {
Self {
leads,
standing: Standing::default(),
hello_state: HelloState::default(),
hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
- sync: SyncSession::new(),
+ sync: SyncSession::new(timing.sync_reply_backoff()),
liveness: Liveness::Active,
added_at: Instant::now(),
last_seen: None,
@@ -350,7 +352,10 @@ impl PeerState {
/// Start Automerge sync with this peer from scratch. A message still being handed to
/// Mycelium was generated against the old state; its outcome is then ignored.
pub(crate) fn reset_sync(&mut self) {
- self.sync = SyncSession::new();
+ // The same waits for answers, from the start.
+ let mut reply_backoff = self.sync.reply_backoff;
+ reply_backoff.reset();
+ self.sync = SyncSession::new(reply_backoff);
}
/// An overdue sync answer from an established, active peer (see
@@ -525,6 +530,9 @@ pub(crate) struct Timing {
pub(crate) dormant_probe: Duration,
/// While catching up: PING the source after this much silence (`swarm::source`).
pub(crate) catch_up_probe: Duration,
+ /// First wait for an answer to a sync message before sending again; doubles while
+ /// answers stay out (`SyncSession::reply_backoff`).
+ pub(crate) sync_reply_timeout: Duration,
}
impl Timing {
@@ -542,18 +550,33 @@ impl Timing {
// WAN, mobile or metered links, or 100+ nodes | 60-120 s | At 100 nodes and 30 s it’s about 6–7 messages/s per node. Fine, but rarely worth it on battery or metered links.
// Several hundred nodes or more | - | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the swarm, which calls for partial views (SWIM-style random probing and syncing with a subset of peers).
pub(crate) fn from_keepalive(keepalive: Duration) -> Self {
+ Self::new(keepalive, SYNC_REPLY_TIMEOUT)
+ }
+
+ /// [`from_keepalive`](Self::from_keepalive) with another first sync reply timeout:
+ /// for tests, whose messages don't travel for seconds.
+ pub(crate) fn new(keepalive: Duration, sync_reply_timeout: Duration) -> Self {
Self {
keepalive,
dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row
// >= 2 keepalives, so a second exchange confirms the difference; and >= twice
// the first sync reply timeout, so a lost message's sender resends it first.
- resync_after: (keepalive * 2).max(SYNC_REPLY_TIMEOUT * 2),
+ resync_after: (keepalive * 2).max(sync_reply_timeout * 2),
dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes
// Seconds, not a keepalive interval: a dead source should cost little time.
// Only the source gets these, and only while catching up.
catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)),
+ sync_reply_timeout,
}
}
+
+ /// Waits for sync answers: `sync_reply_timeout`, doubling up to 30 times as much.
+ pub(crate) fn sync_reply_backoff(&self) -> Backoff {
+ Backoff::new(
+ self.sync_reply_timeout,
+ self.sync_reply_timeout * SYNC_REPLY_TIMEOUT_MAX_FACTOR,
+ )
+ }
}
#[derive(Clone)]
@@ -808,7 +831,7 @@ impl AppState {
/// State for a peer we haven't known so far (not inserted).
pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
let leads = crate::swarm::pair::leads(self.local_id(), pk);
- Arc::new(Mutex::new(PeerState::new(leads)))
+ Arc::new(Mutex::new(PeerState::new(leads, &self.timing)))
}
pub(crate) fn notify_store_changed(&self) {