introduce per-transport sync reply timeout

9 files changed, 56 insertions, 13 deletions

diff --git a/src/main.rs b/src/main.rs
index f170434..6c37872 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -188,6 +188,12 @@ async fn main() -> Result<()> {
         .map(|(_, transport)| transport.delivery_limit())
         .max()
         .unwrap_or_default();
+    // An answer may come over any of them.
+    let sync_reply_timeout = transports
+        .iter()
+        .map(|(_, transport)| transport.sync_reply_timeout())
+        .max()
+        .unwrap_or(state::SYNC_REPLY_TIMEOUT);
     let mut transports = transports.into_iter();
     let (scheme, transport) = transports.next().context("No transports")?;
 
@@ -209,7 +215,7 @@ async fn main() -> Result<()> {
         initial_epoch,
         initial_doc,
         earlier_actors,
-        timing: state::Timing::from_keepalive(args.keepalive, delivery_limit)
+        timing: state::Timing::new(args.keepalive, sync_reply_timeout, delivery_limit)
             .with_probe_max(args.dormant_probe_max),
         neighbours_k: args.neighbours,
         alias: args.alias.clone(),
diff --git a/src/node.rs b/src/node.rs
index 844a44b..b334882 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -40,7 +40,8 @@ pub(crate) struct NodeConfig {
     pub(crate) initial_doc: Document,
     /// Actors this node wrote with in earlier runs (`--document-file`).
     pub(crate) earlier_actors: Vec<ActorId>,
-    /// `Timing::from_keepalive(--keepalive, transport.delivery_limit())`.
+    /// `Timing::new(--keepalive, the transports' sync reply timeout and delivery
+    /// limit)`.
     pub(crate) timing: Timing,
     pub(crate) neighbours_k: usize,
     pub(crate) alias: Option<String>,
diff --git a/src/state/mod.rs b/src/state/mod.rs
index a4ea553..cd96aaf 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -24,7 +24,7 @@ mod timing;
 
 pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
 pub(crate) use read::{ReadSource, StaleView};
-pub(crate) use timing::Timing;
+pub(crate) use timing::{SYNC_REPLY_TIMEOUT, Timing};
 
 use std::collections::HashMap;
 use std::sync::Arc;
diff --git a/src/state/timing.rs b/src/state/timing.rs
index 988e430..af5c0c0 100644
--- a/src/state/timing.rs
+++ b/src/state/timing.rs
@@ -4,9 +4,10 @@ use std::time::Duration;
 
 use crate::backoff::Backoff;
 
-/// First wait for a sync answer; generous, since a transport may take a while with large
-/// messages (Mycelium sends them in many small packets).
-const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
+/// First wait for a sync answer, unless a transport needs longer
+/// (`Transport::sync_reply_timeout`); generous, since a transport may take a while with
+/// large messages (Mycelium sends them in many small packets).
+pub(crate) const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
 /// Upper bound, as a multiple of the first wait: 5 minutes, about as long as a transport
 /// tries to deliver a message (`Transport::delivery_limit`).
 const SYNC_REPLY_TIMEOUT_MAX_FACTOR: u32 = 30;
@@ -47,6 +48,12 @@ impl Timing {
         self.keepalive * 5
     }
 
+    /// [`new`](Self::new) with the default first sync reply timeout.
+    #[cfg(test)]
+    pub(crate) fn from_keepalive(keepalive: Duration, delivery_limit: Duration) -> Self {
+        Self::new(keepalive, SYNC_REPLY_TIMEOUT, delivery_limit)
+    }
+
     /// The timings for `--keepalive`.
     ///
     /// Each node keeps about `k` neighbours (`--neighbours`) whatever the swarm's size,
@@ -63,13 +70,9 @@ impl Timing {
     /// | LAN, servers | 15–30 s | a returning or silent peer is noticed within a minute or two |
     /// | WAN, mobile, metered (the default, 1m) | 60–120 s | about 0.4 messages per second per node at `k` = 12 and 1m |
     ///
-    /// `delivery_limit`: the transport's (`Transport::delivery_limit`).
-    pub(crate) fn from_keepalive(keepalive: Duration, delivery_limit: Duration) -> Self {
-        Self::new(keepalive, SYNC_REPLY_TIMEOUT, delivery_limit)
-    }
-
-    /// [`from_keepalive`](Self::from_keepalive) with another first sync reply timeout:
-    /// for tests, whose messages don't travel for seconds.
+    /// `sync_reply_timeout`: the transports' (`Transport::sync_reply_timeout`), or for
+    /// tests, whose messages don't travel for seconds, a shorter one. `delivery_limit`:
+    /// the transports' (`Transport::delivery_limit`).
     pub(crate) fn new(
         keepalive: Duration,
         sync_reply_timeout: Duration,
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 8ef0535..7b4a8c7 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -495,6 +495,10 @@ impl Transport for IdentityTransport {
         self.shared.paths.delivery_limit()
     }
 
+    fn sync_reply_timeout(&self) -> Duration {
+        self.shared.paths.sync_reply_timeout()
+    }
+
     fn local_addresses(&self) -> Vec<String> {
         self.shared.paths.local_addresses()
     }
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index a802269..1de0633 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -768,6 +768,13 @@ pub(crate) trait Transport: Send + Sync {
     /// a message sent is delivered or lost for good.
     fn delivery_limit(&self) -> std::time::Duration;
 
+    /// The first wait for an answer to a sync message (`Timing::sync_reply_timeout`): by
+    /// default `SYNC_REPLY_TIMEOUT`, longer on a transport whose round trips take
+    /// seconds.
+    fn sync_reply_timeout(&self) -> std::time::Duration {
+        crate::state::SYNC_REPLY_TIMEOUT
+    }
+
     /// Where this node can be reached, as addresses other nodes can use: published in
     /// the member list (`swarm::members`). By default its identity, for a transport
     /// whose identities are addresses (Mycelium: a public key).
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 8bb8533..1f88e0a 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -644,6 +644,16 @@ impl Paths {
             .unwrap_or_default()
     }
 
+    /// The longest of the transports' (`Transport::sync_reply_timeout`): an answer may
+    /// come over any of them.
+    pub(crate) fn sync_reply_timeout(&self) -> Duration {
+        self.links
+            .iter()
+            .map(|link| link.inner.sync_reply_timeout())
+            .max()
+            .unwrap_or(crate::state::SYNC_REPLY_TIMEOUT)
+    }
+
     /// Each transport's (`Transport::local_addresses`), with its scheme: taken in turns,
     /// each transport's first in the configured order, then each one's second, and so on,
     /// so a list cut short (`members::MAX_ADDRESSES`) still has every transport's best.
diff --git a/src/transport/tor.rs b/src/transport/tor.rs
index 26dab14..7ecd624 100644
--- a/src/transport/tor.rs
+++ b/src/transport/tor.rs
@@ -76,6 +76,8 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(60);
 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
 /// How long a message may take, before what its size adds (`stream_timeout`).
 const STREAM_TIMEOUT: Duration = Duration::from_secs(30);
+/// The first wait for an answer to a sync message (`Transport::sync_reply_timeout`).
+const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(30);
 /// The slowest a message is expected to go over Tor, in bytes per second.
 const MIN_RATE: u64 = 64 << 10;
 /// Messages read and not handled yet, at most.
@@ -569,6 +571,12 @@ impl Transport for TorTransport {
         stream_timeout(MAX_MESSAGE)
     }
 
+    /// Round trips over Tor take a second or more, and the first message to a peer
+    /// waits for a connection (`CONNECT_TIMEOUT`, `HANDSHAKE_TIMEOUT`).
+    fn sync_reply_timeout(&self) -> Duration {
+        SYNC_REPLY_TIMEOUT
+    }
+
     fn local_addresses(&self) -> Vec<String> {
         vec![self.shared.local_id.clone()]
     }
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 6385f2e..2e45a31 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -306,6 +306,10 @@ impl Transport for CountingTransport {
         self.inner.delivery_limit()
     }
 
+    fn sync_reply_timeout(&self) -> std::time::Duration {
+        self.inner.sync_reply_timeout()
+    }
+
     fn local_addresses(&self) -> Vec<String> {
         self.inner.local_addresses()
     }