autumn-web 0.9.0

//! Per-request capability quotas (issue #1632).
//!
//! Fuel is the guest's ceiling. It is not the host's: a `kv-set` costs the guest
//! one call frame and costs the host a cache round-trip; a `job-enqueue` costs
//! it a durable write. A plugin whose fuel budget is generous — which every
//! plugin that renders a page has — could spend all of it on host work priced
//! at nothing.
//!
//! So every capability carries a count, and the counts share one budget:
//!
//! * a per-capability ceiling (`kv_reads`, `outbound_calls`, …) bounds one
//!   surface, and
//! * `calls` bounds the *sum*, so a plugin cannot spend every per-capability
//!   ceiling at once and call that staying within its quota.
//!
//! Exceeding one denies that call and records it. It does not fail the request:
//! see the module header on why a denial is an answer.

// autumn-determinism-gate: production code in this module must read time and
// mint identifiers through the framework's injected seams (ClockSource /
// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
// (issue #1797). Justify exceptions with
// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
#![cfg_attr(not(test), deny(clippy::disallowed_methods))]

use std::sync::Mutex;
use std::time::Instant;

use super::super::grants::CapabilityQuotas;
use super::super::manifest::SandboxCapability;
use super::CapabilityCall;

/// What one request has spent.
///
/// Counters rather than a rate: a request is the unit an operator reasons about
/// ("this plugin may touch the cache 64 times to render its panel"), and it is
/// the unit `max_concurrency` already bounds, so the two multiply into a rate
/// without either having to measure time.
#[derive(Debug, Clone)]
pub struct QuotaLedger {
    declared: CapabilityQuotas,
    kv_reads: u32,
    kv_writes: u32,
    outbound_calls: u32,
    db_reads: u32,
    db_writes: u32,
    job_enqueues: u32,
    calls: u32,
}

impl QuotaLedger {
    /// A fresh ledger for one request.
    #[must_use]
    pub const fn new(declared: CapabilityQuotas) -> Self {
        Self {
            declared,
            kv_reads: 0,
            kv_writes: 0,
            outbound_calls: 0,
            db_reads: 0,
            db_writes: 0,
            job_enqueues: 0,
            calls: 0,
        }
    }

    /// The ceilings this ledger enforces.
    #[must_use]
    pub const fn declared(&self) -> &CapabilityQuotas {
        &self.declared
    }

    /// Charge one dispatch against the shared `calls` budget.
    ///
    /// Charged for *every* call, including one about to be refused: a refusal
    /// costs a grant scan, an encoded reply, a ledger entry and a log line, and
    /// leaving refusals unmetered made the cheapest way to spend the host's time
    /// the one nothing counted. See `CapabilityRuntime::dispatch`.
    ///
    /// # Errors
    ///
    /// Names the quota field that is spent.
    pub const fn charge_call(&mut self) -> Result<(), &'static str> {
        if self.calls >= self.declared.calls {
            return Err("calls");
        }
        self.calls = self.calls.saturating_add(1);
        Ok(())
    }

    /// Charge one call against its per-capability counter.
    ///
    /// Split from [`charge_call`](Self::charge_call) so the two ceilings bound
    /// different things: the shared one bounds *dispatches*, and these bound
    /// *backend work*. A call refused for naming an ungranted host never
    /// reaches a backend, so spending its `outbound_calls` unit would let one
    /// manifest mistake starve the calls the plugin is entitled to make.
    ///
    /// # Errors
    ///
    /// Names the quota field that is spent.
    pub const fn charge_capability(&mut self, call: &CapabilityCall) -> Result<(), &'static str> {
        let (counter, ceiling, field) = match call {
            CapabilityCall::KvGet { .. } => {
                (&mut self.kv_reads, self.declared.kv_reads, "kv_reads")
            }
            CapabilityCall::KvSet { .. } | CapabilityCall::KvDelete { .. } => {
                (&mut self.kv_writes, self.declared.kv_writes, "kv_writes")
            }
            CapabilityCall::HttpFetch { .. } => (
                &mut self.outbound_calls,
                self.declared.outbound_calls,
                "outbound_calls",
            ),
            CapabilityCall::DbGet { .. } | CapabilityCall::DbQuery { .. } => {
                (&mut self.db_reads, self.declared.db_reads, "db_reads")
            }
            CapabilityCall::DbInsert { .. }
            | CapabilityCall::DbUpdate { .. }
            | CapabilityCall::DbDelete { .. } => {
                (&mut self.db_writes, self.declared.db_writes, "db_writes")
            }
            CapabilityCall::JobEnqueue { .. } => (
                &mut self.job_enqueues,
                self.declared.job_enqueues,
                "job_enqueues",
            ),
        };
        if *counter >= ceiling {
            return Err(field);
        }
        *counter = counter.saturating_add(1);
        Ok(())
    }
}

// ── Rate ─────────────────────────────────────────────────────────────────

/// Calls per second, per (plugin, capability), across requests.
///
/// The counters above bound one request. They do not bound a *rate*: a plugin
/// whose panel is fetched a thousand times a second spends a thousand times its
/// per-request budget, and every one of those calls is legitimate as far as the
/// ledger can see. This is the ceiling on the aggregate, keyed the way the
/// framework's tiered rate limiting is keyed for clients — by the thing being
/// limited rather than by the thing asking.
///
/// One bucket per capability rather than one per plugin, so a plugin that is
/// busy on the cache does not lose its ability to enqueue a job. And one
/// limiter per plugin, so exceeding it "denies the call without affecting other
/// plugins or host routes" the way the acceptance criterion asks: no other
/// plugin shares this state, and no host route consults it.
#[derive(Debug)]
pub struct CapabilityRateLimiter {
    /// One bucket per capability, indexed by `SandboxCapability::ALL`'s order.
    buckets: Vec<Mutex<Bucket>>,
    per_second: u32,
}

#[derive(Debug)]
struct Bucket {
    /// Tokens available, scaled by `SCALE` so a refill smaller than one token
    /// is not lost to integer truncation.
    tokens: u64,
    last: Instant,
}

/// Fixed-point scale for the token count.
const SCALE: u64 = 1_000;

impl CapabilityRateLimiter {
    /// A limiter allowing `per_second` calls per capability, per second.
    ///
    /// The bucket starts full, so a plugin's first request is never throttled
    /// by a limiter that has only just been built.
    #[must_use]
    pub fn new(per_second: u32) -> Self {
        let full = u64::from(per_second).saturating_mul(SCALE);
        Self {
            buckets: SandboxCapability::ALL
                .iter()
                .map(|_| {
                    Mutex::new(Bucket {
                        tokens: full,
                        last: crate::time::ambient_instant(),
                    })
                })
                .collect(),
            per_second,
        }
    }

    /// The rate this limiter was built for, in calls per second per capability.
    ///
    /// Exposed so a caller handed someone else's limiter can tell whether it is
    /// looser than the one the manifest asked for; see
    /// `SandboxedPlugin::with_services`.
    #[must_use]
    pub const fn per_second(&self) -> u32 {
        self.per_second
    }

    /// Take one token for `capability`, refilling for elapsed time first.
    ///
    /// Returns `false` when the bucket is empty, which the caller turns into a
    /// [`quota-exceeded`](super::DenialReason::QuotaExceeded) denial.
    #[must_use]
    pub fn try_take(&self, capability: SandboxCapability) -> bool {
        let Some(index) = SandboxCapability::ALL
            .iter()
            .position(|known| *known == capability)
        else {
            // A capability this build does not know cannot have been granted —
            // `SandboxCapability::parse` refuses the name — so this is
            // unreachable through the manifest. Refusing rather than allowing
            // keeps it fail-closed if it ever becomes reachable.
            return false;
        };
        let Some(bucket) = self.buckets.get(index) else {
            return false;
        };
        // A poisoned lock means another thread panicked while holding it. The
        // bucket is a pair of plain numbers with no invariant a panic could
        // break, so the count is taken rather than the request being failed —
        // and `PoisonError::into_inner` is how that is spelled without
        // `unwrap`, which this module's panic gate forbids.
        let mut bucket = bucket
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let full = u64::from(self.per_second).saturating_mul(SCALE);
        let now = crate::time::ambient_instant();
        let micros = u64::try_from(now.saturating_duration_since(bucket.last).as_micros())
            .unwrap_or(u64::MAX);
        // `as_micros` rather than `as_secs_f64`: the refill has to be monotone
        // in elapsed time and identical on every platform, and a float divide
        // is neither.
        let rate = u64::from(self.per_second).saturating_mul(SCALE);
        let refill = micros.saturating_mul(rate) / 1_000_000;
        bucket.tokens = bucket.tokens.saturating_add(refill).min(full);
        // Advance the clock only by the time that actually became tokens, and
        // carry the remainder. Assigning `now` unconditionally *discards* every
        // interval shorter than one token's worth — 5 µs at the default 200/s —
        // so a plugin whose calls arrive faster than that quantum refills
        // strictly slower than its declared rate, and under contention could sit
        // at zero tokens while time passed. Clamped at `micros` so the clock can
        // only ever move forward to `now` at most.
        let converted = refill
            .saturating_mul(1_000_000)
            .checked_div(rate)
            .unwrap_or(micros)
            .min(micros);
        bucket.last = bucket
            .last
            .checked_add(std::time::Duration::from_micros(converted))
            .unwrap_or(now);
        if bucket.tokens < SCALE {
            return false;
        }
        bucket.tokens = bucket.tokens.saturating_sub(SCALE);
        true
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use chrono::{TimeZone, Utc};

    use super::{CapabilityRateLimiter, SandboxCapability};
    use crate::time::{TickingClock, install_ambient};

    #[test]
    fn a_bucket_from_a_nested_timeline_refills_on_the_outer_one() {
        let epoch = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let capability = SandboxCapability::ALL[0];
        let outer = TickingClock::starting_at(epoch);
        let _outer = install_ambient(Arc::new(outer.clone()));
        let limiter = CapabilityRateLimiter::new(1);

        // A nested timeline drains the bucket an hour in.
        let inner = TickingClock::starting_at(epoch);
        let guard = install_ambient(Arc::new(inner.clone()));
        inner.advance(Duration::from_secs(3600));
        assert!(limiter.try_take(capability));
        assert!(!limiter.try_take(capability), "drained");
        drop(guard);

        // The outer timeline goes on from the nested one's, so no time has
        // passed. One second here refills one token.
        assert!(!limiter.try_take(capability), "no time has passed yet");
        outer.advance(Duration::from_secs(1));
        assert!(limiter.try_take(capability), "one second refills one token");
    }
}