av_state/

velocity.rs

//! Token-velocity tracking: sliding-window tokens-per-second intended for
//! rate limiting. (The loop-breaker's `N+ tokens` arm uses its own
//! cumulative per-session counter, not this window.)
//!
//! # NOT WIRED INTO ENFORCEMENT
//!
//! There are currently zero production callers of [`TokenVelocity`]; the
//! only users outside this crate are the integration tests in
//! `crates/av-state/tests/e2e_race_velocity.rs`. The type predates the
//! current admission / loop-breaker plumbing and was never connected to
//! either the breaker's `min_tokens` gate (in `av_loopdetect`) or any
//! harness rate-limit path.
//!
//! **Do not wire this into the breaker without a shared per-session
//! lock.** The velocity `record_at` mutation and the breaker's `observe`
//! mutation live under different [`parking_lot::Mutex`] instances, so a
//! blind wire-up races the two counters. If you need per-session
//! velocity in the breaker, extend `av_loopdetect::Breaker` with its own
//! velocity window guarded by the same mutex the breaker already holds.
//!
//! The type is kept public for now (rather than deleted) because the
//! sliding-window arithmetic is exercised by the
//! `e2e_race_velocity` adversarial tests that lock in the
//! `saturating_add` discipline — reusable ground for the
//! future breaker-integrated window when someone builds it.

use parking_lot::Mutex;
use std::collections::VecDeque;

/// Sliding-window token counter.
#[derive(Debug)]
pub struct TokenVelocity {
    window_ms: u64,
    samples: Mutex<VecDeque<(u64, u64)>>, // (timestamp_ms, tokens)
}

impl TokenVelocity {
    /// Create a tracker with the given window size in milliseconds.
    pub fn new(window_ms: u64) -> Self {
        Self {
            window_ms: window_ms.max(1),
            samples: Mutex::new(VecDeque::new()),
        }
    }

    /// Record `tokens` at time `now_ms` and return the windowed total.
    pub fn record_at(&self, now_ms: u64, tokens: u64) -> u64 {
        let mut samples = self.samples.lock();
        samples.push_back((now_ms, tokens));
        let cutoff = now_ms.saturating_sub(self.window_ms);
        // R108 F4: filter across the WHOLE deque (retain), not just
        // the front. Prior shape assumed monotonically-increasing
        // `now_ms` and only evicted via `while front < cutoff`.
        // Under NTP slew / VM live-migration a caller can produce
        // a strictly-earlier now_ms than a previously recorded
        // sample, leaving that older sample stranded mid-deque —
        // subsequent evictions stop as soon as they hit a
        // "future" front, so the stranded sample keeps
        // contributing to the windowed sum long after its own
        // timestamp fell out of the window. Realistically only
        // visible under clock skew, but the crate is `pub` and
        // the module header explicitly frames the sliding-window
        // arithmetic as "reusable ground for the future breaker-
        // integrated window when someone builds it" — future
        // wire-up would inherit the bug. `retain` is O(n) but the
        // deque is small by construction (few samples per window)
        // and this is not on any hot path today.
        samples.retain(|(t, _)| *t >= cutoff);
        samples.iter().fold(0u64, |acc, (_, n)| acc.saturating_add(*n))
    }

    /// Record at the current wall clock.
    pub fn record(&self, tokens: u64) -> u64 {
        self.record_at(av_core::time::now_ms(), tokens)
    }

    /// Windowed total without recording.
    pub fn current_at(&self, now_ms: u64) -> u64 {
        let samples = self.samples.lock();
        let cutoff = now_ms.saturating_sub(self.window_ms);
        // Mirror record_at's saturating_add discipline.
        samples
            .iter()
            .filter(|(t, _)| *t >= cutoff)
            .fold(0u64, |acc, (_, n)| acc.saturating_add(*n))
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

    use super::*;

    #[test]
    fn accumulates_within_window() {
        let v = TokenVelocity::new(1000);
        assert_eq!(v.record_at(0, 100), 100);
        assert_eq!(v.record_at(500, 50), 150);
        assert_eq!(v.record_at(999, 25), 175);
    }

    #[test]
    fn old_samples_expire() {
        let v = TokenVelocity::new(1000);
        v.record_at(0, 100);
        v.record_at(500, 50);
        // t=1400 → cutoff 400: only the t=0 sample expires.
        assert_eq!(v.record_at(1400, 10), 60);
        // t=1600 → cutoff 600: the t=500 sample expires too.
        assert_eq!(v.current_at(1600), 10);
        assert_eq!(v.current_at(3000), 0, "everything expires eventually");
    }

    #[test]
    fn clock_going_backwards_does_not_panic() {
        let v = TokenVelocity::new(1000);
        v.record_at(5000, 10);
        // Skewed clock: earlier timestamp after a later one.
        let total = v.record_at(4000, 5);
        assert!(total >= 5, "must count at least the new sample, got {total}");
    }

    #[test]
    fn record_returns_the_windowed_total_including_the_new_sample() {
        // The wall-clock `record` wrapper must not be reducible to a
        // constant (mutant misses on `record -> 0 / 1`).
        let v = TokenVelocity::new(60_000);
        let first = v.record(7);
        assert_eq!(first, 7, "first record must return the value just written");
        let second = v.record(3);
        assert_eq!(
            second, 10,
            "second record must reflect both samples inside the window"
        );
    }

    /// `Iterator::sum` on `u64` panics in debug and
    /// wraps in release on overflow. Every other counter/spend
    /// site in av_state uses checked/saturating arithmetic;
    /// velocity now does too. Two u64::MAX samples inside the
    /// window must return u64::MAX (saturated), not panic and not
    /// wrap to a small number.
    #[test]
    fn windowed_sum_saturates_instead_of_wrapping_or_panicking() {
        let v = TokenVelocity::new(60_000);
        v.record_at(1_000, u64::MAX);
        let total = v.record_at(2_000, u64::MAX);
        assert_eq!(
            total,
            u64::MAX,
            "windowed sum must saturate at u64::MAX, got {total}"
        );
        // current_at path also.
        let peek = v.current_at(3_000);
        assert_eq!(peek, u64::MAX, "current_at must saturate at u64::MAX, got {peek}");
    }
}