make sync reply timeout part of timing

This enables upcoming lossy-links tests.

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) {