diff --git a/CHANGELOG.md b/CHANGELOG.md index 621da8679..5f9141a81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,6 +72,11 @@ required, but is left to future work. - `rad patch list` includes a `Labels` column, showing any assigned labels to the patch. +- The node can now be configured with a per-stream timeout. A new configuration + key under `node.limits.fetchTimeout` is added and the default value remains 30 + seconds. This default is suitable for TCP connections. However on the Tor and + I2P transports, which are higher latency, a larger timeout avoids spurious + aborts during tunnel rebuilds or congestion. ## Fixed Bugs diff --git a/crates/radicle-cli/examples/rad-config.md b/crates/radicle-cli/examples/rad-config.md index c129289fe..ac51b5bfd 100644 --- a/crates/radicle-cli/examples/rad-config.md +++ b/crates/radicle-cli/examples/rad-config.md @@ -42,7 +42,8 @@ $ rad config "inbound": 128, "outbound": 16 }, - "fetchPackReceive": "500.0 MiB" + "fetchPackReceive": "500.0 MiB", + "fetchTimeout": 30 }, "workers": 8, "seedingPolicy": { @@ -286,7 +287,8 @@ $ rad config schema "inbound": 128, "outbound": 16 }, - "fetchPackReceive": "500.0 MiB" + "fetchPackReceive": "500.0 MiB", + "fetchTimeout": 30 } }, "workers": { @@ -552,6 +554,11 @@ $ rad config schema "description": "Channel limits.", "$ref": "#/$defs/FetchPackSizeLimit", "default": "500.0 MiB" + }, + "fetchTimeout": { + "description": "How long a fetch stream is allowed to stall before being aborted.\n\nThis is a per-stream inactivity timeout, not a cap on total fetch\nduration: if no bytes flow between the peers for longer than this, the\nfetch is aborted. The default value is suitable for direct TCP\nconnections; on high-latency anonymizing transports (Tor, I2P) a larger\nvalue avoids spurious aborts during tunnel rebuilds or congestion.", + "$ref": "#/$defs/FetchTimeout", + "default": 30 } } }, @@ -634,6 +641,10 @@ $ rad config schema "10 MB" ] }, + "FetchTimeout": { + "description": "The duration value provided to each fetch being performed on a node.", + "$ref": "#/$defs/LocalDuration" + }, "DefaultSeedingPolicy": { "description": "Default seeding policy. Applies when no repository policies for the given repo are found.", "oneOf": [ diff --git a/crates/radicle-node/src/wire.rs b/crates/radicle-node/src/wire.rs index 2691626fd..4fbf66d7d 100644 --- a/crates/radicle-node/src/wire.rs +++ b/crates/radicle-node/src/wire.rs @@ -13,7 +13,7 @@ use cyphernet::addr::{HostName, InetHost, NetAddr}; use cyphernet::encrypt::noise::{HandshakePattern, Keyset, NoiseState}; use cyphernet::proxy::socks5; use cyphernet::{Digest, EcSk, Ecdh, Sha256}; -use localtime::LocalTime; +use localtime::{LocalDuration, LocalTime}; use mio::net::TcpStream; use radicle::node::device::Device; @@ -35,7 +35,6 @@ use crate::reactor::{Listener, Transport}; use crate::reactor::{NoiseSession, ProtocolArtifact, SessionEvent, Socks5Session}; use crate::reactor::{Token, Tokens}; use crate::service; -use crate::service::FETCH_TIMEOUT; use crate::service::io::Io; use crate::service::{DisconnectReason, Metrics, Service, session}; use crate::worker; @@ -716,10 +715,13 @@ where log::debug!(target: "wire", "Received `open` command for stream {stream} from {nid}"); metrics.streams_opened += 1; metrics.received_fetch_requests += 1; - let reader_limit = self.service.config().limits.fetch_pack_receive; + let limits = &self.service.config().limits; + let reader_limit = limits.fetch_pack_receive; + let fetch_timeout: time::Duration = + LocalDuration::from(limits.fetch_timeout).into(); let Some(channels) = streams.register( stream, - ChannelsConfig::new(FETCH_TIMEOUT) + ChannelsConfig::new(fetch_timeout) .with_reader_limit(reader_limit), ) else { log::debug!(target: "wire", "Peer attempted to open already-open stream {stream}"); diff --git a/crates/radicle-protocol/src/service.rs b/crates/radicle-protocol/src/service.rs index 8530de5f8..5f541cd2f 100644 --- a/crates/radicle-protocol/src/service.rs +++ b/crates/radicle-protocol/src/service.rs @@ -2664,7 +2664,9 @@ where } fn fetch_config(&self) -> fetcher::FetchConfig { + let timeout: LocalDuration = self.config.limits.fetch_timeout.into(); fetcher::FetchConfig::default() + .with_timeout(timeout.into()) .with_minimum_feature_level(self.config.fetch.feature_level_min()) } } diff --git a/crates/radicle/src/node/config.rs b/crates/radicle/src/node/config.rs index 3d7ab9a3c..b9d488f82 100644 --- a/crates/radicle/src/node/config.rs +++ b/crates/radicle/src/node/config.rs @@ -146,6 +146,15 @@ pub struct Limits { /// Channel limits. pub fetch_pack_receive: FetchPackSizeLimit, + + /// How long a fetch stream is allowed to stall before being aborted. + /// + /// This is a per-stream inactivity timeout, not a cap on total fetch + /// duration: if no bytes flow between the peers for longer than this, the + /// fetch is aborted. The default value is suitable for direct TCP + /// connections; on high-latency anonymizing transports (Tor, I2P) a larger + /// value avoids spurious aborts during tunnel rebuilds or congestion. + pub fetch_timeout: FetchTimeout, } /// Limiter for byte streams. @@ -238,6 +247,30 @@ impl Default for FetchPackSizeLimit { } } +/// The duration value provided to each fetch being performed on a node. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, Eq, PartialEq)] +#[serde(transparent)] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +pub struct FetchTimeout(LocalDuration); + +impl Default for FetchTimeout { + fn default() -> Self { + Self(LocalDuration::from_secs(30)) + } +} + +impl From for LocalDuration { + fn from(t: FetchTimeout) -> Self { + t.0 + } +} + +impl From for FetchTimeout { + fn from(d: LocalDuration) -> Self { + Self(d) + } +} + /// Connection limits. #[derive(Debug, Clone, Serialize, Deserialize, Default)] #[serde(default, rename_all = "camelCase")] diff --git a/crates/radicle/src/profile.rs b/crates/radicle/src/profile.rs index b70867e8d..64f286856 100644 --- a/crates/radicle/src/profile.rs +++ b/crates/radicle/src/profile.rs @@ -914,6 +914,7 @@ mod test { "routingMaxAge": 604800, "gossipMaxAge": 604800, "fetchConcurrency": 1, + "fetchTimeout": 30, "maxOpenFiles": 4096, "rate": { "inbound": { "fillRate": 10.0, "capacity": 2048 },