diff --git a/Cargo.lock b/Cargo.lock index 2572227f0c7..9be49d70ca3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6417,9 +6417,11 @@ dependencies = [ "futures", "getrandom 0.2.17", "gloo-timers", + "h2", "hex", "http", "http-serde", + "hyper-util", "lru", "rand 0.8.6", "serde", @@ -6428,6 +6430,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tonic-web-wasm-client", + "tower 0.5.3", "tracing", "wasm-bindgen-futures", ] diff --git a/packages/rs-dapi-client/Cargo.toml b/packages/rs-dapi-client/Cargo.toml index da97e5b4f98..36043a61304 100644 --- a/packages/rs-dapi-client/Cargo.toml +++ b/packages/rs-dapi-client/Cargo.toml @@ -68,7 +68,11 @@ serde_json = { version = "1.0.140", optional = true } chrono = { version = "0.4.38", features = ["serde"] } [dev-dependencies] -tokio = { version = "1.40", features = ["macros"] } +tokio = { version = "1.40", features = ["macros", "rt", "test-util", "io-util"] } +# In-memory HTTP/2 server and connector for the stalled-response-body tests. +h2 = "0.4" +hyper-util = { version = "0.1", features = ["tokio"] } +tower = { version = "0.5", features = ["util"] } [package.metadata.cargo-machete] diff --git a/packages/rs-dapi-client/src/connection_pool.rs b/packages/rs-dapi-client/src/connection_pool.rs index b8e7b74775a..df2399b8cd7 100644 --- a/packages/rs-dapi-client/src/connection_pool.rs +++ b/packages/rs-dapi-client/src/connection_pool.rs @@ -17,7 +17,30 @@ use crate::{ /// Cloning the pool will create a new reference to the same pool. #[derive(Debug, Clone)] pub struct ConnectionPool { - inner: Arc>>, + inner: Arc>, +} + +#[derive(Debug)] +struct PoolState { + connections: LruCache, + /// Generation of the connection pooled most recently. + generation: u64, +} + +/// A pooled connection and the generation it was pooled at. +#[derive(Debug)] +struct Pooled { + generation: u64, + item: PoolItem, +} + +/// Identity of a pooled connection: the client type, the node, and the +/// connection-affecting settings (`None` when none were given). +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct PoolKey { + prefix: PoolPrefix, + uri: String, + connection: Option, } impl ConnectionPool { @@ -29,9 +52,10 @@ impl ConnectionPool { /// Panics if the capacity is zero. pub fn new(capacity: usize) -> Self { Self { - inner: Arc::new(Mutex::new(LruCache::new( - capacity.try_into().expect("must be non-zero"), - ))), + inner: Arc::new(Mutex::new(PoolState { + connections: LruCache::new(capacity.try_into().expect("must be non-zero")), + generation: 0, + })), } } } @@ -56,7 +80,12 @@ impl ConnectionPool { settings: Option<&AppliedRequestSettings>, ) -> Option { let key = Self::key(prefix, uri, settings); - self.inner.lock().expect("must lock").get(&key).cloned() + self.inner + .lock() + .expect("must lock") + .connections + .get(&key) + .map(|pooled| pooled.item.clone()) } /// Get value from cache or create it using provided closure. @@ -74,38 +103,110 @@ impl ConnectionPool { settings: Option<&AppliedRequestSettings>, create: impl FnOnce() -> Result, ) -> Result { - if let Some(cli) = self.get(prefix, uri, settings) { - return Ok(cli); - } + self.get_or_create_with_generation(prefix, uri, settings, create) + .map(|(item, _)| item) + } - let cli = create(); - if let Ok(cli) = &cli { - self.put(uri, settings, cli.clone()); + /// Like [ConnectionPool::get_or_create], and also returns the generation + /// the returned connection was pooled at (see [ConnectionPool::generation]). + /// + /// The generation is read under the same lock that finds or stores the + /// connection, so it is the returned connection's own even when other + /// threads replace it right afterwards. + pub fn get_or_create_with_generation( + &self, + prefix: PoolPrefix, + uri: &Uri, + settings: Option<&AppliedRequestSettings>, + create: impl FnOnce() -> Result, + ) -> Result<(PoolItem, u64), E> { + let key = Self::key(prefix, uri, settings); + let cached = self + .inner + .lock() + .expect("must lock") + .connections + .get(&key) + .map(|pooled| (pooled.item.clone(), pooled.generation)); + if let Some(cached) = cached { + return Ok(cached); } - cli + + let item = create()?; + let generation = self.put_with_generation(uri, settings, item.clone()); + Ok((item, generation)) } /// Put item into the pool for the given uri and settings. pub fn put(&self, uri: &Uri, settings: Option<&AppliedRequestSettings>, value: PoolItem) { + self.put_with_generation(uri, settings, value); + } + + /// Put item into the pool and return the generation it was pooled at. + fn put_with_generation( + &self, + uri: &Uri, + settings: Option<&AppliedRequestSettings>, + value: PoolItem, + ) -> u64 { let key = Self::key(&value, uri, settings); - self.inner.lock().expect("must lock").put(key, value); + let mut state = self.inner.lock().expect("must lock"); + state.generation += 1; + let generation = state.generation; + state.connections.put( + key, + Pooled { + generation, + item: value, + }, + ); + generation + } + + /// Generation of the connection pooled most recently. Every connection + /// put into the pool afterwards gets a higher generation. + pub fn generation(&self) -> u64 { + self.inner.lock().expect("must lock").generation + } + + /// Drop every connection to `uri` pooled at or before `generation`, + /// whatever its prefix and connection settings. + /// + /// A request that misses its deadline may have been sent over a half-open + /// connection: the network path died after the request left, and nothing + /// on the idle channel would ever notice. Keeping it pooled would stall + /// the next request sent to the same node, so the executor evicts it and + /// the next request dials a fresh connection. + /// + /// Connections pooled after `generation` stay. They were dialed after the + /// timed-out attempt took its connection, for example by a concurrent + /// request that already timed out on the same node and reconnected. + pub fn remove_uri(&self, uri: &Uri, generation: u64) { + let uri = uri.to_string(); + let mut state = self.inner.lock().expect("must lock"); + let stale: Vec = state + .connections + .iter() + .filter(|(key, pooled)| key.uri == uri && pooled.generation <= generation) + .map(|(key, _)| key.clone()) + .collect(); + for key in stale { + state.connections.pop(&key); + } } fn key>( class: C, uri: &Uri, settings: Option<&AppliedRequestSettings>, - ) -> String { - let prefix: PoolPrefix = class.into(); + ) -> PoolKey { // Only connection-affecting settings participate in the key (see // `AppliedRequestSettings::connection_key`), so requests differing only // in per-request knobs (timeout, retries, banning) share a connection. - // The settings segment is always present (and contains no `:`), so the - // two branches cannot produce colliding shapes even for a URI whose - // path mimics a key fragment. - match settings { - Some(settings) => format!("{}:{}:{}", prefix, uri, settings.connection_key()), - None => format!("{}:{}:none", prefix, uri), + PoolKey { + prefix: class.into(), + uri: uri.to_string(), + connection: settings.map(AppliedRequestSettings::connection_key), } } } @@ -161,6 +262,7 @@ impl From for CoreGrpcClient { } /// Prefix for the item in the pool. Used to distinguish between Core and Platform clients. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum PoolPrefix { Core, Platform, @@ -399,6 +501,106 @@ mod tests { assert!(result.is_some()); } + #[tokio::test] + async fn should_remove_every_pooled_connection_to_an_uri() { + let pool = ConnectionPool::new(10); + let uri = test_uri(); + let longer_port = Uri::from_str("http://127.0.0.1:30001").unwrap(); + let connect_timeout = RequestSettings { + connect_timeout: Some(Duration::from_secs(3)), + ..RequestSettings::default() + } + .finalize(); + + pool.put(&uri, None, make_platform_pool_item()); + pool.put(&uri, None, make_core_pool_item()); + pool.put(&uri, Some(&connect_timeout), make_platform_pool_item()); + pool.put(&longer_port, None, make_platform_pool_item()); + + pool.remove_uri(&uri, pool.generation()); + + assert!(pool.get(PoolPrefix::Platform, &uri, None).is_none()); + assert!(pool.get(PoolPrefix::Core, &uri, None).is_none()); + assert!(pool + .get(PoolPrefix::Platform, &uri, Some(&connect_timeout)) + .is_none()); + assert!( + pool.get(PoolPrefix::Platform, &longer_port, None).is_some(), + "a URI that merely starts with the evicted one must stay pooled" + ); + } + + #[tokio::test] + async fn should_remove_only_the_exact_uri_when_another_extends_it_past_a_colon() { + let cases = [ + ("http://node", "http://node:443"), + ("http://node/grpc", "http://node/grpc:8080"), + ]; + for (evicted, kept) in cases { + let pool = ConnectionPool::new(10); + let evicted = Uri::from_str(evicted).unwrap(); + let kept = Uri::from_str(kept).unwrap(); + pool.put(&evicted, None, make_platform_pool_item()); + pool.put(&kept, None, make_platform_pool_item()); + + pool.remove_uri(&evicted, pool.generation()); + + assert!( + pool.get(PoolPrefix::Platform, &evicted, None).is_none(), + "{evicted} must be evicted" + ); + assert!( + pool.get(PoolPrefix::Platform, &kept, None).is_some(), + "{kept} must stay pooled when {evicted} is evicted" + ); + } + } + + #[tokio::test] + async fn should_keep_connections_pooled_after_the_given_generation() { + let pool = ConnectionPool::new(10); + let uri = test_uri(); + pool.put(&uri, None, make_platform_pool_item()); + let used_by_attempt = pool.generation(); + // Replaced after the attempt took its connection. + pool.put(&uri, None, make_platform_pool_item()); + + pool.remove_uri(&uri, used_by_attempt); + assert!( + pool.get(PoolPrefix::Platform, &uri, None).is_some(), + "a connection pooled after the given generation must stay" + ); + + pool.remove_uri(&uri, pool.generation()); + assert!(pool.get(PoolPrefix::Platform, &uri, None).is_none()); + } + + #[tokio::test] + async fn should_return_the_generation_of_the_connection_it_takes() { + let pool = ConnectionPool::new(10); + let uri = test_uri(); + + let (_, created) = pool + .get_or_create_with_generation(PoolPrefix::Platform, &uri, None, || { + Ok::<_, String>(make_platform_pool_item()) + }) + .unwrap(); + assert_eq!(created, pool.generation()); + + let (_, taken) = pool + .get_or_create_with_generation(PoolPrefix::Platform, &uri, None, || { + Err("the pooled connection must be reused".to_string()) + }) + .unwrap(); + pool.put(&uri, None, make_platform_pool_item()); + + assert_eq!(taken, created); + assert!( + taken < pool.generation(), + "a replacement pooled afterwards must get a higher generation" + ); + } + #[tokio::test] async fn test_connection_pool_clone_shares_data() { let pool = ConnectionPool::new(10); diff --git a/packages/rs-dapi-client/src/dapi_client.rs b/packages/rs-dapi-client/src/dapi_client.rs index 6569b1ba4a3..8c883a6fb46 100644 --- a/packages/rs-dapi-client/src/dapi_client.rs +++ b/packages/rs-dapi-client/src/dapi_client.rs @@ -4,6 +4,8 @@ use dapi_grpc::mock::Mockable; use dapi_grpc::tonic::async_trait; #[cfg(not(target_arch = "wasm32"))] use dapi_grpc::tonic::transport::Certificate; +#[cfg(not(target_arch = "wasm32"))] +use dapi_grpc::tonic::Status; use std::fmt::{Debug, Display}; use std::time::Duration; use tracing::Instrument; @@ -709,6 +711,325 @@ mod tests { let display = format!("{}", err); assert!(display.contains("address list error")); } + + /// Executor-level coverage for evicting the pooled connection of a node + /// whose attempt missed its deadline. + #[cfg(not(target_arch = "wasm32"))] + mod deadline_pool_eviction { + use super::*; + use crate::connection_pool::{PoolItem, PoolPrefix}; + use crate::transport::{BoxFuture, PlatformGrpcClient}; + use crate::Uri; + use dapi_grpc::tonic::transport::Channel; + use dapi_grpc::tonic::Code; + use std::sync::{Arc, Mutex}; + + /// Takes its connection from the executor's pool, as the real gRPC + /// clients do, so the pool holds an entry for every node dialed. + struct PooledClient { + uri: Uri, + } + + impl PooledClient { + fn pooled( + uri: Uri, + settings: Option<&AppliedRequestSettings>, + pool: &ConnectionPool, + ) -> Result { + pool.get_or_create(PoolPrefix::Platform, &uri, settings, || { + Ok::<_, TransportError>(PoolItem::Platform(PlatformGrpcClient::new( + Channel::builder(uri.clone()).connect_lazy(), + ))) + })?; + Ok(Self { uri }) + } + } + + impl TransportClient for PooledClient { + fn with_uri(uri: Uri, pool: &ConnectionPool) -> Result { + Self::pooled(uri, None, pool) + } + + fn with_uri_and_settings( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result { + Self::pooled(uri, Some(settings), pool) + } + } + + #[derive(Debug)] + struct Pong; + + impl Mockable for Pong {} + + /// Never answers on the first node it is sent to; answers at once + /// everywhere else. + #[derive(Clone, Debug, Default)] + struct StallFirstRequest { + stalled: Arc>>, + } + + impl Mockable for StallFirstRequest {} + + impl TransportRequest for StallFirstRequest { + type Client = PooledClient; + type Response = Pong; + + const SETTINGS_OVERRIDES: RequestSettings = RequestSettings::default(); + + fn method_name(&self) -> &'static str { + "stall_first" + } + + fn execute_transport<'c>( + self, + client: &'c mut Self::Client, + _settings: &AppliedRequestSettings, + ) -> BoxFuture<'c, Result> { + let mut stalled = self.stalled.lock().expect("stall lock"); + let target = stalled.get_or_insert_with(|| client.uri.clone()); + if *target == client.uri { + Box::pin(futures::future::pending()) + } else { + Box::pin(async { Ok(Pong) }) + } + } + } + + #[tokio::test(start_paused = true)] + async fn should_evict_the_pooled_connection_of_a_node_that_missed_its_deadline() { + let request = StallFirstRequest::default(); + let client = DapiClient::new( + "http://127.0.0.1:10001,http://127.0.0.1:10002" + .parse() + .expect("valid address list"), + RequestSettings::default(), + ); + + let response = client + .execute(request.clone(), RequestSettings::default()) + .await + .expect("the other node must answer"); + + let stalled = request + .stalled + .lock() + .expect("stall lock") + .clone() + .expect("a node stalled"); + // The executor's applied settings for these defaults (no CA + // certificate) produce the same pool key. + let settings = RequestSettings::default().finalize(); + assert!( + client + .pool + .get(PoolPrefix::Platform, &stalled, Some(&settings)) + .is_none(), + "the stalled node's connection must be evicted" + ); + assert!( + client + .pool + .get( + PoolPrefix::Platform, + response.address.uri(), + Some(&settings) + ) + .is_some(), + "the healthy node's connection must stay pooled" + ); + } + + /// Replaces its node's pooled connection, as a concurrent request + /// that timed out on the node and reconnected would, then never + /// answers. + #[derive(Clone, Debug)] + struct ReplaceConnectionThenStall { + pool: ConnectionPool, + } + + impl Mockable for ReplaceConnectionThenStall {} + + impl TransportRequest for ReplaceConnectionThenStall { + type Client = PooledClient; + type Response = Pong; + + const SETTINGS_OVERRIDES: RequestSettings = RequestSettings::default(); + + fn method_name(&self) -> &'static str { + "replace_connection_then_stall" + } + + fn execute_transport<'c>( + self, + client: &'c mut Self::Client, + settings: &AppliedRequestSettings, + ) -> BoxFuture<'c, Result> { + self.pool.put( + &client.uri, + Some(settings), + PoolItem::Platform(PlatformGrpcClient::new( + Channel::builder(client.uri.clone()).connect_lazy(), + )), + ); + Box::pin(futures::future::pending()) + } + } + + #[tokio::test(start_paused = true)] + async fn should_keep_a_connection_pooled_after_the_attempt_took_its_own() { + let client = DapiClient::new( + "http://127.0.0.1:10001" + .parse() + .expect("valid address list"), + RequestSettings { + retries: Some(0), + ..RequestSettings::default() + }, + ); + let request = ReplaceConnectionThenStall { + pool: client.pool.clone(), + }; + + let error = client + .execute(request, RequestSettings::default()) + .await + .expect_err("the only node never answers"); + + assert!( + matches!( + &error.inner, + DapiClientError::Transport(TransportError::Grpc(status)) + if status.code() == Code::DeadlineExceeded + ), + "expected DeadlineExceeded, got {:?}", + error.inner + ); + let uri = error.address.expect("the attempted node").uri().clone(); + let settings = RequestSettings::default().finalize(); + assert!( + client + .pool + .get(PoolPrefix::Platform, &uri, Some(&settings)) + .is_some(), + "a connection pooled after the attempt took its own must stay pooled" + ); + } + + /// Takes its node's pooled connection, which another worker replaces + /// before the constructor returns. + struct ReplacedWhileBuildingClient; + + impl ReplacedWhileBuildingClient { + fn build( + uri: Uri, + settings: Option<&AppliedRequestSettings>, + pool: &ConnectionPool, + ) -> Result<(Self, u64), TransportError> { + let connect = || { + Ok::<_, TransportError>(PoolItem::Platform(PlatformGrpcClient::new( + Channel::builder(uri.clone()).connect_lazy(), + ))) + }; + let (_, generation) = pool.get_or_create_with_generation( + PoolPrefix::Platform, + &uri, + settings, + connect, + )?; + // What a concurrent request that timed out and reconnected + // would pool. + pool.put(&uri, settings, connect()?); + Ok((Self, generation)) + } + } + + impl TransportClient for ReplacedWhileBuildingClient { + fn with_uri(uri: Uri, pool: &ConnectionPool) -> Result { + Self::build(uri, None, pool).map(|(client, _)| client) + } + + fn with_uri_and_settings( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result { + Self::build(uri, Some(settings), pool).map(|(client, _)| client) + } + + fn with_uri_and_settings_and_generation( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result<(Self, u64), TransportError> { + Self::build(uri, Some(settings), pool) + } + } + + /// Never answers. + #[derive(Clone, Debug)] + struct StallOnReplacedConnection; + + impl Mockable for StallOnReplacedConnection {} + + impl TransportRequest for StallOnReplacedConnection { + type Client = ReplacedWhileBuildingClient; + type Response = Pong; + + const SETTINGS_OVERRIDES: RequestSettings = RequestSettings::default(); + + fn method_name(&self) -> &'static str { + "stall_on_replaced_connection" + } + + fn execute_transport<'c>( + self, + _client: &'c mut Self::Client, + _settings: &AppliedRequestSettings, + ) -> BoxFuture<'c, Result> { + Box::pin(futures::future::pending()) + } + } + + #[tokio::test(start_paused = true)] + async fn should_keep_a_connection_pooled_while_the_client_was_being_built() { + let client = DapiClient::new( + "http://127.0.0.1:10001" + .parse() + .expect("valid address list"), + RequestSettings { + retries: Some(0), + ..RequestSettings::default() + }, + ); + + let error = client + .execute(StallOnReplacedConnection, RequestSettings::default()) + .await + .expect_err("the only node never answers"); + + assert!( + matches!( + &error.inner, + DapiClientError::Transport(TransportError::Grpc(status)) + if status.code() == Code::DeadlineExceeded + ), + "expected DeadlineExceeded, got {:?}", + error.inner + ); + let uri = error.address.expect("the attempted node").uri().clone(); + let settings = RequestSettings::default().finalize(); + assert!( + client + .pool + .get(PoolPrefix::Platform, &uri, Some(&settings)) + .is_some(), + "a connection pooled while the attempt's client was being built must stay pooled" + ); + } + } } #[async_trait] @@ -789,13 +1110,17 @@ impl DapiRequestExecutor for DapiClient { let response_name = request.response_name(); // Try to create transport client - let transport_client_result = R::Client::with_uri_and_settings( + let transport_client_result = R::Client::with_uri_and_settings_and_generation( address.uri().clone(), &applied_settings, &self.pool, ); - let mut transport_client = match transport_client_result { + // `pool_generation` is the pool generation of the connection + // this attempt uses. A deadline eviction below keeps + // connections pooled later: a concurrent request may already + // have evicted this one and reconnected. + let (mut transport_client, pool_generation) = match transport_client_result { Ok(client) => client, Err(transport_error) => { let can_retry_error = transport_error.can_retry(); @@ -834,15 +1159,36 @@ impl DapiRequestExecutor for DapiClient { }; // Execute the transport request - let result = transport_request + let attempt = transport_request .execute_transport(&mut transport_client, &applied_settings) .instrument(tracing::trace_span!( "execute_request", ?address, settings = ?applied_settings, method = request.method_name(), - )) - .await; + )); + // tonic enforces the `grpc-timeout` header only until the + // response headers arrive; reading the body has no limit, so an + // attempt over a half-open connection would never return. Bound + // the whole attempt, and drop the pooled connection it used. + #[cfg(not(target_arch = "wasm32"))] + let result = match applied_settings.attempt_deadline() { + Some(deadline) => match tokio::time::timeout(deadline, attempt).await { + Ok(result) => result, + Err(_) => { + self.pool.remove_uri(address.uri(), pool_generation); + Err(TransportError::Grpc(Status::deadline_exceeded(format!( + "no complete response within {deadline:?}" + )))) + } + }, + None => attempt.await, + }; + #[cfg(target_arch = "wasm32")] + let result = { + let _ = pool_generation; + attempt.await + }; let execution_result = match result { Ok(response) => { diff --git a/packages/rs-dapi-client/src/request_settings.rs b/packages/rs-dapi-client/src/request_settings.rs index 17f45d13b65..b503d21e9e5 100644 --- a/packages/rs-dapi-client/src/request_settings.rs +++ b/packages/rs-dapi-client/src/request_settings.rs @@ -21,7 +21,15 @@ const DEFAULT_BAN_FAILED_ADDRESS: bool = true; pub struct RequestSettings { /// Timeout for establishing a connection. pub connect_timeout: Option, - /// Timeout for single request (soft limit). + /// Timeout for a single request attempt. + /// + /// It is sent to the server as the `grpc-timeout` header. On native targets + /// it also bounds the whole attempt on the client: an attempt still running + /// after `timeout + connect_timeout` fails with `DeadlineExceeded` and is + /// retried like any other retryable error. For unary RPCs the attempt runs + /// from dispatch through the response body and trailers; for streaming + /// RPCs it ends when the response headers arrive, so consuming the + /// returned stream is not bounded by it. Zero disables both limits. /// /// Note that the total maximum time of execution can exceed `(timeout + connect_timeout) * retries` /// as it accounts for internal processing time between retries. @@ -88,7 +96,7 @@ impl RequestSettings { pub struct AppliedRequestSettings { /// Timeout for establishing a connection. pub connect_timeout: Option, - /// Timeout for a request. + /// Timeout for a single request attempt; see [RequestSettings::timeout]. pub timeout: Duration, /// Number of retries until returning the last error. pub retries: usize, @@ -110,6 +118,22 @@ impl AppliedRequestSettings { self } + /// Upper bound for one request attempt: `timeout` plus `connect_timeout`, + /// the bound documented on [RequestSettings::timeout]. A unary attempt + /// covers dispatch through the response body and trailers; a streaming + /// attempt ends when the response headers arrive. `None` when `timeout` + /// is zero, which means "no limit" (the transport then omits the + /// `grpc-timeout` header as well). + pub fn attempt_deadline(&self) -> Option { + if self.timeout.is_zero() { + return None; + } + Some( + self.timeout + .saturating_add(self.connect_timeout.unwrap_or_default()), + ) + } + /// Cache key fragment for the [ConnectionPool](crate::ConnectionPool), /// covering only the fields that affect the constructed transport client: /// connect timeout, response decoding limit and CA certificate. @@ -254,6 +278,42 @@ mod tests { assert!(result.ca_certificate.is_some()); } + #[test] + fn should_bound_attempt_by_timeout_plus_connect_timeout() { + let applied = RequestSettings { + timeout: Some(Duration::from_secs(10)), + connect_timeout: Some(Duration::from_secs(3)), + ..RequestSettings::default() + } + .finalize(); + assert_eq!(applied.attempt_deadline(), Some(Duration::from_secs(13))); + + let default = RequestSettings::default().finalize(); + assert_eq!(default.attempt_deadline(), Some(Duration::from_secs(10))); + } + + #[test] + fn should_not_bound_attempt_when_timeout_is_zero() { + let applied = RequestSettings { + timeout: Some(Duration::ZERO), + connect_timeout: Some(Duration::from_secs(3)), + ..RequestSettings::default() + } + .finalize(); + assert_eq!(applied.attempt_deadline(), None); + } + + #[test] + fn should_saturate_attempt_deadline_instead_of_overflowing() { + let applied = RequestSettings { + timeout: Some(Duration::MAX), + connect_timeout: Some(Duration::from_secs(1)), + ..RequestSettings::default() + } + .finalize(); + assert_eq!(applied.attempt_deadline(), Some(Duration::MAX)); + } + #[test] fn test_connection_key_ignores_per_request_settings() { let custom = RequestSettings { diff --git a/packages/rs-dapi-client/src/transport.rs b/packages/rs-dapi-client/src/transport.rs index bea0968fd2f..bad291c6b26 100644 --- a/packages/rs-dapi-client/src/transport.rs +++ b/packages/rs-dapi-client/src/transport.rs @@ -153,6 +153,24 @@ pub trait TransportClient: Send + Sized { settings: &AppliedRequestSettings, pool: &ConnectionPool, ) -> Result; + + /// Build client using node's url and [AppliedRequestSettings], together + /// with the pool generation of the connection it took (see + /// [ConnectionPool::generation]). When an attempt misses its deadline, + /// the executor evicts the node's connections up to that generation. + /// + /// The default reads the pool's generation after building the client, so + /// it can also cover a connection another thread pooled in between. + /// Clients that take their connection from `pool` override it with the + /// generation [ConnectionPool::get_or_create_with_generation] returns. + fn with_uri_and_settings_and_generation( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result<(Self, u64), TransportError> { + let client = Self::with_uri_and_settings(uri, settings, pool)?; + Ok((client, pool.generation())) + } } #[cfg(test)] diff --git a/packages/rs-dapi-client/src/transport/grpc.rs b/packages/rs-dapi-client/src/transport/grpc.rs index 7e419e9930b..1f919c9df2f 100644 --- a/packages/rs-dapi-client/src/transport/grpc.rs +++ b/packages/rs-dapi-client/src/transport/grpc.rs @@ -32,12 +32,17 @@ impl TransportClient for PlatformGrpcClient { settings: &AppliedRequestSettings, pool: &ConnectionPool, ) -> Result { - Ok(pool - .get_or_create( - PoolPrefix::Platform, - &uri, - Some(settings), - || match create_channel(uri.clone(), Some(settings)) { + Self::with_uri_and_settings_and_generation(uri, settings, pool).map(|(client, _)| client) + } + + fn with_uri_and_settings_and_generation( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result<(Self, u64), TransportError> { + let (item, generation) = + pool.get_or_create_with_generation(PoolPrefix::Platform, &uri, Some(settings), || { + match create_channel(uri.clone(), Some(settings)) { Ok(channel) => { let mut client = Self::new(channel); if let Some(max_size) = settings.max_decoding_message_size { @@ -49,9 +54,9 @@ impl TransportClient for PlatformGrpcClient { "Channel creation failed: {}", e ))), - }, - )? - .into()) + } + })?; + Ok((item.into(), generation)) } } @@ -75,12 +80,17 @@ impl TransportClient for CoreGrpcClient { settings: &AppliedRequestSettings, pool: &ConnectionPool, ) -> Result { - Ok(pool - .get_or_create( - PoolPrefix::Core, - &uri, - Some(settings), - || match create_channel(uri.clone(), Some(settings)) { + Self::with_uri_and_settings_and_generation(uri, settings, pool).map(|(client, _)| client) + } + + fn with_uri_and_settings_and_generation( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result<(Self, u64), TransportError> { + let (item, generation) = + pool.get_or_create_with_generation(PoolPrefix::Core, &uri, Some(settings), || { + match create_channel(uri.clone(), Some(settings)) { Ok(channel) => { let mut client = Self::new(channel); if let Some(max_size) = settings.max_decoding_message_size { @@ -92,9 +102,9 @@ impl TransportClient for CoreGrpcClient { "Channel creation failed: {}", e ))), - }, - )? - .into()) + } + })?; + Ok((item.into(), generation)) } } @@ -252,6 +262,14 @@ macro_rules! impl_transport_request_grpc { const STREAMING_TIMEOUT: Duration = Duration::from_secs(5 * 60); +/// Attempt timeout for unary requests whose responses run to megabytes. +/// +/// The timeout bounds the whole attempt including the response body (see +/// `RequestSettings::timeout`), so the 10 s default would fail these responses +/// on a slow link every time and ban the node that was sending them. Dead +/// connections are still caught early by the channel's HTTP/2 keepalive. +const LARGE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5 * 60); + impl_transport_request_grpc!( platform_proto::GetIdentityRequest, platform_proto::GetIdentityResponse, @@ -617,7 +635,12 @@ impl_transport_request_grpc!( platform_proto::GetShieldedEncryptedNotesRequest, platform_proto::GetShieldedEncryptedNotesResponse, PlatformGrpcClient, - RequestSettings::default(), + RequestSettings { + // A full chunk carries thousands of encrypted notes (megabytes), and + // the notes sync fetches several chunks in parallel over one link. + timeout: Some(LARGE_RESPONSE_TIMEOUT), + ..RequestSettings::default() + }, get_shielded_encrypted_notes ); @@ -923,6 +946,7 @@ impl_transport_request_grpc!( // GetRecentCompactedAddressBalanceChangesResponse can have 100 values * 2048 addresses * ~44 bytes each = ~9MB // We set it to 16MB to be safe max_decoding_message_size: Some(16 * 1024 * 1024), + timeout: Some(LARGE_RESPONSE_TIMEOUT), ..RequestSettings::default() }, get_recent_compacted_address_balance_changes diff --git a/packages/rs-dapi-client/src/transport/tonic_channel.rs b/packages/rs-dapi-client/src/transport/tonic_channel.rs index c2df9d92b47..779bdb33f52 100644 --- a/packages/rs-dapi-client/src/transport/tonic_channel.rs +++ b/packages/rs-dapi-client/src/transport/tonic_channel.rs @@ -3,6 +3,7 @@ use crate::{request_settings::AppliedRequestSettings, Uri}; use dapi_grpc::core::v0::core_client::CoreClient; use dapi_grpc::platform::v0::platform_client::PlatformClient; use dapi_grpc::tonic::transport::{Certificate, Channel, ClientTlsConfig}; +use std::time::Duration; /// Platform Client using gRPC transport. pub type PlatformGrpcClient = PlatformClient; @@ -13,6 +14,11 @@ pub type CoreGrpcClient = CoreClient; // #[derive(Default, Clone, Debug)] pub type TokioBackonSleeper = backon::TokioSleeper; +/// HTTP/2 PING interval while a request is in flight on a connection. +const HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); +/// How long to wait for a PING acknowledgement before closing the connection. +const HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10); + /// Create channel (connection) for gRPC transport. pub fn create_channel( uri: Uri, @@ -50,6 +56,15 @@ pub fn create_channel( }; } + // Ping only while a request is in flight: a connection whose network path + // died mid-response is then closed within interval + timeout, failing its + // streams instead of leaving them waiting for data that never comes. + // Idle pooled connections are not pinged. + builder = builder + .http2_keep_alive_interval(HTTP2_KEEP_ALIVE_INTERVAL) + .keep_alive_timeout(HTTP2_KEEP_ALIVE_TIMEOUT) + .keep_alive_while_idle(false); + builder = builder .tls_config(tls_config) .expect("Failed to set TLS config"); diff --git a/packages/rs-dapi-client/tests/request_deadline.rs b/packages/rs-dapi-client/tests/request_deadline.rs new file mode 100644 index 00000000000..05f50b6b573 --- /dev/null +++ b/packages/rs-dapi-client/tests/request_deadline.rs @@ -0,0 +1,200 @@ +//! Client-side attempt deadline: an attempt whose response never completes +//! (a node that stalls mid-response, or a half-open connection) must fail with +//! `DeadlineExceeded` after `timeout + connect_timeout` and be retried on +//! another node, instead of hanging the request forever. +//! +//! The tests run on tokio's paused clock, so the deadlines elapse instantly. + +#[allow(dead_code)] +mod common; + +use std::fmt::Debug; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use common::{FakeClient, FakeResponse}; +use dapi_grpc::mock::Mockable; +use dapi_grpc::tonic::Code; +use rs_dapi_client::transport::{ + AppliedRequestSettings, BoxFuture, TransportError, TransportRequest, +}; +use rs_dapi_client::{ + Address, AddressList, DapiClient, DapiClientError, DapiRequestExecutor, RequestSettings, Uri, +}; + +/// How a fake node answers one attempt. +#[derive(Clone, Copy)] +enum Answer { + /// Respond successfully after this much time. + After(Duration), + /// Never finish the response. + Never, +} + +/// Fake request whose `answer` closure decides, per node, how the attempt +/// behaves over time. `hit_uris` records every node the executor tried. +#[derive(Clone)] +struct DelayedRequest { + answer: Arc Answer + Send + Sync>, + hit_uris: Arc>>, +} + +impl DelayedRequest { + fn new(answer: impl Fn(&Uri) -> Answer + Send + Sync + 'static) -> Self { + Self { + answer: Arc::new(answer), + hit_uris: Default::default(), + } + } + + fn hits(&self) -> Vec { + self.hit_uris.lock().unwrap().clone() + } +} + +impl Debug for DelayedRequest { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DelayedRequest") + .field("hit_uris", &self.hit_uris) + .finish() + } +} + +impl Mockable for DelayedRequest {} + +impl TransportRequest for DelayedRequest { + type Client = FakeClient; + type Response = FakeResponse; + + const SETTINGS_OVERRIDES: RequestSettings = RequestSettings::default(); + + fn method_name(&self) -> &'static str { + "delayed_fake_method" + } + + fn execute_transport<'c>( + self, + client: &'c mut Self::Client, + _settings: &AppliedRequestSettings, + ) -> BoxFuture<'c, Result> { + let uri = client.uri.clone(); + self.hit_uris.lock().unwrap().push(uri.clone()); + match (self.answer)(&uri) { + Answer::After(delay) => Box::pin(async move { + tokio::time::sleep(delay).await; + Ok(FakeResponse) + }), + Answer::Never => Box::pin(futures::future::pending()), + } + } +} + +fn two_nodes() -> AddressList { + "http://127.0.0.1:10001,http://127.0.0.1:10002" + .parse() + .expect("valid address list") +} + +#[tokio::test(start_paused = true)] +async fn should_retry_on_another_node_when_an_attempt_misses_its_deadline() { + // The first node the executor picks stalls forever; every other node + // answers at once. + let stalled: Arc>> = Default::default(); + let stalled_c = stalled.clone(); + let request = DelayedRequest::new(move |uri| { + let mut stalled = stalled_c.lock().unwrap(); + let target = stalled.get_or_insert_with(|| uri.clone()); + if *target == *uri { + Answer::Never + } else { + Answer::After(Duration::ZERO) + } + }); + let client = DapiClient::new(two_nodes(), RequestSettings::default()); + + let started = tokio::time::Instant::now(); + let response = client + .execute(request.clone(), RequestSettings::default()) + .await + .expect("the retry on the healthy node must succeed"); + + let stalled_uri = stalled.lock().unwrap().clone().expect("a node stalled"); + assert_eq!(response.retries, 1); + assert_ne!(response.address.uri(), &stalled_uri); + assert_eq!(request.hits().len(), 2); + assert!( + started.elapsed() >= Duration::from_secs(10), + "the stalled attempt must run until the default 10 s deadline, not be cut earlier" + ); + let stalled_node = Address::try_from(stalled_uri).expect("valid address"); + assert!( + client.address_list().is_banned(&stalled_node), + "the node that missed the deadline must be banned" + ); +} + +#[tokio::test(start_paused = true)] +async fn should_fail_with_deadline_exceeded_when_every_node_stalls() { + let request = DelayedRequest::new(|_| Answer::Never); + let client = DapiClient::new(two_nodes(), RequestSettings::default()); + + let error = client + .execute(request.clone(), RequestSettings::default()) + .await + .expect_err("no node ever completes a response"); + + // Both nodes were tried once and banned, then the address list ran dry. + assert_eq!(request.hits().len(), 2); + match error.inner { + DapiClientError::NoAvailableAddressesToRetry(last) => { + let TransportError::Grpc(status) = *last; + assert_eq!(status.code(), Code::DeadlineExceeded); + } + other => panic!("expected NoAvailableAddressesToRetry, got {other:?}"), + } +} + +#[tokio::test(start_paused = true)] +async fn should_cut_a_response_that_outlasts_timeout_plus_connect_timeout() { + // The response would complete, but only after the attempt deadline: + // headers-only timeouts let this through; the attempt deadline must not. + let request = DelayedRequest::new(|_| Answer::After(Duration::from_secs(14))); + let settings = RequestSettings { + timeout: Some(Duration::from_secs(10)), + connect_timeout: Some(Duration::from_secs(3)), + retries: Some(0), + ..RequestSettings::default() + }; + let client = DapiClient::new(two_nodes(), settings); + + let error = client + .execute(request.clone(), RequestSettings::default()) + .await + .expect_err("a 14 s response exceeds the 13 s attempt deadline"); + + match error.inner { + DapiClientError::Transport(TransportError::Grpc(status)) => { + assert_eq!(status.code(), Code::DeadlineExceeded) + } + other => panic!("expected a gRPC DeadlineExceeded, got {other:?}"), + } +} + +#[tokio::test(start_paused = true)] +async fn should_not_cut_an_attempt_when_timeout_is_zero() { + // Zero means "no limit", for the `grpc-timeout` header and the attempt alike. + let request = DelayedRequest::new(|_| Answer::After(Duration::from_secs(60))); + let settings = RequestSettings { + timeout: Some(Duration::ZERO), + ..RequestSettings::default() + }; + let client = DapiClient::new(two_nodes(), RequestSettings::default()); + + let response = client + .execute(request.clone(), settings) + .await + .expect("a slow response must complete when the timeout is disabled"); + + assert_eq!(response.retries, 0); + assert_eq!(request.hits().len(), 1); +} diff --git a/packages/rs-dapi-client/tests/stalled_response_body.rs b/packages/rs-dapi-client/tests/stalled_response_body.rs new file mode 100644 index 00000000000..39cdff7d3b3 --- /dev/null +++ b/packages/rs-dapi-client/tests/stalled_response_body.rs @@ -0,0 +1,216 @@ +//! Regression through tonic itself: a node that sends the response headers +//! of a unary call and then never sends its body. +//! +//! tonic enforces the `grpc-timeout` header only until the response headers +//! arrive, so such a call never completes on its own. The executor's attempt +//! deadline must cut it with `DeadlineExceeded` and fail over to another +//! node. The server runs in memory over a duplex stream, so the tests need no +//! network access. + +#[allow(dead_code)] +mod common; + +use std::fmt::Debug; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use common::FakeClient; +use dapi_grpc::mock::Mockable; +use dapi_grpc::platform::v0::{GetStatusRequest, GetStatusResponse}; +use dapi_grpc::tonic::transport::{Channel, Endpoint}; +use dapi_grpc::tonic::{Code, IntoRequest}; +use hyper_util::rt::TokioIo; +use rs_dapi_client::transport::{ + AppliedRequestSettings, BoxFuture, PlatformGrpcClient, TransportError, TransportRequest, +}; +use rs_dapi_client::{ + Address, AddressList, DapiClient, DapiClientError, DapiRequestExecutor, RequestSettings, Uri, +}; + +const ATTEMPT_TIMEOUT: Duration = Duration::from_millis(200); + +/// A gRPC client whose channel runs over an in-memory duplex stream to an +/// HTTP/2 server that answers every request with `200` response headers and +/// then never sends a body or trailers. +fn body_stalling_client() -> PlatformGrpcClient { + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + let mut connection = h2::server::handshake(server_io) + .await + .expect("h2 handshake"); + // Keep every response stream open (and the connection polled) so the + // body stays pending instead of the stream being reset. + let mut open_streams = Vec::new(); + while let Some(accepted) = connection.accept().await { + let (_request, mut respond) = accepted.expect("accept request"); + let headers = http::Response::builder() + .status(200) + .header("content-type", "application/grpc") + .body(()) + .expect("response headers"); + open_streams.push( + respond + .send_response(headers, false) + .expect("send response headers"), + ); + } + }); + + let mut io = Some(client_io); + let channel: Channel = Endpoint::from_static("http://body-stalling.in-memory") + .connect_with_connector_lazy(tower::service_fn(move |_: Uri| { + let io = io.take(); + async move { + io.map(TokioIo::new) + .ok_or_else(|| std::io::Error::other("the in-memory stream is single-use")) + } + })); + PlatformGrpcClient::new(channel) +} + +/// `getStatus` request whose first attempted node is served by the +/// body-stalling tonic channel; every other node answers at once. +#[derive(Clone)] +struct StatusRequest { + stalling_client: PlatformGrpcClient, + stalled_node: Arc>>, +} + +impl StatusRequest { + fn new() -> Self { + Self { + stalling_client: body_stalling_client(), + stalled_node: Default::default(), + } + } + + fn stalled_node(&self) -> Option { + self.stalled_node.lock().expect("stalled node lock").clone() + } +} + +impl Debug for StatusRequest { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StatusRequest") + .field("stalled_node", &self.stalled_node) + .finish() + } +} + +impl Mockable for StatusRequest {} + +impl TransportRequest for StatusRequest { + type Client = FakeClient; + type Response = GetStatusResponse; + + const SETTINGS_OVERRIDES: RequestSettings = RequestSettings::default(); + + fn method_name(&self) -> &'static str { + "get_status" + } + + fn execute_transport<'c>( + self, + client: &'c mut Self::Client, + settings: &AppliedRequestSettings, + ) -> BoxFuture<'c, Result> { + let is_stalled_node = { + let mut stalled = self.stalled_node.lock().expect("stalled node lock"); + *stalled.get_or_insert_with(|| client.uri.clone()) == client.uri + }; + if !is_stalled_node { + return Box::pin(async { Ok(GetStatusResponse::default()) }); + } + // The same shape as the production transport: the request timeout + // travels only as the `grpc-timeout` header. + let mut grpc = self.stalling_client; + let mut request = GetStatusRequest::default().into_request(); + request.set_timeout(settings.timeout); + Box::pin(async move { + grpc.get_status(request) + .await + .map(|response| response.into_inner()) + .map_err(TransportError::Grpc) + }) + } +} + +fn settings(retries: usize) -> RequestSettings { + RequestSettings { + timeout: Some(ATTEMPT_TIMEOUT), + retries: Some(retries), + ..RequestSettings::default() + } +} + +/// The premise: with only the `grpc-timeout` header, a unary call whose body +/// never arrives is still pending long after that timeout. +#[tokio::test] +async fn should_leave_a_body_stalled_call_pending_with_only_the_grpc_timeout_header() { + let mut client = body_stalling_client(); + let mut request = GetStatusRequest::default().into_request(); + request.set_timeout(ATTEMPT_TIMEOUT); + + let outcome = tokio::time::timeout(Duration::from_secs(2), client.get_status(request)).await; + + assert!( + outcome.is_err(), + "tonic must still be waiting for the body 2 s after a 200 ms grpc-timeout, got {outcome:?}" + ); +} + +#[tokio::test] +async fn should_cut_a_body_stalled_call_with_deadline_exceeded() { + let request = StatusRequest::new(); + let client = DapiClient::new( + "http://127.0.0.1:20001" + .parse() + .expect("valid address list"), + settings(0), + ); + + let error = client + .execute(request, RequestSettings::default()) + .await + .expect_err("the only node never completes its response"); + + match error.inner { + DapiClientError::Transport(TransportError::Grpc(status)) => { + assert_eq!( + status.code(), + Code::DeadlineExceeded, + "unexpected status: {status:?}" + ) + } + other => panic!("expected a gRPC DeadlineExceeded, got {other:?}"), + } +} + +#[tokio::test] +async fn should_fail_over_when_a_node_stalls_its_response_body() { + let request = StatusRequest::new(); + let address_list: AddressList = "http://127.0.0.1:20001,http://127.0.0.1:20002" + .parse() + .expect("valid address list"); + let client = DapiClient::new(address_list, settings(5)); + + let started = tokio::time::Instant::now(); + let response = client + .execute(request.clone(), RequestSettings::default()) + .await + .expect("the other node must answer"); + let elapsed = started.elapsed(); + + let stalled_uri = request.stalled_node().expect("a node was tried first"); + assert_eq!(response.retries, 1); + assert_ne!(response.address.uri(), &stalled_uri); + assert!( + elapsed >= ATTEMPT_TIMEOUT && elapsed < Duration::from_secs(2), + "the stalled attempt must end at its deadline, took {elapsed:?}" + ); + let stalled_node = Address::try_from(stalled_uri).expect("valid address"); + assert!( + client.address_list().is_banned(&stalled_node), + "the node that stalled its response body must be banned" + ); +}