v0.2.1-pre.004
This commit is contained in:
@@ -1,12 +1,12 @@
|
|||||||
# file: Cargo.toml
|
# file: Cargo.toml
|
||||||
# version: 99
|
# version: 100
|
||||||
|
|
||||||
[workspace]
|
[workspace]
|
||||||
resolver = "3"
|
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"]
|
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]
|
[workspace.package]
|
||||||
version = "0.2.1-pre.3.fix.1"
|
version = "0.2.1-pre.4"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project"
|
repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project"
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
<!-- file: ROADMAP.md -->
|
<!-- file: ROADMAP.md -->
|
||||||
<!-- version: 29 -->
|
<!-- version: 30 -->
|
||||||
|
|
||||||
# Roadmap KSP
|
# Roadmap KSP
|
||||||
|
|
||||||
@@ -41,7 +41,7 @@ Le roadmap décrit les objectifs à atteindre et les grandes étapes prévues. U
|
|||||||
### Cadrage
|
### 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.
|
- [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
|
### Releases fonctionnelles décidées/pressenties
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
# file: crates/ksp-onchain-transport-lib/Cargo.toml
|
# file: crates/ksp-onchain-transport-lib/Cargo.toml
|
||||||
# version: 1
|
# version: 2
|
||||||
|
|
||||||
[package]
|
[package]
|
||||||
name = "ksp-onchain-transport-lib"
|
name = "ksp-onchain-transport-lib"
|
||||||
@@ -13,6 +13,7 @@ ksp-logging-lib = { path = "../ksp-logging-lib" }
|
|||||||
reqwest.workspace = true
|
reqwest.workspace = true
|
||||||
serde.workspace = true
|
serde.workspace = true
|
||||||
serde_json.workspace = true
|
serde_json.workspace = true
|
||||||
|
tokio = { workspace = true, features = ["sync"] }
|
||||||
|
|
||||||
[lints]
|
[lints]
|
||||||
workspace = true
|
workspace = true
|
||||||
|
|||||||
@@ -1,26 +1,35 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/src/client.rs
|
// 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)]
|
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
|
||||||
pub enum HttpEndpointAvailability {
|
pub enum HttpEndpointAvailability {
|
||||||
/// The endpoint is administratively disabled and cannot be selected.
|
/// The endpoint or role is administratively disabled and cannot be selected.
|
||||||
Disabled,
|
Disabled,
|
||||||
/// The endpoint is enabled and currently eligible for selection.
|
/// The endpoint or role is enabled and currently eligible for selection.
|
||||||
Available,
|
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,
|
Degraded,
|
||||||
/// The endpoint is enabled but temporarily excluded after provider rate limiting.
|
/// The endpoint or role is temporarily excluded after provider rate limiting.
|
||||||
RateLimited,
|
RateLimited,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Safe routing snapshot for one configured endpoint role.
|
/// Safe routing and resilience snapshot for one configured endpoint role.
|
||||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
pub struct HttpEndpointRoleSnapshot {
|
pub struct HttpEndpointRoleSnapshot {
|
||||||
role: std::string::String,
|
role: std::string::String,
|
||||||
enabled: bool,
|
enabled: bool,
|
||||||
request_kinds: std::vec::Vec<std::string::String>,
|
request_kinds: std::vec::Vec<std::string::String>,
|
||||||
priority: u32,
|
priority: u32,
|
||||||
|
availability: crate::HttpEndpointAvailability,
|
||||||
|
requests_per_second: std::option::Option<u32>,
|
||||||
|
burst_capacity: std::option::Option<u32>,
|
||||||
|
max_concurrent_requests: std::option::Option<u32>,
|
||||||
|
in_flight_requests: std::option::Option<u32>,
|
||||||
|
cooldown_remaining: std::option::Option<std::time::Duration>,
|
||||||
|
success_count: u64,
|
||||||
|
failure_count: u64,
|
||||||
|
rate_limit_count: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl HttpEndpointRoleSnapshot {
|
impl HttpEndpointRoleSnapshot {
|
||||||
@@ -47,6 +56,60 @@ impl HttpEndpointRoleSnapshot {
|
|||||||
pub const fn priority(&self) -> u32 {
|
pub const fn priority(&self) -> u32 {
|
||||||
return self.priority;
|
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<u32> {
|
||||||
|
return self.requests_per_second;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the configured token-bucket burst capacity.
|
||||||
|
#[must_use]
|
||||||
|
pub const fn burst_capacity(&self) -> std::option::Option<u32> {
|
||||||
|
return self.burst_capacity;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the configured maximum concurrent request count.
|
||||||
|
#[must_use]
|
||||||
|
pub const fn max_concurrent_requests(&self) -> std::option::Option<u32> {
|
||||||
|
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<u32> {
|
||||||
|
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<std::time::Duration> {
|
||||||
|
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.
|
/// Safe metadata snapshot for one logical HTTP endpoint.
|
||||||
@@ -111,12 +174,16 @@ pub struct HttpEndpointClient {
|
|||||||
struct HttpEndpointClientInner {
|
struct HttpEndpointClientInner {
|
||||||
settings: crate::HttpEndpointSettings,
|
settings: crate::HttpEndpointSettings,
|
||||||
_client: reqwest::Client,
|
_client: reqwest::Client,
|
||||||
availability: std::sync::atomic::AtomicU8,
|
role_runtimes: std::vec::Vec<std::sync::Arc<crate::resilience::HttpRoleRuntime>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl HttpEndpointClient {
|
impl HttpEndpointClient {
|
||||||
/// Builds one logical endpoint client from KSP-owned runtime settings.
|
/// Builds one logical endpoint client from KSP-owned runtime settings.
|
||||||
pub fn new(settings: crate::HttpEndpointSettings) -> ksp_core_lib::Result<Self> {
|
pub fn new(settings: crate::HttpEndpointSettings) -> ksp_core_lib::Result<Self> {
|
||||||
|
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<tokio::sync::Notify>) -> ksp_core_lib::Result<Self> {
|
||||||
let validation = crate::settings::validate_endpoint_settings(&settings);
|
let validation = crate::settings::validate_endpoint_settings(&settings);
|
||||||
if let std::result::Result::Err(error) = validation {
|
if let std::result::Result::Err(error) = validation {
|
||||||
return std::result::Result::Err(error);
|
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!(
|
ksp_logging_lib::debug!(
|
||||||
target: env!("CARGO_PKG_NAME"),
|
target: env!("CARGO_PKG_NAME"),
|
||||||
endpoint_name = settings.name(),
|
endpoint_name = settings.name(),
|
||||||
provider = settings.provider().as_str(),
|
provider = settings.provider().as_str(),
|
||||||
cluster = settings.cluster().as_str(),
|
cluster = settings.cluster().as_str(),
|
||||||
enabled = settings.enabled(),
|
enabled = settings.enabled(),
|
||||||
|
role_count = settings.roles().len(),
|
||||||
"created logical HTTP endpoint client"
|
"created logical HTTP endpoint client"
|
||||||
);
|
);
|
||||||
return std::result::Result::Ok(Self {
|
return std::result::Result::Ok(Self { inner: std::sync::Arc::new(HttpEndpointClientInner { settings, _client: client, role_runtimes }) });
|
||||||
inner: std::sync::Arc::new(HttpEndpointClientInner { settings, _client: client, availability: std::sync::atomic::AtomicU8::new(availability) }),
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns the endpoint identity used for safe diagnostics and routing.
|
/// Returns the endpoint identity used for safe diagnostics and routing.
|
||||||
@@ -170,7 +239,13 @@ impl HttpEndpointClient {
|
|||||||
return self.inner.settings.enabled();
|
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]
|
#[must_use]
|
||||||
pub fn supports(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> bool {
|
pub fn supports(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> bool {
|
||||||
return self.matching_role(role, request_kind).is_some();
|
return self.matching_role(role, request_kind).is_some();
|
||||||
@@ -180,17 +255,17 @@ impl HttpEndpointClient {
|
|||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn snapshot(&self) -> crate::HttpEndpointSnapshot {
|
pub fn snapshot(&self) -> crate::HttpEndpointSnapshot {
|
||||||
let mut roles = std::vec::Vec::with_capacity(self.inner.settings.roles().len());
|
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());
|
let mut request_kinds = std::vec::Vec::with_capacity(role.request_kinds().len());
|
||||||
for request_kind in role.request_kinds() {
|
for request_kind in role.request_kinds() {
|
||||||
request_kinds.push(request_kind.as_str().to_owned());
|
request_kinds.push(request_kind.as_str().to_owned());
|
||||||
}
|
}
|
||||||
roles.push(crate::HttpEndpointRoleSnapshot {
|
let runtime = self.inner.role_runtimes.get(role_index);
|
||||||
role: role.role().as_str().to_owned(),
|
let role_snapshot = match runtime {
|
||||||
enabled: role.enabled(),
|
std::option::Option::Some(runtime) => role_snapshot(role, request_kinds, runtime),
|
||||||
request_kinds,
|
std::option::Option::None => fallback_role_snapshot(role, request_kinds),
|
||||||
priority: role.priority(),
|
};
|
||||||
});
|
roles.push(role_snapshot);
|
||||||
}
|
}
|
||||||
return crate::HttpEndpointSnapshot {
|
return crate::HttpEndpointSnapshot {
|
||||||
name: self.name().to_owned(),
|
name: self.name().to_owned(),
|
||||||
@@ -207,7 +282,7 @@ impl HttpEndpointClient {
|
|||||||
role: &crate::HttpRoleName,
|
role: &crate::HttpRoleName,
|
||||||
request_kind: &crate::HttpRequestKind,
|
request_kind: &crate::HttpRequestKind,
|
||||||
) -> std::option::Option<&'a crate::HttpEndpointRoleSettings> {
|
) -> std::option::Option<&'a crate::HttpEndpointRoleSettings> {
|
||||||
if !self.is_selectable() {
|
if !self.enabled() {
|
||||||
return std::option::Option::None;
|
return std::option::Option::None;
|
||||||
}
|
}
|
||||||
for candidate_role in self.inner.settings.roles() {
|
for candidate_role in self.inner.settings.roles() {
|
||||||
@@ -223,12 +298,68 @@ impl HttpEndpointClient {
|
|||||||
return std::option::Option::None;
|
return std::option::Option::None;
|
||||||
}
|
}
|
||||||
|
|
||||||
fn is_selectable(&self) -> bool {
|
pub(crate) fn matching_role_runtime(
|
||||||
return self.enabled() && self.availability() == crate::HttpEndpointAvailability::Available;
|
&self,
|
||||||
|
role: &crate::HttpRoleName,
|
||||||
|
request_kind: &crate::HttpRequestKind,
|
||||||
|
) -> std::option::Option<(u32, std::sync::Arc<crate::resilience::HttpRoleRuntime>)> {
|
||||||
|
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 {
|
pub(crate) fn availability(&self) -> crate::HttpEndpointAvailability {
|
||||||
return availability_from_code(self.inner.availability.load(std::sync::atomic::Ordering::Relaxed));
|
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<std::string::String>,
|
||||||
|
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<std::string::String>) -> 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<reqwest::Client, reqwest::Error> {
|
fn build_reqwest_client(settings: &crate::HttpEndpointSettings) -> std::result::Result<reqwest::Client, reqwest::Error> {
|
||||||
let mut builder = reqwest::Client::builder()
|
let mut builder = reqwest::Client::builder()
|
||||||
.connect_timeout(settings.connect_timeout())
|
.connect_timeout(settings.connect_timeout())
|
||||||
@@ -251,24 +423,6 @@ fn build_reqwest_client(settings: &crate::HttpEndpointSettings) -> std::result::
|
|||||||
return builder.build();
|
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)]
|
#[cfg(test)]
|
||||||
#[path = "../unit_tests/client.rs"]
|
#[path = "../unit_tests/client.rs"]
|
||||||
mod tests;
|
mod tests;
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/src/lib.rs
|
// file: crates/ksp-onchain-transport-lib/src/lib.rs
|
||||||
// version: 2
|
// version: 3
|
||||||
#![warn(missing_docs)]
|
#![warn(missing_docs)]
|
||||||
#![deny(unreachable_pub)]
|
#![deny(unreachable_pub)]
|
||||||
#![forbid(unsafe_code)]
|
#![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
|
//! 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
|
//! 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`
|
//! clients, priority-aware pools, bounded admission limits and retry/no-resend policy are available, while typed Solana method adapters remain staged by
|
||||||
//! prereleases.
|
//! subsequent `0.2.1` prereleases.
|
||||||
|
|
||||||
mod client;
|
mod client;
|
||||||
mod error;
|
mod error;
|
||||||
mod json_rpc;
|
mod json_rpc;
|
||||||
mod pool;
|
mod pool;
|
||||||
|
mod resilience;
|
||||||
mod rpc_method;
|
mod rpc_method;
|
||||||
mod settings;
|
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;
|
pub use self::json_rpc::parse_json_rpc_response_value;
|
||||||
/// Result of one logical endpoint selection.
|
/// Result of one logical endpoint selection.
|
||||||
pub use self::pool::HttpEndpointSelection;
|
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;
|
pub use self::pool::HttpTransportPool;
|
||||||
/// Safe snapshot of the logical HTTP endpoint pool.
|
/// Safe snapshot of the logical HTTP endpoint pool.
|
||||||
pub use self::pool::HttpTransportPoolSnapshot;
|
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.
|
/// Functional category used by the audited Solana HTTP JSON-RPC registry.
|
||||||
pub use self::rpc_method::HttpRpcCategory;
|
pub use self::rpc_method::HttpRpcCategory;
|
||||||
/// Release that owns typed KSP coverage for one audited HTTP RPC method.
|
/// Release that owns typed KSP coverage for one audited HTTP RPC method.
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/src/pool.rs
|
// file: crates/ksp-onchain-transport-lib/src/pool.rs
|
||||||
// version: 2
|
// version: 3
|
||||||
|
|
||||||
/// Safe snapshot of the logical HTTP endpoint pool.
|
/// Safe snapshot of the logical HTTP endpoint pool.
|
||||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
@@ -20,7 +20,7 @@ impl HttpTransportPoolSnapshot {
|
|||||||
return self.endpoints.len();
|
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]
|
#[must_use]
|
||||||
pub fn available_endpoint_count(&self) -> usize {
|
pub fn available_endpoint_count(&self) -> usize {
|
||||||
return self.endpoints.iter().filter(|endpoint| return endpoint.availability() == crate::HttpEndpointAvailability::Available).count();
|
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<crate::resilience::HttpRoleRuntime>,
|
||||||
|
_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>) -> 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)]
|
#[derive(Clone)]
|
||||||
pub struct HttpTransportPool {
|
pub struct HttpTransportPool {
|
||||||
inner: std::sync::Arc<HttpTransportPoolInner>,
|
inner: std::sync::Arc<HttpTransportPoolInner>,
|
||||||
@@ -76,6 +167,8 @@ pub struct HttpTransportPool {
|
|||||||
|
|
||||||
struct HttpTransportPoolInner {
|
struct HttpTransportPoolInner {
|
||||||
clients: std::vec::Vec<crate::HttpEndpointClient>,
|
clients: std::vec::Vec<crate::HttpEndpointClient>,
|
||||||
|
retry: crate::HttpRetrySettings,
|
||||||
|
notify: std::sync::Arc<tokio::sync::Notify>,
|
||||||
cursors: std::sync::Mutex<std::collections::BTreeMap<(std::string::String, std::string::String, u32), usize>>,
|
cursors: std::sync::Mutex<std::collections::BTreeMap<(std::string::String, std::string::String, u32), usize>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -86,9 +179,10 @@ impl HttpTransportPool {
|
|||||||
if let std::result::Result::Err(error) = validation {
|
if let std::result::Result::Err(error) = validation {
|
||||||
return std::result::Result::Err(error);
|
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());
|
let mut clients = std::vec::Vec::with_capacity(settings.endpoints().len());
|
||||||
for endpoint in settings.endpoints() {
|
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 {
|
let client = match client_result {
|
||||||
std::result::Result::Ok(client) => client,
|
std::result::Result::Ok(client) => client,
|
||||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||||
@@ -96,18 +190,32 @@ impl HttpTransportPool {
|
|||||||
clients.push(client);
|
clients.push(client);
|
||||||
}
|
}
|
||||||
let pool = Self {
|
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!(
|
ksp_logging_lib::debug!(
|
||||||
target: env!("CARGO_PKG_NAME"),
|
target: env!("CARGO_PKG_NAME"),
|
||||||
endpoint_count = pool.inner.clients.len(),
|
endpoint_count = pool.inner.clients.len(),
|
||||||
available_endpoint_count = pool.snapshot().available_endpoint_count(),
|
available_endpoint_count = pool.snapshot().available_endpoint_count(),
|
||||||
|
max_retries = pool.inner.retry.max_retries(),
|
||||||
"created logical HTTP endpoint pool"
|
"created logical HTTP endpoint pool"
|
||||||
);
|
);
|
||||||
return std::result::Result::Ok(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<crate::HttpEndpointSelection> {
|
pub fn select_for_method(&self, role: &crate::HttpRoleName, method: &crate::HttpRpcMethodDescriptor) -> ksp_core_lib::Result<crate::HttpEndpointSelection> {
|
||||||
let support = method.ensure_runtime_supported();
|
let support = method.ensure_runtime_supported();
|
||||||
if let std::result::Result::Err(error) = support {
|
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()));
|
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(
|
pub fn select_for_request_kind(
|
||||||
&self,
|
&self,
|
||||||
role: &crate::HttpRoleName,
|
role: &crate::HttpRoleName,
|
||||||
request_kind: &crate::HttpRequestKind,
|
request_kind: &crate::HttpRequestKind,
|
||||||
) -> ksp_core_lib::Result<crate::HttpEndpointSelection> {
|
) -> ksp_core_lib::Result<crate::HttpEndpointSelection> {
|
||||||
|
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<crate::HttpRequestPermit> {
|
||||||
|
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<crate::HttpRequestPermit> {
|
||||||
|
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<crate::HttpRequestPermit> {
|
||||||
|
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<crate::HttpRequestPermit> {
|
||||||
|
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::time::Instant> = 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<std::time::Instant>,
|
||||||
|
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<std::time::Duration> {
|
||||||
|
let mut timeout: std::option::Option<std::time::Duration> = 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<PoolCandidate> {
|
||||||
let mut candidates = std::vec::Vec::new();
|
let mut candidates = std::vec::Vec::new();
|
||||||
for (client_index, client) in self.inner.clients.iter().enumerate() {
|
for (client_index, client) in self.inner.clients.iter().enumerate() {
|
||||||
let matching_role = client.matching_role(role, request_kind);
|
let matching_role = client.matching_role(role, request_kind);
|
||||||
@@ -129,22 +449,29 @@ impl HttpTransportPool {
|
|||||||
candidates.push(PoolCandidate { client_index, priority: matching_role.priority() });
|
candidates.push(PoolCandidate { client_index, priority: matching_role.priority() });
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if candidates.is_empty() {
|
candidates.sort_by_key(|candidate| return candidate.priority);
|
||||||
return selection_failed(role, request_kind);
|
return candidates;
|
||||||
|
}
|
||||||
|
|
||||||
|
fn runtime_candidates(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> std::vec::Vec<RuntimePoolCandidate> {
|
||||||
|
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);
|
candidates.sort_by_key(|candidate| return candidate.priority);
|
||||||
let first_candidate = candidates.first();
|
return candidates;
|
||||||
let best_priority = match first_candidate {
|
}
|
||||||
std::option::Option::Some(candidate) => candidate.priority,
|
|
||||||
std::option::Option::None => return selection_failed(role, request_kind),
|
fn selection_from_candidate(
|
||||||
};
|
&self,
|
||||||
let best_tier: std::vec::Vec<PoolCandidate> = candidates.into_iter().take_while(|candidate| return candidate.priority == best_priority).collect();
|
role: &crate::HttpRoleName,
|
||||||
let selected_position = self.next_position(role, request_kind, best_priority, best_tier.len());
|
request_kind: &crate::HttpRequestKind,
|
||||||
let selected = match best_tier.get(selected_position) {
|
candidate: &PoolCandidate,
|
||||||
std::option::Option::Some(selected) => selected,
|
) -> ksp_core_lib::Result<crate::HttpEndpointSelection> {
|
||||||
std::option::Option::None => return selection_failed(role, request_kind),
|
let selected_client = self.inner.clients.get(candidate.client_index);
|
||||||
};
|
|
||||||
let selected_client = self.inner.clients.get(selected.client_index);
|
|
||||||
let client = match selected_client {
|
let client = match selected_client {
|
||||||
std::option::Option::Some(client) => client.clone(),
|
std::option::Option::Some(client) => client.clone(),
|
||||||
std::option::Option::None => return selection_failed(role, request_kind),
|
std::option::Option::None => return selection_failed(role, request_kind),
|
||||||
@@ -154,23 +481,34 @@ impl HttpTransportPool {
|
|||||||
endpoint_name = client.name(),
|
endpoint_name = client.name(),
|
||||||
role = role.as_str(),
|
role = role.as_str(),
|
||||||
request_kind = request_kind.as_str(),
|
request_kind = request_kind.as_str(),
|
||||||
priority = best_priority,
|
priority = candidate.priority,
|
||||||
tier_size = best_tier.len(),
|
"selected logical HTTP endpoint without runtime admission"
|
||||||
"selected logical HTTP endpoint"
|
|
||||||
);
|
);
|
||||||
return std::result::Result::Ok(crate::HttpEndpointSelection {
|
return std::result::Result::Ok(crate::HttpEndpointSelection {
|
||||||
client,
|
client,
|
||||||
role: role.clone(),
|
role: role.clone(),
|
||||||
request_kind: request_kind.clone(),
|
request_kind: request_kind.clone(),
|
||||||
priority: best_priority,
|
priority: candidate.priority,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns a safe pool snapshot without endpoint URLs.
|
fn selection_from_runtime_candidate(
|
||||||
#[must_use]
|
&self,
|
||||||
pub fn snapshot(&self) -> crate::HttpTransportPoolSnapshot {
|
role: &crate::HttpRoleName,
|
||||||
let endpoints = self.inner.clients.iter().map(crate::HttpEndpointClient::snapshot).collect();
|
request_kind: &crate::HttpRequestKind,
|
||||||
return crate::HttpTransportPoolSnapshot { endpoints };
|
candidate: &RuntimePoolCandidate,
|
||||||
|
) -> ksp_core_lib::Result<crate::HttpEndpointSelection> {
|
||||||
|
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 {
|
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,
|
priority: u32,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct RuntimePoolCandidate {
|
||||||
|
client_index: usize,
|
||||||
|
priority: u32,
|
||||||
|
runtime: std::sync::Arc<crate::resilience::HttpRoleRuntime>,
|
||||||
|
}
|
||||||
|
|
||||||
|
enum RuntimeSelectionAttempt {
|
||||||
|
Ready(crate::HttpRequestPermit),
|
||||||
|
Blocked { earliest_ready: std::option::Option<std::time::Instant>, 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<std::time::Instant>, candidate: std::time::Instant) -> std::option::Option<std::time::Instant> {
|
||||||
|
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<crate::HttpEndpointSelection> {
|
fn selection_failed(role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> ksp_core_lib::Result<crate::HttpEndpointSelection> {
|
||||||
|
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<std::time::Duration> {
|
||||||
|
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<crate::HttpRequestPermit> {
|
||||||
|
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<crate::HttpRequestPermit> {
|
||||||
|
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(
|
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("role", role.as_str())
|
||||||
.with_context("request_kind", request_kind.as_str()),
|
.with_context("request_kind", request_kind.as_str()),
|
||||||
);
|
);
|
||||||
|
|||||||
390
crates/ksp-onchain-transport-lib/src/resilience.rs
Normal file
390
crates/ksp-onchain-transport-lib/src/resilience.rs
Normal file
@@ -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<std::time::Duration> {
|
||||||
|
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<std::time::Duration>,
|
||||||
|
) -> 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<std::option::Option<HttpTokenBucketState>>,
|
||||||
|
semaphore: std::option::Option<std::sync::Arc<tokio::sync::Semaphore>>,
|
||||||
|
notify: std::sync::Arc<tokio::sync::Notify>,
|
||||||
|
cooldown_until: std::sync::Mutex<std::option::Option<std::time::Instant>>,
|
||||||
|
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<tokio::sync::Notify>) -> 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<std::time::Duration> {
|
||||||
|
return self.cooldown_remaining_at(std::time::Instant::now());
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn max_concurrent_requests(&self) -> std::option::Option<u32> {
|
||||||
|
return self.limits.max_concurrent_requests().map(|value| return value.get());
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn in_flight_requests(&self) -> std::option::Option<u32> {
|
||||||
|
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<Self>, 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>) -> 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<std::time::Duration> {
|
||||||
|
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<std::time::Instant> {
|
||||||
|
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<tokio::sync::OwnedSemaphorePermit>,
|
||||||
|
notify: std::sync::Arc<tokio::sync::Notify>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<std::time::Instant> {
|
||||||
|
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;
|
||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/src/settings.rs
|
// file: crates/ksp-onchain-transport-lib/src/settings.rs
|
||||||
// version: 3
|
// version: 4
|
||||||
|
|
||||||
/// Runtime HTTP endpoint URL owned by Transport.
|
/// Runtime HTTP endpoint URL owned by Transport.
|
||||||
///
|
///
|
||||||
@@ -156,7 +156,10 @@ pub struct HttpRoleLimits {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl 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]
|
#[must_use]
|
||||||
pub const fn new(
|
pub const fn new(
|
||||||
requests_per_second: std::option::Option<std::num::NonZeroU32>,
|
requests_per_second: std::option::Option<std::num::NonZeroU32>,
|
||||||
@@ -173,7 +176,7 @@ impl HttpRoleLimits {
|
|||||||
return self.requests_per_second;
|
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]
|
#[must_use]
|
||||||
pub const fn burst_capacity(&self) -> std::option::Option<std::num::NonZeroU32> {
|
pub const fn burst_capacity(&self) -> std::option::Option<std::num::NonZeroU32> {
|
||||||
return self.burst_capacity;
|
return self.burst_capacity;
|
||||||
@@ -185,7 +188,7 @@ impl HttpRoleLimits {
|
|||||||
return self.max_concurrent_requests;
|
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]
|
#[must_use]
|
||||||
pub const fn pause_after_rate_limit(&self) -> std::option::Option<std::time::Duration> {
|
pub const fn pause_after_rate_limit(&self) -> std::option::Option<std::time::Duration> {
|
||||||
return self.pause_after_rate_limit;
|
return self.pause_after_rate_limit;
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/tests/dependency_boundary.rs
|
// 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`.
|
//! 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-core-lib"));
|
||||||
assert!(manifest.contains("ksp-logging-lib"));
|
assert!(manifest.contains("ksp-logging-lib"));
|
||||||
assert!(manifest.contains("reqwest.workspace = true"));
|
assert!(manifest.contains("reqwest.workspace = true"));
|
||||||
|
assert!(manifest.contains("tokio = { workspace = true, features = [\"sync\"] }"));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/tests/public_api.rs
|
// 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.
|
//! 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("SECRET-CANARY"));
|
||||||
assert!(!rendered.contains("provider.invalid"));
|
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"));
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/unit_tests/client.rs
|
// file: crates/ksp-onchain-transport-lib/unit_tests/client.rs
|
||||||
// version: 1
|
// version: 2
|
||||||
|
|
||||||
fn endpoint(enabled: bool, url_text: &str) -> crate::HttpEndpointSettings {
|
fn endpoint(enabled: bool, url_text: &str) -> crate::HttpEndpointSettings {
|
||||||
let url = crate::HttpEndpointUrl::parse(url_text).expect("test endpoint URL must parse");
|
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("default"), &crate::HttpRequestKind::new("get_balance")));
|
||||||
assert!(!client.supports(&crate::HttpRoleName::new("write"), &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);
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
// file: crates/ksp-onchain-transport-lib/unit_tests/pool.rs
|
// 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::HttpRequestKind>) -> crate::HttpEndpointRoleSettings {
|
fn role(name: &str, priority: u32, request_kinds: std::vec::Vec<crate::HttpRequestKind>) -> crate::HttpEndpointRoleSettings {
|
||||||
return crate::HttpEndpointRoleSettings::new(
|
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<u32>,
|
||||||
|
burst_capacity: std::option::Option<u32>,
|
||||||
|
max_concurrent_requests: std::option::Option<u32>,
|
||||||
|
cooldown: std::option::Option<std::time::Duration>,
|
||||||
|
) -> 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::HttpEndpointSettings>) -> crate::HttpTransportSettings {
|
fn settings(endpoints: std::vec::Vec<crate::HttpEndpointSettings>) -> crate::HttpTransportSettings {
|
||||||
return crate::HttpTransportSettings::new(
|
return crate::HttpTransportSettings::new(
|
||||||
endpoints,
|
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");
|
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);
|
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);
|
||||||
|
}
|
||||||
|
|||||||
186
crates/ksp-onchain-transport-lib/unit_tests/resilience.rs
Normal file
186
crates/ksp-onchain-transport-lib/unit_tests/resilience.rs
Normal file
@@ -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<u32>,
|
||||||
|
burst_capacity: std::option::Option<u32>,
|
||||||
|
max_concurrent_requests: std::option::Option<u32>,
|
||||||
|
cooldown: std::option::Option<std::time::Duration>,
|
||||||
|
) -> 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);
|
||||||
|
}
|
||||||
231
deltas/0.2.1/pre.004.md
Normal file
231
deltas/0.2.1/pre.004.md
Normal file
@@ -0,0 +1,231 @@
|
|||||||
|
<!-- file: deltas/0.2.1/pre.004.md -->
|
||||||
|
<!-- version: 1 -->
|
||||||
|
|
||||||
|
# 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`.
|
||||||
@@ -1,5 +1,5 @@
|
|||||||
<!-- file: docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md -->
|
<!-- file: docs/plans/008-V0_2_1_ONCHAIN_HTTP_PLAN.md -->
|
||||||
<!-- version: 4 -->
|
<!-- version: 5 -->
|
||||||
|
|
||||||
# `0.2.1-pre.001` — plan `ksp-onchain-transport-lib` HTTP Solana foundation
|
# `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.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.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.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.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.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` |
|
| `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`.
|
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
|
## 23. Séquence `0.2.x` recalibrée
|
||||||
|
|
||||||
```text
|
```text
|
||||||
|
|||||||
Reference in New Issue
Block a user