qrusty 0.25.1

// src/counter_reconcile.rs

//! Background counter-drift self-heal (issue #142, SYS-0031).
//!
//! The per-queue `QueueCounters` cache (SYS-0018) is maintained by ~130
//! independent delta sites across the storage backends.  There is no single
//! chokepoint, so any one state-changing path that forgets its delta drifts the
//! reported depth *permanently* — the symptom operators see as "messages that
//! never drain."  SYS-0018 item 7 added a drift detector (`diagnose_queue`) and
//! an operator-manual repair (`reseed_queue_counters`); this task closes the
//! loop by running that detect→repair cycle automatically, on a rotating
//! budget, so a wrong depth can never persist regardless of which site slipped.
//!
//! The root-cause sweep of the delta sites (A2) removes the actual holes; this
//! sweeper is the net that catches whatever the audit misses or a future edit
//! reintroduces.
//!
//! # Rotating budget
//!
//! Each tick diagnoses at most `batch` queues, advancing a cursor through the
//! (sorted) queue list so every queue is visited once per
//! `ceil(n / batch)` ticks.  This bounds the per-tick scan cost: a full
//! `diagnose_queue` is a queue-prefix scan, and SYS-0018 item 2 reserves the
//! only *full-store* scan for startup.  We never scan more than `batch` queues
//! per tick, so the sweep cost stays flat as queue count grows.
//
// ponytail: O(batch) prefix scans per tick — fine for the self-heal net.  If a
// queue is enormous and diagnosis itself becomes costly, add a per-queue scan
// budget (chunked diagnose) rather than widening this loop.

use crate::api::StorageApi;
use crate::drain_estimate::DrainEstimates;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

/// Default reconcile period: diagnose a batch of queues every 30s.
const DEFAULT_PERIOD_SECS: u64 = 30;
/// Default queues diagnosed per tick.
const DEFAULT_BATCH: usize = 64;

/// Configuration for the counter-reconcile sweep, read once from the
/// environment at startup.
#[derive(Debug, Clone, Copy)]
pub struct ReconcileConfig {
    /// When false, the sweep task is not spawned at all.
    pub enabled: bool,
    /// Interval between sweep ticks.
    pub period: Duration,
    /// Maximum number of queues diagnosed per tick.
    pub batch: usize,
}

impl Default for ReconcileConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            period: Duration::from_secs(DEFAULT_PERIOD_SECS),
            batch: DEFAULT_BATCH,
        }
    }
}

impl ReconcileConfig {
    /// Build from env:
    /// - `QRUSTY_COUNTER_RECONCILE_ENABLED` — `0`/`false` disables (default on).
    /// - `QRUSTY_COUNTER_RECONCILE_SECS`    — tick period in seconds (default 30).
    /// - `QRUSTY_COUNTER_RECONCILE_BATCH`   — queues per tick (default 64).
    ///
    /// Invalid or zero values fall back to the default, and the period is
    /// floored at 1s so a typo can't busy-loop the task.
    pub fn from_env() -> Self {
        let d = Self::default();
        let enabled = match std::env::var("QRUSTY_COUNTER_RECONCILE_ENABLED") {
            Ok(v) => !matches!(v.trim().to_ascii_lowercase().as_str(), "0" | "false" | "no"),
            Err(_) => d.enabled,
        };
        let period = std::env::var("QRUSTY_COUNTER_RECONCILE_SECS")
            .ok()
            .and_then(|v| v.parse::<u64>().ok())
            .filter(|&s| s > 0)
            .map(Duration::from_secs)
            .unwrap_or(d.period);
        let batch = std::env::var("QRUSTY_COUNTER_RECONCILE_BATCH")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .filter(|&b| b > 0)
            .unwrap_or(d.batch);
        Self {
            enabled,
            period,
            batch,
        }
    }
}

/// Pure rotating-budget selector: given `total` queues and the current
/// `cursor`, return the indices to diagnose this tick (at most `batch`,
/// distinct, wrapping) and the next cursor.
///
/// Wrapping means a `batch` larger than `total` collapses to the whole list
/// once (no duplicate indices in a single tick).
fn select_batch(total: usize, cursor: usize, batch: usize) -> (Vec<usize>, usize) {
    if total == 0 || batch == 0 {
        return (Vec::new(), 0);
    }
    let take = batch.min(total);
    let start = cursor % total;
    let indices: Vec<usize> = (0..take).map(|i| (start + i) % total).collect();
    let next = (start + take) % total;
    (indices, next)
}

/// Run the counter-drift self-heal loop forever.
///
/// On each tick: list queues, diagnose the next `batch` of them, and for any
/// whose cache disagrees with a fresh scan, reseed the counters, log a warning
/// with before/after, and increment `repairs`.
pub async fn counter_reconciler(
    storage: Arc<dyn StorageApi>,
    repairs: Arc<AtomicU64>,
    estimates: Arc<DrainEstimates>,
    cfg: ReconcileConfig,
) {
    let mut interval = tokio::time::interval(cfg.period);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    let mut cursor: usize = 0;
    loop {
        interval.tick().await;
        if let Err(e) = sweep_once(
            storage.as_ref(),
            &repairs,
            &estimates,
            cfg.batch,
            &mut cursor,
        )
        .await
        {
            // A transient storage error must not kill the self-heal task.
            tracing::error!("counter_reconciler: sweep failed: {}", e);
        }
    }
}

/// One sweep tick, factored out so tests can drive it deterministically.
/// Returns the number of queues repaired this tick.
async fn sweep_once(
    storage: &dyn StorageApi,
    repairs: &AtomicU64,
    estimates: &DrainEstimates,
    batch: usize,
    cursor: &mut usize,
) -> anyhow::Result<u64> {
    let queues = storage.list_queues().await?;
    // Drop drain state for queues that no longer exist so the map stays bounded.
    estimates.retain_queues(&queues);
    let (indices, next) = select_batch(queues.len(), *cursor, batch);
    *cursor = next;

    let mut repaired = 0u64;
    for i in indices {
        let name = &queues[i];
        let diag = match storage.diagnose_queue(name).await? {
            Some(d) => d,
            None => continue, // backend doesn't support diagnostics
        };
        // SYS-0032: feed the fresh scanned depth into the drain estimator on
        // every visit (not just on drift), so the EWMA sees a regular sample.
        let depth =
            diag.scanned_available + diag.scanned_locked_active + diag.scanned_locked_expired;
        estimates.observe(name, depth);
        // Structural drift check against the SYS-0018 counter model (#142):
        // available == unlocked messages; locked == active + expired-unswept
        // locks.  We deliberately do NOT trigger on `diag.discrepancies`
        // emptiness alone — an expired-but-unswept lock is a legitimate
        // transient the normal unlock-expired sweep clears within seconds, and
        // reseeding it here would fight that sweep.  Only persistent counter
        // drift (e.g. a missed ack delta) is repaired.
        let expected_available = diag.scanned_available;
        let expected_locked = diag.scanned_locked_active + diag.scanned_locked_expired;
        if diag.counter_available == expected_available && diag.counter_locked == expected_locked {
            continue;
        }
        // Drift found — reseed from a fresh scan.
        let before = (diag.counter_available, diag.counter_locked);
        match storage.reseed_queue_counters(name).await? {
            Some(after) => {
                repaired += 1;
                repairs.fetch_add(1, Ordering::Relaxed);
                tracing::warn!(
                    event_type = "counter_drift_repaired",
                    queue = %name,
                    before_available = before.0,
                    before_locked = before.1,
                    after_available = after.0,
                    after_locked = after.1,
                    discrepancies = ?diag.discrepancies,
                    "counter drift detected and repaired by reseed",
                );
            }
            None => {
                tracing::warn!(
                    queue = %name,
                    discrepancies = ?diag.discrepancies,
                    "counter drift detected but backend does not support reseed",
                );
            }
        }
    }
    Ok(repaired)
}

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

    /// A throwaway drain-estimate store for sweep tests that don't assert on it.
    fn test_estimates() -> DrainEstimates {
        DrainEstimates::new(crate::drain_estimate::DrainConfig::default())
    }
    use crate::memory_storage::MemoryStorage;
    use crate::message::{Message, Priority};
    use chrono::Utc;

    #[test]
    fn select_batch_wraps_and_advances() {
        // Normal step.
        assert_eq!(select_batch(10, 0, 3), (vec![0, 1, 2], 3));
        assert_eq!(select_batch(10, 3, 3), (vec![3, 4, 5], 6));
        // Wrap across the end.
        assert_eq!(select_batch(10, 8, 3), (vec![8, 9, 0], 1));
    }

    #[test]
    fn select_batch_batch_ge_total_takes_all_once() {
        let (idx, next) = select_batch(4, 2, 10);
        assert_eq!(idx, vec![2, 3, 0, 1]); // distinct, whole list, starting at cursor
        assert_eq!(next, 2);
    }

    #[test]
    fn select_batch_empty_and_zero() {
        assert_eq!(select_batch(0, 5, 4), (Vec::new(), 0));
        assert_eq!(select_batch(5, 0, 0), (Vec::new(), 0));
    }

    fn msg(queue: &str, payload: &str) -> Message {
        Message {
            id: uuid::Uuid::new_v4().to_string(),
            queue: queue.to_string(),
            payload: payload.to_string(),
            priority: Priority::Numeric(100),
            created_at: Utc::now(),
            locked_until: None,
            locked_by: None,
            retry_count: 0,
            max_retries: 3,
            payload_ref: None,
            payload_hash: None,
        }
    }

    // B3: injected drift self-heals in a single sweep tick, and the repair
    // metric is bumped exactly once.
    #[tokio::test]
    async fn sweep_repairs_injected_drift() {
        let storage = MemoryStorage::new();
        storage.push(msg("q", "a")).await.unwrap();
        storage.push(msg("q", "b")).await.unwrap();
        let s = storage.get_queue_stats("q").await.unwrap();
        assert_eq!((s.available, s.locked), (2, 0));

        // Corrupt the cache behind the storage layer's back.
        storage.__test_set_queue_counters("q", 99, 7).await;
        let s = storage.get_queue_stats("q").await.unwrap();
        assert_eq!((s.available, s.locked), (99, 7), "drift injected");

        let repairs = AtomicU64::new(0);
        let mut cursor = 0usize;
        let n = sweep_once(&storage, &repairs, &test_estimates(), 64, &mut cursor)
            .await
            .unwrap();

        assert_eq!(n, 1, "exactly one queue repaired");
        assert_eq!(repairs.load(Ordering::Relaxed), 1);
        let s = storage.get_queue_stats("q").await.unwrap();
        assert_eq!(
            (s.available, s.locked),
            (2, 0),
            "counters self-healed to the true population"
        );
    }

    // #142 regression: an expired-but-unswept lock is a legitimate transient,
    // NOT drift.  The sweep must leave the counters alone (and not fight the
    // normal unlock-expired sweep) — the earlier `discrepancies`-based trigger
    // would have wrongly "repaired" here.
    #[tokio::test]
    async fn sweep_ignores_expired_unswept_lock() {
        let storage = MemoryStorage::new();
        storage.push(msg("q", "a")).await.unwrap();
        // Lock with a 0s timeout, then let it expire without running the
        // unlock-expired sweep.  Counter is (available=0, locked=1) per
        // SYS-0018; the scan sees the lock as expired-unswept.
        let _ = storage.pop("q", "c1", 0).await.unwrap().unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        let s = storage.get_queue_stats("q").await.unwrap();
        assert_eq!(
            (s.available, s.locked),
            (0, 1),
            "counters correct per SYS-0018"
        );

        let repairs = AtomicU64::new(0);
        let mut cursor = 0usize;
        let n = sweep_once(&storage, &repairs, &test_estimates(), 64, &mut cursor)
            .await
            .unwrap();

        assert_eq!(n, 0, "expired-unswept lock is not drift — no repair");
        assert_eq!(repairs.load(Ordering::Relaxed), 0);
        let s = storage.get_queue_stats("q").await.unwrap();
        assert_eq!(
            (s.available, s.locked),
            (0, 1),
            "counters untouched by sweep"
        );
    }

    // A clean queue is not counted as a repair.
    #[tokio::test]
    async fn sweep_noop_when_accurate() {
        let storage = MemoryStorage::new();
        storage.push(msg("q", "a")).await.unwrap();
        let repairs = AtomicU64::new(0);
        let mut cursor = 0usize;
        let n = sweep_once(&storage, &repairs, &test_estimates(), 64, &mut cursor)
            .await
            .unwrap();
        assert_eq!(n, 0);
        assert_eq!(repairs.load(Ordering::Relaxed), 0);
    }

    // --- SYS-0032 drain-estimator wiring (#151) ---------------------------
    // #146 shipped the estimator with pure unit tests only; nothing exercised
    // the real `sweep_once -> observe -> state` path against a live backend.
    // These drive a real MemoryStorage so the depth fed to the estimator is a
    // genuine scan (available + locked), not a synthetic number. Depth moves
    // monotonically, so the classification is fixed by the *sign* of the net
    // rate and is robust to sleep jitter; a small `tau` keeps the EWMA reactive
    // across closely-spaced samples.
    use crate::drain_estimate::{DrainConfig, DrainState};

    fn reactive_estimates() -> DrainEstimates {
        DrainEstimates::new(DrainConfig {
            tau_secs: 0.05,
            min_samples: 3,
            noise_floor: 0.01,
        })
    }

    /// Pop then ack `n` messages so queue depth actually drops (a bare pop only
    /// locks — the message still counts toward depth).
    async fn drain_n(storage: &MemoryStorage, queue: &str, n: usize) {
        for _ in 0..n {
            if let Some(m) = storage.pop(queue, "drain-test", 30).await.unwrap() {
                storage.ack(queue, &m.id, "drain-test").await.unwrap();
            }
        }
    }

    /// Sweep, then wait so the next sample has a strictly positive Δt.
    async fn sweep_tick(storage: &MemoryStorage, est: &DrainEstimates, cursor: &mut usize) {
        let repairs = AtomicU64::new(0);
        sweep_once(storage, &repairs, est, 64, cursor)
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }

    #[tokio::test]
    async fn sweep_feeds_estimator_draining() {
        let storage = MemoryStorage::new();
        for i in 0..30 {
            storage.push(msg("q", &format!("m{i}"))).await.unwrap();
        }
        let est = reactive_estimates();
        let mut cursor = 0usize;
        // Depths observed: 30, 25, 20, 15, 10, 5 — strictly falling.
        for _ in 0..5 {
            sweep_tick(&storage, &est, &mut cursor).await;
            drain_n(&storage, "q", 5).await;
        }
        sweep_tick(&storage, &est, &mut cursor).await;
        match est.state("q") {
            DrainState::Draining { eta_secs } => {
                assert!(eta_secs.is_finite() && eta_secs >= 0.0, "eta={eta_secs}")
            }
            other => panic!("falling queue should classify Draining, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn sweep_feeds_estimator_growing() {
        let storage = MemoryStorage::new();
        storage.push(msg("q", "seed")).await.unwrap();
        let est = reactive_estimates();
        let mut cursor = 0usize;
        // Depths observed: 1, 6, 11, 16, 21 — strictly rising.
        for i in 0..4 {
            sweep_tick(&storage, &est, &mut cursor).await;
            for j in 0..5 {
                storage.push(msg("q", &format!("g{i}-{j}"))).await.unwrap();
            }
        }
        sweep_tick(&storage, &est, &mut cursor).await;
        assert_eq!(
            est.state("q"),
            DrainState::Growing,
            "a queue whose depth keeps rising must report Growing, never an ETA"
        );
    }

    #[tokio::test]
    async fn sweep_feeds_estimator_stalled() {
        let storage = MemoryStorage::new();
        for i in 0..20 {
            storage.push(msg("q", &format!("s{i}"))).await.unwrap();
        }
        let est = reactive_estimates();
        let mut cursor = 0usize;
        // Depth never changes across samples => net rate 0 => Stalled (idle).
        for _ in 0..5 {
            sweep_tick(&storage, &est, &mut cursor).await;
        }
        assert_eq!(
            est.state("q"),
            DrainState::Stalled,
            "an idle queue with unchanging depth must report Stalled"
        );
    }
}