make sync reply timeout part of timing

This enables upcoming lossy-links tests.

4 files changed, 42 insertions, 16 deletions

diff --git a/src/main.rs b/src/main.rs
index 6d2329f..e494497 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -167,7 +167,7 @@ async fn main() -> Result<()> {
         initial_epoch,
         initial_doc,
         earlier_actors,
-        keepalive: Duration::from_secs(args.keepalive_secs),
+        timing: state::Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)),
         neighbours_k: args.neighbours,
         alias: args.alias.clone(),
         electorate_window: Duration::from_secs(args.electorate_window_days * 24 * 3600),
diff --git a/src/node.rs b/src/node.rs
index 82b99fc..8a49c56 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -33,7 +33,8 @@ pub(crate) struct NodeConfig {
     pub(crate) initial_doc: AutoCommit,
     /// Actors this node wrote with in earlier runs (`--document-file`).
     pub(crate) earlier_actors: Vec<ActorId>,
-    pub(crate) keepalive: Duration,
+    /// `Timing::from_keepalive(--keepalive)`.
+    pub(crate) timing: Timing,
     pub(crate) neighbours_k: usize,
     pub(crate) alias: Option<String>,
     pub(crate) electorate_window: Duration,
@@ -89,7 +90,7 @@ impl Node {
             change_tx,
             initial_sync_done: Arc::new(AtomicBool::new(false)),
             change_events,
-            timing: Timing::from_keepalive(config.keepalive),
+            timing: config.timing,
             remake: Arc::new(remake::RemakeState::default()),
             swarm_key: config.swarm_key,
             run_id: uuid::Uuid::new_v4().into_bytes(),
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 2e33310..f0fee02 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -24,14 +24,16 @@ use crate::document::replica::Epoch;
 use crate::document::store;
 use crate::files::Files;
 use crate::node::{self, Node, NodeConfig};
-use crate::state::{AppState, PeerState};
+use crate::state::{AppState, PeerState, Timing};
 use crate::swarm::bootstrap::{self, Start};
 use crate::transport::sim::SimNet;
 
 /// Keepalive of simulated nodes: dormancy after 1 s, probes of dormant peers every
-/// 2.5 s (`Timing::from_keepalive`). Sync reply timeouts and heals don't scale with it
-/// (10 s and 20 s at least).
+/// 2.5 s (`Timing::new`).
 pub(crate) const KEEPALIVE: Duration = Duration::from_millis(250);
+/// First wait for a sync answer before sending again (10 s in production, where large
+/// messages take a while): a lost sync message costs a test 1 s, a heal 2 s.
+pub(crate) const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(1);
 
 /// A simulated swarm: the net and the nodes started on it.
 pub(crate) struct Sim {
@@ -69,7 +71,7 @@ impl Sim {
             initial_epoch: Epoch::GENESIS,
             initial_doc: AutoCommit::new(),
             earlier_actors: Vec::new(),
-            keepalive: KEEPALIVE,
+            timing: Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT),
             neighbours_k: 8,
             alias: Some(id.to_owned()),
             electorate_window: Duration::from_secs(30 * 24 * 3600),
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) {