khadhroony-solana-project/crates/ksp-onchain-transport-lib/src/http_resilience.rs

// 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;