v0.2.9-pre.006
This commit is contained in:
405
crates/ksp-onchain-transport-lib/src/http_resilience.rs
Normal file
405
crates/ksp-onchain-transport-lib/src/http_resilience.rs
Normal file
@@ -0,0 +1,405 @@
|
||||
// file: crates/ksp-onchain-transport-lib/src/http_resilience.rs
|
||||
// version: 4
|
||||
|
||||
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),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// Crate-internal `HttpRoleRuntime` state shared across the owning crate.
|
||||
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 {
|
||||
/// Creates a new `HttpRoleRuntime` value.
|
||||
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),
|
||||
};
|
||||
}
|
||||
|
||||
/// Returns the current availability.
|
||||
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;
|
||||
}
|
||||
|
||||
/// Returns the current cooldown remaining.
|
||||
pub(crate) fn cooldown_remaining(&self) -> std::option::Option<std::time::Duration> {
|
||||
return self.cooldown_remaining_at(std::time::Instant::now());
|
||||
}
|
||||
|
||||
/// Returns the current max concurrent requests.
|
||||
pub(crate) fn max_concurrent_requests(&self) -> std::option::Option<u32> {
|
||||
return self.limits.max_concurrent_requests().map(|value| return value.get());
|
||||
}
|
||||
|
||||
/// Returns the current in flight requests.
|
||||
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));
|
||||
}
|
||||
|
||||
/// Returns the current success count.
|
||||
pub(crate) fn success_count(&self) -> u64 {
|
||||
return self.success_count.load(std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Returns the current failure count.
|
||||
pub(crate) fn failure_count(&self) -> u64 {
|
||||
return self.failure_count.load(std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Returns the current rate limit count.
|
||||
pub(crate) fn rate_limit_count(&self) -> u64 {
|
||||
return self.rate_limit_count.load(std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Attempts to acquire.
|
||||
pub(crate) fn try_acquire(self: &std::sync::Arc<Self>, now: std::time::Instant) -> crate::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 crate::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 crate::RoleAdmissionAttempt::ConcurrencySaturated,
|
||||
std::result::Result::Err(tokio::sync::TryAcquireError::Closed) => return crate::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 crate::RoleAdmissionAttempt::BlockedUntil(ready_at);
|
||||
}
|
||||
return crate::RoleAdmissionAttempt::Ready(HttpConcurrencyPermit { semaphore_permit, notify: std::sync::Arc::clone(&self.notify) });
|
||||
}
|
||||
|
||||
/// Records success.
|
||||
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;
|
||||
}
|
||||
|
||||
/// Records failure.
|
||||
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;
|
||||
}
|
||||
|
||||
/// Records rate limited.
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// Crate-internal `RoleAdmissionAttempt` variants used by the owning crate.
|
||||
pub(crate) enum RoleAdmissionAttempt {
|
||||
Ready(HttpConcurrencyPermit),
|
||||
BlockedUntil(std::time::Instant),
|
||||
ConcurrencySaturated,
|
||||
Unavailable,
|
||||
}
|
||||
|
||||
/// Crate-internal `HttpConcurrencyPermit` state shared across the owning crate.
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
/// 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);
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../unit_tests/http_resilience.rs"]
|
||||
mod tests;
|
||||
Reference in New Issue
Block a user