From 7ba19f7859f75f25fe2d0fa230ee686f7171df28 Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 31 Aug 2026 15:53:22 +0200 Subject: [PATCH 1/2] perf(dapi-client): reuse connections via sticky address rotation Address selection previously picked a uniformly random DAPI node from the full list (~259 hosts on mainnet) on every request attempt, so nearly every request landed on a cold host and paid a fresh TCP + TLS handshake, defeating the connection pool entirely (and, on WASM, the browser's per-origin connection reuse). AddressList now rotates round-robin over a small sticky active set (default 5, configurable via with_active_set_size). Banned or removed addresses are pruned from the set on the next selection and random live standby addresses are promoted in their place, so the existing ban ladder remains the only health signal and failover behavior is unchanged. The connection pool key now covers only connection-affecting settings (connect timeout, decode limit, CA certificate) instead of the whole applied-settings debug string, so requests differing only in per-request knobs (timeout, retries, banning) share one channel per host - e.g. broadcastStateTransition and waitForStateTransitionResult no longer handshake separately. Co-Authored-By: Claude Fable 5 --- packages/rs-dapi-client/Cargo.toml | 1 + packages/rs-dapi-client/src/address_list.rs | 245 ++++++++++++++++-- .../rs-dapi-client/src/connection_pool.rs | 46 +++- .../rs-dapi-client/src/request_settings.rs | 76 ++++++ 4 files changed, 344 insertions(+), 24 deletions(-) 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..57e8c5e9e9c 100644 --- a/packages/rs-dapi-client/src/address_list.rs +++ b/packages/rs-dapi-client/src/address_list.rs @@ -14,6 +14,13 @@ use std::time::Duration; const DEFAULT_BASE_BAN_PERIOD: Duration = Duration::from_secs(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; + /// DAPI address. #[derive(Debug, Clone, Eq)] #[cfg_attr(feature = "mocks", derive(serde::Serialize, serde::Deserialize))] @@ -148,6 +155,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,12 +181,26 @@ pub enum AddressListError { InvalidAddressUri(String), } +/// Sticky rotation state: the addresses currently receiving traffic, and the +/// most recently served one that round-robin selection advances from. +#[derive(Debug, Default)] +struct Rotation { + active: Vec
, + last_served: Option
, +} + /// 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 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, + active_set_size: usize, } impl Default for AddressList { @@ -196,10 +225,21 @@ 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, + active_set_size: DEFAULT_ACTIVE_SET_SIZE, } } + /// Set how many addresses receive traffic at a time (minimum 1). + /// + /// Smaller values maximize connection reuse, larger values spread load over + /// more nodes. + pub fn with_active_set_size(mut self, size: usize) -> Self { + self.active_set_size = size.max(1); + self + } + /// Bans address /// Returns false if the address is not in the list. /// @@ -294,7 +334,14 @@ 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. + /// + /// 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 or removed is dropped from the set here + /// and a random live standby address is promoted in its place, so the ban + /// ladder remains the only health signal. /// /// An address is considered live when it has never been banned or when its /// ban period has already expired. @@ -303,19 +350,52 @@ impl AddressList { // poisoned lock; adopt poison-tolerant locking consistently (SEC-003). 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()) + // 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 or no longer in the list. + rotation.active.retain(|address| { + guard + .get(address) + .map(|status| status.is_live(now)) + .unwrap_or(false) + }); + + // Refill vacancies with random live standby addresses. + let vacancies = self.active_set_size.saturating_sub(rotation.active.len()); + if vacancies > 0 { + let promoted = guard + .iter() + .filter(|&(address, status)| { + status.is_live(now) && !rotation.active.contains(address) + }) + .choose_multiple(&mut SmallRng::from_entropy(), vacancies); + + rotation + .active + .extend(promoted.into_iter().map(|(address, _)| address.clone())); + } + + if rotation.active.is_empty() { + return None; + } + + // Advance relative to the last-served address rather than a bare index: + // eviction shifts indices, and an index-based cursor could serve the + // same address twice in a row after 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(|address| address == last)); + + let index = last_position.map_or(0, |position| (position + 1) % rotation.active.len()); + let address = rotation.active[index].clone(); + rotation.last_served = Some(address.clone()); + Some(address) } /// Get all not banned addresses. @@ -344,12 +424,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 +445,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 +755,134 @@ 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() { + // Regression: with an index-based cursor, evicting an active member + // other than the one just served shifted indices and could serve the + // same address twice in a row. + 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_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..e7428ce7176 100644 --- a/packages/rs-dapi-client/src/connection_pool.rs +++ b/packages/rs-dapi-client/src/connection_pool.rs @@ -97,7 +97,13 @@ 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. + match settings { + Some(settings) => format!("{}:{}:{}", prefix, uri, settings.connection_key()), + None => format!("{}:{}", prefix, uri), + } } } @@ -176,8 +182,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 +344,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/request_settings.rs b/packages/rs-dapi-client/src/request_settings.rs index 9ffbc7109ff..0840cbaa763 100644 --- a/packages/rs-dapi-client/src/request_settings.rs +++ b/packages/rs-dapi-client/src/request_settings.rs @@ -105,6 +105,29 @@ 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 fn connection_key(&self) -> String { + #[cfg(not(target_arch = "wasm32"))] + let ca_certificate = self.ca_certificate.as_ref().map(|cert| { + use std::hash::{DefaultHasher, Hash, Hasher}; + let mut hasher = DefaultHasher::new(); + cert.as_ref().hash(&mut hasher); + hasher.finish() + }); + #[cfg(target_arch = "wasm32")] + let ca_certificate: Option = None; + + format!( + "connect_timeout={:?},max_decoding_message_size={:?},ca_certificate={:?}", + self.connect_timeout, self.max_decoding_message_size, ca_certificate + ) + } } #[cfg(test)] @@ -199,4 +222,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()); + } } From 155fc49214b8511e1a7887b6459a11a8ac941735 Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 31 Aug 2026 17:42:52 +0200 Subject: [PATCH 2/2] fix(sdk): harden sticky rotation per review feedback Active-set slots now expire after a jittered 5-7.5 minute lifetime, so no small set of nodes observes a client's whole query stream for the process lifetime while connections still stay warm for minutes at a time. Failover no longer depends on the ban ladder: when banning is disabled (ban_failed_address=false, e.g. FFI token operations), a failing node is evicted from the rotation without touching its ban state, instead of keeping its slot forever. The exponential ban ladder and server-advertised ban windows are capped at 24h, closing a DateTime overflow panic (reachable around ban_count 26) that poisoned the shared address-list lock. Also: active_set_size moved into the shared rotation state so all clones agree (and shrinking now trims the set); promotion count clamped to the list length so an oversized value cannot over-allocate; RNG seeded outside the rotation write lock so an entropy failure cannot poison it; pool key embeds full CA certificate bytes instead of a 64-bit non-cryptographic hash; connection_key narrowed to pub(crate) with exhaustive destructuring; connect_timeout excluded from the wasm pool key (wasm transport ignores it); pool key settings segment always present so the two branches cannot collide. --- packages/rs-dapi-client/src/address_list.rs | 382 ++++++++++++++++-- .../rs-dapi-client/src/connection_pool.rs | 5 +- packages/rs-dapi-client/src/dapi_client.rs | 6 +- .../rs-dapi-client/src/request_settings.rs | 55 ++- 4 files changed, 395 insertions(+), 53 deletions(-) diff --git a/packages/rs-dapi-client/src/address_list.rs b/packages/rs-dapi-client/src/address_list.rs index 57e8c5e9e9c..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,11 @@ 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 @@ -21,6 +26,13 @@ const DEFAULT_BASE_BAN_PERIOD: Duration = Duration::from_secs(60); /// 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))] @@ -98,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 @@ -111,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; @@ -121,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 @@ -138,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 @@ -181,26 +205,47 @@ pub enum AddressListError { InvalidAddressUri(String), } -/// Sticky rotation state: the addresses currently receiving traffic, and the -/// most recently served one that round-robin selection advances from. -#[derive(Debug, Default)] +/// 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
, + 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 and the rest of the list serves as failover standby (see +/// 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, - active_set_size: usize, } impl Default for AddressList { @@ -227,16 +272,22 @@ impl AddressList { addresses: Arc::new(RwLock::new(HashMap::new())), rotation: Arc::new(RwLock::new(Rotation::default())), base_ban_period, - active_set_size: DEFAULT_ACTIVE_SET_SIZE, } } - /// Set how many addresses receive traffic at a time (minimum 1). + /// Set how many addresses receive traffic at a time. /// - /// Smaller values maximize connection reuse, larger values spread load over - /// more nodes. - pub fn with_active_set_size(mut self, size: usize) -> Self { - self.active_set_size = size.max(1); + /// 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 } @@ -336,68 +387,112 @@ impl AddressList { /// 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 or removed is dropped from the set here - /// and a random live standby address is promoted in its place, so the ban - /// ladder remains the only health signal. + /// 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 now = chrono::Utc::now(); + // 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 or no longer in the list. - rotation.active.retain(|address| { - guard - .get(address) - .map(|status| status.is_live(now)) - .unwrap_or(false) + // 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) }); - // Refill vacancies with random live standby addresses. - let vacancies = self.active_set_size.saturating_sub(rotation.active.len()); + // 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.contains(address) + status.is_live(now) + && !rotation + .active + .iter() + .any(|member| member.address == *address) }) - .choose_multiple(&mut SmallRng::from_entropy(), vacancies); + .choose_multiple(&mut rng, vacancies); rotation .active - .extend(promoted.into_iter().map(|(address, _)| address.clone())); + .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 relative to the last-served address rather than a bare index: - // eviction shifts indices, and an index-based cursor could serve the - // same address twice in a row after 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(|address| address == last)); + // 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].clone(); + 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. /// /// Returns a vector of addresses that are not currently banned or whose ban period has expired. @@ -823,9 +918,6 @@ mod tests { #[test] fn test_get_live_address_no_immediate_repeat_after_other_member_evicted() { - // Regression: with an index-based cursor, evicting an active member - // other than the one just served shifted indices and could serve the - // same address twice in a row. 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()); @@ -883,6 +975,210 @@ mod tests { 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 e7428ce7176..b8e7b74775a 100644 --- a/packages/rs-dapi-client/src/connection_pool.rs +++ b/packages/rs-dapi-client/src/connection_pool.rs @@ -100,9 +100,12 @@ impl ConnectionPool { // 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!("{}:{}", prefix, uri), + None => format!("{}:{}:none", prefix, uri), } } } 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 0840cbaa763..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. @@ -112,20 +116,55 @@ impl AppliedRequestSettings { /// Per-request knobs (request timeout, retries, address banning) are /// deliberately excluded so requests that differ only in those reuse the /// same pooled connection. - pub fn connection_key(&self) -> String { + 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 = self.ca_certificate.as_ref().map(|cert| { - use std::hash::{DefaultHasher, Hash, Hasher}; - let mut hasher = DefaultHasher::new(); - cert.as_ref().hash(&mut hasher); - hasher.finish() + 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; + let ca_certificate: Option = None; format!( "connect_timeout={:?},max_decoding_message_size={:?},ca_certificate={:?}", - self.connect_timeout, self.max_decoding_message_size, ca_certificate + connect_timeout, max_decoding_message_size, ca_certificate ) } }