From 7af8f58d46737bb5835aa1625485f22ecf374ebe Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 27 Oct 2026 00:33:03 +0100 Subject: add Transport::delivery_limit --- src/main.rs | 3 ++- src/node.rs | 2 +- src/simtest/mod.rs | 5 +---- src/state/peer.rs | 2 +- src/state/timing.rs | 29 +++++++++++++++++------------ src/transport/firewall.rs | 4 ++++ src/transport/mod.rs | 4 ++++ src/transport/mycelium/mod.rs | 6 ++++++ src/transport/sim.rs | 4 ++++ src/transport/traffic.rs | 4 ++++ 10 files changed, 44 insertions(+), 19 deletions(-) (limited to 'src') 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, - /// `Timing::from_keepalive(--keepalive)`. + /// `Timing::from_keepalive(--keepalive, transport.delivery_limit())`. pub(crate) timing: Timing, pub(crate) neighbours_k: usize, pub(crate) alias: Option, 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 { 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; 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 { 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 { 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 { self.inner.resolve_seed(seed).await } -- cgit v1.3.1