qrusty 0.25.0

//! Per-queue drain-time estimate (SYS-0032).
//!
//! Estimates how long a queue will take to drain at the recent rate, from the
//! net rate of change of its depth (`departures − arrivals`) smoothed by a
//! time-decayed EWMA. O(1) state per queue, no stored history. See
//! `docs/drain-eta.md` for the rationale (including why this is separate from
//! the short-horizon `QueueRateTracker`).

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

use serde::{Deserialize, Serialize};

/// Tuning for the drain estimator. Env-sourced, alongside `ReconcileConfig`.
#[derive(Debug, Clone, Copy)]
pub struct DrainConfig {
    /// EWMA time constant (seconds): larger = smoother/slower to react.
    pub tau_secs: f64,
    /// Samples required before any state other than `Unknown` is reported.
    pub min_samples: u32,
    /// |net rate| (msgs/sec) at or below which the queue is `Stalled`.
    pub noise_floor: f64,
}

impl Default for DrainConfig {
    fn default() -> Self {
        // tau defaults to a few reconcile periods (reconcile default = 30s).
        Self {
            tau_secs: 120.0,
            min_samples: 3,
            noise_floor: 0.01,
        }
    }
}

impl DrainConfig {
    /// Build from env; invalid or non-positive values fall back to the default.
    /// - `QRUSTY_DRAIN_TAU_SECS`
    /// - `QRUSTY_DRAIN_MIN_SAMPLES`
    /// - `QRUSTY_DRAIN_NOISE_FLOOR`
    pub fn from_env() -> Self {
        let d = Self::default();
        let tau_secs = std::env::var("QRUSTY_DRAIN_TAU_SECS")
            .ok()
            .and_then(|v| v.parse::<f64>().ok())
            .filter(|&v| v.is_finite() && v > 0.0)
            .unwrap_or(d.tau_secs);
        let min_samples = std::env::var("QRUSTY_DRAIN_MIN_SAMPLES")
            .ok()
            .and_then(|v| v.parse::<u32>().ok())
            .filter(|&v| v > 0)
            .unwrap_or(d.min_samples);
        let noise_floor = std::env::var("QRUSTY_DRAIN_NOISE_FLOOR")
            .ok()
            .and_then(|v| v.parse::<f64>().ok())
            .filter(|&v| v.is_finite() && v >= 0.0)
            .unwrap_or(d.noise_floor);
        Self {
            tau_secs,
            min_samples,
            noise_floor,
        }
    }
}

/// Reported drain state for a queue (serialized into `QueueStats`).
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum DrainState {
    /// Depth is decreasing; `eta_secs` is the projected time to empty.
    Draining { eta_secs: f64 },
    /// Arrivals meet or exceed departures — not draining; no ETA.
    Growing,
    /// Activity below the noise floor (idle / no consumers); no ETA.
    Stalled,
    /// Not enough samples yet to estimate.
    Unknown,
}

/// Per-queue EWMA state. `rate` is the smoothed net drain rate in msgs/sec,
/// positive when the queue is shrinking.
#[derive(Debug, Clone, Copy)]
struct QueueDrain {
    rate: f64,
    last_depth: u64,
    last_ts: f64,
    samples: u32,
}

impl QueueDrain {
    fn new(depth: u64, now_secs: f64) -> Self {
        // First sample only establishes a baseline; no rate yet.
        Self {
            rate: 0.0,
            last_depth: depth,
            last_ts: now_secs,
            samples: 1,
        }
    }

    /// Fold a new `(depth, now_secs)` observation into the EWMA. A non-positive
    /// `Δt` (duplicate or backwards clock) is ignored so the rate can't blow up.
    fn observe(&mut self, depth: u64, now_secs: f64, cfg: &DrainConfig) {
        let dt = now_secs - self.last_ts;
        if dt <= 0.0 {
            // Keep the newest depth but don't derive a rate from zero elapsed.
            self.last_depth = depth;
            return;
        }
        let inst = (self.last_depth as f64 - depth as f64) / dt; // + = draining
        let alpha = 1.0 - (-dt / cfg.tau_secs).exp();
        self.rate = alpha * inst + (1.0 - alpha) * self.rate;
        self.last_depth = depth;
        self.last_ts = now_secs;
        self.samples = self.samples.saturating_add(1);
    }

    fn classify(&self, cfg: &DrainConfig) -> DrainState {
        if self.samples < cfg.min_samples {
            return DrainState::Unknown;
        }
        if self.rate > cfg.noise_floor {
            // Draining. Current depth is the last observed depth.
            let eta = self.last_depth as f64 / self.rate;
            DrainState::Draining { eta_secs: eta }
        } else if self.rate < -cfg.noise_floor {
            DrainState::Growing
        } else {
            DrainState::Stalled
        }
    }
}

/// Shared, concurrent per-queue drain estimates. The reconcile sweep calls
/// [`DrainEstimates::observe`]; the stats handlers call [`DrainEstimates::state`].
#[derive(Debug)]
pub struct DrainEstimates {
    inner: Mutex<HashMap<String, QueueDrain>>,
    cfg: DrainConfig,
    base: Instant,
}

impl DrainEstimates {
    pub fn new(cfg: DrainConfig) -> Self {
        Self {
            inner: Mutex::new(HashMap::new()),
            cfg,
            base: Instant::now(),
        }
    }

    fn now_secs(&self) -> f64 {
        self.base.elapsed().as_secs_f64()
    }

    /// Record a fresh depth sample for `queue` using the process monotonic
    /// clock. Called from the reconcile sweep.
    pub fn observe(&self, queue: &str, depth: u64) {
        let now = self.now_secs();
        self.observe_at(queue, depth, now);
    }

    /// Clock-injected form for deterministic tests.
    fn observe_at(&self, queue: &str, depth: u64, now_secs: f64) {
        let mut map = self.inner.lock().expect("drain estimates mutex poisoned");
        map.entry(queue.to_string())
            .and_modify(|q| q.observe(depth, now_secs, &self.cfg))
            .or_insert_with(|| QueueDrain::new(depth, now_secs));
    }

    /// Current drain state for `queue`, or `Unknown` if never observed.
    pub fn state(&self, queue: &str) -> DrainState {
        let map = self.inner.lock().expect("drain estimates mutex poisoned");
        map.get(queue)
            .map(|q| q.classify(&self.cfg))
            .unwrap_or(DrainState::Unknown)
    }

    /// Drop state for queues that no longer exist, so the map can't grow
    /// without bound as queues come and go.
    pub fn retain_queues(&self, live: &[String]) {
        let mut map = self.inner.lock().expect("drain estimates mutex poisoned");
        map.retain(|k, _| live.iter().any(|l| l == k));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn cfg() -> DrainConfig {
        DrainConfig {
            tau_secs: 60.0,
            min_samples: 3,
            noise_floor: 0.01,
        }
    }

    fn est() -> DrainEstimates {
        DrainEstimates::new(cfg())
    }

    #[test]
    fn unknown_until_min_samples() {
        let e = est();
        e.observe_at("q", 100, 0.0);
        assert_eq!(e.state("q"), DrainState::Unknown); // 1 sample
        e.observe_at("q", 90, 1.0);
        assert_eq!(e.state("q"), DrainState::Unknown); // 2 samples
        e.observe_at("q", 80, 2.0);
        assert!(!matches!(e.state("q"), DrainState::Unknown)); // 3 samples
    }

    #[test]
    fn unknown_for_never_observed_queue() {
        assert_eq!(est().state("nope"), DrainState::Unknown);
    }

    #[test]
    fn steady_drain_reports_eta() {
        let e = est();
        // Drain 10 msgs/sec, sampled once a second from 100 down.
        let mut depth = 100i64;
        for t in 0..=10 {
            e.observe_at("q", depth.max(0) as u64, t as f64);
            depth -= 10;
        }
        match e.state("q") {
            DrainState::Draining { eta_secs } => {
                // rate ~10/sec; remaining depth small near the end, eta finite & >= 0.
                assert!(eta_secs >= 0.0 && eta_secs.is_finite(), "eta={eta_secs}");
            }
            other => panic!("expected Draining, got {other:?}"),
        }
    }

    #[test]
    fn eta_tracks_depth_over_rate() {
        // A time-decayed EWMA converges to the true rate over ~several τ, so to
        // check `eta ≈ depth / rate` we must observe for well beyond τ. Steady
        // 5 msgs/sec drain, τ=5s, sampled 1/sec for 40s (8τ): rate → 5, last
        // depth = 1000 − 5·40 = 800, so eta → 800/5 = 160s.
        let e = DrainEstimates::new(DrainConfig {
            tau_secs: 5.0,
            min_samples: 3,
            noise_floor: 0.01,
        });
        for t in 0..=40 {
            let depth = 1000 - 5 * t;
            e.observe_at("q", depth as u64, t as f64);
        }
        match e.state("q") {
            DrainState::Draining { eta_secs } => {
                assert!(
                    (eta_secs - 160.0).abs() < 20.0,
                    "eta={eta_secs}, expected ~160s (800 / 5)"
                );
            }
            other => panic!("expected Draining, got {other:?}"),
        }
    }

    #[test]
    fn growing_queue_reports_growing_never_eta() {
        let e = est();
        let mut depth = 10u64;
        for t in 0..=10 {
            e.observe_at("q", depth, t as f64);
            depth += 10; // arrivals outpace departures
        }
        assert_eq!(e.state("q"), DrainState::Growing);
    }

    #[test]
    fn idle_queue_reports_stalled() {
        let e = est();
        for t in 0..=10 {
            e.observe_at("q", 50, t as f64); // depth never changes
        }
        assert_eq!(e.state("q"), DrainState::Stalled);
    }

    #[test]
    fn zero_dt_is_ignored_no_nan() {
        let e = est();
        e.observe_at("q", 100, 5.0);
        e.observe_at("q", 90, 5.0); // same timestamp => dt=0, must not blow up
        e.observe_at("q", 80, 6.0);
        e.observe_at("q", 70, 7.0);
        match e.state("q") {
            DrainState::Draining { eta_secs } => assert!(eta_secs.is_finite()),
            DrainState::Stalled => {} // acceptable: few effective samples
            other => panic!("unexpected {other:?}"),
        }
    }

    #[test]
    fn irregular_dt_handled() {
        let e = est();
        // Wildly uneven spacing; must stay finite and classify as draining.
        for (t, d) in [
            (0.0, 1000),
            (0.5, 990),
            (30.0, 700),
            (31.0, 695),
            (90.0, 400),
        ] {
            e.observe_at("q", d, t);
        }
        match e.state("q") {
            DrainState::Draining { eta_secs } => assert!(eta_secs.is_finite() && eta_secs > 0.0),
            other => panic!("expected Draining, got {other:?}"),
        }
    }

    #[test]
    fn retain_drops_dead_queues() {
        let e = est();
        e.observe_at("keep", 10, 0.0);
        e.observe_at("drop", 10, 0.0);
        e.retain_queues(&["keep".to_string()]);
        // "drop" is gone => Unknown; "keep" still tracked.
        assert_eq!(e.state("drop"), DrainState::Unknown);
        e.observe_at("keep", 5, 1.0);
        e.observe_at("keep", 1, 2.0);
        assert!(!matches!(e.state("keep"), DrainState::Unknown));
    }

    #[test]
    fn drain_state_serializes_tagged() {
        let j = serde_json::to_string(&DrainState::Draining { eta_secs: 12.5 }).unwrap();
        assert!(j.contains("\"state\":\"draining\""), "{j}");
        assert!(j.contains("\"eta_secs\":12.5"), "{j}");
        let g = serde_json::to_string(&DrainState::Growing).unwrap();
        assert_eq!(g, "{\"state\":\"growing\"}");
    }
}