add Transport::delivery_limit

10 files changed, 44 insertions, 19 deletions

diff --git a/src/main.rs b/src/main.rs
index 76086ca..dade8bc 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -159,6 +159,7 @@ async fn main() -> Result<()> {
 
     let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?);
     debug!(local_id = mycelium_transport.local_id(), "Transport ready");
+    let delivery_limit = mycelium_transport.delivery_limit();
 
     let document_file = args.document_file.clone().map(|path| {
         Arc::new(document::file::DocumentFile::new(
@@ -173,7 +174,7 @@ async fn main() -> Result<()> {
         initial_epoch,
         initial_doc,
         earlier_actors,
-        timing: state::Timing::from_keepalive(args.keepalive),
+        timing: state::Timing::from_keepalive(args.keepalive, delivery_limit),
         neighbours_k: args.neighbours,
         alias: args.alias.clone(),
         electorate_window: args.electorate_window,
diff --git a/src/node.rs b/src/node.rs
index 945f5e7..9330bb6 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -34,7 +34,7 @@ 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>,
-    /// `Timing::from_keepalive(--keepalive)`.
+    /// `Timing::from_keepalive(--keepalive, transport.delivery_limit())`.
     pub(crate) timing: Timing,
     pub(crate) neighbours_k: usize,
     pub(crate) alias: Option<String>,
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 04b5396..531d1c6 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -78,10 +78,7 @@ impl Sim {
             initial_epoch: Epoch::GENESIS,
             initial_doc: AutoCommit::new(),
             earlier_actors: Vec::new(),
-            timing: Timing {
-                delivery_limit: DELIVERY_LIMIT,
-                ..Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT)
-            },
+            timing: Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT),
             neighbours_k: 8,
             alias: Some(id.to_owned()),
             electorate_window: Duration::from_secs(30 * 24 * 3600),
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 344bcdb..92329b2 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -382,7 +382,7 @@ mod tests {
     /// probes back off meanwhile.
     #[test]
     fn unreachable_until_heard_from() {
-        let timing = Timing::from_keepalive(Duration::from_secs(60));
+        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
         let mut peer = PeerState::new(true, &timing);
         let start = Instant::now();
         for _ in 0..UNREACHABLE_AFTER_HELLOS {
diff --git a/src/state/timing.rs b/src/state/timing.rs
index 3c4afe1..0dab3da 100644
--- a/src/state/timing.rs
+++ b/src/state/timing.rs
@@ -4,18 +4,16 @@ use std::time::Duration;
 
 use crate::backoff::Backoff;
 
-/// First wait for a sync answer; generous, since Mycelium delivers large messages in
-/// many small packets.
+/// 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);
-/// Upper bound, as a multiple of the first wait: 5 minutes (Mycelium gives up on a
-/// message after five minutes).
+/// 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;
 /// Upper bound on the wait between probes of an unreachable peer, as a multiple of
 /// `dormant_probe`: an hour at the default keepalive (1m). A returning node HELLOs its own
 /// neighbours, so this only bounds how long one we never hear from stays out.
 const PROBE_MAX_FACTOR: u32 = 6;
-/// How long Mycelium tries to deliver a message before dropping it.
-const MYCELIUM_DELIVERY_LIMIT: Duration = Duration::from_secs(300);
 
 #[derive(Clone, Copy, Debug)]
 pub(crate) struct Timing {
@@ -28,8 +26,9 @@ pub(crate) struct Timing {
     /// 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,
-    /// How long the transport tries to deliver a message before dropping it (Mycelium:
-    /// five minutes): a peer silent for longer has nothing on the way to us any more.
+    /// How long the transport tries to deliver a message before dropping it
+    /// (`Transport::delivery_limit`): a peer silent for longer has nothing on the way to
+    /// us any more.
     pub(crate) delivery_limit: Duration,
 }
 
@@ -54,13 +53,19 @@ impl Timing {
     /// | Tests | 250 ms – 2 s | dormancy within seconds keeps tests short |
     /// | 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 |
-    pub(crate) fn from_keepalive(keepalive: Duration) -> Self {
-        Self::new(keepalive, SYNC_REPLY_TIMEOUT)
+    ///
+    /// `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.
-    pub(crate) fn new(keepalive: Duration, sync_reply_timeout: Duration) -> Self {
+    pub(crate) fn new(
+        keepalive: Duration,
+        sync_reply_timeout: Duration,
+        delivery_limit: Duration,
+    ) -> Self {
         Self {
             keepalive,
             dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row
@@ -72,7 +77,7 @@ impl Timing {
             // 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,
-            delivery_limit: MYCELIUM_DELIVERY_LIMIT,
+            delivery_limit,
         }
     }
 
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index 2cc1dba..df82aa7 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -68,6 +68,10 @@ impl Transport for FirewallTransport {
         self.inner.is_peer_id(id)
     }
 
+    fn delivery_limit(&self) -> std::time::Duration {
+        self.inner.delivery_limit()
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index febbdf0..d97c51c 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -486,6 +486,10 @@ pub(crate) trait Transport: Send + Sync {
     /// key). Identities other peers name to us (in a HELLO_REPLY) are checked with it.
     fn is_peer_id(&self, id: &str) -> bool;
 
+    /// How long the transport tries to deliver a message before dropping it: after that,
+    /// a message sent is delivered or lost for good.
+    fn delivery_limit(&self) -> std::time::Duration;
+
     /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's
     /// identity, as `local_id` gives it on that node.
     async fn resolve_seed(&self, seed: &str) -> Result<String>;
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 6304819..17344c7 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -15,6 +15,8 @@ pub(crate) mod daemon;
 pub(crate) use daemon::{MyceliumSetup, MyceliumSpawnConfig};
 
 const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
+/// How long Mycelium tries to deliver a message before dropping it.
+const DELIVERY_LIMIT: Duration = Duration::from_secs(300);
 /// The largest inbound message read (its JSON, base64 payload included: about 190 MiB of
 /// payload). Anyone can send us anything; this bounds what one message costs before it
 /// is even parsed. Generous, because a sync message can carry a whole history until
@@ -188,6 +190,10 @@ impl Transport for MyceliumTransport {
         is_pubkey(id)
     }
 
+    fn delivery_limit(&self) -> Duration {
+        DELIVERY_LIMIT
+    }
+
     async fn resolve_seed(&self, hint: &str) -> Result<String> {
         if is_pubkey(hint) {
             return Ok(hint.to_owned());
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 05160f8..149798b 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -260,6 +260,10 @@ impl Transport for SimTransport {
         !id.is_empty()
     }
 
+    fn delivery_limit(&self) -> std::time::Duration {
+        crate::simtest::DELIVERY_LIMIT
+    }
+
     /// A seed is a node ID; one no node has doesn't resolve (yet).
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         if self.net.is_known(seed) {
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 6c3d683..0d93525 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -141,6 +141,10 @@ impl Transport for CountingTransport {
         self.inner.is_peer_id(id)
     }
 
+    fn delivery_limit(&self) -> std::time::Duration {
+        self.inner.delivery_limit()
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }