From e1b5e3c19dc1779ef247003ba5c3854344e7d1b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexandre=20Bald=C3=A9?= Date: Wed, 20 May 2026 02:06:43 +0000 Subject: [PATCH 1/3] Expose per-upstream client timeouts and retries in `ClientConfig` `Client::new` already accepts `request_timeout`, `connection_timeout`, and `retries` arguments, but `from_config` hardcodes all three to `None` because `ClientConfig` only exposes `endpoints` and `shuffle_endpoints`. As a result the only way to override the 30s per-upstream request timeout (and the 30s connection timeout, and the default retry count) is to construct `Client` directly in Rust, which isn't reachable from the YAML-driven config. Adds three optional fields to `ClientConfig`: - `request_timeout_seconds` - `connection_timeout_seconds` - `retries` `from_config` plumbs them into `Client::new`. None of the existing defaults change when the fields are omitted. The motivating case is heavy storage queries against slow public RPCs (Acala under load is the case that surfaced this in `polkadot-fellows/runtimes#1180` / `open-web3-stack/polkadot-ecosystem-tests#621`) where 30s per upstream is not enough and Subway exhausts its endpoint cycle without serving a response. --- src/extensions/client/mod.rs | 23 +++++++++++++++++++---- src/server.rs | 3 +++ src/tests/merge_subscription.rs | 3 +++ src/tests/upstream.rs | 3 +++ 4 files changed, 28 insertions(+), 4 deletions(-) diff --git a/src/extensions/client/mod.rs b/src/extensions/client/mod.rs index f3cfc6a..225225d 100644 --- a/src/extensions/client/mod.rs +++ b/src/extensions/client/mod.rs @@ -57,6 +57,18 @@ pub struct ClientConfig { pub endpoints: Vec, #[serde(default = "bool_true")] pub shuffle_endpoints: bool, + /// Per-upstream request timeout in seconds. Defaults to 30s if unset. + /// Increase when upstream chains may take longer than 30s to respond to + /// heavy storage queries. + #[serde(default)] + pub request_timeout_seconds: Option, + /// Per-upstream connection (handshake) timeout in seconds. Defaults to 30s if unset. + #[serde(default)] + pub connection_timeout_seconds: Option, + /// Number of retries to apply per upstream call before rotating endpoints. + /// Defaults to the internal client default if unset. + #[serde(default)] + pub retries: Option, } fn validate_endpoint(endpoint: &str, _context: &()) -> garde::Result { @@ -134,13 +146,16 @@ impl Extension for Client { type Config = ClientConfig; async fn from_config(config: &Self::Config, _registry: &ExtensionRegistry) -> Result { - if config.shuffle_endpoints { + let request_timeout = config.request_timeout_seconds.map(Duration::from_secs); + let connection_timeout = config.connection_timeout_seconds.map(Duration::from_secs); + let endpoints = if config.shuffle_endpoints { let mut endpoints = config.endpoints.clone(); endpoints.shuffle(&mut thread_rng()); - Ok(Self::new(endpoints, None, None, None)?) + endpoints } else { - Ok(Self::new(config.endpoints.clone(), None, None, None)?) - } + config.endpoints.clone() + }; + Ok(Self::new(endpoints, request_timeout, connection_timeout, config.retries)?) } } diff --git a/src/server.rs b/src/server.rs index 4fccd1c..a1c259c 100644 --- a/src/server.rs +++ b/src/server.rs @@ -254,6 +254,9 @@ mod tests { client: Some(ClientConfig { endpoints: vec![endpoint], shuffle_endpoints: false, + request_timeout_seconds: None, + connection_timeout_seconds: None, + retries: None, }), server: Some(ServerConfig { listen_address: "127.0.0.1".to_string(), diff --git a/src/tests/merge_subscription.rs b/src/tests/merge_subscription.rs index 558d820..7748a9c 100644 --- a/src/tests/merge_subscription.rs +++ b/src/tests/merge_subscription.rs @@ -49,6 +49,9 @@ async fn merge_subscription_works() { client: Some(ClientConfig { endpoints: vec![format!("ws://{addr}")], shuffle_endpoints: false, + request_timeout_seconds: None, + connection_timeout_seconds: None, + retries: None, }), server: Some(ServerConfig { listen_address: "0.0.0.0".to_string(), diff --git a/src/tests/upstream.rs b/src/tests/upstream.rs index 2afc10e..71e5ec4 100644 --- a/src/tests/upstream.rs +++ b/src/tests/upstream.rs @@ -31,6 +31,9 @@ async fn upstream_error_propagate() { client: Some(ClientConfig { endpoints: vec![format!("ws://{addr}")], shuffle_endpoints: false, + request_timeout_seconds: None, + connection_timeout_seconds: None, + retries: None, }), server: Some(ServerConfig { listen_address: "0.0.0.0".to_string(), From b2e071e3a6f2c54b45c60e304bc6cfc7f32082c4 Mon Sep 17 00:00:00 2001 From: Bryan Chen Date: Wed, 20 May 2026 16:14:17 +1200 Subject: [PATCH 2/3] cargo fmt --- src/extensions/client/mod.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/extensions/client/mod.rs b/src/extensions/client/mod.rs index 225225d..4486fcf 100644 --- a/src/extensions/client/mod.rs +++ b/src/extensions/client/mod.rs @@ -155,7 +155,12 @@ impl Extension for Client { } else { config.endpoints.clone() }; - Ok(Self::new(endpoints, request_timeout, connection_timeout, config.retries)?) + Ok(Self::new( + endpoints, + request_timeout, + connection_timeout, + config.retries, + )?) } } From e5000a2033801097bedf5bb6802f07b01657891c Mon Sep 17 00:00:00 2001 From: Bryan Chen Date: Wed, 20 May 2026 16:23:52 +1200 Subject: [PATCH 3/3] feat(bench): Add client config options for connection timeout, request timeout, and retries --- benches/bench/main.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/benches/bench/main.rs b/benches/bench/main.rs index 81f9651..4745590 100644 --- a/benches/bench/main.rs +++ b/benches/bench/main.rs @@ -217,6 +217,9 @@ fn config() -> Config { format!("ws://{}", SERVER_TWO_ENDPOINT), ], shuffle_endpoints: false, + connection_timeout_seconds: None, + request_timeout_seconds: None, + retries: None, }), server: Some(ServerConfig { listen_address: SUBWAY_SERVER_ADDR.to_string(),