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(), diff --git a/src/extensions/client/mod.rs b/src/extensions/client/mod.rs index f3cfc6a..4486fcf 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,21 @@ 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(),