zenkey-fleet 0.14.0

//! Declared versus observed `kind` (#422): the sensor that publishes
//! `oom_kills_total` as a gauge, and nothing could have caught it.
//!
//! RFC 08 §2 (v1.32) lets a subject declare what its leaf value *is* —
//! `counter | gauge | text | bool` — and RFC 13 §3 says what a judge owes
//! that declaration:
//!
//! - **Not asked** when the entry declares no `kind`. Nothing here runs for
//!   such a key; the doctor's listen phase never calls
//!   [`KindObservation::observe`] for it.
//! - **Established(no)** — `kind-mismatch`, severity Error — when a
//!   self-describing payload's `type` tag disagrees with the declared kind,
//!   or when a `counter` decreased between two samples of one origin with
//!   no `alive` cycle of that origin in between. The restart is the one
//!   sanctioned reset, and it is on the wire (RFC 04 §5): the listen phase
//!   watches the liveliness planes and reports a down-then-up through
//!   [`KindObservation::alive_cycled`].
//! - **Unobservable** when the payload could not be decoded, said with the
//!   reason — one Warning per key, never folded into a pass.
//!
//! Per origin, never pooled: a key already names its origin, so the table
//! is per key and the counter baseline is that key's own series. The window
//! rides in every finding (O5/O6) — a window in which a counter did not
//! decrease has not established that it is one, which is why there is no
//! `Established(yes)` here at all.
//!
//! Pure, the house pattern of [`crate::judge::field`] and
//! [`crate::judge::budget`]: nothing here takes a session, so the same
//! observation judges a `.zrec` replay.

use std::collections::BTreeMap;

use serde_json::Value;
use zenkey::{SliceToken, SubjectKind};

use crate::judge::common::{EXPANSION_CAP, FINDING_CAP};
use crate::model::examples::Examples;
use crate::report::{CheckId, DoctorFinding, DoctorSeverity};

/// The producer identity a liveliness token and a data key share:
/// `(origin, producer)`, with a service origin's producer being the origin
/// minus its `@` — exactly [`crate::bus::roster::token_identity`]'s reading
/// of `v1/@catalog/state/alive`, so a cycle seen on the token plane lands on
/// the keys it restarted.
pub type ProducerId = (String, String);

/// What one key's series has shown against its declared kind.
#[derive(Debug)]
pub struct KeyKind {
    /// Whose series this is — the cycle bookkeeping is per producer.
    pub producer: ProducerId,
    pub declared: SubjectKind,
    /// Samples that reached a verdict (decoded and read).
    pub judged: u64,
    /// Samples with a self-describing tag that disagreed with `declared`.
    pub tag_mismatches: u64,
    /// Samples whose value was not of the declared kind — a negative or
    /// non-numeric `counter`, a non-boolean `bool`, a non-string `text`.
    pub value_mismatches: u64,
    /// `counter` only: decreases with no `alive` cycle in between.
    pub decreases: u64,
    /// Samples that could not be decoded at all — unobservable, said so.
    pub undecoded: u64,
    /// Up to [`EXPANSION_CAP`] mismatches, spelled out.
    examples: Examples<String>,
    /// `counter` only: the last value read, the next sample's baseline.
    last: Option<f64>,
    /// The producer's cycle generation this key's baseline belongs to.
    generation: u64,
}

/// The per-key observation the listen phase feeds and [`judge_kind`] reads.
#[derive(Debug, Default)]
pub struct KindObservation {
    keys: BTreeMap<String, KeyKind>,
    /// How many `alive` cycles each producer has shown. A key whose baseline
    /// is from an older generation resets on its next sample instead of
    /// being judged against a series the restart ended — kept per producer
    /// rather than as a one-shot mark so every key of a restarted producer
    /// resets, not just the first one to publish (#422).
    cycles: BTreeMap<ProducerId, u64>,
}

/// The tag a self-describing payload carries, when it does
/// (`{"type": "<kind>", "value": …}`, RFC 08 §2 / RFC 11 §4). A tag outside
/// the four is not a disagreement — it is a payload this rule does not read.
fn payload_tag(doc: &Value) -> Option<SubjectKind> {
    doc.get("type")
        .and_then(Value::as_str)
        .and_then(SubjectKind::from_payload_tag)
}

/// The value the declaration is about: `value` when the payload is an
/// object carrying one, the payload itself otherwise.
fn leaf(doc: &Value) -> &Value {
    match doc.get("value") {
        Some(v) if doc.is_object() => v,
        _ => doc,
    }
}

/// The boundaries a histogram value states, when it states them: a
/// `buckets` array of numbers, or of objects each carrying an `le` bound
/// (the two common spellings; the shape is the profile's). A trailing
/// `+Inf` is the implicit bound and is dropped. `None` when the value states
/// none this rule can read — silence is not a disagreement.
fn stated_bounds(v: &Value) -> Option<Vec<f64>> {
    let items = v.get("buckets")?.as_array()?;
    let mut out = Vec::with_capacity(items.len());
    for item in items {
        let bound = match item {
            Value::Number(n) => n.as_f64()?,
            Value::Object(o) => match o.get("le")? {
                Value::Number(n) => n.as_f64()?,
                Value::String(s) if s == "+Inf" || s == "inf" || s == "Infinity" => f64::INFINITY,
                _ => return None,
            },
            Value::String(s) if s == "+Inf" || s == "inf" || s == "Infinity" => f64::INFINITY,
            _ => return None,
        };
        out.push(bound);
    }
    if out.last().is_some_and(|b| b.is_infinite()) {
        out.pop();
    }
    Some(out)
}

/// The first disagreement between declared and stated bounds, spelled for
/// the evidence line; `None` when they are equal.
fn bounds_disagree(declared: &[f64], stated: &[f64]) -> Option<String> {
    if let Some(i) = declared
        .iter()
        .zip(stated)
        .position(|(d, s)| d.to_bits() != s.to_bits())
    {
        return Some(format!(
            "histogram bound {} is {}, registry declares {}",
            i + 1,
            stated[i],
            declared[i]
        ));
    }
    (declared.len() != stated.len()).then(|| {
        format!(
            "histogram states {} bound(s), registry declares {}",
            stated.len(),
            declared.len()
        )
    })
}

/// A short spelling of a JSON value for the evidence line.
fn describe(v: &Value) -> String {
    match v {
        Value::Null => "null".into(),
        Value::Bool(b) => format!("boolean {b}"),
        Value::Number(n) => format!("number {n}"),
        Value::String(s) if s.len() > 24 => format!("string {:?}…", &s[..24]),
        Value::String(s) => format!("string {s:?}"),
        Value::Array(a) => format!("array of {}", a.len()),
        Value::Object(o) => format!("object with {} field(s)", o.len()),
    }
}

impl KindObservation {
    pub fn new() -> KindObservation {
        KindObservation::default()
    }

    /// The producer `(origin, producer)` went down and came back — the
    /// sanctioned counter reset (RFC 08 §2). A lone `NodeUp` at window start
    /// is a token seen, not a cycle; the caller decides that.
    pub fn alive_cycled(&mut self, origin: &str, producer: &str) {
        *self
            .cycles
            .entry((origin.to_string(), producer.to_string()))
            .or_default() += 1;
    }

    /// One `Put` sample on a key whose registry entry declares `kind`.
    /// `doc` is the structural document, `None` when the payload could not
    /// be decoded (counted, reported as unobservable, never judged).
    pub fn observe(
        &mut self,
        key: &str,
        origin: &str,
        producer: &str,
        declared: SubjectKind,
        doc: Option<&Value>,
    ) {
        self.observe_declared(key, origin, producer, declared, None, doc);
    }

    /// [`observe`](Self::observe), with a `histogram` entry's declared
    /// `buckets` (RFC 08 §2, v1.36): a payload that states its boundaries
    /// must state these. `None` for every other kind, and for a histogram
    /// declared by a slice this build could not read the bounds of.
    pub fn observe_declared(
        &mut self,
        key: &str,
        origin: &str,
        producer: &str,
        declared: SubjectKind,
        declared_buckets: Option<&[f64]>,
        doc: Option<&Value>,
    ) {
        let producer_id = (origin.to_string(), producer.to_string());
        let generation = self.cycles.get(&producer_id).copied().unwrap_or(0);
        let entry = self.keys.entry(key.to_string()).or_insert_with(|| KeyKind {
            producer: producer_id,
            declared,
            judged: 0,
            tag_mismatches: 0,
            value_mismatches: 0,
            decreases: 0,
            undecoded: 0,
            examples: Examples::new(EXPANSION_CAP),
            last: None,
            generation,
        });
        let Some(doc) = doc else {
            entry.undecoded += 1;
            return;
        };
        entry.judged += 1;

        // (a) The tag, when the payload self-describes. A disagreeing tag is
        // the finding on its own; the value is not judged twice.
        if let Some(tag) = payload_tag(doc)
            && tag != declared
        {
            entry.tag_mismatches += 1;
            entry.examples.push_with(|| {
                format!(
                    "payload tags itself `{}`, registry declares `{}`",
                    tag.payload_tag(),
                    declared.token()
                )
            });
            return;
        }

        // (b) The value.
        let v = leaf(doc);
        match declared {
            SubjectKind::Gauge => {
                if v.as_f64().is_none() {
                    entry.value_mismatches += 1;
                    entry
                        .examples
                        .push_with(|| format!("gauge value is {}", describe(v)));
                }
            }
            SubjectKind::Bool => {
                if !v.is_boolean() {
                    entry.value_mismatches += 1;
                    entry
                        .examples
                        .push_with(|| format!("bool value is {}", describe(v)));
                }
            }
            SubjectKind::Text => {
                if !v.is_string() {
                    entry.value_mismatches += 1;
                    entry
                        .examples
                        .push_with(|| format!("text value is {}", describe(v)));
                }
            }
            SubjectKind::Histogram => {
                // The payload shape is the application profile's (RFC 11 §4);
                // 08 §2 fixes only that it is a distribution and that stated
                // boundaries equal the declared ones.
                if !v.is_object() {
                    entry.value_mismatches += 1;
                    entry
                        .examples
                        .push_with(|| format!("histogram value is {}", describe(v)));
                    return;
                }
                if let (Some(declared), Some(stated)) = (declared_buckets, stated_bounds(v))
                    && let Some(why) = bounds_disagree(declared, &stated)
                {
                    entry.value_mismatches += 1;
                    entry.examples.push_with(|| why);
                }
            }
            SubjectKind::Counter => {
                let Some(n) = v.as_f64() else {
                    entry.value_mismatches += 1;
                    entry
                        .examples
                        .push_with(|| format!("counter value is {}", describe(v)));
                    return;
                };
                if n < 0.0 {
                    entry.value_mismatches += 1;
                    entry
                        .examples
                        .push_with(|| format!("counter value is negative ({n})"));
                }
                if entry.generation != generation {
                    // The producer cycled its `alive` token since this
                    // key's baseline: the series restarted, and this sample
                    // is the new baseline (RFC 08 §2, RFC 13 §3).
                    entry.generation = generation;
                    entry.last = Some(n);
                    return;
                }
                if let Some(prev) = entry.last
                    && n < prev
                {
                    entry.decreases += 1;
                    entry.examples.push_with(|| {
                        format!("counter decreased {prev} → {n} with no `alive` cycle in between")
                    });
                }
                entry.last = Some(n);
            }
        }
    }
}

impl KeyKind {
    /// The mismatches spelled out, up to [`EXPANSION_CAP`] of them.
    pub fn examples(&self) -> &[String] {
        self.examples.as_slice()
    }
}

impl KindObservation {
    /// Every key observed, with what its series showed.
    pub fn iter(&self) -> impl Iterator<Item = (&str, &KeyKind)> {
        self.keys.iter().map(|(k, v)| (k.as_str(), v))
    }

    pub fn keys_seen(&self) -> usize {
        self.keys.len()
    }
}

/// The `kind-mismatch` findings for one listen window: one Error per key
/// whose series disagreed with its declaration, one Warning per key whose
/// payloads could not be judged, both capped at the doctor's per-check
/// finding cap with the remainder counted. Filter **then** cap, like every
/// listen check.
pub fn judge_kind(observation: &KindObservation, window_s: f64) -> Vec<DoctorFinding> {
    let mut findings = Vec::new();
    let mut bad: Examples<DoctorFinding> = Examples::new(FINDING_CAP);
    let mut unjudged: Examples<DoctorFinding> = Examples::new(FINDING_CAP);
    for (key, k) in observation.iter() {
        let mismatches = k.tag_mismatches + k.value_mismatches + k.decreases;
        if mismatches > 0 {
            bad.push_with(|| {
                let mut parts = Vec::new();
                if k.tag_mismatches > 0 {
                    parts.push(format!("{} tag disagreement(s)", k.tag_mismatches));
                }
                if k.decreases > 0 {
                    parts.push(format!("{} decrease(s)", k.decreases));
                }
                if k.value_mismatches > 0 {
                    parts.push(format!("{} value(s) not of that kind", k.value_mismatches));
                }
                let examples = k.examples.as_slice().join("; ");
                DoctorFinding {
                    severity: DoctorSeverity::Error,
                    check: CheckId::KindMismatch,
                    subject: key.to_string(),
                    evidence: format!(
                        "declared `{}`, and {} of {} sample(s) from origin {} in {window_s:.0}s \
                         disagree: {} — e.g. {examples}",
                        k.declared.token(),
                        mismatches,
                        k.judged,
                        k.producer.0,
                        parts.join(", "),
                    ),
                    citation: Some("RFC 08 §2".into()),
                }
            });
        }
        if k.undecoded > 0 {
            unjudged.push_with(|| DoctorFinding {
                severity: DoctorSeverity::Warning,
                check: CheckId::KindMismatch,
                subject: key.to_string(),
                evidence: format!(
                    "kind not judged: {} payload(s) from origin {} in {window_s:.0}s could \
                     not be decoded, so the declared `{}` is unobservable for them",
                    k.undecoded,
                    k.producer.0,
                    k.declared.token(),
                ),
                citation: Some("RFC 13 §3".into()),
            });
        }
    }
    for (ex, tail) in [
        (bad, "more key(s) with the same finding"),
        (unjudged, "more key(s) whose kind was not judged"),
    ] {
        let more = ex.more(tail);
        findings.extend(ex.into_vec());
        if let Some(evidence) = more {
            findings.push(DoctorFinding {
                severity: DoctorSeverity::Info,
                check: CheckId::KindMismatch,
                subject: "fleet".into(),
                evidence,
                citation: None,
            });
        }
    }
    findings
}

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

    const KEY: &str = "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/memory/oom_kills_total";

    fn observe(obs: &mut KindObservation, declared: SubjectKind, doc: Value) {
        obs.observe(KEY, "h-aaaaaaaaaaaa", "sysinfo", declared, Some(&doc));
    }

    /// v1.36: a histogram is an object, and the boundaries it states —
    /// as numbers or as `le` objects, `+Inf` implicit — equal the declared.
    #[test]
    fn a_histogram_is_judged_by_its_shape_and_its_stated_bounds() {
        let key = "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/system/runqlat";
        let declared = [0.001, 0.01, 0.1];
        let judge = |doc: Value| {
            let mut obs = KindObservation::new();
            obs.observe_declared(
                key,
                "h-aaaaaaaaaaaa",
                "sysinfo",
                SubjectKind::Histogram,
                Some(&declared),
                Some(&doc),
            );
            judge_kind(&obs, 10.0)
                .into_iter()
                .filter(|f| f.check == CheckId::KindMismatch)
                .map(|f| f.evidence)
                .collect::<Vec<_>>()
        };
        // Equal bounds, both spellings, the tag agreeing: no finding.
        assert!(judge(json!({"type": "histogram", "value": {"buckets": [0.001, 0.01, 0.1], "counts": [1, 2, 3, 4]}})).is_empty());
        assert!(
            judge(json!({"buckets": [{"le": 0.001}, {"le": 0.01}, {"le": 0.1}, {"le": "+Inf"}]}))
                .is_empty()
        );
        // A distribution that states no bounds is not a disagreement.
        assert!(judge(json!({"count": 4, "sum": 0.3})).is_empty());
        // A different bound names the first that differs.
        let f = judge(json!({"buckets": [0.001, 0.05, 0.1]}));
        assert_eq!(f.len(), 1, "{f:?}");
        assert!(
            f[0].contains("histogram bound 2 is 0.05, registry declares 0.01"),
            "{}",
            f[0]
        );
        // A missing bound is a count disagreement.
        let f = judge(json!({"buckets": [0.001, 0.01]}));
        assert!(
            f[0].contains("states 2 bound(s), registry declares 3"),
            "{}",
            f[0]
        );
        // A scalar where a distribution was declared.
        let f = judge(json!(12.5));
        assert!(f[0].contains("histogram value is number 12.5"), "{}", f[0]);
        // A tag naming another kind is the tag finding, not a value one.
        let f = judge(json!({"type": "gauge", "value": 1.0}));
        assert!(
            f[0].contains("payload tags itself `gauge`, registry declares `histogram`"),
            "{}",
            f[0]
        );
    }

    fn mismatches(obs: &KindObservation) -> Vec<DoctorFinding> {
        judge_kind(obs, 10.0)
            .into_iter()
            .filter(|f| f.severity == DoctorSeverity::Error)
            .collect()
    }

    /// A counter that goes down is the finding, named per origin with the
    /// window; a counter that only goes up is not a pass — it is nothing.
    #[test]
    fn a_decreasing_counter_is_a_finding_and_a_rising_one_is_nothing() {
        let mut obs = KindObservation::new();
        for n in [10, 20, 30] {
            observe(&mut obs, SubjectKind::Counter, json!(n));
        }
        assert!(
            judge_kind(&obs, 10.0).is_empty(),
            "no Established(yes) exists"
        );

        observe(&mut obs, SubjectKind::Counter, json!(5));
        let f = mismatches(&obs);
        assert_eq!(f.len(), 1, "{f:?}");
        assert_eq!(f[0].check, CheckId::KindMismatch);
        assert_eq!(f[0].subject, KEY);
        assert!(
            f[0].evidence.contains("h-aaaaaaaaaaaa"),
            "{}",
            f[0].evidence
        );
        assert!(f[0].evidence.contains("10s"), "the window is stated");
        assert!(f[0].evidence.contains("30 → 5"), "{}", f[0].evidence);
        assert_eq!(f[0].citation.as_deref(), Some("RFC 08 §2"));
    }

    /// The restart is the one sanctioned reset (RFC 08 §2): a cycle of the
    /// producer's `alive` token between the samples makes the drop a new
    /// baseline, for every key of that producer, and a cycle of *another*
    /// producer excuses nothing.
    #[test]
    fn a_reset_across_an_alive_cycle_is_not_a_finding() {
        let mut obs = KindObservation::new();
        let other = "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/memory/page_faults_total";
        observe(&mut obs, SubjectKind::Counter, json!(10));
        obs.observe(
            other,
            "h-aaaaaaaaaaaa",
            "sysinfo",
            SubjectKind::Counter,
            Some(&json!(7)),
        );
        obs.alive_cycled("h-aaaaaaaaaaaa", "sysinfo");
        observe(&mut obs, SubjectKind::Counter, json!(0));
        obs.observe(
            other,
            "h-aaaaaaaaaaaa",
            "sysinfo",
            SubjectKind::Counter,
            Some(&json!(0)),
        );
        assert!(mismatches(&obs).is_empty(), "{:?}", judge_kind(&obs, 10.0));

        // After the reset the new baseline holds: a later drop is judged.
        observe(&mut obs, SubjectKind::Counter, json!(3));
        observe(&mut obs, SubjectKind::Counter, json!(1));
        assert_eq!(mismatches(&obs).len(), 1);

        // Another producer's cycle is not this one's restart.
        let mut obs = KindObservation::new();
        observe(&mut obs, SubjectKind::Counter, json!(10));
        obs.alive_cycled("h-aaaaaaaaaaaa", "netring");
        observe(&mut obs, SubjectKind::Counter, json!(0));
        assert_eq!(mismatches(&obs).len(), 1);
    }

    /// A self-describing payload's tag must agree (RFC 08 §2); `boolean` is
    /// how `bool` is spelled on the wire (RFC 11 §4), and an unknown tag is
    /// not a disagreement.
    #[test]
    fn a_disagreeing_tag_is_a_finding_and_an_agreeing_or_foreign_one_is_not() {
        let mut obs = KindObservation::new();
        observe(
            &mut obs,
            SubjectKind::Counter,
            json!({"type": "gauge", "value": 1}),
        );
        let f = mismatches(&obs);
        assert_eq!(f.len(), 1, "{f:?}");
        assert!(
            f[0].evidence.contains("tags itself `gauge`"),
            "{}",
            f[0].evidence
        );

        let mut obs = KindObservation::new();
        observe(
            &mut obs,
            SubjectKind::Bool,
            json!({"type": "boolean", "value": true}),
        );
        // A tag outside the vocabulary is foreign, not a disagreement.
        // (`histogram` was this test's foreign tag until v1.36 made it a
        // kind; `summary` is one v1.36 deliberately did not ratify.)
        observe(
            &mut obs,
            SubjectKind::Bool,
            json!({"type": "summary", "value": true}),
        );
        assert!(mismatches(&obs).is_empty());

        // …and since v1.36 a `histogram` tag on a `bool` subject disagrees.
        let mut obs = KindObservation::new();
        observe(
            &mut obs,
            SubjectKind::Bool,
            json!({"type": "histogram", "value": {}}),
        );
        assert_eq!(mismatches(&obs).len(), 1);
    }

    /// The value itself: a `text` that is a number, a `bool` that is a
    /// string, a negative `counter` — bare or under `value`.
    #[test]
    fn a_value_not_of_the_declared_kind_is_a_finding() {
        let mut obs = KindObservation::new();
        observe(&mut obs, SubjectKind::Text, json!(3));
        observe(&mut obs, SubjectKind::Text, json!({"value": "ok"}));
        assert_eq!(mismatches(&obs).len(), 1);
        let mut obs = KindObservation::new();
        observe(&mut obs, SubjectKind::Bool, json!({"value": "true"}));
        assert_eq!(mismatches(&obs).len(), 1);
        let mut obs = KindObservation::new();
        observe(&mut obs, SubjectKind::Counter, json!(-1));
        assert_eq!(mismatches(&obs).len(), 1);
        let mut obs = KindObservation::new();
        observe(&mut obs, SubjectKind::Gauge, json!(-1.5));
        observe(&mut obs, SubjectKind::Gauge, json!({"value": 2}));
        assert!(mismatches(&obs).is_empty(), "a gauge is any number");
    }

    /// An undecodable payload is unobservable, said so per key as a Warning
    /// — never a pass, never an Error (RFC 13 §3).
    #[test]
    fn an_undecodable_payload_is_reported_unobservable_not_passed() {
        let mut obs = KindObservation::new();
        obs.observe(KEY, "h-aaaaaaaaaaaa", "sysinfo", SubjectKind::Counter, None);
        obs.observe(KEY, "h-aaaaaaaaaaaa", "sysinfo", SubjectKind::Counter, None);
        let f = judge_kind(&obs, 10.0);
        assert_eq!(f.len(), 1, "{f:?}");
        assert_eq!(f[0].severity, DoctorSeverity::Warning);
        assert_eq!(f[0].check, CheckId::KindMismatch);
        assert!(
            f[0].evidence.contains("kind not judged: 2 payload(s)"),
            "{}",
            f[0].evidence
        );
        assert!(f[0].evidence.contains("10s"));
    }

    /// The cap bounds the findings and the remainder counts what it hid —
    /// the note cannot disagree with the population it summarises.
    #[test]
    fn the_cap_bites_with_a_counted_remainder() {
        let mut obs = KindObservation::new();
        for i in 0..(FINDING_CAP + 3) {
            let key = format!("v1/h-aaaaaaaaaaaa/telemetry/sysinfo/k{i}");
            obs.observe(
                &key,
                "h-aaaaaaaaaaaa",
                "sysinfo",
                SubjectKind::Text,
                Some(&json!(1)),
            );
        }
        let f = judge_kind(&obs, 10.0);
        assert_eq!(f.len(), FINDING_CAP + 1, "{f:?}");
        let tail = f.last().unwrap();
        assert_eq!(tail.severity, DoctorSeverity::Info);
        assert!(tail.evidence.contains("… and 3 more"), "{}", tail.evidence);
    }
}