506 lines
19 KiB
Rust
506 lines
19 KiB
Rust
// file: crates/ksp-onchain-transport-lib/src/client.rs
|
|
// version: 5
|
|
|
|
/// Passive runtime availability reported for one logical HTTP endpoint or role.
|
|
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
|
|
pub enum HttpEndpointAvailability {
|
|
/// The endpoint or role is administratively disabled and cannot be selected.
|
|
Disabled,
|
|
/// The endpoint or role is enabled and currently eligible for selection.
|
|
Available,
|
|
/// The endpoint or role is enabled but degraded by recent passive runtime observations.
|
|
Degraded,
|
|
/// The endpoint or role is temporarily excluded after provider rate limiting.
|
|
RateLimited,
|
|
}
|
|
|
|
/// Safe routing and resilience snapshot for one configured endpoint role.
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct HttpEndpointRoleSnapshot {
|
|
role: std::string::String,
|
|
enabled: bool,
|
|
request_kinds: std::vec::Vec<std::string::String>,
|
|
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 {
|
|
/// Returns the logical role name.
|
|
#[must_use]
|
|
pub fn role(&self) -> &str {
|
|
return self.role.as_str();
|
|
}
|
|
|
|
/// Returns whether the role is enabled.
|
|
#[must_use]
|
|
pub const fn enabled(&self) -> bool {
|
|
return self.enabled;
|
|
}
|
|
|
|
/// Returns request-kind descriptors accepted by the role.
|
|
#[must_use]
|
|
pub fn request_kinds(&self) -> &[std::string::String] {
|
|
return self.request_kinds.as_slice();
|
|
}
|
|
|
|
/// Returns the routing priority where lower values are preferred.
|
|
#[must_use]
|
|
pub const fn priority(&self) -> u32 {
|
|
return self.priority;
|
|
}
|
|
|
|
/// Returns the passive runtime availability of this role.
|
|
#[must_use]
|
|
pub const fn availability(&self) -> crate::HttpEndpointAvailability {
|
|
return self.availability;
|
|
}
|
|
|
|
/// Returns the configured requests-per-second limit.
|
|
#[must_use]
|
|
pub const fn requests_per_second(&self) -> std::option::Option<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.
|
|
///
|
|
/// Endpoint URLs are intentionally absent because they can contain provider credentials.
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct HttpEndpointSnapshot {
|
|
name: std::string::String,
|
|
provider: std::string::String,
|
|
cluster: std::string::String,
|
|
enabled: bool,
|
|
availability: crate::HttpEndpointAvailability,
|
|
roles: std::vec::Vec<crate::HttpEndpointRoleSnapshot>,
|
|
}
|
|
|
|
impl HttpEndpointSnapshot {
|
|
/// Returns the configured endpoint identity.
|
|
#[must_use]
|
|
pub fn name(&self) -> &str {
|
|
return self.name.as_str();
|
|
}
|
|
|
|
/// Returns the provider descriptor.
|
|
#[must_use]
|
|
pub fn provider(&self) -> &str {
|
|
return self.provider.as_str();
|
|
}
|
|
|
|
/// Returns the cluster descriptor.
|
|
#[must_use]
|
|
pub fn cluster(&self) -> &str {
|
|
return self.cluster.as_str();
|
|
}
|
|
|
|
/// Returns whether the endpoint is administratively enabled.
|
|
#[must_use]
|
|
pub const fn enabled(&self) -> bool {
|
|
return self.enabled;
|
|
}
|
|
|
|
/// Returns the passive runtime availability.
|
|
#[must_use]
|
|
pub const fn availability(&self) -> crate::HttpEndpointAvailability {
|
|
return self.availability;
|
|
}
|
|
|
|
/// Returns safe role snapshots in declaration order.
|
|
#[must_use]
|
|
pub fn roles(&self) -> &[crate::HttpEndpointRoleSnapshot] {
|
|
return self.roles.as_slice();
|
|
}
|
|
}
|
|
|
|
pub(crate) struct HttpEndpointHttpResponse {
|
|
status: u16,
|
|
retry_after: std::option::Option<std::time::Duration>,
|
|
body: std::vec::Vec<u8>,
|
|
}
|
|
|
|
impl HttpEndpointHttpResponse {
|
|
pub(crate) const fn status(&self) -> u16 {
|
|
return self.status;
|
|
}
|
|
|
|
pub(crate) const fn retry_after(&self) -> std::option::Option<std::time::Duration> {
|
|
return self.retry_after;
|
|
}
|
|
|
|
pub(crate) fn body(&self) -> &[u8] {
|
|
return self.body.as_slice();
|
|
}
|
|
}
|
|
|
|
/// Shareable logical HTTP endpoint client owned by KSP Transport.
|
|
///
|
|
/// The underlying `reqwest::Client` owns socket pooling. KSP keeps the configured URL private from diagnostics and exposes only safe routing metadata.
|
|
#[derive(Clone)]
|
|
pub struct HttpEndpointClient {
|
|
inner: std::sync::Arc<HttpEndpointClientInner>,
|
|
}
|
|
|
|
struct HttpEndpointClientInner {
|
|
settings: crate::HttpEndpointSettings,
|
|
client: reqwest::Client,
|
|
role_runtimes: std::vec::Vec<std::sync::Arc<crate::resilience::HttpRoleRuntime>>,
|
|
}
|
|
|
|
impl HttpEndpointClient {
|
|
/// Builds one logical endpoint client from KSP-owned runtime settings.
|
|
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);
|
|
if let std::result::Result::Err(error) = validation {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let client_result = build_reqwest_client(&settings);
|
|
let client = match client_result {
|
|
std::result::Result::Ok(client) => client,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_HTTP_CONNECTION_FAILED, "HTTP endpoint client could not be initialized")
|
|
.with_context("endpoint_name", settings.name())
|
|
.with_source(error.without_url()),
|
|
);
|
|
},
|
|
};
|
|
let mut role_runtimes = std::vec::Vec::with_capacity(settings.roles().len());
|
|
for role in settings.roles() {
|
|
role_runtimes.push(std::sync::Arc::new(crate::resilience::HttpRoleRuntime::new(role, std::sync::Arc::clone(¬ify))));
|
|
}
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
endpoint_name = settings.name(),
|
|
provider = settings.provider().as_str(),
|
|
cluster = settings.cluster().as_str(),
|
|
enabled = settings.enabled(),
|
|
role_count = settings.roles().len(),
|
|
"created logical HTTP endpoint client"
|
|
);
|
|
return std::result::Result::Ok(Self { inner: std::sync::Arc::new(HttpEndpointClientInner { settings, client, role_runtimes }) });
|
|
}
|
|
|
|
/// Returns the endpoint identity used for safe diagnostics and routing.
|
|
#[must_use]
|
|
pub fn name(&self) -> &str {
|
|
return self.inner.settings.name();
|
|
}
|
|
|
|
/// Returns the provider descriptor.
|
|
#[must_use]
|
|
pub fn provider(&self) -> &crate::HttpProviderName {
|
|
return self.inner.settings.provider();
|
|
}
|
|
|
|
/// Returns the cluster descriptor.
|
|
#[must_use]
|
|
pub fn cluster(&self) -> &crate::HttpClusterName {
|
|
return self.inner.settings.cluster();
|
|
}
|
|
|
|
/// Returns whether the endpoint is administratively enabled.
|
|
#[must_use]
|
|
pub fn enabled(&self) -> bool {
|
|
return self.inner.settings.enabled();
|
|
}
|
|
|
|
/// Returns the configured end-to-end request timeout.
|
|
#[must_use]
|
|
pub fn request_timeout(&self) -> std::time::Duration {
|
|
return self.inner.settings.request_timeout();
|
|
}
|
|
|
|
pub(crate) async fn post_json_rpc(&self, payload: &str, timeout: std::time::Duration) -> ksp_core_lib::Result<HttpEndpointHttpResponse> {
|
|
if timeout.is_zero() {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, "HTTP JSON-RPC request budget expired before dispatch")
|
|
.with_context("endpoint_name", self.name()),
|
|
);
|
|
}
|
|
let send_result = self
|
|
.inner
|
|
.client
|
|
.post(self.inner.settings.url().as_str())
|
|
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
|
.body(payload.to_owned())
|
|
.timeout(timeout)
|
|
.send()
|
|
.await;
|
|
let response = match send_result {
|
|
std::result::Result::Ok(response) => response,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(map_reqwest_error(self.name(), error)),
|
|
};
|
|
let status = response.status().as_u16();
|
|
let retry_after = parse_retry_after(response.headers());
|
|
let body_result = response.bytes().await;
|
|
let body = match body_result {
|
|
std::result::Result::Ok(body) => body.to_vec(),
|
|
std::result::Result::Err(error) => return std::result::Result::Err(map_reqwest_error(self.name(), error)),
|
|
};
|
|
return std::result::Result::Ok(HttpEndpointHttpResponse { status, retry_after, body });
|
|
}
|
|
|
|
/// Returns whether one enabled role can serve the requested capability structurally.
|
|
#[must_use]
|
|
pub fn supports(&self, role: &crate::HttpRoleName, request_kind: &crate::HttpRequestKind) -> bool {
|
|
return self.matching_role(role, request_kind).is_some();
|
|
}
|
|
|
|
/// Returns a safe endpoint snapshot with no URL or provider credential material.
|
|
#[must_use]
|
|
pub fn snapshot(&self) -> crate::HttpEndpointSnapshot {
|
|
let mut roles = std::vec::Vec::with_capacity(self.inner.settings.roles().len());
|
|
for (role_index, role) in self.inner.settings.roles().iter().enumerate() {
|
|
let mut request_kinds = std::vec::Vec::with_capacity(role.request_kinds().len());
|
|
for request_kind in role.request_kinds() {
|
|
request_kinds.push(request_kind.as_str().to_owned());
|
|
}
|
|
let runtime = self.inner.role_runtimes.get(role_index);
|
|
let role_snapshot = match runtime {
|
|
std::option::Option::Some(runtime) => role_snapshot(role, request_kinds, runtime),
|
|
std::option::Option::None => fallback_role_snapshot(role, request_kinds),
|
|
};
|
|
roles.push(role_snapshot);
|
|
}
|
|
return crate::HttpEndpointSnapshot {
|
|
name: self.name().to_owned(),
|
|
provider: self.provider().as_str().to_owned(),
|
|
cluster: self.cluster().as_str().to_owned(),
|
|
enabled: self.enabled(),
|
|
availability: self.availability(),
|
|
roles,
|
|
};
|
|
}
|
|
|
|
pub(crate) fn matching_role<'a>(
|
|
&'a self,
|
|
role: &crate::HttpRoleName,
|
|
request_kind: &crate::HttpRequestKind,
|
|
) -> std::option::Option<&'a crate::HttpEndpointRoleSettings> {
|
|
if !self.enabled() {
|
|
return std::option::Option::None;
|
|
}
|
|
for candidate_role in self.inner.settings.roles() {
|
|
if !candidate_role.enabled() || candidate_role.role() != role {
|
|
continue;
|
|
}
|
|
for capability in candidate_role.request_kinds() {
|
|
if capability.is_wildcard() || capability == request_kind {
|
|
return std::option::Option::Some(candidate_role);
|
|
}
|
|
}
|
|
}
|
|
return std::option::Option::None;
|
|
}
|
|
|
|
pub(crate) fn matching_role_runtime(
|
|
&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;
|
|
}
|
|
|
|
pub(crate) fn availability(&self) -> crate::HttpEndpointAvailability {
|
|
if !self.enabled() {
|
|
return crate::HttpEndpointAvailability::Disabled;
|
|
}
|
|
let now = std::time::Instant::now();
|
|
let mut enabled_role_count = 0_usize;
|
|
let mut rate_limited_count = 0_usize;
|
|
let mut degraded = false;
|
|
for (role_index, role) in self.inner.settings.roles().iter().enumerate() {
|
|
if !role.enabled() {
|
|
continue;
|
|
}
|
|
enabled_role_count = enabled_role_count.saturating_add(1);
|
|
let runtime = self.inner.role_runtimes.get(role_index);
|
|
let availability = match runtime {
|
|
std::option::Option::Some(runtime) => runtime.availability(now),
|
|
std::option::Option::None => crate::HttpEndpointAvailability::Degraded,
|
|
};
|
|
if availability == crate::HttpEndpointAvailability::RateLimited {
|
|
rate_limited_count = rate_limited_count.saturating_add(1);
|
|
}
|
|
if availability == crate::HttpEndpointAvailability::Degraded || availability == crate::HttpEndpointAvailability::RateLimited {
|
|
degraded = true;
|
|
}
|
|
}
|
|
if enabled_role_count > 0 && rate_limited_count == enabled_role_count {
|
|
return crate::HttpEndpointAvailability::RateLimited;
|
|
}
|
|
if degraded || enabled_role_count == 0 {
|
|
return crate::HttpEndpointAvailability::Degraded;
|
|
}
|
|
return crate::HttpEndpointAvailability::Available;
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for HttpEndpointClient {
|
|
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
return formatter.debug_struct("HttpEndpointClient").field("snapshot", &self.snapshot()).finish();
|
|
}
|
|
}
|
|
|
|
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 map_reqwest_error(endpoint_name: &str, error: reqwest::Error) -> ksp_core_lib::Error {
|
|
let code = if error.is_timeout() {
|
|
crate::ERROR_CODE_TIMEOUT
|
|
} else if error.is_connect() {
|
|
crate::ERROR_CODE_HTTP_CONNECTION_FAILED
|
|
} else {
|
|
crate::ERROR_CODE_HTTP_REQUEST_FAILED
|
|
};
|
|
return ksp_core_lib::Error::new(code, "HTTP JSON-RPC request failed").with_context("endpoint_name", endpoint_name).with_source(error.without_url());
|
|
}
|
|
|
|
fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> std::option::Option<std::time::Duration> {
|
|
let value = match headers.get(reqwest::header::RETRY_AFTER) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::option::Option::None,
|
|
};
|
|
let text = match value.to_str() {
|
|
std::result::Result::Ok(text) => text.trim(),
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let seconds = match text.parse::<u64>() {
|
|
std::result::Result::Ok(seconds) => seconds,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
return std::option::Option::Some(std::time::Duration::from_secs(seconds));
|
|
}
|
|
|
|
fn build_reqwest_client(settings: &crate::HttpEndpointSettings) -> std::result::Result<reqwest::Client, reqwest::Error> {
|
|
let mut builder = reqwest::Client::builder()
|
|
.connect_timeout(settings.connect_timeout())
|
|
.timeout(settings.request_timeout())
|
|
.redirect(reqwest::redirect::Policy::none())
|
|
.no_proxy()
|
|
.user_agent(concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION")));
|
|
if let std::option::Option::Some(max_idle) = settings.max_idle_connections_per_host() {
|
|
builder = builder.pool_max_idle_per_host(max_idle);
|
|
}
|
|
return builder.build();
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/client.rs"]
|
|
mod tests;
|