Expose per-upstream client timeouts and retries in ClientConfig (#203)

* 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.

* cargo fmt

* feat(bench): Add client config options for connection timeout, request timeout, and retries

---------

Co-authored-by: Bryan Chen <xlchen1291@gmail.com>

Author: rockbmb

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(),

src/extensions/client/mod.rs

@@ -57,6 +57,18 @@ pub struct ClientConfig {
     pub endpoints: Vec<String>,
     #[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<u64>,
+    /// Per-upstream connection (handshake) timeout in seconds. Defaults to 30s if unset.
+    #[serde(default)]
+    pub connection_timeout_seconds: Option<u64>,
+    /// Number of retries to apply per upstream call before rotating endpoints.
+    /// Defaults to the internal client default if unset.
+    #[serde(default)]
+    pub retries: Option<u32>,
 }
 
 fn validate_endpoint(endpoint: &str, _context: &()) -> garde::Result {
@@ -134,13 +146,21 @@ impl Extension for Client {
     type Config = ClientConfig;
 
     async fn from_config(config: &Self::Config, _registry: &ExtensionRegistry) -> Result<Self, anyhow::Error> {
-        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,
+        )?)
     }
 }
 

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(),

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(),

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(),