qrusty 0.25.1

//! 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\"}");
    }

    // --- #151: slow-queue adequacy, the motivation never validated in #146 ---

    #[test]
    fn slow_clean_drain_yields_multiday_eta() {
        // SYS-0032 exists for queues that drain a small fraction per day.
        // Sampled every 30s (reconcile default) at the default tau=120, a
        // *clean* slow drain of 2 msgs/sample (5760/day) from 50_000 must read
        // as Draining with a plausible multi-day ETA — not Unknown/Stalled.
        let e = DrainEstimates::new(DrainConfig::default());
        let dt = 30.0;
        let mut depth = 50_000i64;
        for k in 0..=240 {
            // ~2h of samples: well past the 120s tau, so the EWMA converges.
            e.observe_at("slow", depth.max(0) as u64, k as f64 * dt);
            depth -= 2;
        }
        match e.state("slow") {
            DrainState::Draining { eta_secs } => {
                let days = eta_secs / 86_400.0;
                assert!(
                    (1.0..=30.0).contains(&days),
                    "eta={eta_secs}s ({days} days)"
                );
            }
            other => panic!("clean slow drain should be Draining, got {other:?}"),
        }
    }

    #[test]
    fn slow_drain_under_churn_is_unreliable() {
        // Same clean trend (-2 msgs/sample) but each 30s observation carries
        // +/-40 msgs of arrival/departure churn in the *depth* — a realistic
        // busy-but-slowly-draining queue. A single-tau EWMA can't separate the
        // tiny trend from the churn, so it does NOT stay reliably Draining.
        // This pins *why* the multi-tau horizon ladder was deferred as the
        // follow-up; if that lands and makes this reliable, raise the bound.
        let e = DrainEstimates::new(DrainConfig::default());
        let dt = 30.0;
        let mut depth = 50_000i64;
        let mut seed = 0x1234_5678u64;
        let (mut draining, mut samples) = (0u32, 0u32);
        for k in 0..=240 {
            seed = seed
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            let churn = (seed >> 33) as i64 % 81 - 40; // deterministic [-40, 40]
            let observed = (depth + churn).max(0) as u64;
            e.observe_at("churn", observed, k as f64 * dt);
            if k > 10 {
                samples += 1;
                if matches!(e.state("churn"), DrainState::Draining { .. }) {
                    draining += 1;
                }
            }
            depth -= 2;
        }
        let frac = draining as f64 / samples as f64;
        println!("slow+churn: Draining {draining}/{samples} = {frac:.2} of samples");
        assert!(
            frac < 0.9,
            "single-tau unexpectedly reliable under churn (frac={frac:.2}) — revisit the bound"
        );
    }
}