Expose per-upstream client timeouts and retries in ClientConfig

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