From c1cea6e8135ab9ef9359987e09f5661c63ba5372 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Mon, 17 Aug 2026 19:26:26 +0200 Subject: [PATCH] v0.2.1-pre.004 --- Cargo.toml | 4 +- ROADMAP.md | 4 +- crates/ksp-onchain-transport-lib/Cargo.toml | 3 +- .../ksp-onchain-transport-lib/src/client.rs | 240 ++++++++-- crates/ksp-onchain-transport-lib/src/lib.rs | 19 +- crates/ksp-onchain-transport-lib/src/pool.rs | 451 ++++++++++++++++-- .../src/resilience.rs | 390 +++++++++++++++ .../ksp-onchain-transport-lib/src/settings.rs | 11 +- .../tests/dependency_boundary.rs | 3 +- .../tests/public_api.rs | 61 ++- .../unit_tests/client.rs | 15 +- .../unit_tests/pool.rs | 164 ++++++- .../unit_tests/resilience.rs | 186 ++++++++ deltas/0.2.1/pre.004.md | 231 +++++++++ docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md | 24 +- 15 files changed, 1713 insertions(+), 93 deletions(-) create mode 100644 crates/ksp-onchain-transport-lib/src/resilience.rs create mode 100644 crates/ksp-onchain-transport-lib/unit_tests/resilience.rs create mode 100644 deltas/0.2.1/pre.004.md diff --git a/Cargo.toml b/Cargo.toml index 02fb78a..7d5d794 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 99 +# version: 100 [workspace] resolver = "3" members = ["crates/ksp-app-config-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-logging-lib", "crates/ksp-onchain-transport-lib"] [workspace.package] -version = "0.2.1-pre.3.fix.1" +version = "0.2.1-pre.4" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/ROADMAP.md b/ROADMAP.md index 666c703..b5f2a19 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,5 +1,5 @@ - + # Roadmap KSP @@ -41,7 +41,7 @@ Le roadmap décrit les objectifs à atteindre et les grandes étapes prévues. U ### Cadrage - [X] `0.2.0` — Audit bot3, ordre fonctionnel de `0.2.x`, architecture durable, discipline de sizing et pipeline RAW/CORE/DECODE/SPECIALIZED stabilisés. -- [/] `0.2.1` — HTTP foundation en cours : audit/sizing et matrice 52+14 stabilisés ; crate/settings/JSON-RPC/registry acquis ; client endpoint/pool logique et routing priority/fairness acquis ; prochaine étape : limites, concurrence, cooldown, timeout/retry effectif ; 4 canaris, Config standard et clôture restent planifiés. +- [/] `0.2.1` — HTTP foundation en cours : matrice 52+14, crate/settings/JSON-RPC/registry, client/pool/routing et résilience runtime (RPS/burst, concurrence, cooldown, deadline et retry borné/no-resend) acquis ; 4 canaris, Config standard et clôture restent planifiés. ### Releases fonctionnelles décidées/pressenties diff --git a/crates/ksp-onchain-transport-lib/Cargo.toml b/crates/ksp-onchain-transport-lib/Cargo.toml index 811a779..fa95723 100644 --- a/crates/ksp-onchain-transport-lib/Cargo.toml +++ b/crates/ksp-onchain-transport-lib/Cargo.toml @@ -1,5 +1,5 @@ # file: crates/ksp-onchain-transport-lib/Cargo.toml -# version: 1 +# version: 2 [package] name = "ksp-onchain-transport-lib" @@ -13,6 +13,7 @@ ksp-logging-lib = { path = "../ksp-logging-lib" } reqwest.workspace = true serde.workspace = true serde_json.workspace = true +tokio = { workspace = true, features = ["sync"] } [lints] workspace = true diff --git a/crates/ksp-onchain-transport-lib/src/client.rs b/crates/ksp-onchain-transport-lib/src/client.rs index c46f805..4bc1a41 100644 --- a/crates/ksp-onchain-transport-lib/src/client.rs +++ b/crates/ksp-onchain-transport-lib/src/client.rs @@ -1,26 +1,35 @@ // file: crates/ksp-onchain-transport-lib/src/client.rs -// version: 1 +// version: 2 -/// Passive runtime availability reported for one logical HTTP endpoint. +/// Passive runtime availability reported for one logical HTTP endpoint or role. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum HttpEndpointAvailability { - /// The endpoint is administratively disabled and cannot be selected. + /// The endpoint or role is administratively disabled and cannot be selected. Disabled, - /// The endpoint is enabled and currently eligible for selection. + /// The endpoint or role is enabled and currently eligible for selection. Available, - /// The endpoint is enabled but temporarily degraded by runtime observations. + /// The endpoint or role is enabled but degraded by recent passive runtime observations. Degraded, - /// The endpoint is enabled but temporarily excluded after provider rate limiting. + /// The endpoint or role is temporarily excluded after provider rate limiting. RateLimited, } -/// Safe routing snapshot for one configured endpoint role. +/// Safe routing and resilience snapshot for one configured endpoint role. #[derive(Clone, Debug, Eq, PartialEq)] pub struct HttpEndpointRoleSnapshot { role: std::string::String, enabled: bool, request_kinds: std::vec::Vec, priority: u32, + availability: crate::HttpEndpointAvailability, + requests_per_second: std::option::Option, + burst_capacity: std::option::Option, + max_concurrent_requests: std::option::Option, + in_flight_requests: std::option::Option, + cooldown_remaining: std::option::Option, + success_count: u64, + failure_count: u64, + rate_limit_count: u64, } impl HttpEndpointRoleSnapshot { @@ -47,6 +56,60 @@ impl HttpEndpointRoleSnapshot { pub const fn priority(&self) -> u32 { return self.priority; } + + /// Returns the passive runtime availability of this role. + #[must_use] + pub const fn availability(&self) -> crate::HttpEndpointAvailability { + return self.availability; + } + + /// Returns the configured requests-per-second limit. + #[must_use] + pub const fn requests_per_second(&self) -> std::option::Option { + return self.requests_per_second; + } + + /// Returns the configured token-bucket burst capacity. + #[must_use] + pub const fn burst_capacity(&self) -> std::option::Option { + return self.burst_capacity; + } + + /// Returns the configured maximum concurrent request count. + #[must_use] + pub const fn max_concurrent_requests(&self) -> std::option::Option { + return self.max_concurrent_requests; + } + + /// Returns the number of in-flight requests when concurrency is bounded. + #[must_use] + pub const fn in_flight_requests(&self) -> std::option::Option { + return self.in_flight_requests; + } + + /// Returns the remaining provider cooldown when this role is rate-limited. + #[must_use] + pub const fn cooldown_remaining(&self) -> std::option::Option { + return self.cooldown_remaining; + } + + /// Returns the number of successful requests passively recorded for this role. + #[must_use] + pub const fn success_count(&self) -> u64 { + return self.success_count; + } + + /// Returns the number of failed requests passively recorded for this role. + #[must_use] + pub const fn failure_count(&self) -> u64 { + return self.failure_count; + } + + /// Returns the number of provider rate-limit observations recorded for this role. + #[must_use] + pub const fn rate_limit_count(&self) -> u64 { + return self.rate_limit_count; + } } /// Safe metadata snapshot for one logical HTTP endpoint. @@ -111,12 +174,16 @@ pub struct HttpEndpointClient { struct HttpEndpointClientInner { settings: crate::HttpEndpointSettings, _client: reqwest::Client, - availability: std::sync::atomic::AtomicU8, + role_runtimes: std::vec::Vec>, } impl HttpEndpointClient { /// Builds one logical endpoint client from KSP-owned runtime settings. pub fn new(settings: crate::HttpEndpointSettings) -> ksp_core_lib::Result { + return Self::new_with_notify(settings, std::sync::Arc::new(tokio::sync::Notify::new())); + } + + pub(crate) fn new_with_notify(settings: crate::HttpEndpointSettings, notify: std::sync::Arc) -> ksp_core_lib::Result { let validation = crate::settings::validate_endpoint_settings(&settings); if let std::result::Result::Err(error) = validation { return std::result::Result::Err(error); @@ -132,18 +199,20 @@ impl HttpEndpointClient { ); }, }; - let availability = if settings.enabled() { AVAILABILITY_AVAILABLE } else { AVAILABILITY_DISABLED }; + let mut role_runtimes = std::vec::Vec::with_capacity(settings.roles().len()); + for role in settings.roles() { + role_runtimes.push(std::sync::Arc::new(crate::resilience::HttpRoleRuntime::new(role, std::sync::Arc::clone(¬ify)))); + } ksp_logging_lib::debug!( target: env!("CARGO_PKG_NAME"), endpoint_name = settings.name(), provider = settings.provider().as_str(), cluster = settings.cluster().as_str(), enabled = settings.enabled(), + role_count = settings.roles().len(), "created logical HTTP endpoint client" ); - return std::result::Result::Ok(Self { - inner: std::sync::Arc::new(HttpEndpointClientInner { settings, _client: client, availability: std::sync::atomic::AtomicU8::new(availability) }), - }); + return std::result::Result::Ok(Self { inner: std::sync::Arc::new(HttpEndpointClientInner { settings, _client: client, role_runtimes }) }); } /// Returns the endpoint identity used for safe diagnostics and routing. @@ -170,7 +239,13 @@ impl HttpEndpointClient { return self.inner.settings.enabled(); } - /// Returns whether one enabled role can serve the requested capability. + /// Returns the configured end-to-end request timeout. + #[must_use] + pub fn request_timeout(&self) -> std::time::Duration { + return self.inner.settings.request_timeout(); + } + + /// Returns whether one enabled role can serve the requested capability structurally. #[must_use] pub fn supports(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> bool { return self.matching_role(role, request_kind).is_some(); @@ -180,17 +255,17 @@ impl HttpEndpointClient { #[must_use] pub fn snapshot(&self) -> crate::HttpEndpointSnapshot { let mut roles = std::vec::Vec::with_capacity(self.inner.settings.roles().len()); - for role in self.inner.settings.roles() { + for (role_index, role) in self.inner.settings.roles().iter().enumerate() { let mut request_kinds = std::vec::Vec::with_capacity(role.request_kinds().len()); for request_kind in role.request_kinds() { request_kinds.push(request_kind.as_str().to_owned()); } - roles.push(crate::HttpEndpointRoleSnapshot { - role: role.role().as_str().to_owned(), - enabled: role.enabled(), - request_kinds, - priority: role.priority(), - }); + let runtime = self.inner.role_runtimes.get(role_index); + let role_snapshot = match runtime { + std::option::Option::Some(runtime) => role_snapshot(role, request_kinds, runtime), + std::option::Option::None => fallback_role_snapshot(role, request_kinds), + }; + roles.push(role_snapshot); } return crate::HttpEndpointSnapshot { name: self.name().to_owned(), @@ -207,7 +282,7 @@ impl HttpEndpointClient { role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind, ) -> std::option::Option<&'a crate::HttpEndpointRoleSettings> { - if !self.is_selectable() { + if !self.enabled() { return std::option::Option::None; } for candidate_role in self.inner.settings.roles() { @@ -223,12 +298,68 @@ impl HttpEndpointClient { return std::option::Option::None; } - fn is_selectable(&self) -> bool { - return self.enabled() && self.availability() == crate::HttpEndpointAvailability::Available; + pub(crate) fn matching_role_runtime( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + ) -> std::option::Option<(u32, std::sync::Arc)> { + if !self.enabled() { + return std::option::Option::None; + } + for (role_index, candidate_role) in self.inner.settings.roles().iter().enumerate() { + if !candidate_role.enabled() || candidate_role.role() != role { + continue; + } + let mut handles = false; + for capability in candidate_role.request_kinds() { + if capability.is_wildcard() || capability == request_kind { + handles = true; + break; + } + } + if !handles { + continue; + } + let runtime = self.inner.role_runtimes.get(role_index); + if let std::option::Option::Some(runtime) = runtime { + return std::option::Option::Some((candidate_role.priority(), std::sync::Arc::clone(runtime))); + } + } + return std::option::Option::None; } - fn availability(&self) -> crate::HttpEndpointAvailability { - return availability_from_code(self.inner.availability.load(std::sync::atomic::Ordering::Relaxed)); + pub(crate) fn availability(&self) -> crate::HttpEndpointAvailability { + if !self.enabled() { + return crate::HttpEndpointAvailability::Disabled; + } + let now = std::time::Instant::now(); + let mut enabled_role_count = 0_usize; + let mut rate_limited_count = 0_usize; + let mut degraded = false; + for (role_index, role) in self.inner.settings.roles().iter().enumerate() { + if !role.enabled() { + continue; + } + enabled_role_count = enabled_role_count.saturating_add(1); + let runtime = self.inner.role_runtimes.get(role_index); + let availability = match runtime { + std::option::Option::Some(runtime) => runtime.availability(now), + std::option::Option::None => crate::HttpEndpointAvailability::Degraded, + }; + if availability == crate::HttpEndpointAvailability::RateLimited { + rate_limited_count = rate_limited_count.saturating_add(1); + } + if availability == crate::HttpEndpointAvailability::Degraded || availability == crate::HttpEndpointAvailability::RateLimited { + degraded = true; + } + } + if enabled_role_count > 0 && rate_limited_count == enabled_role_count { + return crate::HttpEndpointAvailability::RateLimited; + } + if degraded || enabled_role_count == 0 { + return crate::HttpEndpointAvailability::Degraded; + } + return crate::HttpEndpointAvailability::Available; } } @@ -238,6 +369,47 @@ impl std::fmt::Debug for HttpEndpointClient { } } +fn role_snapshot( + role: &crate::HttpEndpointRoleSettings, + request_kinds: std::vec::Vec, + runtime: &crate::resilience::HttpRoleRuntime, +) -> crate::HttpEndpointRoleSnapshot { + let availability = if role.enabled() { runtime.availability(std::time::Instant::now()) } else { crate::HttpEndpointAvailability::Disabled }; + return crate::HttpEndpointRoleSnapshot { + role: role.role().as_str().to_owned(), + enabled: role.enabled(), + request_kinds, + priority: role.priority(), + availability, + requests_per_second: role.limits().requests_per_second().map(|value| return value.get()), + burst_capacity: role.limits().burst_capacity().map(|value| return value.get()), + max_concurrent_requests: runtime.max_concurrent_requests(), + in_flight_requests: runtime.in_flight_requests(), + cooldown_remaining: runtime.cooldown_remaining(), + success_count: runtime.success_count(), + failure_count: runtime.failure_count(), + rate_limit_count: runtime.rate_limit_count(), + }; +} + +fn fallback_role_snapshot(role: &crate::HttpEndpointRoleSettings, request_kinds: std::vec::Vec) -> crate::HttpEndpointRoleSnapshot { + return crate::HttpEndpointRoleSnapshot { + role: role.role().as_str().to_owned(), + enabled: role.enabled(), + request_kinds, + priority: role.priority(), + availability: crate::HttpEndpointAvailability::Degraded, + requests_per_second: role.limits().requests_per_second().map(|value| return value.get()), + burst_capacity: role.limits().burst_capacity().map(|value| return value.get()), + max_concurrent_requests: role.limits().max_concurrent_requests().map(|value| return value.get()), + in_flight_requests: std::option::Option::None, + cooldown_remaining: std::option::Option::None, + success_count: 0, + failure_count: 0, + rate_limit_count: 0, + }; +} + fn build_reqwest_client(settings: &crate::HttpEndpointSettings) -> std::result::Result { let mut builder = reqwest::Client::builder() .connect_timeout(settings.connect_timeout()) @@ -251,24 +423,6 @@ fn build_reqwest_client(settings: &crate::HttpEndpointSettings) -> std::result:: return builder.build(); } -const AVAILABILITY_DISABLED: u8 = 0; -const AVAILABILITY_AVAILABLE: u8 = 1; -const AVAILABILITY_DEGRADED: u8 = 2; -const AVAILABILITY_RATE_LIMITED: u8 = 3; - -fn availability_from_code(code: u8) -> crate::HttpEndpointAvailability { - if code == AVAILABILITY_DISABLED { - return crate::HttpEndpointAvailability::Disabled; - } - if code == AVAILABILITY_DEGRADED { - return crate::HttpEndpointAvailability::Degraded; - } - if code == AVAILABILITY_RATE_LIMITED { - return crate::HttpEndpointAvailability::RateLimited; - } - return crate::HttpEndpointAvailability::Available; -} - #[cfg(test)] #[path = "../unit_tests/client.rs"] mod tests; diff --git a/crates/ksp-onchain-transport-lib/src/lib.rs b/crates/ksp-onchain-transport-lib/src/lib.rs index 01a8de3..9d56189 100644 --- a/crates/ksp-onchain-transport-lib/src/lib.rs +++ b/crates/ksp-onchain-transport-lib/src/lib.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/lib.rs -// version: 2 +// version: 3 #![warn(missing_docs)] #![deny(unreachable_pub)] #![forbid(unsafe_code)] @@ -8,13 +8,14 @@ //! //! This crate owns runtime HTTP transport settings, Solana HTTP JSON-RPC envelopes and the audited standard method registry. It deliberately remains //! independent from `ksp-config-lib`, Store and Program layers. Config may later construct these public settings through a one-way adapter. Logical endpoint -//! clients and priority-aware pools are now available, while resilience limits and typed Solana method adapters remain staged by subsequent `0.2.1` -//! prereleases. +//! clients, priority-aware pools, bounded admission limits and retry/no-resend policy are available, while typed Solana method adapters remain staged by +//! subsequent `0.2.1` prereleases. mod client; mod error; mod json_rpc; mod pool; +mod resilience; mod rpc_method; mod settings; @@ -66,10 +67,20 @@ pub use self::json_rpc::parse_json_rpc_response_text; pub use self::json_rpc::parse_json_rpc_response_value; /// Result of one logical endpoint selection. pub use self::pool::HttpEndpointSelection; -/// Shareable logical HTTP endpoint pool with priority routing and round-robin fairness. +/// Runtime admission permit for one HTTP request. +pub use self::pool::HttpRequestPermit; +/// Shareable logical HTTP endpoint pool with priority routing, admission limits and bounded deadlines. pub use self::pool::HttpTransportPool; /// Safe snapshot of the logical HTTP endpoint pool. pub use self::pool::HttpTransportPoolSnapshot; +/// Dispatch knowledge used to prevent ambiguous automatic resubmission. +pub use self::resilience::HttpDispatchState; +/// Transport-level cause considered by the bounded retry policy. +pub use self::resilience::HttpRetryCause; +/// Result of evaluating one bounded transport retry opportunity. +pub use self::resilience::HttpRetryDecision; +/// Evaluates the centralized bounded HTTP retry policy for one audited RPC method. +pub use self::resilience::evaluate_transport_retry; /// Functional category used by the audited Solana HTTP JSON-RPC registry. pub use self::rpc_method::HttpRpcCategory; /// Release that owns typed KSP coverage for one audited HTTP RPC method. diff --git a/crates/ksp-onchain-transport-lib/src/pool.rs b/crates/ksp-onchain-transport-lib/src/pool.rs index 854e2ad..02b4a2b 100644 --- a/crates/ksp-onchain-transport-lib/src/pool.rs +++ b/crates/ksp-onchain-transport-lib/src/pool.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/pool.rs -// version: 2 +// version: 3 /// Safe snapshot of the logical HTTP endpoint pool. #[derive(Clone, Debug, Eq, PartialEq)] @@ -20,7 +20,7 @@ impl HttpTransportPoolSnapshot { return self.endpoints.len(); } - /// Returns the number of endpoints currently eligible for routing. + /// Returns the number of endpoints currently eligible for normal routing. #[must_use] pub fn available_endpoint_count(&self) -> usize { return self.endpoints.iter().filter(|endpoint| return endpoint.availability() == crate::HttpEndpointAvailability::Available).count(); @@ -68,7 +68,98 @@ impl HttpEndpointSelection { } } -/// Shareable logical HTTP endpoint pool with priority routing and per-capability round-robin fairness. +/// Runtime admission permit for one HTTP request. +/// +/// The permit reserves configured concurrency capacity and carries the common request deadline. Dropping it releases any semaphore capacity immediately. +pub struct HttpRequestPermit { + selection: crate::HttpEndpointSelection, + deadline: std::time::Instant, + role_runtime: std::sync::Arc, + _concurrency_permit: crate::resilience::HttpConcurrencyPermit, +} + +impl HttpRequestPermit { + /// Returns the selected logical endpoint and role. + #[must_use] + pub const fn selection(&self) -> &crate::HttpEndpointSelection { + return &self.selection; + } + + /// Returns the selected endpoint client. + #[must_use] + pub fn client(&self) -> &crate::HttpEndpointClient { + return self.selection.client(); + } + + /// Returns the remaining duration in the common request budget. + #[must_use] + pub fn remaining_timeout(&self) -> std::time::Duration { + let now = std::time::Instant::now(); + if now >= self.deadline { + return std::time::Duration::ZERO; + } + return self.deadline.duration_since(now); + } + + /// Records a successful request and clears the passive degraded state for this endpoint role. + pub fn record_success(&self) { + self.role_runtime.record_success(); + ksp_logging_lib::trace!( + target: env!("CARGO_PKG_NAME"), + endpoint_name = self.selection.endpoint_name(), + role = self.selection.role().as_str(), + request_kind = self.selection.request_kind().as_str(), + "recorded successful HTTP endpoint observation" + ); + return; + } + + /// Records a transport failure and marks the endpoint role degraded until a later success. + pub fn record_failure(&self) { + self.role_runtime.record_failure(); + ksp_logging_lib::warn!( + target: env!("CARGO_PKG_NAME"), + endpoint_name = self.selection.endpoint_name(), + provider = self.selection.client().provider().as_str(), + cluster = self.selection.client().cluster().as_str(), + role = self.selection.role().as_str(), + request_kind = self.selection.request_kind().as_str(), + "HTTP endpoint role marked degraded after transport failure" + ); + return; + } + + /// Records provider rate limiting and applies the role cooldown. + /// + /// A provider delay can extend the configured cooldown but is defensively capped by the Transport runtime before use. + pub fn record_rate_limited(&self, provider_retry_after: std::option::Option) -> std::time::Duration { + let pause = self.role_runtime.record_rate_limited(provider_retry_after); + let cooldown_ms = duration_millis_u64(pause); + ksp_logging_lib::warn!( + target: env!("CARGO_PKG_NAME"), + endpoint_name = self.selection.endpoint_name(), + provider = self.selection.client().provider().as_str(), + cluster = self.selection.client().cluster().as_str(), + role = self.selection.role().as_str(), + request_kind = self.selection.request_kind().as_str(), + cooldown_ms, + "HTTP endpoint role entered provider rate-limit cooldown" + ); + return pause; + } +} + +impl std::fmt::Debug for HttpRequestPermit { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter + .debug_struct("HttpRequestPermit") + .field("selection", &self.selection) + .field("remaining_timeout", &self.remaining_timeout()) + .finish(); + } +} + +/// Shareable logical HTTP endpoint pool with priority routing, admission limits and bounded request deadlines. #[derive(Clone)] pub struct HttpTransportPool { inner: std::sync::Arc, @@ -76,6 +167,8 @@ pub struct HttpTransportPool { struct HttpTransportPoolInner { clients: std::vec::Vec, + retry: crate::HttpRetrySettings, + notify: std::sync::Arc, cursors: std::sync::Mutex>, } @@ -86,9 +179,10 @@ impl HttpTransportPool { if let std::result::Result::Err(error) = validation { return std::result::Result::Err(error); } + let notify = std::sync::Arc::new(tokio::sync::Notify::new()); let mut clients = std::vec::Vec::with_capacity(settings.endpoints().len()); for endpoint in settings.endpoints() { - let client_result = crate::HttpEndpointClient::new(endpoint.clone()); + let client_result = crate::HttpEndpointClient::new_with_notify(endpoint.clone(), std::sync::Arc::clone(¬ify)); let client = match client_result { std::result::Result::Ok(client) => client, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -96,18 +190,32 @@ impl HttpTransportPool { clients.push(client); } let pool = Self { - inner: std::sync::Arc::new(HttpTransportPoolInner { clients, cursors: std::sync::Mutex::new(std::collections::BTreeMap::new()) }), + inner: std::sync::Arc::new(HttpTransportPoolInner { + clients, + retry: settings.retry().clone(), + notify, + cursors: std::sync::Mutex::new(std::collections::BTreeMap::new()), + }), }; ksp_logging_lib::debug!( target: env!("CARGO_PKG_NAME"), endpoint_count = pool.inner.clients.len(), available_endpoint_count = pool.snapshot().available_endpoint_count(), + max_retries = pool.inner.retry.max_retries(), "created logical HTTP endpoint pool" ); return std::result::Result::Ok(pool); } - /// Selects an endpoint for one standard audited RPC method. + /// Returns the bounded transport retry settings owned by this pool. + #[must_use] + pub fn retry_settings(&self) -> &crate::HttpRetrySettings { + return &self.inner.retry; + } + + /// Selects an endpoint for one standard audited RPC method without reserving runtime capacity. + /// + /// Request execution should use `acquire_for_method` so rate, cooldown and concurrency limits are enforced. pub fn select_for_method(&self, role: &crate::HttpRoleName, method: &crate::HttpRpcMethodDescriptor) -> ksp_core_lib::Result { let support = method.ensure_runtime_supported(); if let std::result::Result::Err(error) = support { @@ -116,12 +224,224 @@ impl HttpTransportPool { return self.select_for_request_kind(role, &crate::HttpRequestKind::new(method.request_kind())); } - /// Selects an endpoint for an open request-kind descriptor, including provider extensions. + /// Selects an endpoint for an open request-kind descriptor without reserving runtime capacity. pub fn select_for_request_kind( &self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind, ) -> ksp_core_lib::Result { + let candidates = self.static_candidates(role, request_kind); + if candidates.is_empty() { + return selection_failed(role, request_kind); + } + let best_priority = candidates[0].priority; + let mut tier_size = 0_usize; + for candidate in &candidates { + if candidate.priority != best_priority { + break; + } + tier_size = tier_size.saturating_add(1); + } + let selected_position = self.next_position(role, request_kind, best_priority, tier_size); + let selected = match candidates.get(selected_position) { + std::option::Option::Some(selected) => selected, + std::option::Option::None => return selection_failed(role, request_kind), + }; + return self.selection_from_candidate(role, request_kind, selected); + } + + /// Acquires runtime capacity for one standard audited RPC method using the common timeout of matching endpoints. + pub async fn acquire_for_method( + &self, + role: &crate::HttpRoleName, + method: &crate::HttpRpcMethodDescriptor, + ) -> ksp_core_lib::Result { + let support = method.ensure_runtime_supported(); + if let std::result::Result::Err(error) = support { + return std::result::Result::Err(error); + } + return self.acquire_for_request_kind(role, &crate::HttpRequestKind::new(method.request_kind())).await; + } + + /// Acquires runtime capacity for an open request-kind descriptor using the shortest configured request timeout among matching endpoints as the common + /// deadline. + pub async fn acquire_for_request_kind( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + ) -> ksp_core_lib::Result { + let timeout_result = self.common_request_timeout(role, request_kind); + let timeout = match timeout_result { + std::result::Result::Ok(timeout) => timeout, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return self.acquire_for_request_kind_with_timeout(role, request_kind, timeout).await; + } + + /// Acquires runtime capacity with an explicit end-to-end admission budget. + /// + /// This is primarily useful when a higher layer already owns a stricter request deadline. A zero timeout is rejected as immediately expired. + pub async fn acquire_for_request_kind_with_timeout( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + timeout: std::time::Duration, + ) -> ksp_core_lib::Result { + if timeout.is_zero() { + return request_timeout(role, request_kind, "HTTP request admission deadline expired before selection"); + } + let now = std::time::Instant::now(); + let deadline = match now.checked_add(timeout) { + std::option::Option::Some(value) => value, + std::option::Option::None => return request_timeout(role, request_kind, "HTTP request admission deadline could not be represented"), + }; + return self.acquire_until(role, request_kind, deadline).await; + } + + /// Returns a safe pool snapshot without endpoint URLs. + #[must_use] + pub fn snapshot(&self) -> crate::HttpTransportPoolSnapshot { + let endpoints = self.inner.clients.iter().map(|client| return client.snapshot()).collect(); + return crate::HttpTransportPoolSnapshot { endpoints }; + } + + async fn acquire_until( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + deadline: std::time::Instant, + ) -> ksp_core_lib::Result { + let candidates = self.runtime_candidates(role, request_kind); + if candidates.is_empty() { + return request_selection_failed(role, request_kind); + } + loop { + let now = std::time::Instant::now(); + if now >= deadline { + return request_timeout(role, request_kind, "HTTP request admission deadline expired while waiting for endpoint capacity"); + } + let attempt = self.try_candidates(role, request_kind, candidates.as_slice(), now, deadline); + match attempt { + RuntimeSelectionAttempt::Ready(permit) => return std::result::Result::Ok(permit), + RuntimeSelectionAttempt::Blocked { earliest_ready, concurrency_saturated } => { + let wait_result = self.wait_for_capacity(earliest_ready, concurrency_saturated, deadline).await; + if !wait_result { + return request_timeout(role, request_kind, "HTTP request admission deadline expired while waiting for endpoint capacity"); + } + }, + RuntimeSelectionAttempt::Unavailable => return request_selection_failed(role, request_kind), + } + } + } + + fn try_candidates( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + candidates: &[RuntimePoolCandidate], + now: std::time::Instant, + deadline: std::time::Instant, + ) -> RuntimeSelectionAttempt { + let mut earliest_ready: std::option::Option = std::option::Option::None; + let mut concurrency_saturated = false; + let mut tier_start = 0_usize; + while tier_start < candidates.len() { + let priority = candidates[tier_start].priority; + let mut tier_end = tier_start; + while tier_end < candidates.len() && candidates[tier_end].priority == priority { + tier_end = tier_end.saturating_add(1); + } + let tier_size = tier_end.saturating_sub(tier_start); + let start_position = self.next_position(role, request_kind, priority, tier_size); + let mut offset = 0_usize; + while offset < tier_size { + let position = tier_start.saturating_add((start_position.saturating_add(offset)) % tier_size); + let candidate = &candidates[position]; + let admission = candidate.runtime.try_acquire(now); + match admission { + crate::resilience::RoleAdmissionAttempt::Ready(concurrency_permit) => { + let selection_result = self.selection_from_runtime_candidate(role, request_kind, candidate); + let selection = match selection_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return RuntimeSelectionAttempt::Unavailable, + }; + ksp_logging_lib::debug!( + target: env!("CARGO_PKG_NAME"), + endpoint_name = selection.endpoint_name(), + role = role.as_str(), + request_kind = request_kind.as_str(), + priority, + remaining_deadline_ms = duration_millis_u64(deadline.saturating_duration_since(now)), + "admitted HTTP request through logical endpoint pool" + ); + return RuntimeSelectionAttempt::Ready(crate::HttpRequestPermit { + selection, + deadline, + role_runtime: std::sync::Arc::clone(&candidate.runtime), + _concurrency_permit: concurrency_permit, + }); + }, + crate::resilience::RoleAdmissionAttempt::BlockedUntil(ready_at) => { + earliest_ready = earlier_instant(earliest_ready, ready_at); + }, + crate::resilience::RoleAdmissionAttempt::ConcurrencySaturated => { + concurrency_saturated = true; + }, + crate::resilience::RoleAdmissionAttempt::Unavailable => {}, + } + offset = offset.saturating_add(1); + } + tier_start = tier_end; + } + if earliest_ready.is_none() && !concurrency_saturated { + return RuntimeSelectionAttempt::Unavailable; + } + return RuntimeSelectionAttempt::Blocked { earliest_ready, concurrency_saturated }; + } + + async fn wait_for_capacity( + &self, + earliest_ready: std::option::Option, + concurrency_saturated: bool, + deadline: std::time::Instant, + ) -> bool { + let now = std::time::Instant::now(); + if now >= deadline { + return false; + } + let wake_at = match earliest_ready { + std::option::Option::Some(ready_at) => std::cmp::min(ready_at, deadline), + std::option::Option::None => deadline, + }; + if concurrency_saturated { + tokio::select! { + () = self.inner.notify.notified() => {}, + () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => {}, + } + } else { + tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)).await; + } + return std::time::Instant::now() < deadline; + } + + fn common_request_timeout(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> ksp_core_lib::Result { + let mut timeout: std::option::Option = std::option::Option::None; + for client in &self.inner.clients { + if client.matching_role(role, request_kind).is_none() { + continue; + } + timeout = match timeout { + std::option::Option::Some(current) => std::option::Option::Some(std::cmp::min(current, client.request_timeout())), + std::option::Option::None => std::option::Option::Some(client.request_timeout()), + }; + } + return match timeout { + std::option::Option::Some(value) => std::result::Result::Ok(value), + std::option::Option::None => selection_failed_duration(role, request_kind), + }; + } + + fn static_candidates(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> std::vec::Vec { let mut candidates = std::vec::Vec::new(); for (client_index, client) in self.inner.clients.iter().enumerate() { let matching_role = client.matching_role(role, request_kind); @@ -129,22 +449,29 @@ impl HttpTransportPool { candidates.push(PoolCandidate { client_index, priority: matching_role.priority() }); } } - if candidates.is_empty() { - return selection_failed(role, request_kind); + candidates.sort_by_key(|candidate| return candidate.priority); + return candidates; + } + + fn runtime_candidates(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> std::vec::Vec { + let mut candidates = std::vec::Vec::new(); + for (client_index, client) in self.inner.clients.iter().enumerate() { + let runtime_match = client.matching_role_runtime(role, request_kind); + if let std::option::Option::Some((priority, runtime)) = runtime_match { + candidates.push(RuntimePoolCandidate { client_index, priority, runtime }); + } } candidates.sort_by_key(|candidate| return candidate.priority); - let first_candidate = candidates.first(); - let best_priority = match first_candidate { - std::option::Option::Some(candidate) => candidate.priority, - std::option::Option::None => return selection_failed(role, request_kind), - }; - let best_tier: std::vec::Vec = candidates.into_iter().take_while(|candidate| return candidate.priority == best_priority).collect(); - let selected_position = self.next_position(role, request_kind, best_priority, best_tier.len()); - let selected = match best_tier.get(selected_position) { - std::option::Option::Some(selected) => selected, - std::option::Option::None => return selection_failed(role, request_kind), - }; - let selected_client = self.inner.clients.get(selected.client_index); + return candidates; + } + + fn selection_from_candidate( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + candidate: &PoolCandidate, + ) -> ksp_core_lib::Result { + let selected_client = self.inner.clients.get(candidate.client_index); let client = match selected_client { std::option::Option::Some(client) => client.clone(), std::option::Option::None => return selection_failed(role, request_kind), @@ -154,23 +481,34 @@ impl HttpTransportPool { endpoint_name = client.name(), role = role.as_str(), request_kind = request_kind.as_str(), - priority = best_priority, - tier_size = best_tier.len(), - "selected logical HTTP endpoint" + priority = candidate.priority, + "selected logical HTTP endpoint without runtime admission" ); return std::result::Result::Ok(crate::HttpEndpointSelection { client, role: role.clone(), request_kind: request_kind.clone(), - priority: best_priority, + priority: candidate.priority, }); } - /// Returns a safe pool snapshot without endpoint URLs. - #[must_use] - pub fn snapshot(&self) -> crate::HttpTransportPoolSnapshot { - let endpoints = self.inner.clients.iter().map(crate::HttpEndpointClient::snapshot).collect(); - return crate::HttpTransportPoolSnapshot { endpoints }; + fn selection_from_runtime_candidate( + &self, + role: &crate::HttpRoleName, + request_kind: &crate::HttpRequestKind, + candidate: &RuntimePoolCandidate, + ) -> ksp_core_lib::Result { + let selected_client = self.inner.clients.get(candidate.client_index); + let client = match selected_client { + std::option::Option::Some(client) => client.clone(), + std::option::Option::None => return selection_failed(role, request_kind), + }; + return std::result::Result::Ok(crate::HttpEndpointSelection { + client, + role: role.clone(), + request_kind: request_kind.clone(), + priority: candidate.priority, + }); } fn next_position(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind, priority: u32, tier_size: usize) -> usize { @@ -202,9 +540,60 @@ struct PoolCandidate { priority: u32, } +struct RuntimePoolCandidate { + client_index: usize, + priority: u32, + runtime: std::sync::Arc, +} + +enum RuntimeSelectionAttempt { + Ready(crate::HttpRequestPermit), + Blocked { earliest_ready: std::option::Option, concurrency_saturated: bool }, + Unavailable, +} + +fn duration_millis_u64(duration: std::time::Duration) -> u64 { + let converted = u64::try_from(duration.as_millis()); + return match converted { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => u64::MAX, + }; +} + +fn earlier_instant(current: std::option::Option, candidate: std::time::Instant) -> std::option::Option { + return match current { + std::option::Option::Some(value) => std::option::Option::Some(std::cmp::min(value, candidate)), + std::option::Option::None => std::option::Option::Some(candidate), + }; +} + fn selection_failed(role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> ksp_core_lib::Result { + return std::result::Result::Err(selection_error(role, request_kind)); +} + +fn selection_failed_duration(role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> ksp_core_lib::Result { + return std::result::Result::Err(selection_error(role, request_kind)); +} + +fn request_selection_failed(role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> ksp_core_lib::Result { + return std::result::Result::Err(selection_error(role, request_kind)); +} + +fn selection_error(role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> ksp_core_lib::Error { + return ksp_core_lib::Error::new(crate::ERROR_CODE_ENDPOINT_SELECTION_FAILED, "no HTTP endpoint can satisfy the requested role and request kind") + .with_context("role", role.as_str()) + .with_context("request_kind", request_kind.as_str()); +} + +fn request_timeout(role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind, message: &str) -> ksp_core_lib::Result { + ksp_logging_lib::warn!( + target: env!("CARGO_PKG_NAME"), + role = role.as_str(), + request_kind = request_kind.as_str(), + "HTTP request admission deadline expired" + ); return std::result::Result::Err( - ksp_core_lib::Error::new(crate::ERROR_CODE_ENDPOINT_SELECTION_FAILED, "no HTTP endpoint can satisfy the requested role and request kind") + ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, message) .with_context("role", role.as_str()) .with_context("request_kind", request_kind.as_str()), ); diff --git a/crates/ksp-onchain-transport-lib/src/resilience.rs b/crates/ksp-onchain-transport-lib/src/resilience.rs new file mode 100644 index 0000000..a185d35 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/src/resilience.rs @@ -0,0 +1,390 @@ +// file: crates/ksp-onchain-transport-lib/src/resilience.rs +// version: 1 + +const DEFAULT_RATE_LIMIT_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(1); +const MAX_PROVIDER_RETRY_AFTER: std::time::Duration = std::time::Duration::from_secs(60); + +/// Transport-level cause considered by the bounded retry policy. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum HttpRetryCause { + /// A connection could not be established and no usable response exists. + Connection, + /// The request exceeded its transport deadline without a usable response. + Timeout, + /// The provider returned HTTP 429 or an equivalent transport-level rate-limit signal. + RateLimited, + /// The provider returned an HTTP status classified by the caller as temporary. + TemporaryHttp, + /// A generic request failure is not known to be safe to retry automatically. + Request, + /// A JSON-RPC application error was returned by the provider. + RpcApplication, + /// The response violated the KSP transport contract. + InvalidResponse, +} + +impl HttpRetryCause { + #[must_use] + const fn is_retryable(self) -> bool { + return match self { + Self::Connection | Self::Timeout | Self::RateLimited | Self::TemporaryHttp => true, + Self::Request | Self::RpcApplication | Self::InvalidResponse => false, + }; + } +} + +/// Dispatch knowledge used to prevent ambiguous automatic resubmission. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum HttpDispatchState { + /// The transport knows that the request was not dispatched to the provider. + NotDispatched, + /// The transport cannot prove whether a dispatched request was processed remotely. + DispatchedAmbiguous, +} + +/// Result of evaluating one bounded transport retry opportunity. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum HttpRetryDecision { + /// Stop retrying this transport request. + Stop, + /// Retry after the bounded delay. + RetryAfter(std::time::Duration), +} + +impl HttpRetryDecision { + /// Returns whether the decision authorizes another transport attempt. + #[must_use] + pub const fn should_retry(self) -> bool { + return match self { + Self::Stop => false, + Self::RetryAfter(_) => true, + }; + } + + /// Returns the retry delay when another attempt is authorized. + #[must_use] + pub const fn delay(self) -> std::option::Option { + return match self { + Self::Stop => std::option::Option::None, + Self::RetryAfter(delay) => std::option::Option::Some(delay), + }; + } +} + +/// Evaluates the centralized bounded HTTP retry policy for one audited RPC method. +/// +/// `completed_retries` counts retries already performed after the initial attempt. Provider `Retry-After` values are defensively bounded to sixty seconds +/// before they can extend the local exponential backoff. RPC application errors are never converted into transport retries. +#[must_use] +pub fn evaluate_transport_retry( + method: &crate::HttpRpcMethodDescriptor, + settings: &crate::HttpRetrySettings, + cause: crate::HttpRetryCause, + dispatch_state: crate::HttpDispatchState, + completed_retries: u32, + provider_retry_after: std::option::Option, +) -> crate::HttpRetryDecision { + if completed_retries >= settings.max_retries() || !cause.is_retryable() { + return crate::HttpRetryDecision::Stop; + } + if method.transport_retry_class() == crate::TransportRetryClass::NotApplicable { + return crate::HttpRetryDecision::Stop; + } + if method.transport_retry_class() == crate::TransportRetryClass::NeverAfterDispatch && dispatch_state == crate::HttpDispatchState::DispatchedAmbiguous { + return crate::HttpRetryDecision::Stop; + } + let retry_number = completed_retries.saturating_add(1); + let mut delay = retry_backoff(settings, retry_number); + if cause == crate::HttpRetryCause::RateLimited + && let std::option::Option::Some(provider_delay) = provider_retry_after + { + let bounded_provider_delay = std::cmp::min(provider_delay, MAX_PROVIDER_RETRY_AFTER); + if bounded_provider_delay > delay { + delay = bounded_provider_delay; + } + } + return crate::HttpRetryDecision::RetryAfter(delay); +} + +pub(crate) fn retry_backoff(settings: &crate::HttpRetrySettings, retry_number: u32) -> std::time::Duration { + let mut delay = settings.initial_backoff(); + if retry_number <= 1 { + return std::cmp::min(delay, settings.max_backoff()); + } + let mut step = 1_u32; + while step < retry_number { + let doubled = match delay.checked_mul(2) { + std::option::Option::Some(value) => value, + std::option::Option::None => settings.max_backoff(), + }; + delay = std::cmp::min(doubled, settings.max_backoff()); + if delay >= settings.max_backoff() { + return settings.max_backoff(); + } + step = step.saturating_add(1); + } + return delay; +} + +pub(crate) struct HttpRoleRuntime { + limits: crate::HttpRoleLimits, + bucket: std::sync::Mutex>, + semaphore: std::option::Option>, + notify: std::sync::Arc, + cooldown_until: std::sync::Mutex>, + degraded: std::sync::atomic::AtomicBool, + success_count: std::sync::atomic::AtomicU64, + failure_count: std::sync::atomic::AtomicU64, + rate_limit_count: std::sync::atomic::AtomicU64, +} + +impl HttpRoleRuntime { + pub(crate) fn new(settings: &crate::HttpEndpointRoleSettings, notify: std::sync::Arc) -> Self { + let bucket = match settings.limits().requests_per_second() { + std::option::Option::Some(requests_per_second) => { + let burst_capacity = match settings.limits().burst_capacity() { + std::option::Option::Some(capacity) => capacity, + std::option::Option::None => requests_per_second, + }; + std::option::Option::Some(HttpTokenBucketState::new(requests_per_second.get(), burst_capacity.get(), std::time::Instant::now())) + }, + std::option::Option::None => std::option::Option::None, + }; + let semaphore = match settings.limits().max_concurrent_requests() { + std::option::Option::Some(max_concurrent) => { + std::option::Option::Some(std::sync::Arc::new(tokio::sync::Semaphore::new(max_concurrent.get() as usize))) + }, + std::option::Option::None => std::option::Option::None, + }; + return Self { + limits: settings.limits().clone(), + bucket: std::sync::Mutex::new(bucket), + semaphore, + notify, + cooldown_until: std::sync::Mutex::new(std::option::Option::None), + degraded: std::sync::atomic::AtomicBool::new(false), + success_count: std::sync::atomic::AtomicU64::new(0), + failure_count: std::sync::atomic::AtomicU64::new(0), + rate_limit_count: std::sync::atomic::AtomicU64::new(0), + }; + } + + pub(crate) fn availability(&self, now: std::time::Instant) -> crate::HttpEndpointAvailability { + if self.cooldown_remaining_at(now).is_some() { + return crate::HttpEndpointAvailability::RateLimited; + } + if self.degraded.load(std::sync::atomic::Ordering::Relaxed) { + return crate::HttpEndpointAvailability::Degraded; + } + return crate::HttpEndpointAvailability::Available; + } + + pub(crate) fn cooldown_remaining(&self) -> std::option::Option { + return self.cooldown_remaining_at(std::time::Instant::now()); + } + + pub(crate) fn max_concurrent_requests(&self) -> std::option::Option { + return self.limits.max_concurrent_requests().map(|value| return value.get()); + } + + pub(crate) fn in_flight_requests(&self) -> std::option::Option { + let semaphore = match &self.semaphore { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let maximum = match self.limits.max_concurrent_requests() { + std::option::Option::Some(value) => value.get(), + std::option::Option::None => return std::option::Option::None, + }; + let available = semaphore.available_permits(); + let available_u32 = match u32::try_from(available) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => maximum, + }; + return std::option::Option::Some(maximum.saturating_sub(available_u32)); + } + + pub(crate) fn success_count(&self) -> u64 { + return self.success_count.load(std::sync::atomic::Ordering::Relaxed); + } + + pub(crate) fn failure_count(&self) -> u64 { + return self.failure_count.load(std::sync::atomic::Ordering::Relaxed); + } + + pub(crate) fn rate_limit_count(&self) -> u64 { + return self.rate_limit_count.load(std::sync::atomic::Ordering::Relaxed); + } + + pub(crate) fn try_acquire(self: &std::sync::Arc, now: std::time::Instant) -> RoleAdmissionAttempt { + if let std::option::Option::Some(remaining) = self.cooldown_remaining_at(now) { + let ready_at = match now.checked_add(remaining) { + std::option::Option::Some(value) => value, + std::option::Option::None => now, + }; + return RoleAdmissionAttempt::BlockedUntil(ready_at); + } + let semaphore_permit = match &self.semaphore { + std::option::Option::Some(semaphore) => { + let permit_result = std::sync::Arc::clone(semaphore).try_acquire_owned(); + match permit_result { + std::result::Result::Ok(permit) => std::option::Option::Some(permit), + std::result::Result::Err(tokio::sync::TryAcquireError::NoPermits) => return RoleAdmissionAttempt::ConcurrencySaturated, + std::result::Result::Err(tokio::sync::TryAcquireError::Closed) => return RoleAdmissionAttempt::Unavailable, + } + }, + std::option::Option::None => std::option::Option::None, + }; + let token_result = self.try_consume_token(now); + if let std::option::Option::Some(ready_at) = token_result { + drop(semaphore_permit); + self.notify.notify_one(); + return RoleAdmissionAttempt::BlockedUntil(ready_at); + } + return RoleAdmissionAttempt::Ready(HttpConcurrencyPermit { semaphore_permit, notify: std::sync::Arc::clone(&self.notify) }); + } + + pub(crate) fn record_success(&self) { + self.success_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.degraded.store(false, std::sync::atomic::Ordering::Relaxed); + self.notify.notify_waiters(); + return; + } + + pub(crate) fn record_failure(&self) { + self.failure_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.degraded.store(true, std::sync::atomic::Ordering::Relaxed); + self.notify.notify_waiters(); + return; + } + + pub(crate) fn record_rate_limited(&self, provider_retry_after: std::option::Option) -> std::time::Duration { + self.failure_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.rate_limit_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.degraded.store(true, std::sync::atomic::Ordering::Relaxed); + let configured_pause = match self.limits.pause_after_rate_limit() { + std::option::Option::Some(value) => value, + std::option::Option::None => DEFAULT_RATE_LIMIT_COOLDOWN, + }; + let provider_pause = match provider_retry_after { + std::option::Option::Some(value) => std::cmp::min(value, MAX_PROVIDER_RETRY_AFTER), + std::option::Option::None => std::time::Duration::ZERO, + }; + let effective_pause = std::cmp::max(configured_pause, provider_pause); + let now = std::time::Instant::now(); + let candidate = match now.checked_add(effective_pause) { + std::option::Option::Some(value) => value, + std::option::Option::None => now, + }; + let lock_result = self.cooldown_until.lock(); + let mut cooldown_until = match lock_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + let replace = match *cooldown_until { + std::option::Option::Some(current) => candidate > current, + std::option::Option::None => true, + }; + if replace { + *cooldown_until = std::option::Option::Some(candidate); + } + drop(cooldown_until); + self.notify.notify_waiters(); + return effective_pause; + } + + fn cooldown_remaining_at(&self, now: std::time::Instant) -> std::option::Option { + let lock_result = self.cooldown_until.lock(); + let mut cooldown_until = match lock_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + let deadline = match *cooldown_until { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + if deadline <= now { + *cooldown_until = std::option::Option::None; + return std::option::Option::None; + } + return std::option::Option::Some(deadline.duration_since(now)); + } + + fn try_consume_token(&self, now: std::time::Instant) -> std::option::Option { + let lock_result = self.bucket.lock(); + let mut bucket = match lock_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + let state = match bucket.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + return state.try_consume_at(now); + } +} + +pub(crate) enum RoleAdmissionAttempt { + Ready(HttpConcurrencyPermit), + BlockedUntil(std::time::Instant), + ConcurrencySaturated, + Unavailable, +} + +pub(crate) struct HttpConcurrencyPermit { + semaphore_permit: std::option::Option, + notify: std::sync::Arc, +} + +impl Drop for HttpConcurrencyPermit { + fn drop(&mut self) { + let permit = self.semaphore_permit.take(); + drop(permit); + self.notify.notify_one(); + } +} + +#[derive(Debug)] +struct HttpTokenBucketState { + available_tokens: f64, + requests_per_second: u32, + burst_capacity: u32, + last_refill: std::time::Instant, +} + +impl HttpTokenBucketState { + fn new(requests_per_second: u32, burst_capacity: u32, now: std::time::Instant) -> Self { + return Self { available_tokens: f64::from(burst_capacity), requests_per_second, burst_capacity, last_refill: now }; + } + + fn try_consume_at(&mut self, now: std::time::Instant) -> std::option::Option { + self.refill_at(now); + if self.available_tokens >= 1.0 { + self.available_tokens -= 1.0; + return std::option::Option::None; + } + let missing_tokens = 1.0 - self.available_tokens; + let wait_seconds = missing_tokens / f64::from(self.requests_per_second); + let wait = std::time::Duration::from_secs_f64(wait_seconds); + let ready_at = match now.checked_add(wait) { + std::option::Option::Some(value) => value, + std::option::Option::None => now, + }; + return std::option::Option::Some(ready_at); + } + + fn refill_at(&mut self, now: std::time::Instant) { + if now <= self.last_refill { + return; + } + let elapsed_seconds = now.duration_since(self.last_refill).as_secs_f64(); + let refill = elapsed_seconds * f64::from(self.requests_per_second); + self.available_tokens = (self.available_tokens + refill).min(f64::from(self.burst_capacity)); + self.last_refill = now; + return; + } +} + +#[cfg(test)] +#[path = "../unit_tests/resilience.rs"] +mod tests; diff --git a/crates/ksp-onchain-transport-lib/src/settings.rs b/crates/ksp-onchain-transport-lib/src/settings.rs index 245c85c..21bc163 100644 --- a/crates/ksp-onchain-transport-lib/src/settings.rs +++ b/crates/ksp-onchain-transport-lib/src/settings.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/settings.rs -// version: 3 +// version: 4 /// Runtime HTTP endpoint URL owned by Transport. /// @@ -156,7 +156,10 @@ pub struct HttpRoleLimits { } impl HttpRoleLimits { - /// Creates explicit role limits. `None` leaves the corresponding limit unbounded by this KSP transport layer. + /// Creates explicit role limits. + /// + /// When RPS is configured and burst capacity is absent, runtime burst defaults to one second of RPS capacity. An absent concurrency limit is + /// unbounded by this KSP transport layer. An absent rate-limit cooldown uses the Transport runtime fallback cooldown. #[must_use] pub const fn new( requests_per_second: std::option::Option, @@ -173,7 +176,7 @@ impl HttpRoleLimits { return self.requests_per_second; } - /// Returns the configured token-bucket burst capacity. + /// Returns the configured token-bucket burst capacity. `None` means the runtime derives capacity from configured RPS. #[must_use] pub const fn burst_capacity(&self) -> std::option::Option { return self.burst_capacity; @@ -185,7 +188,7 @@ impl HttpRoleLimits { return self.max_concurrent_requests; } - /// Returns the configured cooldown applied after rate limiting. + /// Returns the configured cooldown applied after rate limiting. `None` delegates to the Transport runtime fallback cooldown. #[must_use] pub const fn pause_after_rate_limit(&self) -> std::option::Option { return self.pause_after_rate_limit; diff --git a/crates/ksp-onchain-transport-lib/tests/dependency_boundary.rs b/crates/ksp-onchain-transport-lib/tests/dependency_boundary.rs index 9907098..bd53abd 100644 --- a/crates/ksp-onchain-transport-lib/tests/dependency_boundary.rs +++ b/crates/ksp-onchain-transport-lib/tests/dependency_boundary.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/dependency_boundary.rs -// version: 2 +// version: 3 //! Integration canary for the direct dependency firewall of `ksp-onchain-transport-lib`. @@ -13,4 +13,5 @@ fn transport_manifest_preserves_ksp_dependency_firewall() { assert!(manifest.contains("ksp-core-lib")); assert!(manifest.contains("ksp-logging-lib")); assert!(manifest.contains("reqwest.workspace = true")); + assert!(manifest.contains("tokio = { workspace = true, features = [\"sync\"] }")); } diff --git a/crates/ksp-onchain-transport-lib/tests/public_api.rs b/crates/ksp-onchain-transport-lib/tests/public_api.rs index f8ab059..cc93e00 100644 --- a/crates/ksp-onchain-transport-lib/tests/public_api.rs +++ b/crates/ksp-onchain-transport-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/public_api.rs -// version: 3 +// version: 4 //! Integration tests for the public `ksp-onchain-transport-lib` consumer contract. @@ -101,3 +101,62 @@ fn public_pool_contract_selects_a_standard_method_without_exposing_url() { assert!(!rendered.contains("SECRET-CANARY")); assert!(!rendered.contains("provider.invalid")); } + +#[test] +fn public_retry_policy_preserves_no_resend_after_ambiguous_write_dispatch() { + let method = ksp_onchain_transport_lib::find_http_rpc_method("sendTransaction").expect("sendTransaction must be audited"); + let settings = ksp_onchain_transport_lib::HttpRetrySettings::new(2, std::time::Duration::from_millis(100), std::time::Duration::from_secs(1)); + let decision = ksp_onchain_transport_lib::evaluate_transport_retry( + method, + &settings, + ksp_onchain_transport_lib::HttpRetryCause::Timeout, + ksp_onchain_transport_lib::HttpDispatchState::DispatchedAmbiguous, + 0, + std::option::Option::None, + ); + assert_eq!(decision, ksp_onchain_transport_lib::HttpRetryDecision::Stop); +} + +#[tokio::test] +async fn public_async_admission_exposes_bounded_permit_without_endpoint_url() { + let url = ksp_onchain_transport_lib::HttpEndpointUrl::parse("https://provider.invalid/rpc?token=ASYNC-SECRET-CANARY") + .expect("public URL parser must accept HTTPS endpoint"); + let role = ksp_onchain_transport_lib::HttpEndpointRoleSettings::new( + ksp_onchain_transport_lib::HttpRoleName::new("default"), + true, + std::vec![ksp_onchain_transport_lib::HttpRequestKind::new("get_balance")], + 10, + ksp_onchain_transport_lib::HttpRoleLimits::new( + std::option::Option::None, + std::option::Option::None, + std::num::NonZeroU32::new(1), + std::option::Option::None, + ), + ); + let endpoint = ksp_onchain_transport_lib::HttpEndpointSettings::new( + "primary", + true, + ksp_onchain_transport_lib::HttpProviderName::new("provider"), + ksp_onchain_transport_lib::HttpClusterName::new("devnet"), + url, + std::time::Duration::from_secs(2), + std::time::Duration::from_secs(10), + std::option::Option::Some(8), + std::vec![role], + ); + let settings = ksp_onchain_transport_lib::HttpTransportSettings::new( + std::vec![endpoint], + ksp_onchain_transport_lib::HttpRetrySettings::new(1, std::time::Duration::from_millis(10), std::time::Duration::from_millis(50)), + ); + let pool = ksp_onchain_transport_lib::HttpTransportPool::new(settings).expect("public pool constructor must succeed"); + let method = ksp_onchain_transport_lib::find_http_rpc_method("getBalance").expect("audited method must exist"); + let permit = pool + .acquire_for_method(&ksp_onchain_transport_lib::HttpRoleName::new("default"), method) + .await + .expect("public async admission must acquire capacity"); + assert_eq!(permit.selection().endpoint_name(), "primary"); + assert!(permit.remaining_timeout() > std::time::Duration::ZERO); + let rendered = format!("{permit:?}"); + assert!(!rendered.contains("ASYNC-SECRET-CANARY")); + assert!(!rendered.contains("provider.invalid")); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/client.rs b/crates/ksp-onchain-transport-lib/unit_tests/client.rs index fc525cf..6ff436a 100644 --- a/crates/ksp-onchain-transport-lib/unit_tests/client.rs +++ b/crates/ksp-onchain-transport-lib/unit_tests/client.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/unit_tests/client.rs -// version: 1 +// version: 2 fn endpoint(enabled: bool, url_text: &str) -> crate::HttpEndpointSettings { let url = crate::HttpEndpointUrl::parse(url_text).expect("test endpoint URL must parse"); @@ -47,3 +47,16 @@ fn endpoint_client_matches_exact_and_wildcard_capabilities() { assert!(client.supports(&crate::HttpRoleName::new("default"), &crate::HttpRequestKind::new("get_balance"))); assert!(!client.supports(&crate::HttpRoleName::new("write"), &crate::HttpRequestKind::new("get_balance"))); } + +#[test] +fn endpoint_role_snapshot_exposes_safe_resilience_state() { + let client = super::HttpEndpointClient::new(endpoint(true, "https://api.devnet.solana.com")).expect("client must build"); + let snapshot = client.snapshot(); + let role = &snapshot.roles()[0]; + assert_eq!(role.availability(), crate::HttpEndpointAvailability::Available); + assert_eq!(role.in_flight_requests(), std::option::Option::None); + assert_eq!(role.cooldown_remaining(), std::option::Option::None); + assert_eq!(role.success_count(), 0); + assert_eq!(role.failure_count(), 0); + assert_eq!(role.rate_limit_count(), 0); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/pool.rs b/crates/ksp-onchain-transport-lib/unit_tests/pool.rs index 4ccc3c9..89bbb8c 100644 --- a/crates/ksp-onchain-transport-lib/unit_tests/pool.rs +++ b/crates/ksp-onchain-transport-lib/unit_tests/pool.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/unit_tests/pool.rs -// version: 2 +// version: 3 fn role(name: &str, priority: u32, request_kinds: std::vec::Vec) -> crate::HttpEndpointRoleSettings { return crate::HttpEndpointRoleSettings::new( @@ -25,6 +25,38 @@ fn endpoint(name: &str, enabled: bool, priority: u32, request_kinds: std::vec::V ); } +fn non_zero(value: u32) -> std::num::NonZeroU32 { + return std::num::NonZeroU32::new(value).expect("test limit must be non-zero"); +} + +fn limited_endpoint( + name: &str, + priority: u32, + requests_per_second: std::option::Option, + burst_capacity: std::option::Option, + max_concurrent_requests: std::option::Option, + cooldown: std::option::Option, +) -> crate::HttpEndpointSettings { + let limits = crate::HttpRoleLimits::new( + requests_per_second.map(|value| return non_zero(value)), + burst_capacity.map(|value| return non_zero(value)), + max_concurrent_requests.map(|value| return non_zero(value)), + cooldown, + ); + let role = crate::HttpEndpointRoleSettings::new(crate::HttpRoleName::new("default"), true, std::vec![crate::HttpRequestKind::wildcard()], priority, limits); + return crate::HttpEndpointSettings::new( + name, + true, + crate::HttpProviderName::new("provider"), + crate::HttpClusterName::new("devnet"), + crate::HttpEndpointUrl::parse(format!("https://{name}.invalid/rpc?token=SECRET-CANARY")).expect("test URL must parse"), + std::time::Duration::from_secs(1), + std::time::Duration::from_secs(2), + std::option::Option::Some(4), + std::vec![role], + ); +} + fn settings(endpoints: std::vec::Vec) -> crate::HttpTransportSettings { return crate::HttpTransportSettings::new( endpoints, @@ -172,3 +204,133 @@ fn removed_standard_method_is_rejected_before_endpoint_routing() { let error = pool.select_for_method(&crate::HttpRoleName::new("default"), method).expect_err("removed standard method must be rejected before routing"); assert_eq!(error.code(), crate::ERROR_CODE_METHOD_REMOVED); } + +#[tokio::test] +async fn runtime_concurrency_saturation_falls_back_to_lower_priority_tier() { + let pool = super::HttpTransportPool::new(settings(std::vec![ + limited_endpoint("primary", 1, std::option::Option::None, std::option::Option::None, std::option::Option::Some(1), std::option::Option::None), + limited_endpoint("fallback", 20, std::option::Option::None, std::option::Option::None, std::option::Option::Some(1), std::option::Option::None), + ])) + .expect("pool must build"); + let role = crate::HttpRoleName::new("default"); + let kind = crate::HttpRequestKind::new("get_balance"); + let first = pool.acquire_for_request_kind(&role, &kind).await.expect("first request must acquire primary"); + assert_eq!(first.selection().endpoint_name(), "primary"); + let second = pool.acquire_for_request_kind(&role, &kind).await.expect("second request must fall back while primary is saturated"); + assert_eq!(second.selection().endpoint_name(), "fallback"); +} + +#[tokio::test] +async fn runtime_token_bucket_exhaustion_falls_back_without_busy_waiting() { + let pool = super::HttpTransportPool::new(settings(std::vec![ + limited_endpoint("primary", 1, std::option::Option::Some(1), std::option::Option::Some(1), std::option::Option::None, std::option::Option::None), + limited_endpoint("fallback", 20, std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::None), + ])) + .expect("pool must build"); + let role = crate::HttpRoleName::new("default"); + let kind = crate::HttpRequestKind::new("get_balance"); + let first = pool.acquire_for_request_kind(&role, &kind).await.expect("first request must consume primary token"); + assert_eq!(first.selection().endpoint_name(), "primary"); + drop(first); + let second = pool.acquire_for_request_kind(&role, &kind).await.expect("fallback must be used while primary token bucket refills"); + assert_eq!(second.selection().endpoint_name(), "fallback"); +} + +#[tokio::test] +async fn provider_cooldown_excludes_rate_limited_role_and_uses_fallback() { + let pool = super::HttpTransportPool::new(settings(std::vec![ + limited_endpoint( + "primary", + 1, + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(std::time::Duration::from_millis(50)), + ), + limited_endpoint("fallback", 20, std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::None), + ])) + .expect("pool must build"); + let role = crate::HttpRoleName::new("default"); + let kind = crate::HttpRequestKind::new("get_balance"); + let primary = pool.acquire_for_request_kind(&role, &kind).await.expect("primary must be acquired"); + assert_eq!(primary.selection().endpoint_name(), "primary"); + let pause = primary.record_rate_limited(std::option::Option::None); + assert_eq!(pause, std::time::Duration::from_millis(50)); + drop(primary); + let fallback = pool.acquire_for_request_kind(&role, &kind).await.expect("fallback must be selected during primary cooldown"); + assert_eq!(fallback.selection().endpoint_name(), "fallback"); +} + +#[tokio::test] +async fn admission_waits_for_released_concurrency_without_holding_a_sync_mutex_across_await() { + let pool = super::HttpTransportPool::new(settings(std::vec![limited_endpoint( + "primary", + 1, + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(1), + std::option::Option::None, + )])) + .expect("pool must build"); + let role = crate::HttpRoleName::new("default"); + let kind = crate::HttpRequestKind::new("get_balance"); + let first = pool.acquire_for_request_kind(&role, &kind).await.expect("first permit must be acquired"); + let release_task = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + drop(first); + }); + let second = pool + .acquire_for_request_kind_with_timeout(&role, &kind, std::time::Duration::from_millis(100)) + .await + .expect("second permit must wake after concurrency release"); + assert_eq!(second.selection().endpoint_name(), "primary"); + release_task.await.expect("release task must complete"); +} + +#[tokio::test] +async fn admission_timeout_is_bounded_when_concurrency_never_becomes_available() { + let pool = super::HttpTransportPool::new(settings(std::vec![limited_endpoint( + "primary", + 1, + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(1), + std::option::Option::None, + )])) + .expect("pool must build"); + let role = crate::HttpRoleName::new("default"); + let kind = crate::HttpRequestKind::new("get_balance"); + let _held = pool.acquire_for_request_kind(&role, &kind).await.expect("first permit must be acquired"); + let error = pool + .acquire_for_request_kind_with_timeout(&role, &kind, std::time::Duration::from_millis(20)) + .await + .expect_err("second permit must time out while concurrency remains saturated"); + assert_eq!(error.code(), crate::ERROR_CODE_TIMEOUT); +} + +#[tokio::test] +async fn passive_health_snapshot_moves_from_degraded_back_to_available_after_success() { + let pool = super::HttpTransportPool::new(settings(std::vec![limited_endpoint( + "primary", + 1, + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + )])) + .expect("pool must build"); + let role = crate::HttpRoleName::new("default"); + let kind = crate::HttpRequestKind::new("get_balance"); + let first = pool.acquire_for_request_kind(&role, &kind).await.expect("request permit must be acquired"); + first.record_failure(); + drop(first); + let degraded = pool.snapshot(); + assert_eq!(degraded.endpoints()[0].availability(), crate::HttpEndpointAvailability::Degraded); + assert_eq!(degraded.endpoints()[0].roles()[0].failure_count(), 1); + let second = pool.acquire_for_request_kind(&role, &kind).await.expect("degraded endpoint remains eligible for passive recovery"); + second.record_success(); + drop(second); + let recovered = pool.snapshot(); + assert_eq!(recovered.endpoints()[0].availability(), crate::HttpEndpointAvailability::Available); + assert_eq!(recovered.endpoints()[0].roles()[0].success_count(), 1); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/resilience.rs b/crates/ksp-onchain-transport-lib/unit_tests/resilience.rs new file mode 100644 index 0000000..0977ce6 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/unit_tests/resilience.rs @@ -0,0 +1,186 @@ +// file: crates/ksp-onchain-transport-lib/unit_tests/resilience.rs +// version: 1 + +fn non_zero(value: u32) -> std::num::NonZeroU32 { + return std::num::NonZeroU32::new(value).expect("test limit must be non-zero"); +} + +fn retry_settings() -> crate::HttpRetrySettings { + return crate::HttpRetrySettings::new(4, std::time::Duration::from_millis(100), std::time::Duration::from_millis(500)); +} + +fn method(name: &str) -> &'static crate::HttpRpcMethodDescriptor { + return crate::find_http_rpc_method(name).expect("audited test method must exist"); +} + +fn role_limits( + requests_per_second: std::option::Option, + burst_capacity: std::option::Option, + max_concurrent_requests: std::option::Option, + cooldown: std::option::Option, +) -> crate::HttpEndpointRoleSettings { + return crate::HttpEndpointRoleSettings::new( + crate::HttpRoleName::new("default"), + true, + std::vec![crate::HttpRequestKind::wildcard()], + 10, + crate::HttpRoleLimits::new( + requests_per_second.map(|value| return non_zero(value)), + burst_capacity.map(|value| return non_zero(value)), + max_concurrent_requests.map(|value| return non_zero(value)), + cooldown, + ), + ); +} + +#[test] +fn retry_backoff_is_exponential_and_bounded() { + let settings = retry_settings(); + assert_eq!(super::retry_backoff(&settings, 1), std::time::Duration::from_millis(100)); + assert_eq!(super::retry_backoff(&settings, 2), std::time::Duration::from_millis(200)); + assert_eq!(super::retry_backoff(&settings, 3), std::time::Duration::from_millis(400)); + assert_eq!(super::retry_backoff(&settings, 4), std::time::Duration::from_millis(500)); + assert_eq!(super::retry_backoff(&settings, 32), std::time::Duration::from_millis(500)); +} + +#[test] +fn retry_safe_timeout_is_retried_until_budget_is_exhausted() { + let settings = retry_settings(); + let first = super::evaluate_transport_retry( + method("getBalance"), + &settings, + super::HttpRetryCause::Timeout, + super::HttpDispatchState::DispatchedAmbiguous, + 0, + std::option::Option::None, + ); + assert_eq!(first, super::HttpRetryDecision::RetryAfter(std::time::Duration::from_millis(100))); + let exhausted = super::evaluate_transport_retry( + method("getBalance"), + &settings, + super::HttpRetryCause::Timeout, + super::HttpDispatchState::DispatchedAmbiguous, + settings.max_retries(), + std::option::Option::None, + ); + assert_eq!(exhausted, super::HttpRetryDecision::Stop); +} + +#[test] +fn write_submission_never_retries_after_ambiguous_dispatch() { + let decision = super::evaluate_transport_retry( + method("sendTransaction"), + &retry_settings(), + super::HttpRetryCause::Connection, + super::HttpDispatchState::DispatchedAmbiguous, + 0, + std::option::Option::None, + ); + assert_eq!(decision, super::HttpRetryDecision::Stop); +} + +#[test] +fn write_submission_can_retry_when_transport_proves_no_dispatch() { + let decision = super::evaluate_transport_retry( + method("sendTransaction"), + &retry_settings(), + super::HttpRetryCause::Connection, + super::HttpDispatchState::NotDispatched, + 0, + std::option::Option::None, + ); + assert!(decision.should_retry()); +} + +#[test] +fn rpc_application_and_invalid_response_are_not_transport_retries() { + for cause in [super::HttpRetryCause::RpcApplication, super::HttpRetryCause::InvalidResponse, super::HttpRetryCause::Request] { + let decision = super::evaluate_transport_retry( + method("getBalance"), + &retry_settings(), + cause, + super::HttpDispatchState::NotDispatched, + 0, + std::option::Option::None, + ); + assert_eq!(decision, super::HttpRetryDecision::Stop); + } +} + +#[test] +fn provider_retry_after_can_extend_backoff_but_is_defensively_bounded() { + let settings = retry_settings(); + let extended = super::evaluate_transport_retry( + method("getBalance"), + &settings, + super::HttpRetryCause::RateLimited, + super::HttpDispatchState::DispatchedAmbiguous, + 0, + std::option::Option::Some(std::time::Duration::from_secs(3)), + ); + assert_eq!(extended.delay(), std::option::Option::Some(std::time::Duration::from_secs(3))); + let bounded = super::evaluate_transport_retry( + method("getBalance"), + &settings, + super::HttpRetryCause::RateLimited, + super::HttpDispatchState::DispatchedAmbiguous, + 0, + std::option::Option::Some(std::time::Duration::from_secs(600)), + ); + assert_eq!(bounded.delay(), std::option::Option::Some(std::time::Duration::from_secs(60))); +} + +#[test] +fn token_bucket_consumes_burst_then_refills_from_elapsed_time() { + let start = std::time::Instant::now(); + let mut bucket = super::HttpTokenBucketState::new(2, 2, start); + assert!(bucket.try_consume_at(start).is_none()); + assert!(bucket.try_consume_at(start).is_none()); + assert!(bucket.try_consume_at(start).is_some()); + let later = start.checked_add(std::time::Duration::from_millis(500)).expect("test instant must advance"); + assert!(bucket.try_consume_at(later).is_none()); +} + +#[test] +fn absent_burst_capacity_defaults_to_one_second_of_rps_capacity() { + let role = role_limits(std::option::Option::Some(2), std::option::Option::None, std::option::Option::None, std::option::Option::None); + let runtime = std::sync::Arc::new(super::HttpRoleRuntime::new(&role, std::sync::Arc::new(tokio::sync::Notify::new()))); + let now = std::time::Instant::now(); + let first = runtime.try_acquire(now); + let second = runtime.try_acquire(now); + let third = runtime.try_acquire(now); + assert!(matches!(first, super::RoleAdmissionAttempt::Ready(_))); + assert!(matches!(second, super::RoleAdmissionAttempt::Ready(_))); + assert!(matches!(third, super::RoleAdmissionAttempt::BlockedUntil(_))); +} + +#[test] +fn concurrency_semaphore_releases_capacity_when_permit_is_dropped() { + let role = role_limits(std::option::Option::None, std::option::Option::None, std::option::Option::Some(1), std::option::Option::None); + let runtime = std::sync::Arc::new(super::HttpRoleRuntime::new(&role, std::sync::Arc::new(tokio::sync::Notify::new()))); + let now = std::time::Instant::now(); + let first = runtime.try_acquire(now); + let held = match first { + super::RoleAdmissionAttempt::Ready(permit) => permit, + _ => panic!("first concurrency permit must be available"), + }; + assert!(matches!(runtime.try_acquire(now), super::RoleAdmissionAttempt::ConcurrencySaturated)); + drop(held); + assert!(matches!(runtime.try_acquire(now), super::RoleAdmissionAttempt::Ready(_))); +} + +#[test] +fn rate_limit_cooldown_marks_role_and_caps_provider_delay() { + let role = role_limits( + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(std::time::Duration::from_millis(10)), + ); + let runtime = super::HttpRoleRuntime::new(&role, std::sync::Arc::new(tokio::sync::Notify::new())); + let pause = runtime.record_rate_limited(std::option::Option::Some(std::time::Duration::from_secs(600))); + assert_eq!(pause, std::time::Duration::from_secs(60)); + assert_eq!(runtime.rate_limit_count(), 1); + assert_eq!(runtime.failure_count(), 1); + assert_eq!(runtime.availability(std::time::Instant::now()), crate::HttpEndpointAvailability::RateLimited); +} diff --git a/deltas/0.2.1/pre.004.md b/deltas/0.2.1/pre.004.md new file mode 100644 index 0000000..94a370b --- /dev/null +++ b/deltas/0.2.1/pre.004.md @@ -0,0 +1,231 @@ + + + +# Delta `v0.2.1-pre.004` + +## Base + +Base attendue : + +```text +v0.2.1-pre.003-fix.001 +``` + +Cette base a été validée localement par le user avec `cargo fmt`, `cargo check --workspace`, `cargo clippy --workspace --all-targets`, les tests ciblés Transport et `cargo test --workspace`. + +Version Cargo cible : + +```text +0.2.1-pre.4 +``` + +## Objectif + +Matérialiser la résilience runtime prévue par le plan `008` autour du client/pool HTTP déjà introduit : + +- token bucket RPS/burst par rôle ; +- limite de concurrence par semaphore ; +- cooldown après rate-limit provider ; +- admission async avec deadline commune ; +- fallback vers les pairs et tiers moins prioritaires lorsqu'un candidat est temporairement indisponible ; +- états passifs `Degraded` / `RateLimited` et snapshots sûrs ; +- retry/backoff transport borné ; +- interdiction centralisée d'un resend après dispatch ambigu pour les opérations `NeverAfterDispatch`. + +Cette tranche ne réalise toujours pas de POST JSON-RPC ni de méthode Solana typée ; le raccordement réseau appartient à `pre.005`. + +## Modifications + +### Workspace et dépendances + +- `workspace.package.version` passe de `0.2.1-pre.3.fix.1` à `0.2.1-pre.4` ; +- aucune nouvelle dépendance tierce n'est introduite ; +- `ksp-onchain-transport-lib` consomme désormais `tokio.workspace = true` avec la feature locale `sync`, nécessaire au semaphore et à `Notify` ; +- les features runtime/time Tokio restent possédées par la déclaration workspace existante ; +- le firewall Transport -> Config/Store/Program et l'absence de `tracing` direct restent inchangés. + +### Admission RPS / burst + +Chaque rôle endpoint possède son état runtime indépendant. + +Lorsque `requests_per_second` est configuré : + +- un token bucket est créé au niveau endpoint/rôle ; +- `burst_capacity` fixe la capacité maximale ; +- si le burst n'est pas explicite, la capacité dérive de la valeur RPS, soit une seconde de capacité ; +- les tokens se reforment proportionnellement au temps écoulé ; +- une admission sans token disponible expose l'instant de prochaine admissibilité au pool plutôt que de bloquer un thread. + +Sans RPS, le rôle n'est pas limité par le token bucket KSP. + +### Concurrence + +`max_concurrent_requests` est matérialisé par un `tokio::sync::Semaphore` par rôle. + +Le nouveau `HttpRequestPermit` : + +- réserve la capacité de concurrence ; +- conserve la sélection endpoint/rôle/request-kind ; +- transporte la deadline commune ; +- libère automatiquement la capacité à son drop ; +- réveille les waiters lorsque de la capacité redevient disponible. + +Un candidat saturé n'empêche pas d'essayer les autres candidats du même tier puis les tiers moins prioritaires. + +### Cooldown et rate-limit + +`HttpRequestPermit::record_rate_limited()` : + +- incrémente les compteurs failure/rate-limit ; +- marque le rôle dégradé ; +- applique le cooldown configuré par `pause_after_rate_limit` ; +- utilise un fallback runtime de 1 seconde lorsque ce cooldown n'est pas configuré ; +- accepte un délai provider déjà résolu et le borne à 60 secondes avant de pouvoir prolonger le cooldown local ; +- notifie le pool de la nouvelle disponibilité temporelle. + +Le parsing concret du header HTTP `Retry-After` reste au futur exécuteur HTTP de `pre.005`; `pre.004` stabilise le contrat en `Duration`. + +### Admission async et deadline commune + +Nouvelles surfaces publiques : + +```text +HttpRequestPermit +HttpTransportPool::acquire_for_method(...) +HttpTransportPool::acquire_for_request_kind(...) +HttpTransportPool::acquire_for_request_kind_with_timeout(...) +``` + +Algorithme runtime : + +1. filtrer endpoint/rôle/capability comme en `pre.003` ; +2. ordonner les candidats par priorité ; +3. appliquer le round-robin dans chaque tier ; +4. tenter l'admission cooldown + concurrence + token bucket ; +5. essayer les pairs puis les tiers inférieurs lorsqu'un candidat est temporairement bloqué ; +6. si tous les candidats sont temporairement bloqués, attendre notification ou prochain instant de refill/cooldown ; +7. ne jamais dépasser la deadline commune. + +Par défaut, la deadline commune utilise le plus petit `request_timeout` des endpoints structurellement compatibles. Une surcharge permet à une couche supérieure d'imposer un budget plus strict. + +### Retry / backoff / no-resend + +Nouveaux contrats publics : + +```text +HttpRetryCause +HttpDispatchState +HttpRetryDecision +evaluate_transport_retry(...) +``` + +La policy centralisée : + +- respecte `HttpRetrySettings::max_retries` ; +- applique un backoff exponentiel borné entre `initial_backoff` et `max_backoff` ; +- autorise seulement les causes transport classées retryables (`Connection`, `Timeout`, `RateLimited`, `TemporaryHttp`) ; +- exclut `Request`, `RpcApplication` et `InvalidResponse` du retry automatique ; +- respecte `TransportRetryClass::NotApplicable` ; +- interdit tout retry d'une méthode `NeverAfterDispatch` lorsque l'état est `DispatchedAmbiguous` ; +- autorise encore une tentative si le transport sait explicitement que la requête n'a pas été dispatchée ; +- peut prolonger le backoff d'un rate-limit par un délai provider borné à 60 secondes. + +Aucune logique de retry métier/exécution Solana n'est introduite. + +### Santé passive et snapshots + +Les rôles endpoint exposent maintenant de manière sûre : + +- `availability` ; +- RPS/burst configurés ; +- concurrence maximale et nombre en vol ; +- cooldown restant ; +- compteurs success/failure/rate-limit. + +Transitions : + +```text +Available --failure--> Degraded +Available/Degraded --rate-limit--> RateLimited +RateLimited --cooldown écoulé--> Degraded +Degraded --success--> Available +``` + +Les snapshots n'exposent toujours aucune URL ni credential provider. + +### ROADMAP et plan + +`ROADMAP.md` reçoit uniquement la mise à jour synthétique normale de l'état de `0.2.1` : la résilience runtime devient acquise et les prochains jalons majeurs restent les 4 canaris, Config standard et la clôture. Aucun historique de prerelease/fix n'y est ajouté. + +Le plan `008` enregistre `pre.004` comme réalisé et décrit l'état précis dans sa section de suivi. + +## Tests ajoutés + +La suite Transport passe de **53 à 72 tests déclarés**. + +Les nouveaux tests couvrent notamment : + +- backoff exponentiel et borne maximale ; +- budget `max_retries` ; +- no-resend après dispatch ambigu ; +- retry permis lorsqu'un non-dispatch est prouvé ; +- RPC application errors et réponses invalides non retryées ; +- extension/cap du délai provider ; +- token bucket burst/refill déterministe ; +- burst implicite dérivé du RPS ; +- libération de semaphore ; +- cooldown/rate-limit ; +- fallback vers un tier inférieur sur saturation concurrence, RPS ou cooldown ; +- réveil après libération de concurrence ; +- timeout d'admission borné ; +- transition passive degraded -> available ; +- snapshot public de résilience ; +- admission async depuis l'API publique sans fuite de l'URL ; +- consommation publique de la policy de retry. + +## Hors périmètre conservé + +Restent à `pre.005` : + +- exécution HTTP POST JSON-RPC réelle ; +- mapping concret des erreurs `reqwest`/HTTP vers `HttpRetryCause` ; +- parsing de `Retry-After` HTTP ; +- méthodes typées `getHealth`, `getVersion`, `getGenesisHash`, `getBalance` ; +- fixtures RPC de ces méthodes. + +Restent à `pre.006+` : + +- document/schema Config standard Transport ; +- adapter Config -> settings Transport ; +- smoke tests réseau opt-in et documentation finale. + +## Validation + +Non exécutée dans le sandbox de génération : `cargo` et `rustc` n'y sont pas installés. + +Après application : + +```bash +cargo fmt --all +cargo check --workspace +cargo clippy --workspace --all-targets +cargo test -p ksp-onchain-transport-lib +cargo test --workspace +``` + +Le graphe de dépendances n'ajoute aucune crate nouvelle, mais la feature locale `tokio/sync` devient directement utilisée par Transport. Pour auditer le feature-set effectif de cette tranche : + +```bash +cargo tree -p ksp-onchain-transport-lib -e features +cargo tree -p ksp-onchain-transport-lib -e normal +``` + +## Suite + +Tranche suivante prévue : + +```text +0.2.1-pre.005 +``` + +Périmètre : exécuteur HTTP JSON-RPC réel + quatre méthodes canari typées `getHealth`, `getVersion`, `getGenesisHash`, `getBalance`, avec fixtures déterministes et raccordement des timeouts/429/erreurs transport à la résilience de `pre.004`. diff --git a/docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md b/docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md index 97b77b3..87fa7e1 100644 --- a/docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md +++ b/docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md @@ -1,5 +1,5 @@ - + # `0.2.1-pre.001` — plan `ksp-onchain-transport-lib` HTTP Solana foundation @@ -872,7 +872,7 @@ La release peut devenir stable seulement si : | `pre.001` | audit KSP + bot3 + docs officielles, matrice 52+14, architecture, split et sizing | | `pre.002` | **réalisé** : crate/workspace, codes erreur, settings/validation, JSON-RPC, descriptors/status, base Logging | | `pre.003` | **réalisé** : endpoint client + pool logique + rôles/capabilities + priorité/fairness/fallback + snapshots sûrs | -| `pre.004` | RPS/burst/concurrence/cooldown + timeout + retry/backoff + classification retry/no-resend | +| `pre.004` | **réalisé** : RPS/burst/concurrence/cooldown + deadline commune + retry/backoff + classification retry/no-resend | | `pre.005` | méthodes canari `getHealth`, `getVersion`, `getGenesisHash`, `getBalance` + fixtures déterministes | | `pre.006` | `std.transport` schema/document/example + registry Config + adapter Config -> Transport + sensibilité/env tests | | `pre.007` | completeness/canaries, smoke opt-in, `cargo tree`, README/USAGE, docs finales, prompt `0.2.2`, préparation `rel.001` | @@ -914,6 +914,26 @@ Les éléments annoncés pour `pre.003` dans cet état historique sont désormai Restent volontairement à `pre.004` : token bucket RPS/burst, semaphore de concurrence, cooldown/429, deadline effective, retry/backoff et mutations passives `Degraded/RateLimited`. +### 22.3 État après `0.2.1-pre.004` + +`pre.004` matérialise la résilience runtime du pool sans encore exécuter les quatre méthodes JSON-RPC canari : + +- `workspace.package.version = 0.2.1-pre.4` ; +- limiteur token-bucket par couple endpoint/rôle avec RPS et burst ; lorsque RPS est configuré sans burst explicite, la capacité initiale dérive d'une seconde de RPS ; +- semaphore Tokio par rôle pour `max_concurrent_requests`, avec permit détenu pendant la durée de l'admission/exécution et notification des waiters à la libération ; +- cooldown par rôle après rate-limit provider, utilisant la valeur configurée ou un fallback runtime borné ; un `Retry-After` déjà résolu en durée peut prolonger ce cooldown dans une borne défensive ; +- sélection runtime par priorité puis round-robin, avec tentative des pairs et tiers inférieurs lorsqu'un candidat est temporairement bloqué par RPS, cooldown ou concurrence ; +- si tous les candidats sont temporairement bloqués, attente du premier signal/candidat admissible sans dépasser une deadline commune ; par défaut, cette deadline utilise le plus petit `request_timeout` des endpoints structurellement compatibles ; +- snapshots sûrs enrichis avec availability de rôle, limites, requêtes en vol, cooldown restant et compteurs success/failure/rate-limit, sans URL ni secret ; +- transitions passives `Available -> Degraded/RateLimited` sur observations runtime et retour `Degraded -> Available` après succès ; +- policy centralisée `evaluate_transport_retry()` : backoff exponentiel borné, budget `max_retries`, causes transport explicitement retryables, RPC application errors hors retry, et respect strict de `TransportRetryClass::NeverAfterDispatch` pour empêcher un resend automatique après dispatch ambigu ; +- `tokio` est consommé directement par Transport uniquement via la dépendance workspace avec feature locale `sync`; `time`/runtime restent fournis par le feature-set workspace existant ; +- la suite Transport compte désormais **72 tests déclarés**, avec preuves supplémentaires sur limiter, concurrence, cooldown, deadline, fallback runtime, santé passive, policy de retry et consommation publique de l'admission async. + +La tranche ne parse pas encore elle-même le header HTTP `Retry-After` et n'exécute pas de POST JSON-RPC : ces signaux seront raccordés à l'exécuteur HTTP avec les méthodes canari de `pre.005`. + +Restent à `pre.005` : exécution HTTP JSON-RPC réelle et wrappers typés `getHealth`, `getVersion`, `getGenesisHash`, `getBalance`, avec fixtures déterministes et raccordement des erreurs HTTP/429/timeout à la résilience maintenant disponible. + ## 23. Séquence `0.2.x` recalibrée ```text