v0.2.11-pre.003

2026-08-25 21:00:19 +02:00
parent 2dbddfe367
commit be2a06bc80
18 changed files with 1415 additions and 45 deletions
--- a/crates/ksp-offchain-transport-lib/src/http_admission.rs
+++ b/crates/ksp-offchain-transport-lib/src/http_admission.rs
@@ -0,0 +1,218 @@
+// file: crates/ksp-offchain-transport-lib/src/http_admission.rs
+// version: 1
+
+//! 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 },
+    /// No local cadence limit is configured; provider-driven cooldown still applies.
+    Unlimited,
+}
+
+impl 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 || burst > requests {
+            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,
+    policy: HttpAdmissionPolicy,
+    token_bucket: std::sync::Mutex<std::option::Option<HttpTokenBucketState>>,
+}
+
+impl HttpAdmissionController {
+    /// Creates one limiter from a provider-owned local admission policy.
+    pub(crate) fn new(policy: 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 > 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 {
+            HttpAdmissionPolicy::Fixed { requests, window, burst } => {
+                std::option::Option::Some(HttpTokenBucketState::new(requests, window, burst, std::time::Instant::now()))
+            },
+            HttpAdmissionPolicy::Dynamic | HttpAdmissionPolicy::Unlimited => std::option::Option::None,
+        };
+        return std::result::Result::Ok(Self {
+            cooldown_until: std::sync::Mutex::new(std::option::Option::None),
+            fallback_cooldown,
+            policy,
+            token_bucket: std::sync::Mutex::new(token_bucket),
+        });
+    }
+
+    /// Returns the configured local policy.
+    #[must_use]
+    pub(crate) const fn policy(&self) -> HttpAdmissionPolicy {
+        return self.policy;
+    }
+
+    /// Tries to admit one request immediately without sleeping.
+    pub(crate) fn try_admit(&self) -> 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, 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, if any.
+    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) -> HttpAdmissionDecision {
+        if let std::option::Option::Some(remaining) = self.cooldown_remaining_at(now) {
+            return 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 HttpAdmissionDecision::Ready,
+        };
+        return match state.try_consume_at(now) {
+            std::option::Option::Some(delay) => HttpAdmissionDecision::Deferred(delay),
+            std::option::Option::None => 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;