treewide: Add configurable per-stream fetch timeout

This introduces a `fetch_timeout` setting to the node configuration,
replacing the hardcoded constant. It defines the duration a fetch stream
is allowed to stall before being aborted. The default remains 30s, but
users on anonymizing transports like Tor can now increase it to avoid
spurious timeouts during network congestion.
This commit is contained in:
zebus
2026-07-17 17:05:11 +02:00
committed by Lorenz Leutgeb
parent 027de89cdb
commit 66ab7f35c9
6 changed files with 60 additions and 6 deletions
+5
View File
@@ -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
+13 -2
View File
@@ -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": [
+6 -4
View File
@@ -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}");
+2
View File
@@ -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())
}
}
+33
View File
@@ -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<FetchTimeout> for LocalDuration {
fn from(t: FetchTimeout) -> Self {
t.0
}
}
impl From<LocalDuration> for FetchTimeout {
fn from(d: LocalDuration) -> Self {
Self(d)
}
}
/// Connection limits.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(default, rename_all = "camelCase")]
+1
View File
@@ -914,6 +914,7 @@ mod test {
"routingMaxAge": 604800,
"gossipMaxAge": 604800,
"fetchConcurrency": 1,
"fetchTimeout": 30,
"maxOpenFiles": 4096,
"rate": {
"inbound": { "fillRate": 10.0, "capacity": 2048 },