zenkey-fleet 0.11.1

//! RFC 09 §6 cutover acceptance, half one (issue #59): prove the retired key
//! family **silent** while the new planes carry traffic.
//!
//! One without the other is not evidence — a quiet old root on a dead fleet
//! proves only that the fleet is dead, which is why the verdict has three
//! states and not two.
//!
//! The leak check states its meaning explicitly — *anything outside
//! `<base>/v1/`* — rather than riding on key algebra, because the version
//! chunk is plain (`v1`, not `@v1`) and `<base>/**` reaches the new keys too
//! (RFC 09 §6's own note). And the scope is honest per RFC 09 §5.1 O5: a `**`
//! subscription cannot cross `@`-chunks, so the verbatim planes and the admin
//! space are outside this check by construction.
//!
//! Lives here rather than in a frontend (issue #206) because every line of it
//! is judgement over bus traffic: the sample-bucketing rule, the tie-break and
//! the verdict ladder are the check, and a second explorer must not have to
//! re-derive them. The frontend keeps the session, the rendering and the exit
//! code.

use std::collections::BTreeMap;
use std::time::Duration;

use crate::{Error, Result};

use crate::judge::common::{FINDING_CAP, new_prefix};
use crate::model::examples::Examples;
use crate::report::{CutoverReport, CutoverVerdict};

/// The scope sentence this check operates under — rendered by the caller
/// before the window opens, because a user watching a 30-second silence
/// deserves to know what was and was not being watched (O5).
pub fn scope_note(old_root: &str, new_prefix: &str, window: Duration) -> String {
    let window = window.as_secs_f64();
    format!(
        "cutover check: {window}s window — asserting {old_root} silent while \
         {new_prefix}** carries traffic (RFC 09 §6). `**` cannot cross \
         `@`-chunks: verbatim planes and the admin space are outside this \
         check by construction (O5)."
    )
}

/// Watch the whole bus for `window` seconds and judge the cutover.
pub async fn run_cutover(
    fleet: &crate::Fleet<'_>,
    old_root: &str,
    window: Duration,
) -> Result<CutoverReport> {
    let old_expr = zenoh::key_expr::KeyExpr::try_from(old_root.to_string())
        .map_err(|e| Error::unaskable_from(format!("--old-root {old_root:?}"), e))?;
    let new_prefix = new_prefix(fleet.base());

    let monitor = crate::Monitor::start(fleet.session(), crate::MonitorSpec::default()).await?;
    let mut events = monitor.events();
    // `**` is the whole bus: leaving this one to `Drop` on an error path is
    // the loudest version of the leak (#336).
    let monitor = monitor.watching(["**"]).await?;

    let mut old_keys: BTreeMap<String, u64> = BTreeMap::new();
    let mut leaked: BTreeMap<String, u64> = BTreeMap::new();
    let (mut old_samples, mut new_samples, mut leak_samples, mut dropped) =
        (0u64, 0u64, 0u64, 0u64);
    let deadline = tokio::time::Instant::now() + window;
    // One timer for the whole window, not one per iteration (#346).
    // `sleep_until` builds a future and registers a timer each time it
    // is evaluated, and a `select!` in a loop evaluates it on every
    // pass — at 100k samples/s that is 100k registrations a second for
    // a deadline that never moves.
    let window_over = tokio::time::sleep_until(deadline);
    tokio::pin!(window_over);
    loop {
        let item = tokio::select! {
            item = events.recv() => item,
            () = &mut window_over => break,
        };
        match item {
            Some(crate::StreamItem::Event(crate::FleetEvent::Sample(s))) => {
                // Old-root membership is key-expression inclusion (the root
                // may be a wildcard family); the new plane is a stated
                // prefix. Old wins ties: a root inside <base>/v1/ is being
                // *retired*, and its traffic is the failure.
                if zenoh::key_expr::KeyExpr::try_from(s.key.as_str())
                    .map(|k| old_expr.includes(&k))
                    .unwrap_or(false)
                {
                    old_samples += 1;
                    *old_keys.entry(s.key.clone()).or_default() += 1;
                } else if s.key.starts_with(&new_prefix) {
                    new_samples += 1;
                } else {
                    leak_samples += 1;
                    *leaked.entry(s.key.clone()).or_default() += 1;
                }
            }
            Some(crate::StreamItem::Dropped(n)) => dropped += n,
            Some(_) => continue,
            None => break,
        }
    }
    monitor.shutdown().await?;

    // The keys-seen counts beside these are the exact totals, so the buckets
    // name examples and leave the arithmetic to the counter.
    let cap = |m: &BTreeMap<String, u64>| {
        let mut ex = Examples::new(FINDING_CAP);
        for (k, n) in m {
            ex.push_with(|| format!("{k} ({n})"));
        }
        ex.into_vec()
    };
    Ok(CutoverReport {
        old_root: old_root.to_string(),
        new_prefix,
        window_s: window.as_secs_f64(),
        old_samples,
        old_keys_seen: old_keys.len(),
        old_examples: cap(&old_keys),
        new_samples,
        leak_samples,
        leaked_keys_seen: leaked.len(),
        leak_examples: cap(&leaked),
        dropped,
        verdict: verdict(old_samples, new_samples),
    })
}

/// The three-state ladder, pure so it can be exercised without a bus.
///
/// Order matters: the old root speaking is a failure whatever else is true,
/// and a silent old root on a silent bus is *unproven* rather than a pass —
/// a dead fleet passes the silence half for free (RFC 05 §3.1).
pub fn verdict(old_samples: u64, new_samples: u64) -> CutoverVerdict {
    if old_samples > 0 {
        CutoverVerdict::OldStillSpeaks
    } else if new_samples == 0 {
        CutoverVerdict::Unproven
    } else {
        CutoverVerdict::Pass
    }
}

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

    /// The ladder's three states, including the one nothing exercised: a
    /// quiet old root on a quiet bus is not evidence of a finished migration.
    #[test]
    fn silence_on_both_planes_is_unproven_not_a_pass() {
        assert_eq!(verdict(0, 12), CutoverVerdict::Pass);
        assert_eq!(verdict(3, 12), CutoverVerdict::OldStillSpeaks);
        assert_eq!(verdict(0, 0), CutoverVerdict::Unproven);
        // Old wins ties: traffic on the retired family is the failure even
        // when the new plane is busy.
        assert_eq!(verdict(1, 10_000), CutoverVerdict::OldStillSpeaks);
    }
    #[test]
    fn the_scope_note_states_what_it_cannot_see() {
        let note = scope_note("old/**", "acme/v1/", Duration::from_secs(30));
        assert!(note.contains("30s window"));
        assert!(
            note.contains("cannot cross"),
            "a wildcard scope must not be presented as total coverage (O5): {note}"
        );
    }
}