diff --git a/packages/rs-dapi-client/Cargo.toml b/packages/rs-dapi-client/Cargo.toml index a52c12400fb..da97e5b4f98 100644 --- a/packages/rs-dapi-client/Cargo.toml +++ b/packages/rs-dapi-client/Cargo.toml @@ -55,6 +55,7 @@ http-serde = { version = "2.1", optional = true } rand = { version = "0.8.5", features = [ "small_rng", "getrandom", + "alloc", ], default-features = false } thiserror = "2.0.17" tracing = "0.1.41" diff --git a/packages/rs-dapi-client/src/address_list.rs b/packages/rs-dapi-client/src/address_list.rs index 5bac1674666..5ca2e7e0856 100644 --- a/packages/rs-dapi-client/src/address_list.rs +++ b/packages/rs-dapi-client/src/address_list.rs @@ -3,7 +3,7 @@ use crate::address_ban_info::AddressBanInfo; use crate::Uri; use chrono::Utc; -use rand::{rngs::SmallRng, seq::IteratorRandom, SeedableRng}; +use rand::{rngs::SmallRng, seq::IteratorRandom, Rng, SeedableRng}; use std::collections::hash_map::Entry; use std::collections::HashMap; use std::hash::{Hash, Hasher}; @@ -14,6 +14,25 @@ use std::time::Duration; const DEFAULT_BASE_BAN_PERIOD: Duration = Duration::from_secs(60); +/// Longest ban window either ban path can produce. Bounds `e^ban_count`, +/// which otherwise overflows `DateTime + Duration` arithmetic (a panic that +/// poisons the shared lock) once `ban_count` reaches the mid-20s. +const MAX_BAN_PERIOD: Duration = Duration::from_secs(24 * 60 * 60); + +/// Default number of addresses that receive traffic at a time. +/// +/// Kept small so requests reuse warm connections instead of sampling the whole +/// list (hundreds of nodes on mainnet), where nearly every request would land +/// on a cold host and pay a fresh TCP + TLS handshake. +const DEFAULT_ACTIVE_SET_SIZE: usize = 5; + +/// How long an address may hold an active-set slot before it is retired and a +/// random live standby is promoted in its place. Bounding slot tenure keeps +/// connections warm for minutes at a time while preventing any small set of +/// nodes from observing the client's entire query stream for the whole +/// process lifetime. +const SLOT_LIFETIME: Duration = Duration::from_secs(5 * 60); + /// DAPI address. #[derive(Debug, Clone, Eq)] #[cfg_attr(feature = "mocks", derive(serde::Serialize, serde::Deserialize))] @@ -91,7 +110,8 @@ impl AddressStatus { /// Ban the [Address] and record the `reason` for the ban. /// /// Applies exponential backoff: the ban window is `base × e^ban_count` - /// (where `ban_count` is the value *before* this call), and `banned_until` + /// (where `ban_count` is the value *before* this call), capped at 24 hours + /// (or `base` itself if larger), and `banned_until` /// is always re-based to `now + window` unconditionally, regardless of any /// existing active ban. Concretely, a health failure on a node that already /// holds a longer rate-limit window (set via [`AddressStatus::ban_for`]) will @@ -104,7 +124,14 @@ impl AddressStatus { /// The counter resets to 0 on [`AddressStatus::unban`]. pub fn ban_with_reason(&mut self, base_ban_period: &Duration, reason: Option) { let coefficient = (self.ban_count as f64).exp(); - let ban_period = Duration::from_secs_f64(base_ban_period.as_secs_f64() * coefficient); + let max_ban_period = MAX_BAN_PERIOD.max(*base_ban_period); + let ban_secs = base_ban_period.as_secs_f64() * coefficient; + // NaN/inf compare false, so any overflowing window falls to the cap. + let ban_period = if ban_secs < max_ban_period.as_secs_f64() { + Duration::from_secs_f64(ban_secs) + } else { + max_ban_period + }; self.banned_until = Some(chrono::Utc::now() + ban_period); self.ban_count += 1; @@ -114,7 +141,8 @@ impl AddressStatus { /// Ban the address for an exact `period` (server-advertised), bypassing the /// exponential ladder used by [`AddressStatus::ban_with_reason`]. /// - /// The ban window is flat (not exponential). `banned_until` is advanced to + /// The ban window is flat (not exponential) and capped at 24 hours. + /// `banned_until` is advanced to /// `now + period` only when that timestamp is **later** than the current /// `banned_until`, so a short-reset call never shortens a longer active ban /// (health ban or a prior longer rate-limit ban). `ban_reason` is updated @@ -131,6 +159,9 @@ impl AddressStatus { /// sequences. [`AddressStatus::ban_with_reason`] re-bases `banned_until` /// unconditionally — see its docs for the intentional cross-method semantics. pub fn ban_for(&mut self, period: Duration, reason: Option) { + // A server-advertised window is clamped like the ladder: a hostile or + // buggy period must not overflow `DateTime + Duration`. + let period = period.min(MAX_BAN_PERIOD); let advertised_until = chrono::Utc::now() + period; if self .banned_until @@ -148,6 +179,14 @@ impl AddressStatus { self.ban_count > 0 } + /// Check if [Address] is live at `now`: never banned, or its ban period has + /// already expired. + fn is_live(&self, now: chrono::DateTime) -> bool { + self.banned_until + .map(|banned_until| banned_until < now) + .unwrap_or(true) + } + /// Clears ban record. pub fn unban(&mut self) { self.ban_count = 0; @@ -166,11 +205,46 @@ pub enum AddressListError { InvalidAddressUri(String), } +/// One member of the sticky active set: the address plus the moment its slot +/// expires and a random standby replaces it. +#[derive(Debug)] +struct ActiveMember { + address: Address, + slot_expires_at: chrono::DateTime, +} + +/// Sticky rotation state: the addresses currently receiving traffic, the most +/// recently served one that round-robin selection advances from, and the +/// configured active-set size. Shared (behind one lock) by every clone of an +/// [AddressList] so all clones drive the same rotation. +#[derive(Debug)] +struct Rotation { + active: Vec, + last_served: Option
, + active_set_size: usize, +} + +impl Default for Rotation { + fn default() -> Self { + Rotation { + active: Vec::new(), + last_served: None, + active_set_size: DEFAULT_ACTIVE_SET_SIZE, + } + } +} + /// A structure to manage DAPI addresses to select from /// for [DapiRequest](crate::DapiRequest) execution. +/// +/// Address selection is sticky: requests rotate over a small active set of +/// addresses (5 by default, see [AddressList::with_active_set_size]) and the +/// rest of the list serves as failover standby (see +/// [AddressList::get_live_address]). #[derive(Debug, Clone)] pub struct AddressList { addresses: Arc>>, + rotation: Arc>, base_ban_period: Duration, } @@ -196,10 +270,27 @@ impl AddressList { pub fn with_settings(base_ban_period: Duration) -> Self { AddressList { addresses: Arc::new(RwLock::new(HashMap::new())), + rotation: Arc::new(RwLock::new(Rotation::default())), base_ban_period, } } + /// Set how many addresses receive traffic at a time. + /// + /// Defaults to 5. `0` is clamped to 1. Smaller values maximize connection + /// reuse, larger values spread load over more nodes. The effective size is + /// additionally capped by the number of live addresses, so a very large + /// value (e.g. `usize::MAX`) disables stickiness and round-robins over the + /// whole list. + /// + /// The size lives in the rotation state shared by every clone of this + /// list, so it applies to all clones and takes effect on the next + /// selection (shrinking drops the excess members). + pub fn with_active_set_size(self, size: usize) -> Self { + self.rotation.write().unwrap().active_set_size = size.max(1); + self + } + /// Bans address /// Returns false if the address is not in the list. /// @@ -294,28 +385,112 @@ impl AddressList { self.add(Address::try_from(uri).expect("valid uri")) } - /// Randomly select a not-banned address. + /// Select a not-banned address to send the next request to. + /// + /// Not a pure getter: every call advances the shared rotation cursor and + /// may promote or retire active-set members, steering traffic for all + /// clones of this list. + /// + /// Selection is sticky: requests rotate round-robin over a small active + /// set of addresses (see [AddressList::with_active_set_size]) instead of + /// sampling the whole list, so connections to those hosts stay warm. An + /// active address that got banned, removed or evicted on failover (see + /// [AddressList::evict_from_rotation]) is dropped from the set here and a + /// random live standby address is promoted in its place. Each slot also + /// expires after a jittered lifetime (5–7.5 minutes), so no node holds a + /// slot — and a view of this client's query stream — indefinitely. /// /// An address is considered live when it has never been banned or when its /// ban period has already expired. pub fn get_live_address(&self) -> Option
{ // TODO(low): module-wide `.read()/.write().unwrap()` panics on a - // poisoned lock; adopt poison-tolerant locking consistently (SEC-003). + // poisoned lock; adopt poison-tolerant locking consistently. let guard = self.addresses.read().unwrap(); - let mut rng = SmallRng::from_entropy(); let now = chrono::Utc::now(); - guard - .iter() - .filter(|(_, status)| { - status - .banned_until - .map(|banned_until| banned_until < now) - .unwrap_or(true) - }) - .choose(&mut rng) - .map(|(addr, _)| addr.clone()) + // Seeded outside the critical section: `from_entropy` panics if the OS + // entropy source fails, and a panic while holding the write lock would + // poison it and permanently disable address selection. + let mut rng = SmallRng::from_entropy(); + + // Lock ordering: `addresses` before `rotation`; this is the only place + // both locks are held at once. + let mut rotation = self.rotation.write().unwrap(); + + // Drop active addresses that are banned, no longer in the list, or + // whose slot lifetime expired. + rotation.active.retain(|member| { + now < member.slot_expires_at + && guard + .get(&member.address) + .map(|status| status.is_live(now)) + .unwrap_or(false) + }); + + // Honor a shrunken size (the rotation state is shared, so it may have + // been reconfigured through any clone). + let size = rotation.active_set_size; + rotation.active.truncate(size); + + // Refill vacancies with random live standby addresses. Bounded by the + // list length so an oversized configured value cannot over-allocate in + // `choose_multiple`. + let vacancies = size.saturating_sub(rotation.active.len()).min(guard.len()); + if vacancies > 0 { + let promoted = guard + .iter() + .filter(|&(address, status)| { + status.is_live(now) + && !rotation + .active + .iter() + .any(|member| member.address == *address) + }) + .choose_multiple(&mut rng, vacancies); + + rotation + .active + .extend(promoted.into_iter().map(|(address, _)| ActiveMember { + address: address.clone(), + // Jitter staggers expiries so slots retire one at a time + // instead of the whole set at once. + slot_expires_at: now + SLOT_LIFETIME.mul_f64(rng.gen_range(1.0..1.5)), + })); + } + + if rotation.active.is_empty() { + return None; + } + + // Advance from the last-served address: eviction re-orders the active + // set, so a positional cursor would not survive churn. Start from the + // head when the last-served address is gone (or nothing has been + // served yet). + let last_position = rotation.last_served.as_ref().and_then(|last| { + rotation + .active + .iter() + .position(|member| member.address == *last) + }); + + let index = last_position.map_or(0, |position| (position + 1) % rotation.active.len()); + let address = rotation.active[index].address.clone(); + rotation.last_served = Some(address.clone()); + Some(address) + } + + /// Drop `address` from the sticky active set, leaving its ban state + /// untouched; the next selection promotes a random live standby in its + /// place. + /// + /// This is the failover path for callers that disable banning + /// ([RequestSettings::ban_failed_address](crate::RequestSettings)): a + /// failing node must stop receiving its slot's traffic even when it is + /// never banned. + pub fn evict_from_rotation(&self, address: &Address) { + let mut rotation = self.rotation.write().unwrap(); + rotation.active.retain(|member| member.address != *address); } /// Get all not banned addresses. @@ -344,12 +519,7 @@ impl AddressList { guard .iter() - .filter(|(_, status)| { - status - .banned_until - .map(|banned_until| banned_until < now) - .unwrap_or(true) - }) + .filter(|(_, status)| status.is_live(now)) .map(|(addr, _)| addr.clone()) .collect() } @@ -370,11 +540,7 @@ impl AddressList { guard .iter() .map(|(addr, status)| { - let banned = status.ban_count > 0 - && status - .banned_until - .map(|banned_until| banned_until >= now) - .unwrap_or(false); + let banned = status.ban_count > 0 && !status.is_live(now); AddressBanInfo { uri: addr.to_string(), banned, @@ -684,6 +850,335 @@ mod tests { assert!(list.get_live_address().is_none()); } + #[test] + fn test_get_live_address_sticks_to_small_active_set() { + let mut list = AddressList::new(); + for i in 0..50 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + let distinct: std::collections::HashSet = (0..200) + .map(|_| list.get_live_address().unwrap().to_string()) + .collect(); + + assert_eq!( + distinct.len(), + DEFAULT_ACTIVE_SET_SIZE, + "all traffic must rotate over exactly the active set" + ); + } + + #[test] + fn test_get_live_address_round_robins_over_active_set() { + let mut list = AddressList::new().with_active_set_size(2); + for i in 0..5 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + let picks: Vec = (0..6) + .map(|_| list.get_live_address().unwrap().to_string()) + .collect(); + + // Strict alternation between the two active members. + assert_ne!(picks[0], picks[1]); + assert_eq!(picks[0], picks[2]); + assert_eq!(picks[1], picks[3]); + assert_eq!(picks[0], picks[4]); + assert_eq!(picks[1], picks[5]); + } + + #[test] + fn test_get_live_address_ban_evicts_active_and_promotes_standby() { + let mut list = AddressList::new().with_active_set_size(1); + for i in 0..3 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + let first = list.get_live_address().unwrap(); + for _ in 0..5 { + assert_eq!( + list.get_live_address().unwrap(), + first, + "selection must be sticky until the active address fails" + ); + } + + list.ban(&first); + + let second = list.get_live_address().unwrap(); + assert_ne!(second, first, "banned address must leave the active set"); + for _ in 0..5 { + assert_eq!( + list.get_live_address().unwrap(), + second, + "selection must stick to the promoted standby" + ); + } + } + + #[test] + fn test_get_live_address_no_immediate_repeat_after_other_member_evicted() { + let mut list = AddressList::new().with_active_set_size(2); + for i in 0..3 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + let first = list.get_live_address().unwrap(); + let second = list.get_live_address().unwrap(); + let third = list.get_live_address().unwrap(); + assert_eq!(first, third, "two-member set alternates"); + + // Ban the member that was NOT just served; a standby gets promoted. + list.ban(&second); + + let next = list.get_live_address().unwrap(); + assert_ne!( + next, third, + "must not serve the same address twice in a row after eviction" + ); + assert_ne!(next, second, "banned address must not be served"); + } + + #[test] + fn test_get_live_address_removed_address_pruned_from_active_set() { + let mut list = AddressList::new().with_active_set_size(1); + list.add("http://127.0.0.1:3000".parse().unwrap()); + list.add("http://127.0.0.1:3001".parse().unwrap()); + + let first = list.get_live_address().unwrap(); + list.remove(&first); + + let second = list.get_live_address().unwrap(); + assert_ne!(second, first); + } + + #[test] + fn test_get_live_address_with_fewer_live_addresses_than_active_set() { + let mut list = AddressList::new(); // default active set size 5 + list.add("http://127.0.0.1:3000".parse().unwrap()); + list.add("http://127.0.0.1:3001".parse().unwrap()); + + let distinct: std::collections::HashSet = (0..10) + .map(|_| list.get_live_address().unwrap().to_string()) + .collect(); + assert_eq!(distinct.len(), 2, "both live addresses rotate"); + } + + #[test] + fn test_get_live_address_all_banned_returns_none() { + let mut list = AddressList::new().with_active_set_size(1); + let addr: Address = "http://127.0.0.1:3000".parse().unwrap(); + list.add(addr.clone()); + + assert!(list.get_live_address().is_some()); + list.ban(&addr); + assert!(list.get_live_address().is_none()); + } + + #[test] + fn test_get_live_address_expired_slot_is_recycled() { + let mut list = AddressList::new(); + for i in 0..10 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + // Populate the active set, then back-date every slot's expiry. + list.get_live_address().unwrap(); + { + let mut rotation = list.rotation.write().unwrap(); + assert_eq!(rotation.active.len(), DEFAULT_ACTIVE_SET_SIZE); + for member in rotation.active.iter_mut() { + member.slot_expires_at = chrono::Utc::now() - Duration::from_secs(1); + } + } + + let before = chrono::Utc::now(); + assert!( + list.get_live_address().is_some(), + "selection must survive whole-set expiry" + ); + + let rotation = list.rotation.read().unwrap(); + assert_eq!(rotation.active.len(), DEFAULT_ACTIVE_SET_SIZE); + assert!( + rotation + .active + .iter() + .all(|member| member.slot_expires_at > before), + "expired slots must be retired and re-issued with fresh expiries" + ); + } + + #[test] + fn test_get_live_address_sole_address_survives_slot_expiry() { + let mut list = AddressList::new().with_active_set_size(1); + let addr: Address = "http://127.0.0.1:3000".parse().unwrap(); + list.add(addr.clone()); + + assert_eq!(list.get_live_address().unwrap(), addr); + list.rotation.write().unwrap().active[0].slot_expires_at = + chrono::Utc::now() - Duration::from_secs(1); + + assert_eq!( + list.get_live_address().unwrap(), + addr, + "the only live address must be re-promoted after its slot expires" + ); + } + + #[test] + fn test_evict_from_rotation_removes_member_without_ban() { + let mut list = AddressList::new().with_active_set_size(2); + for i in 0..3 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + let served = list.get_live_address().unwrap(); + list.evict_from_rotation(&served); + + assert!( + !list + .rotation + .read() + .unwrap() + .active + .iter() + .any(|member| member.address == served), + "evicted address must leave the active set" + ); + assert!( + !list.is_banned(&served), + "eviction must not touch ban state" + ); + assert!(list.get_live_address().is_some()); + } + + #[test] + fn test_update_address_ban_status_evicts_when_banning_disabled() { + use crate::{transport::AppliedRequestSettings, CanRetry, ExecutionError, ExecutionResult}; + + #[derive(Debug)] + struct RetryableError; + impl std::fmt::Display for RetryableError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "retryable") + } + } + impl CanRetry for RetryableError { + fn can_retry(&self) -> bool { + true + } + } + + let mut list = AddressList::new().with_active_set_size(2); + for i in 0..3 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + let failed = list.get_live_address().unwrap(); + + let result: ExecutionResult = Err(ExecutionError { + inner: RetryableError, + retries: 0, + address: Some(failed.clone()), + }); + let settings = AppliedRequestSettings { + connect_timeout: None, + timeout: Duration::from_secs(10), + retries: 5, + ban_failed_address: false, + max_decoding_message_size: None, + #[cfg(not(target_arch = "wasm32"))] + ca_certificate: None, + }; + crate::update_address_ban_status(&list, &result, &settings); + + assert!( + !list + .rotation + .read() + .unwrap() + .active + .iter() + .any(|member| member.address == failed), + "failed address must leave the rotation even when banning is disabled" + ); + assert!(!list.is_banned(&failed), "banning stays disabled"); + } + + #[test] + fn test_with_active_set_size_is_shared_across_clones_and_shrinks() { + let mut list = AddressList::new(); + for i in 0..10 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + // Fill the default-sized active set through the original handle. + list.get_live_address().unwrap(); + assert_eq!( + list.rotation.read().unwrap().active.len(), + DEFAULT_ACTIVE_SET_SIZE + ); + + // Reconfiguring through a clone applies to the shared rotation. + let _clone = list.clone().with_active_set_size(2); + + let distinct: std::collections::HashSet = (0..20) + .map(|_| list.get_live_address().unwrap().to_string()) + .collect(); + assert_eq!( + distinct.len(), + 2, + "shrunken size set through a clone must constrain every handle" + ); + } + + #[test] + fn test_with_active_set_size_max_uses_whole_list() { + let mut list = AddressList::new().with_active_set_size(usize::MAX); + for i in 0..3 { + list.add(format!("http://127.0.0.1:{}", 3000 + i).parse().unwrap()); + } + + // Must not over-allocate or panic; effectively disables stickiness. + let distinct: std::collections::HashSet = (0..9) + .map(|_| list.get_live_address().unwrap().to_string()) + .collect(); + assert_eq!(distinct.len(), 3, "all live addresses rotate"); + } + + #[test] + fn test_ban_ladder_window_is_capped() { + let mut status = AddressStatus::default(); + let base = Duration::from_secs(60); + + // Pre-cap, ban #27 overflowed `DateTime + Duration` and panicked. + for _ in 0..40 { + status.ban(&base); + } + + let until = status.banned_until.expect("banned_until set"); + let window = until - chrono::Utc::now(); + assert!( + window <= chrono::TimeDelta::from_std(MAX_BAN_PERIOD).unwrap(), + "ban window must be capped at MAX_BAN_PERIOD" + ); + } + + #[test] + fn test_ban_for_huge_period_is_clamped() { + let mut status = AddressStatus::default(); + + // Pre-clamp this overflowed `DateTime + Duration` and panicked. + status.ban_for(Duration::from_secs(u64::MAX), None); + + let until = status.banned_until.expect("banned_until set"); + let window = until - chrono::Utc::now(); + assert!( + window <= chrono::TimeDelta::from_std(MAX_BAN_PERIOD).unwrap(), + "advertised ban window must be clamped to MAX_BAN_PERIOD" + ); + } + #[test] fn test_address_list_get_live_address_returns_some_when_available() { let mut list = AddressList::new(); diff --git a/packages/rs-dapi-client/src/connection_pool.rs b/packages/rs-dapi-client/src/connection_pool.rs index 5a3dcc3dc43..b8e7b74775a 100644 --- a/packages/rs-dapi-client/src/connection_pool.rs +++ b/packages/rs-dapi-client/src/connection_pool.rs @@ -97,7 +97,16 @@ impl ConnectionPool { settings: Option<&AppliedRequestSettings>, ) -> String { let prefix: PoolPrefix = class.into(); - format!("{}:{}{:?}", prefix, uri, settings) + // 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), + } } } @@ -176,8 +185,10 @@ impl From<&PoolItem> for PoolPrefix { #[cfg(test)] mod tests { use super::*; + use crate::RequestSettings; use dapi_grpc::tonic::transport::Channel; use std::str::FromStr; + use std::time::Duration; fn test_uri() -> Uri { Uri::from_str("http://127.0.0.1:3000").unwrap() @@ -336,6 +347,42 @@ mod tests { let _client: CoreGrpcClient = item.into(); } + #[tokio::test] + async fn test_connection_pool_shares_client_across_per_request_settings() { + let pool = ConnectionPool::new(10); + let uri = test_uri(); + + // Settings differing only in per-request knobs (timeout, retries, + // banning) must map to the same pooled connection... + let stored = RequestSettings { + timeout: Some(Duration::from_secs(30)), + retries: Some(3), + ban_failed_address: Some(false), + ..RequestSettings::default() + } + .finalize(); + pool.put(&uri, Some(&stored), make_platform_pool_item()); + + let default = RequestSettings::default().finalize(); + assert!( + pool.get(PoolPrefix::Platform, &uri, Some(&default)) + .is_some(), + "per-request settings must not split pooled connections" + ); + + // ...while connection-affecting settings still get their own entry. + let connect = RequestSettings { + connect_timeout: Some(Duration::from_secs(3)), + ..RequestSettings::default() + } + .finalize(); + assert!( + pool.get(PoolPrefix::Platform, &uri, Some(&connect)) + .is_none(), + "connection-affecting settings must key separate connections" + ); + } + #[tokio::test] async fn test_connection_pool_different_prefixes_different_keys() { 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 aa7aea5dbd2..6569b1ba4a3 100644 --- a/packages/rs-dapi-client/src/dapi_client.rs +++ b/packages/rs-dapi-client/src/dapi_client.rs @@ -239,10 +239,14 @@ pub fn update_address_ban_status( ); } } else { + // Banning is disabled for this request, but failover + // must still move traffic away from the failing node: + // drop it from the sticky rotation, ban state untouched. + address_list.evict_from_rotation(address); tracing::debug!( ?error, ?address, - "we should ban the address {address} due to the error but banning is disabled" + "banning is disabled; evicted address {address} from rotation due to the error" ); } } else { diff --git a/packages/rs-dapi-client/src/request_settings.rs b/packages/rs-dapi-client/src/request_settings.rs index 9ffbc7109ff..5685855b24c 100644 --- a/packages/rs-dapi-client/src/request_settings.rs +++ b/packages/rs-dapi-client/src/request_settings.rs @@ -80,6 +80,10 @@ impl RequestSettings { } /// DAPI settings ready to use. +/// +/// When adding a field, decide whether it affects the constructed transport +/// client and update `connection_key` accordingly (its exhaustive +/// destructuring will not compile until you do). #[derive(Debug, Clone)] pub struct AppliedRequestSettings { /// Timeout for establishing a connection. @@ -105,6 +109,64 @@ impl AppliedRequestSettings { self.ca_certificate = ca_cert; self } + + /// 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. + /// Per-request knobs (request timeout, retries, address banning) are + /// deliberately excluded so requests that differ only in those reuse the + /// same pooled connection. + pub(crate) fn connection_key(&self) -> String { + // Exhaustive destructuring: adding a settings field breaks this + // binding, forcing an explicit connection-affecting-or-not decision. + #[cfg(not(target_arch = "wasm32"))] + let Self { + connect_timeout, + timeout: _, + retries: _, + ban_failed_address: _, + max_decoding_message_size, + ca_certificate, + } = self; + #[cfg(target_arch = "wasm32")] + let Self { + connect_timeout, + timeout: _, + retries: _, + ban_failed_address: _, + max_decoding_message_size, + } = self; + + // The wasm transport ignores all settings when building its client + // (see `wasm_channel::create_channel`), so nothing may split the key + // there. + #[cfg(target_arch = "wasm32")] + let connect_timeout = { + let _ = connect_timeout; + &None:: + }; + + // The full certificate bytes (hex), not a short hash: two trust + // anchors must never share a pool key, or a request pinned to one CA + // silently reuses a channel built against the other. + #[cfg(not(target_arch = "wasm32"))] + let ca_certificate = ca_certificate.as_ref().map(|cert| { + use std::fmt::Write; + let bytes = cert.as_ref(); + let mut hex = String::with_capacity(bytes.len() * 2); + for byte in bytes { + write!(hex, "{byte:02x}").expect("writing to a String cannot fail"); + } + hex + }); + #[cfg(target_arch = "wasm32")] + let ca_certificate: Option = None; + + format!( + "connect_timeout={:?},max_decoding_message_size={:?},ca_certificate={:?}", + connect_timeout, max_decoding_message_size, ca_certificate + ) + } } #[cfg(test)] @@ -199,4 +261,57 @@ mod tests { let result = applied.with_ca_certificate(Some(cert)); assert!(result.ca_certificate.is_some()); } + + #[test] + fn test_connection_key_ignores_per_request_settings() { + let custom = RequestSettings { + timeout: Some(Duration::from_secs(30)), + retries: Some(1), + ban_failed_address: Some(false), + ..RequestSettings::default() + } + .finalize(); + let default = RequestSettings::default().finalize(); + + assert_eq!( + custom.connection_key(), + default.connection_key(), + "timeout/retries/banning must not split pooled connections" + ); + } + + #[test] + fn test_connection_key_differs_on_connection_settings() { + let default = RequestSettings::default().finalize(); + + let connect_timeout = RequestSettings { + connect_timeout: Some(Duration::from_secs(3)), + ..RequestSettings::default() + } + .finalize(); + assert_ne!(default.connection_key(), connect_timeout.connection_key()); + + let decode_limit = RequestSettings { + max_decoding_message_size: Some(16 * 1024 * 1024), + ..RequestSettings::default() + } + .finalize(); + assert_ne!(default.connection_key(), decode_limit.connection_key()); + } + + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn test_connection_key_differs_on_ca_certificate() { + let default = RequestSettings::default().finalize(); + let with_ca = RequestSettings::default() + .finalize() + .with_ca_certificate(Some(Certificate::from_pem("fake-pem-data"))); + + assert_ne!(default.connection_key(), with_ca.connection_key()); + + let with_other_ca = RequestSettings::default() + .finalize() + .with_ca_certificate(Some(Certificate::from_pem("other-pem-data"))); + assert_ne!(with_ca.connection_key(), with_other_ca.connection_key()); + } }