khadhroony-solana-project/crates/ksp-offchain-transport-lib/src/http_admission.rs

// file: crates/ksp-offchain-transport-lib/src/http_admission.rs
// version: 3

//! Provider-neutral local request admission and rate-limit cooldown primitives.

/// Maximum provider-directed cooldown accepted from `Retry-After`.
pub(crate) const HTTP_MAX_RETRY_AFTER: std::time::Duration = std::time::Duration::from_secs(3_600);

const DEFAULT_RATE_LIMIT_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(1);
const MAX_RATE_LIMIT_WINDOW: std::time::Duration = std::time::Duration::from_secs(3_600);

/// Crate-internal local request-admission policy.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum HttpAdmissionPolicy {
    /// No stable local cadence is known; only provider-driven cooldown is enforced.
    Dynamic,
    /// Enforce a smooth token bucket for a documented request budget and window.
    Fixed { requests: u32, window: std::time::Duration, burst: u32 },
}

impl crate::HttpAdmissionPolicy {
    /// Creates a validated fixed-window admission policy.
    pub(crate) fn fixed(requests: u32, window: std::time::Duration, burst: std::option::Option<u32>) -> ksp_core_lib::Result<Self> {
        let burst = match burst {
            std::option::Option::Some(value) => value,
            std::option::Option::None => 1,
        };
        if requests == 0 || window.is_zero() || window > MAX_RATE_LIMIT_WINDOW || burst == 0 {
            return std::result::Result::Err(
                ksp_core_lib::Error::new(crate::ERROR_CODE_HTTP_RATE_LIMIT_INVALID, "HTTP request-admission policy is invalid")
                    .with_context("field", "rate_limit"),
            );
        }
        return std::result::Result::Ok(Self::Fixed { requests, window, burst });
    }
}

/// Result of one non-blocking local admission attempt.
pub(crate) enum HttpAdmissionDecision {
    /// The request may dispatch now and one local token has been consumed when applicable.
    Ready,
    /// The request must be deferred for at least this duration.
    Deferred(std::time::Duration),
}

/// Shared non-blocking limiter used by provider adapters and later refresh orchestration.
pub(crate) struct HttpAdmissionController {
    cooldown_until: std::sync::Mutex<std::option::Option<std::time::Instant>>,
    fallback_cooldown: std::time::Duration,
    token_bucket: std::sync::Mutex<std::option::Option<HttpTokenBucketState>>,
}

impl crate::HttpAdmissionController {
    /// Creates one limiter from a provider-owned local admission policy.
    pub(crate) fn new(policy: crate::HttpAdmissionPolicy, fallback_cooldown: std::option::Option<std::time::Duration>) -> ksp_core_lib::Result<Self> {
        let fallback_cooldown = match fallback_cooldown {
            std::option::Option::Some(value) => value,
            std::option::Option::None => DEFAULT_RATE_LIMIT_COOLDOWN,
        };
        if fallback_cooldown.is_zero() || fallback_cooldown > crate::HTTP_MAX_RETRY_AFTER {
            return std::result::Result::Err(
                ksp_core_lib::Error::new(crate::ERROR_CODE_HTTP_RATE_LIMIT_INVALID, "HTTP fallback cooldown is outside the supported bounds")
                    .with_context("field", "fallback_cooldown"),
            );
        }
        let token_bucket = match policy {
            crate::HttpAdmissionPolicy::Fixed { requests, window, burst } => {
                std::option::Option::Some(HttpTokenBucketState::new(requests, window, burst, std::time::Instant::now()))
            },
            crate::HttpAdmissionPolicy::Dynamic => std::option::Option::None,
        };
        return std::result::Result::Ok(Self {
            cooldown_until: std::sync::Mutex::new(std::option::Option::None),
            fallback_cooldown,
            token_bucket: std::sync::Mutex::new(token_bucket),
        });
    }

    /// Tries to admit one request immediately without sleeping.
    pub(crate) fn try_admit(&self) -> crate::HttpAdmissionDecision {
        return self.try_admit_at(std::time::Instant::now());
    }

    /// Records a provider 429 and extends cooldown using a bounded `Retry-After` value when present.
    pub(crate) fn record_rate_limited(&self, provider_retry_after: std::option::Option<std::time::Duration>) -> std::time::Duration {
        let provider_delay = match provider_retry_after {
            std::option::Option::Some(value) => std::cmp::min(value, crate::HTTP_MAX_RETRY_AFTER),
            std::option::Option::None => std::time::Duration::ZERO,
        };
        let effective = std::cmp::max(self.fallback_cooldown, provider_delay);
        self.record_cooldown_until(std::time::Instant::now(), effective);
        ksp_logging_lib::debug!(
            target: crate::TRACING_TARGET,
            cooldown_ms = duration_millis_u64(effective),
            provider_retry_after_present = provider_retry_after.is_some(),
            "recorded off-chain HTTP provider cooldown"
        );
        return effective;
    }

    /// Returns the remaining provider cooldown for deterministic limiter tests.
    #[cfg(test)]
    pub(crate) fn cooldown_remaining(&self) -> std::option::Option<std::time::Duration> {
        return self.cooldown_remaining_at(std::time::Instant::now());
    }

    fn try_admit_at(&self, now: std::time::Instant) -> crate::HttpAdmissionDecision {
        if let std::option::Option::Some(remaining) = self.cooldown_remaining_at(now) {
            return crate::HttpAdmissionDecision::Deferred(remaining);
        }
        let lock_result = self.token_bucket.lock();
        let mut token_bucket = match lock_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(poisoned) => poisoned.into_inner(),
        };
        let state = match token_bucket.as_mut() {
            std::option::Option::Some(value) => value,
            std::option::Option::None => return crate::HttpAdmissionDecision::Ready,
        };
        return match state.try_consume_at(now) {
            std::option::Option::Some(delay) => crate::HttpAdmissionDecision::Deferred(delay),
            std::option::Option::None => crate::HttpAdmissionDecision::Ready,
        };
    }

    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 record_cooldown_until(&self, now: std::time::Instant, delay: std::time::Duration) {
        let candidate = match now.checked_add(delay) {
            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);
        }
        return;
    }
}

#[derive(Debug)]
struct HttpTokenBucketState {
    available_tokens: f64,
    burst: u32,
    last_refill: std::time::Instant,
    refill_per_second: f64,
}

impl HttpTokenBucketState {
    fn new(requests: u32, window: std::time::Duration, burst: u32, now: std::time::Instant) -> Self {
        let refill_per_second = f64::from(requests) / window.as_secs_f64();
        return Self { available_tokens: f64::from(burst), burst, last_refill: now, refill_per_second };
    }

    fn try_consume_at(&mut self, now: std::time::Instant) -> std::option::Option<std::time::Duration> {
        self.refill_at(now);
        if self.available_tokens >= 1.0 {
            self.available_tokens -= 1.0;
            return std::option::Option::None;
        }
        let missing = 1.0 - self.available_tokens;
        let wait_seconds = missing / self.refill_per_second;
        return std::option::Option::Some(std::time::Duration::from_secs_f64(wait_seconds));
    }

    fn refill_at(&mut self, now: std::time::Instant) {
        if now <= self.last_refill {
            return;
        }
        let elapsed = now.duration_since(self.last_refill).as_secs_f64();
        self.available_tokens = (self.available_tokens + elapsed * self.refill_per_second).min(f64::from(self.burst));
        self.last_refill = now;
        return;
    }
}

fn duration_millis_u64(duration: std::time::Duration) -> u64 {
    return match u64::try_from(duration.as_millis()) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => u64::MAX,
    };
}

#[cfg(test)]
#[path = "../unit_tests/http_admission.rs"]
mod tests;