diff --git a/Cargo.lock b/Cargo.lock index 0e834cd11b2..8b42a7c6948 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -223,8 +223,11 @@ dependencies = [ "antithesis_sdk", "anyhow", "clap", + "ddsketch", "harness", + "proptest", "reqwest", + "rustls", "serde", "serde_json", ] diff --git a/test/antithesis/bin/launch.sh b/test/antithesis/bin/launch.sh index 357d6c2652c..a57f589795d 100755 --- a/test/antithesis/bin/launch.sh +++ b/test/antithesis/bin/launch.sh @@ -51,6 +51,7 @@ SCENARIO_DIR="$ANTITHESIS_DIR/scenarios/$SCENARIO" SCENARIO_TEST_NAME="" SCENARIO_DESCRIPTION="" SCENARIO_FAULT_NODES="" +SCENARIO_FORCE_DISABLE_ALL_FAULTS="" SCENARIO_WEBHOOK="" # shellcheck source=/dev/null . "$SCENARIO_DIR/launch.env" @@ -74,7 +75,7 @@ FAULT_NODES="${FAULT_NODES-$SCENARIO_FAULT_NODES}" # run_test global fault toggles. Defaults match the endpoint defaults so a plain # shot behaves as it did before the move off persistent_storage. SIMULTANEOUS_FAULTS="${SIMULTANEOUS_FAULTS:-false}" -FORCE_DISABLE_ALL_FAULTS="${FORCE_DISABLE_ALL_FAULTS:-false}" +FORCE_DISABLE_ALL_FAULTS="${FORCE_DISABLE_ALL_FAULTS:-${SCENARIO_FORCE_DISABLE_ALL_FAULTS:-false}}" # antithesis.source. On the unified run_test endpoint this one identifier both # selects the datadog_agent customization block -- title "Datadog Agent" and the diff --git a/test/antithesis/scenarios/differential/Cargo.toml b/test/antithesis/scenarios/differential/Cargo.toml index 0891f8daab4..42fe33de4d9 100644 --- a/test/antithesis/scenarios/differential/Cargo.toml +++ b/test/antithesis/scenarios/differential/Cargo.toml @@ -13,7 +13,12 @@ workspace = true antithesis_sdk = { workspace = true, features = ["full", "rand_v0_10"] } anyhow = { workspace = true, features = ["std"] } clap = { workspace = true, features = ["derive", "env", "error-context", "help", "std", "usage"] } +ddsketch = { workspace = true } harness = { path = "../../harness" } reqwest = { workspace = true, features = ["blocking", "json"] } +rustls = { workspace = true, features = ["aws_lc_rs"] } serde = { workspace = true } serde_json = { workspace = true } + +[dev-dependencies] +proptest = { workspace = true } diff --git a/test/antithesis/scenarios/differential/Dockerfile b/test/antithesis/scenarios/differential/Dockerfile index 341e2b69657..8594c6a68d1 100644 --- a/test/antithesis/scenarios/differential/Dockerfile +++ b/test/antithesis/scenarios/differential/Dockerfile @@ -62,8 +62,7 @@ RUN --mount=type=cache,target=/tools/target,id=antithesis-differential-tools-tar cargo build --release --package harness --bin first_sample_config --bin eventually_adp_alive && \ cp /tools/target/release/antithesis-intake /usr/local/bin/antithesis-intake && \ cp /tools/target/release/parallel_driver_send_dogstatsd_differential /usr/local/bin/parallel_driver_send_dogstatsd_differential && \ - cp /tools/target/release/eventually_differential_contexts /usr/local/bin/eventually_differential_contexts && \ - cp /tools/target/release/finally_differential_contexts /usr/local/bin/finally_differential_contexts && \ + cp /tools/target/release/eventually_differential_curves /usr/local/bin/eventually_differential_curves && \ cp /tools/target/release/eventually_adp_alive /usr/local/bin/eventually_adp_alive && \ cp /tools/target/release/first_sample_config /usr/local/bin/first_sample_config @@ -122,8 +121,7 @@ RUN mkdir -p /opt/antithesis/test/v1/main COPY --chmod=755 test/antithesis/scenarios/differential/workload/setup-complete.sh /opt/antithesis/setup-complete.sh COPY --from=tools-builder --chmod=755 /usr/local/bin/first_sample_config /opt/antithesis/test/v1/main/first_sample_config COPY --from=tools-builder --chmod=755 /usr/local/bin/parallel_driver_send_dogstatsd_differential /opt/antithesis/test/v1/main/parallel_driver_send_dogstatsd_differential -COPY --from=tools-builder --chmod=755 /usr/local/bin/eventually_differential_contexts /opt/antithesis/test/v1/main/eventually_differential_contexts -COPY --from=tools-builder --chmod=755 /usr/local/bin/finally_differential_contexts /opt/antithesis/test/v1/main/finally_differential_contexts +COPY --from=tools-builder --chmod=755 /usr/local/bin/eventually_differential_curves /opt/antithesis/test/v1/main/eventually_differential_curves COPY --from=tools-builder --chmod=755 /usr/local/bin/eventually_adp_alive /opt/antithesis/test/v1/main/eventually_adp_alive # Datadog Agent debug symbols. Antithesis scans /symbols in every pushed image and # matches by build-id, so this symbolizes the Core Agent running in the SUT lanes. diff --git a/test/antithesis/scenarios/differential/launch.env b/test/antithesis/scenarios/differential/launch.env index 2b9f0b30bcb..e7514a5e433 100644 --- a/test/antithesis/scenarios/differential/launch.env +++ b/test/antithesis/scenarios/differential/launch.env @@ -3,3 +3,7 @@ SCENARIO_DESCRIPTION="ADP differential: Datadog Agent ADP-on vs Datadog Agent AD # No node faults. Terminating, hanging, or throttling can't be done as a group # so there's no fair way to fault the lanes. SCENARIO_FAULT_NODES="" +# Disable ALL fault injection for the differential A/B. Network clog and clock skew hit one lane +# before the other, which shows up as a false divergence, not an ADP bug. The master switch is the +# only canonical knob that turns off the global network and clock fault classes. +SCENARIO_FORCE_DISABLE_ALL_FAULTS="true" diff --git a/test/antithesis/scenarios/differential/src/bin/eventually_differential_contexts.rs b/test/antithesis/scenarios/differential/src/bin/eventually_differential_contexts.rs deleted file mode 100644 index dffc64731c2..00000000000 --- a/test/antithesis/scenarios/differential/src/bin/eventually_differential_contexts.rs +++ /dev/null @@ -1,56 +0,0 @@ -//! Antithesis `eventually_` check: fails when a context has sat in the symmetric difference of the -//! two lanes longer than `ACCEPTABLE_FLUSH_DELAY`. A member still within the budget is in flight. - -use std::time::Duration; - -use antithesis_scenario_differential::contexts::{Difference, Lane, LaneView}; -use antithesis_sdk::prelude::*; -use anyhow::Context; -use clap::Parser; -use harness::ACCEPTABLE_FLUSH_DELAY; -use reqwest::blocking::Client; -use serde_json::json; - -/// Cap on diverging contexts listed in the assertion details, to bound the payload. -const SAMPLE_LIMIT: usize = 25; - -#[derive(Debug, Parser)] -struct Config { - #[arg(long = "intake-addr", env = "INTAKE_CONTROL_ADDR", default_value = "intake:2049")] - intake_addr: String, -} - -fn main() -> anyhow::Result<()> { - antithesis_init(); - let config = Config::parse(); - let client = Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .context("build HTTP client")?; - - let agent = LaneView::fetch(&client, &config.intake_addr, Lane::Agent)?; - let adp = LaneView::fetch(&client, &config.intake_addr, Lane::Adp)?; - let difference = Difference::between(&agent, &adp); - let delayed = difference.delayed(ACCEPTABLE_FLUSH_DELAY); - let budget = ACCEPTABLE_FLUSH_DELAY.as_secs() as i64; - let mut overdue: Vec<_> = difference - .diverging() - .into_iter() - .filter(|d| d.age_secs > budget) - .collect(); - let adp_only = overdue.iter().filter(|d| matches!(d.lane, Lane::Adp)).count(); - let agent_only = overdue.iter().filter(|d| matches!(d.lane, Lane::Agent)).count(); - overdue.truncate(SAMPLE_LIMIT); - - let details = json!({ - "difference": difference.len(), - "delayed": delayed, - "acceptable_flush_delay_secs": ACCEPTABLE_FLUSH_DELAY.as_secs(), - "adp_only": adp_only, - "agent_only": agent_only, - "sample": overdue, - }); - - assert_always!(delayed == 0, "differential.contexts_eventually_equivalent", &details); - Ok(()) -} diff --git a/test/antithesis/scenarios/differential/src/bin/eventually_differential_curves.rs b/test/antithesis/scenarios/differential/src/bin/eventually_differential_curves.rs new file mode 100644 index 00000000000..0d3fd374547 --- /dev/null +++ b/test/antithesis/scenarios/differential/src/bin/eventually_differential_curves.rs @@ -0,0 +1,119 @@ +//! The single differential correctness assertion: the online curve oracle. Each tick it polls both +//! lanes' settled aggregation curves over `/antithesis/curves`, runs the per-context banded-Frechet +//! aligner plus the decision layer, unions the per-context trips into one `defect_count`, and fires one +//! `assert_always!` that the count is zero. The details carry the defect tally and a bounded triage +//! sample sorted by continuous worst gap. +//! +//! ADP-off (Agent) is the normative reference and ADP-on (data plane) is the SUT, so any surviving trip +//! in either direction is an ADP defect. The extent rides as triage metadata only and never decides the +//! verdict. + +use std::collections::BTreeSet; +use std::thread::sleep; +use std::time::Duration; + +use antithesis_scenario_differential::contexts::{Bucket, Context, Lane, LaneView}; +use antithesis_scenario_differential::decision::{ + bounded_triage, defect_count, evaluate, ContextVerdict, DecisionParams, Extent, +}; +use antithesis_scenario_differential::ground::RuleSet; +use antithesis_sdk::prelude::*; +use clap::Parser; +use reqwest::blocking::Client; +use serde_json::json; + +/// How often the two lanes are polled and the assertion re-evaluated. +const POLL_INTERVAL: Duration = Duration::from_secs(15); +/// Cap on defect contexts listed in the assertion details, to bound the payload. +const SAMPLE_LIMIT: usize = 25; + +#[derive(Debug, Parser)] +struct Config { + #[arg(long = "intake-addr", env = "INTAKE_CONTROL_ADDR", default_value = "intake:2049")] + intake_addr: String, +} + +/// One defect context's triage row: its identity, the mechanisms that fired, and its extent. Sorted by +/// the extent's continuous worst gap. +struct Triage { + context: Context, + trips: Vec, + extent: Extent, +} + +fn main() -> anyhow::Result<()> { + antithesis_init(); + let config = Config::parse(); + // reqwest resolves with rustls-no-provider across the workspace, so install a + // crypto provider before the Client is built, even for plain HTTP, or the build + // panics on its internal runtime thread. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let client = Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .map_err(|e| anyhow::anyhow!("build HTTP client: {e}"))?; + + loop { + if let Err(error) = poll(&client, &config.intake_addr) { + // A momentary fetch failure is not a divergence. Retry on the next tick. + eprintln!("poll failed, retrying: {error:#}"); + } + sleep(POLL_INTERVAL); + } +} + +fn poll(client: &Client, intake_addr: &str) -> anyhow::Result<()> { + let reference = LaneView::fetch(client, intake_addr, Lane::Agent)?.curves(); + let sut = LaneView::fetch(client, intake_addr, Lane::Adp)?.curves(); + let params = DecisionParams::launch(); + let rules = RuleSet::default_rules(); + let empty: Vec = Vec::new(); + + let contexts: BTreeSet<&Context> = reference.keys().chain(sut.keys()).collect(); + let mut verdicts: Vec = Vec::with_capacity(contexts.len()); + let mut triage: Vec = Vec::new(); + for context in contexts { + let reference_curve = reference.get(context).unwrap_or(&empty); + let sut_curve = sut.get(context).unwrap_or(&empty); + let verdict = evaluate(reference_curve, sut_curve, &rules, ¶ms); + if verdict.tripped { + triage.push(Triage { + context: context.clone(), + trips: verdict.trips.iter().map(|t| t.rule.clone()).collect(), + extent: verdict.extent, + }); + } + verdicts.push(verdict); + } + + let defects = defect_count(&verdicts); + let sample: Vec<_> = bounded_triage(triage, |t| t.extent.worst_gap, SAMPLE_LIMIT) + .into_iter() + .map(|t| { + json!({ + "context": t.context, + "trips": t.trips, + "run_length": t.extent.run_length, + "worst_gap": finite(t.extent.worst_gap), + "accumulated_gap": finite(t.extent.accumulated_gap), + }) + }) + .collect(); + + let details = json!({ + "contexts": verdicts.len(), + "defects": defects, + "band_alpha": params.band.alpha, + "band_floor": params.band.floor, + "sample": sample, + }); + + assert_always!(defects == 0, "differential.contexts_curve_equivalent", &details); + Ok(()) +} + +/// A finite magnitude as itself, a non-finite one as JSON null: an infinite present/absent leash serdes +/// cleanly rather than truncating the payload. +fn finite(value: f64) -> Option { + value.is_finite().then_some(value) +} diff --git a/test/antithesis/scenarios/differential/src/bin/finally_differential_contexts.rs b/test/antithesis/scenarios/differential/src/bin/finally_differential_contexts.rs deleted file mode 100644 index 608e030008f..00000000000 --- a/test/antithesis/scenarios/differential/src/bin/finally_differential_contexts.rs +++ /dev/null @@ -1,55 +0,0 @@ -//! Antithesis `finally_` check: after drivers and faults stop, sleeps one `ACCEPTABLE_FLUSH_DELAY` -//! to drain in-flight contexts, then fails if the symmetric difference `D` is not empty. - -use std::thread::sleep; -use std::time::Duration; - -use antithesis_scenario_differential::contexts::{Difference, Lane, LaneView}; -use antithesis_sdk::prelude::*; -use anyhow::Context; -use clap::Parser; -use harness::ACCEPTABLE_FLUSH_DELAY; -use reqwest::blocking::Client; -use serde_json::json; - -/// Cap on diverging contexts listed in the assertion details, to bound the payload. -const SAMPLE_LIMIT: usize = 25; - -#[derive(Debug, Parser)] -struct Config { - #[arg(long = "intake-addr", env = "INTAKE_CONTROL_ADDR", default_value = "intake:2049")] - intake_addr: String, -} - -fn main() -> anyhow::Result<()> { - antithesis_init(); - let config = Config::parse(); - let client = Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .context("build HTTP client")?; - - sleep(ACCEPTABLE_FLUSH_DELAY); - - let agent = LaneView::fetch(&client, &config.intake_addr, Lane::Agent)?; - let adp = LaneView::fetch(&client, &config.intake_addr, Lane::Adp)?; - let difference = Difference::between(&agent, &adp); - let mut residual = difference.diverging(); - let adp_only = residual.iter().filter(|d| matches!(d.lane, Lane::Adp)).count(); - let agent_only = residual.iter().filter(|d| matches!(d.lane, Lane::Agent)).count(); - residual.truncate(SAMPLE_LIMIT); - - let details = json!({ - "difference": difference.len(), - "adp_only": adp_only, - "agent_only": agent_only, - "sample": residual, - }); - - assert_always!( - difference.is_empty(), - "differential.contexts_finally_converged", - &details - ); - Ok(()) -} diff --git a/test/antithesis/scenarios/differential/src/contexts.rs b/test/antithesis/scenarios/differential/src/contexts.rs index 88393668b32..5a1006d784e 100644 --- a/test/antithesis/scenarios/differential/src/contexts.rs +++ b/test/antithesis/scenarios/differential/src/contexts.rs @@ -1,11 +1,9 @@ -//! Context comparison shared by the `eventually_` and `finally_` checks. +//! The two lanes' aggregation curves, fetched from the intake and keyed by context identity. //! -//! Each check fetches the two lanes from the intake and builds a [`Difference`], the symmetric -//! difference `D` of their context sets. `eventually_` fails an overdue member. `finally_` fails any -//! residual after a drain. +//! The curve oracle pairs the two lanes' [`LaneView::curves`] maps by context and hands each pair to +//! the decision layer. use std::collections::{BTreeMap, BTreeSet}; -use std::time::Duration; use reqwest::blocking::Client; use serde::{Deserialize, Serialize}; @@ -19,17 +17,48 @@ pub struct Context { kind: String, } -/// A context and the time it first arrived on a lane, the flat wire shape the intake serves. +/// A DDSketch summary plus its log-grid bins as `(key, count)`. The differential-crate mirror of the +/// intake's `capture::SketchValue`; both sides serde the same wire shape so no second decoder sits on +/// either lane. +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub struct SketchValue { + pub count: i64, + pub sum: f64, + pub min: f64, + pub max: f64, + pub bins: Vec<(i32, u32)>, +} + +/// One aggregated bucket value, kind-agnostic. The mirror of the intake's `capture::BucketValue`. +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub enum BucketValue { + /// A count, rate, or gauge scalar. + Scalar(f64), + /// A DDSketch summary plus its log-grid bins. + Sketch(SketchValue), +} + +/// One bucket on a context's aggregation curve: the wire bucket-start, its value, and whether a +/// second differing value landed on the same bucket-start on the same lane. The mirror of the +/// intake's `capture::Bucket`. +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub struct Bucket { + pub bucket_start: u64, + pub value: BucketValue, + pub conflict: bool, +} + +/// A context and its captured curve: the curve DTO the intake serves and this bin consumes. `kind` +/// rides in `context`. #[derive(Clone, Debug, Deserialize)] struct Captured { #[serde(flatten)] context: Context, - first_seen: i64, + /// The context's aggregation curve in bucket-start order. + #[serde(default)] + buckets: Vec, } -/// A lane's cumulative context set with each context's first-seen time. The README's `C_ADP`/`C_DA`. -type Cumulative = BTreeMap; - /// The lane to read from the intake control API. #[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "snake_case")] @@ -47,16 +76,14 @@ impl Lane { } } -/// One lane's contexts and the intake's current time. `now` ages `first_seen` against the same clock -/// that stamped it. +/// One lane's captured contexts, fetched from the intake control API. #[derive(Clone, Debug, Deserialize)] pub struct LaneView { - now: i64, contexts: Vec, } impl LaneView { - /// Fetches a lane's contexts and the intake's current time from the control API. + /// Fetches a lane's contexts from the control API. /// /// # Errors /// @@ -73,107 +100,14 @@ impl LaneView { .map_err(|e| anyhow::anyhow!("parse JSON response from {url}: {e}")) } - fn cumulative(&self) -> Cumulative { - self.contexts - .iter() - .map(|c| (c.context.clone(), c.first_seen)) - .collect() - } -} - -/// A diverging context's lane and first-seen time, before it is aged against `now`. -#[derive(Clone, Copy, Debug)] -struct Member { - lane: Lane, - first_seen: i64, -} - -/// A diverging context tagged with the lane that carries it and how long it has sat in `D`. -/// -/// `agent` means ADP dropped or mangled a context the Datadog Agent emitted; `adp` means ADP emitted -/// one the Agent did not. The Agent is normative, so either direction is an ADP defect. -#[derive(Clone, Debug, Serialize)] -pub struct Diverging { - pub lane: Lane, - #[serde(flatten)] - pub context: Context, - pub age_secs: i64, -} - -/// The symmetric difference `D` of `C_ADP` and `C_DA`: contexts on one lane but not both, aged -/// against the intake's clock. -pub struct Difference { - members: BTreeMap, - now: i64, -} - -impl Difference { - /// Computes `D` from the two lanes' cumulative sets. - #[must_use] - pub fn between(agent: &LaneView, adp: &LaneView) -> Self { - let c_da = agent.cumulative(); - let c_adp = adp.cumulative(); - let mut members = BTreeMap::new(); - for (context, &first_seen) in &c_da { - if !c_adp.contains_key(context) { - members.insert( - context.clone(), - Member { - lane: Lane::Agent, - first_seen, - }, - ); - } - } - for (context, &first_seen) in &c_adp { - if !c_da.contains_key(context) { - members.insert( - context.clone(), - Member { - lane: Lane::Adp, - first_seen, - }, - ); - } - } - Self { - members, - now: agent.now.max(adp.now), - } - } - - /// The number of diverging contexts. + /// Each context's captured aggregation curve keyed by context identity. The curve oracle pairs the + /// two lanes' maps by context; a context on only one map is a presence concern the decision layer + /// settles against the band. #[must_use] - pub fn len(&self) -> usize { - self.members.len() - } - - /// Whether the lanes agree. - #[must_use] - pub fn is_empty(&self) -> bool { - self.members.is_empty() - } - - /// How many members are `delayed`: sat in `D` longer than `budget` by the intake's clock. - #[must_use] - pub fn delayed(&self, budget: Duration) -> usize { - let budget = budget.as_secs() as i64; - self.members - .values() - .filter(|m| self.now - m.first_seen > budget) - .count() - } - - /// The diverging contexts, each tagged with its lane and its age in `D` by the intake's clock. - #[must_use] - pub fn diverging(&self) -> Vec { - self.members + pub fn curves(&self) -> BTreeMap> { + self.contexts .iter() - .map(|(context, member)| Diverging { - lane: member.lane, - context: context.clone(), - age_secs: self.now - member.first_seen, - }) + .map(|c| (c.context.clone(), c.buckets.clone())) .collect() } } @@ -184,77 +118,78 @@ mod tests { use super::*; - fn captured(name: &str, kind: &str, tags: &[&str], first_seen: i64) -> Captured { - Captured { - context: Context { - name: name.to_string(), - tagset: tags.iter().map(|t| (*t).to_string()).collect(), - kind: kind.to_string(), - }, - first_seen, - } - } - - fn lane(now: i64, contexts: Vec) -> LaneView { - LaneView { now, contexts } - } - #[test] fn flushed_type_is_part_of_identity() { - // Same name and tags but different flushed type are distinct contexts. - let agent = lane(10, vec![captured("adp.requests", "count", &["host:h"], 10)]); - let adp = lane(10, vec![captured("adp.requests", "gauge", &["host:h"], 10)]); - - assert_eq!(Difference::between(&agent, &adp).len(), 2); + // Same name and tags but different flushed type are distinct contexts, so the curve keys differ. + let view: LaneView = serde_json::from_value(json!({ + "contexts": [ + { "name": "adp.requests", "tagset": ["host:h"], "kind": "count" }, + { "name": "adp.requests", "tagset": ["host:h"], "kind": "gauge" }, + ], + })) + .expect("deserialize LaneView"); + + assert_eq!(view.curves().len(), 2); } #[test] fn tag_order_does_not_decide_identity() { - // The same tag set in any order is one context, so the lanes agree. - let agent = lane(10, vec![captured("adp.requests", "count", &["b:2", "a:1"], 10)]); - let adp = lane(11, vec![captured("adp.requests", "count", &["a:1", "b:2"], 11)]); - - assert!(Difference::between(&agent, &adp).is_empty()); + // The same tag set in any order is one context, so both lanes key their curves the same. + let agent: LaneView = serde_json::from_value(json!({ + "contexts": [{ "name": "adp.requests", "tagset": ["b:2", "a:1"], "kind": "count" }], + })) + .expect("deserialize agent LaneView"); + let adp: LaneView = serde_json::from_value(json!({ + "contexts": [{ "name": "adp.requests", "tagset": ["a:1", "b:2"], "kind": "count" }], + })) + .expect("deserialize adp LaneView"); + + assert_eq!( + agent.curves().keys().collect::>(), + adp.curves().keys().collect::>() + ); } #[test] - fn diverging_tags_each_member_with_its_lane() { - // A context only the Agent emitted is agent-side; one only ADP emitted is adp-side. Age is the - // intake clock minus first_seen. - let agent = lane(10, vec![captured("agent.only", "count", &["host:h"], 4)]); - let adp = lane(10, vec![captured("adp.only", "gauge", &["host:h"], 7)]); - - let mut members = Difference::between(&agent, &adp).diverging(); - members.sort_by(|a, b| a.context.name.cmp(&b.context.name)); - - assert!(matches!(members[0].lane, Lane::Adp)); - assert_eq!(members[0].context.name, "adp.only"); - assert_eq!(members[0].age_secs, 3); - assert!(matches!(members[1].lane, Lane::Agent)); - assert_eq!(members[1].context.name, "agent.only"); - assert_eq!(members[1].age_secs, 6); - } - - #[test] - fn delayed_counts_members_past_the_budget() { - // A member is delayed once the intake clock has moved more than the budget past first_seen. - let present = vec![captured("adp.requests", "count", &["host:h"], 100)]; - let fresh = Difference::between(&lane(140, present.clone()), &lane(140, vec![])); - let stale = Difference::between(&lane(200, present), &lane(200, vec![])); + fn deserializes_the_flat_wire_shape() -> Result<(), serde_json::Error> { + let view: LaneView = serde_json::from_value(json!({ + "contexts": [{ "name": "adp.requests", "tagset": ["host:h"], "kind": "count" }], + }))?; - assert_eq!(fresh.delayed(Duration::from_secs(60)), 0); - assert_eq!(stale.delayed(Duration::from_secs(60)), 1); + assert_eq!(view.contexts.len(), 1); + Ok(()) } + // The curve DTO the intake serves deserializes into the consumer mirror: a context carrying a mix + // of scalar and sketch buckets, one flagged as a within-lane conflict. #[test] - fn deserializes_the_flat_wire_shape() -> Result<(), serde_json::Error> { + fn deserializes_the_curve_dto() -> Result<(), serde_json::Error> { let view: LaneView = serde_json::from_value(json!({ - "now": 2000, - "contexts": [{ "name": "adp.requests", "tagset": ["host:h"], "kind": "count", "first_seen": 1000 }], + "contexts": [{ + "name": "app.dist", + "tagset": ["env:prod"], + "kind": "sketch", + "buckets": [ + { "bucket_start": 100, "value": { "Scalar": 1.5 }, "conflict": false }, + { "bucket_start": 110, "value": { "Sketch": { + "count": 5, "sum": 30.0, "min": 1.0, "max": 9.0, "bins": [[3, 2], [5, 3]] + } }, "conflict": true }, + ], + }], }))?; - assert_eq!(view.now, 2000); - assert_eq!(view.contexts.len(), 1); + let captured = &view.contexts[0]; + assert_eq!(captured.buckets.len(), 2); + assert_eq!( + captured.buckets[0], + Bucket { + bucket_start: 100, + value: BucketValue::Scalar(1.5), + conflict: false, + } + ); + assert!(captured.buckets[1].conflict); + assert!(matches!(captured.buckets[1].value, BucketValue::Sketch(_))); Ok(()) } } diff --git a/test/antithesis/scenarios/differential/src/decision.rs b/test/antithesis/scenarios/differential/src/decision.rs new file mode 100644 index 00000000000..0a11d528dda --- /dev/null +++ b/test/antithesis/scenarios/differential/src/decision.rs @@ -0,0 +1,394 @@ +//! The online decision layer: the clipped signed CUSUM, the per-context verdict, the triage extent, +//! and the single union defect count the always-assertion reads. +//! +//! Every accumulator here is a running/anytime quantity with an absolute bound. Nothing is divided by +//! panel length: the O(1/N) dilution that sank DTW is banned. + +use std::cmp::Ordering; + +use harness::{CUSUM_H, CUSUM_K_BASE_SCALAR, CUSUM_K_BASE_SKETCH, CUSUM_PHI_MAX, K_MERGE, SAKOE_CHIBA_BAND}; + +use crate::contexts::{Bucket, BucketValue, SketchValue}; +use crate::frechet::{Frontier, Trip}; +use crate::ground::{bins_total, quantile_from_bins, Band, RuleContext, RuleSet}; + +/// The displayed quantiles the sketch channels take a signed CUSUM over. Read off the shared log grid, +/// never a probability-axis average, so no channel is the forbidden 1-Wasserstein dilution. +const DISPLAYED_QUANTILES: [f64; 4] = [0.5, 0.9, 0.95, 0.99]; + +/// The clipped signed CUSUM over one channel's per-column normalized gap. Both arms are +/// running/anytime quantities that fire at an absolute threshold `h`, never length-normalized. +#[derive(Clone, Copy, Debug)] +pub struct Cusum { + c_plus: f64, + c_minus: f64, + h: f64, +} + +impl Cusum { + /// A CUSUM with firing threshold `h`, both arms zeroed. + #[must_use] + pub fn new(h: f64) -> Self { + Self { + c_plus: 0.0, + c_minus: 0.0, + h, + } + } + + /// Folds one column: the clipped normalized gap `u = clip(gap/b, -1, 1)` with trend-aware slack + /// `k`. `C+ = max(0, C+ + u - k)`, `C- = max(0, C- - u - k)`. + pub fn update(&mut self, u: f64, k: f64) { + self.c_plus = (self.c_plus + u - k).max(0.0); + self.c_minus = (self.c_minus - u - k).max(0.0); + } + + /// Whether either arm has crossed the firing threshold `h`. + #[must_use] + pub fn fired(&self) -> bool { + self.c_plus > self.h || self.c_minus > self.h + } + + /// The larger of the two arms, for triage. + #[must_use] + pub fn peak(&self) -> f64 { + self.c_plus.max(self.c_minus) + } +} + +/// Continuous triage magnitude for a context, sorted on `worst_gap` descending. Rides as metadata +/// only; NEVER feeds the verdict. +#[derive(Clone, Copy, Debug, Default, PartialEq)] +pub struct Extent { + /// The longest run of consecutive over-band columns. + pub run_length: u64, + /// The largest single over-band gap magnitude seen. + pub worst_gap: f64, + /// The running sum of over-band gap magnitudes. + pub accumulated_gap: f64, +} + +/// The per-context verdict: whether the context is a defect (any guard-survived sup trip, any CUSUM +/// arm, any absolute-law violation, or a past-horizon presence mismatch) and its triage extent. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct ContextVerdict { + /// Whether the context counts as a defect. The union of every trip source, once per context. + pub tripped: bool, + /// The mechanisms that fired, for triage. + pub trips: Vec, + /// Continuous triage magnitude; never decides `tripped`. + pub extent: Extent, +} + +/// The single always-assertion quantity: the number of contexts that tripped. Counts a context once +/// regardless of how many mechanisms fired, and is never length-normalized. +#[must_use] +pub fn defect_count(verdicts: &[ContextVerdict]) -> usize { + verdicts.iter().filter(|verdict| verdict.tripped).count() +} + +/// The trip a surviving sup fold or present/absent bottom carries, forwarded from the aligner. +/// The value-channel CUSUM trip. +const VALUE_CUSUM: &str = "value_cusum"; +/// The per-quantile sketch CUSUM trip. +const QUANTILE_CUSUM: &str = "quantile_cusum"; +/// A context present on one lane and absent on the other past what the band absorbs. +const PRESENT_ABSENT: &str = "present_absent"; +/// A malformed align: the two lanes disagree on a bucket's kind, which is a real divergence. +const ALIGN_ERROR: &str = "align_error"; + +/// The provisional launch parameters for the online decision: the ddsketch-derived value band, the +/// banded-Frechet aligner, and the CUSUM slacks. All are baked defaults, not calibrated values. +#[derive(Clone, Copy, Debug)] +pub struct DecisionParams { + /// The per-value band `max(alpha*|v|, floor)` derived from the ddsketch API. + pub band: Band, + /// The banded-Frechet aligner: the single coupling the sup verdict reads. + pub frontier: Frontier, + /// The CUSUM firing threshold `h`. + pub h: f64, + /// The scalar channels' base slack `k_base`. + pub k_base_scalar: f64, + /// The sketch mean channel's base slack `k_base`. + pub k_base_sketch: f64, + /// The per-quantile channels' merge slack `k_merge`. + pub k_merge: f64, + /// The trend-aware slack coefficient `phi_max`. + pub phi_max: f64, +} + +impl DecisionParams { + /// The launch defaults, reading the provisional harness constants. + #[must_use] + pub fn launch() -> Self { + Self { + band: Band::ddsketch(), + frontier: Frontier::new(SAKOE_CHIBA_BAND), + h: CUSUM_H, + k_base_scalar: CUSUM_K_BASE_SCALAR, + k_base_sketch: CUSUM_K_BASE_SKETCH, + k_merge: K_MERGE, + phi_max: CUSUM_PHI_MAX, + } + } +} + +/// Evaluates one context's two lane curves into a [`ContextVerdict`]: the union of every trip source +/// (the guard-survived sup fold, the value and per-quantile CUSUM arms, the present/absent bottom, and +/// the per-lane absolute-law detectors) plus the triage extent that never feeds the verdict. +/// +/// `reference` is the normative Agent curve, `sut` the ADP curve. The sup fold reads the single banded +/// coupling the aligner computes. The CUSUM channels fold the per-column signed gap of the positional +/// coupling: at the shared flush interval the two lanes settle equal-length and column-aligned, so the +/// positional gap is the coupled gap; a within-band slide is absorbed by the sup's band and is a +/// transient the anytime CUSUM's slack rides out. +#[must_use] +pub fn evaluate(reference: &[Bucket], sut: &[Bucket], rules: &RuleSet, params: &DecisionParams) -> ContextVerdict { + if reference.is_empty() || sut.is_empty() { + return presence_verdict(reference.len().max(sut.len()), params.frontier.band()); + } + + let mut trips = Vec::new(); + + // The single aligner: the guard-survived sup fold and the present/absent bottom. Every other + // mechanism reads the positional coupling, never a second alignment. + match params.frontier.align(reference, sut, params.band, rules) { + Ok(verdict) => trips.extend(verdict.trips), + Err(_) => trips.push(trip(ALIGN_ERROR)), + } + + // The value channel: counter/rate summed value, gauge last value, sketch mean, each projected to a + // scalar track. The clipped signed CUSUM subsumes the k=3 debounce and extends to sub-band bias. + let reference_track = scalar_track(reference); + let sut_track = scalar_track(sut); + let is_sketch = reference + .iter() + .chain(sut) + .any(|b| matches!(b.value, BucketValue::Sketch(_))); + let k_base = if is_sketch { + params.k_base_sketch + } else { + params.k_base_scalar + }; + if cusum_fires( + &reference_track, + &sut_track, + params.band, + params.h, + k_base, + params.phi_max, + rules, + params.frontier.band(), + ) { + trips.push(trip(VALUE_CUSUM)); + } + + let extent = extent(&reference_track, &sut_track, params.band); + + if is_sketch { + // The sketch quantile channels: a signed CUSUM per displayed quantile off the shared grid, its + // slack the merge tolerance. The strict count gate rides in the aligner's sup as an infinite + // leash, so no separate count channel repeats it. + let empty_rules = RuleSet::new(); + for &q in &DISPLAYED_QUANTILES { + let reference_q = quantile_track(reference, q); + let sut_q = quantile_track(sut, q); + if cusum_fires( + &reference_q, + &sut_q, + params.band, + params.h, + params.k_merge, + params.phi_max, + &empty_rules, + params.frontier.band(), + ) { + trips.push(trip(QUANTILE_CUSUM)); + break; + } + } + // The per-lane absolute laws: each lane's sketch must be internally consistent regardless of + // the other lane. A violation is add-only, never suppressed. + if let Some(name) = first_absolute_law_violation(reference, sut, params.band) { + trips.push(trip(&format!("absolute_law:{name}"))); + } + } + + ContextVerdict { + tripped: !trips.is_empty(), + trips, + extent, + } +} + +/// A context on one lane only: an infinite-leash present/absent bottom once the present curve carries +/// more buckets than the band can absorb, a transient one-bucket absence otherwise. The extent's +/// infinite worst gap sorts it first in triage. +fn presence_verdict(present_len: usize, band: usize) -> ContextVerdict { + if present_len > band { + ContextVerdict { + tripped: true, + trips: vec![trip(PRESENT_ABSENT)], + extent: Extent { + run_length: present_len as u64, + worst_gap: f64::INFINITY, + accumulated_gap: f64::INFINITY, + }, + } + } else { + ContextVerdict::default() + } +} + +/// Folds the clipped signed CUSUM over the two scalar tracks' positional gaps and reports whether either +/// arm crossed `h`. `u_t = clip(gap_t/b_t, -1, 1)`; the slack is trend-aware, `k_t = k_base + phi_max * +/// |a_t - a_{t-1}| / b_t`, so a legitimate reference transition widens the slack rather than tripping. +/// A guard-suppressed over-band column contributes zero, keeping the fold monotone green-ward. A +/// non-finite column (an unknown kind's placeholder) contributes zero and breaks the trend. +fn cusum_fires( + reference: &[f64], sut: &[f64], band: Band, h: f64, k_base: f64, phi_max: f64, rules: &RuleSet, window: usize, +) -> bool { + let mut cusum = Cusum::new(h); + let mut previous: Option = None; + for t in 0..reference.len().min(sut.len()) { + let (a, s) = (reference[t], sut[t]); + if !a.is_finite() || !s.is_finite() { + cusum.update(0.0, k_base); + previous = None; + continue; + } + let b = band.at(a.abs().max(s.abs())); + let gap = s - a; + let suppressed = gap.abs() > b + && rules.suppressed(&RuleContext { + reference, + sut, + ref_index: t, + sut_index: t, + band, + window, + }); + let u = if suppressed { 0.0 } else { (gap / b).clamp(-1.0, 1.0) }; + let trend = previous.map_or(0.0, |p| phi_max * (a - p).abs() / b); + cusum.update(u, k_base + trend); + if cusum.fired() { + return true; + } + previous = Some(a); + } + cusum.fired() +} + +/// The triage extent over the positional coupling's raw gaps: the longest over-band run, the largest +/// single gap magnitude, and the summed over-band magnitude. Computed over every finite column +/// regardless of guard suppression, so it reflects the raw divergence for triage while never deciding +/// the verdict. +fn extent(reference: &[f64], sut: &[f64], band: Band) -> Extent { + let mut extent = Extent::default(); + let mut run = 0u64; + for t in 0..reference.len().min(sut.len()) { + let (a, s) = (reference[t], sut[t]); + if !a.is_finite() || !s.is_finite() { + run = 0; + continue; + } + let gap = (s - a).abs(); + extent.worst_gap = extent.worst_gap.max(gap); + if gap > band.at(a.abs().max(s.abs())) { + run += 1; + extent.run_length = extent.run_length.max(run); + extent.accumulated_gap += gap; + } else { + run = 0; + } + } + extent +} + +/// Each bucket projected to the scalar track the value channel and the extent read: a scalar's value, +/// a sketch's mean. An empty sketch folds to zero. +fn scalar_track(buckets: &[Bucket]) -> Vec { + buckets + .iter() + .map(|bucket| match &bucket.value { + BucketValue::Scalar(value) => *value, + BucketValue::Sketch(sketch) if sketch.count > 0 => sketch.sum / sketch.count as f64, + BucketValue::Sketch(_) => 0.0, + }) + .collect() +} + +/// Each bucket's displayed quantile `q` read off the shared log grid. A scalar or empty sketch column +/// carries a non-finite placeholder the CUSUM treats as a break. +fn quantile_track(buckets: &[Bucket], q: f64) -> Vec { + buckets + .iter() + .map(|bucket| match &bucket.value { + BucketValue::Sketch(sketch) => quantile_from_bins(&sketch.bins, q).unwrap_or(f64::NAN), + BucketValue::Scalar(_) => f64::NAN, + }) + .collect() +} + +/// The first per-lane absolute-law violation on either curve's sketch buckets, or `None` when every +/// sketch is internally consistent. Each law is a within-lane invariant a faithful decode always +/// satisfies, so a violation is a real decode or aggregation break. +fn first_absolute_law_violation(reference: &[Bucket], sut: &[Bucket], band: Band) -> Option<&'static str> { + reference + .iter() + .chain(sut) + .filter_map(|bucket| match &bucket.value { + BucketValue::Sketch(sketch) => sketch_law_violation(sketch, band), + BucketValue::Scalar(_) => None, + }) + .next() +} + +/// The first violated within-lane sketch law: count non-negative, count equal to its bins' total, +/// min at or below max, mean within the band-widened `[min, max]`, and quantile monotonicity across the +/// displayed quantiles. +fn sketch_law_violation(sketch: &SketchValue, band: Band) -> Option<&'static str> { + if sketch.count < 0 { + return Some("count_nonneg"); + } + if bins_total(&sketch.bins) != sketch.count as u64 { + return Some("count_equals_bins"); + } + if sketch.count == 0 { + return None; + } + if sketch.min > sketch.max { + return Some("min_le_max"); + } + let mean = sketch.sum / sketch.count as f64; + if mean < sketch.min - band.at(sketch.min.abs()) || mean > sketch.max + band.at(sketch.max.abs()) { + return Some("mean_in_range"); + } + let mut previous = f64::NEG_INFINITY; + for &q in &DISPLAYED_QUANTILES { + if let Some(value) = quantile_from_bins(&sketch.bins, q) { + if value + band.at(value.abs()) < previous { + return Some("quantile_monotone"); + } + previous = value; + } + } + None +} + +/// A trip naming the mechanism that fired. +fn trip(rule: &str) -> Trip { + Trip { rule: rule.to_string() } +} + +/// Orders triage entries by continuous worst gap descending and bounds them to `limit`. The ordering is +/// triage metadata only; it never decides the verdict. +#[must_use] +pub fn bounded_triage(mut entries: Vec, worst_gap: impl Fn(&T) -> f64, limit: usize) -> Vec { + entries.sort_by(|a, b| worst_gap(b).partial_cmp(&worst_gap(a)).unwrap_or(Ordering::Equal)); + entries.truncate(limit); + entries +} + +#[cfg(test)] +mod tests; diff --git a/test/antithesis/scenarios/differential/src/decision/tests.rs b/test/antithesis/scenarios/differential/src/decision/tests.rs new file mode 100644 index 00000000000..2afa6c91199 --- /dev/null +++ b/test/antithesis/scenarios/differential/src/decision/tests.rs @@ -0,0 +1,178 @@ +//! Tests for the online decision layer: the clipped signed CUSUM's debounce, the sub-band-bias repair +//! (C1), the triage extent that never feeds the verdict, the per-lane absolute-law detectors, and the +//! bounded sorted triage sample. + +use super::{bounded_triage, defect_count, evaluate, ContextVerdict, DecisionParams, Extent}; +use crate::contexts::{Bucket, BucketValue, SketchValue}; +use crate::frechet::Trip; +use crate::ground::RuleSet; + +/// A scalar curve: one bucket per value, ten wire-seconds apart in bucket-start order. +fn scalar_curve(values: &[f64]) -> Vec { + values + .iter() + .enumerate() + .map(|(index, &value)| Bucket { + bucket_start: index as u64 * 10, + value: BucketValue::Scalar(value), + conflict: false, + }) + .collect() +} + +/// A single-bucket sketch curve carrying the given summary and one log-grid bin. +fn sketch_curve(count: i64, sum: f64, min: f64, max: f64, bins: Vec<(i32, u32)>) -> Vec { + vec![Bucket { + bucket_start: 0, + value: BucketValue::Sketch(SketchValue { + count, + sum, + min, + max, + bins, + }), + conflict: false, + }] +} + +/// Whether any trip carries the named mechanism. +fn has_trip(verdict: &ContextVerdict, rule: &str) -> bool { + verdict.trips.iter().any(|trip| trip.rule == rule) +} + +#[test] +fn lone_straddle_is_suppressed() { + // A single column deviates by just under its band; the rest are identical. The sup does not fire + // (sub-band) and the CUSUM's single excursion decays below `h`: the debounce the CUSUM subsumes. + let params = DecisionParams::launch(); + let base = 1000.0; + let band = params.band.at(base); + let mut sut_values = vec![base; 12]; + sut_values[5] = base + 0.99 * band; + let verdict = evaluate( + &scalar_curve(&[base; 12]), + &scalar_curve(&sut_values), + &RuleSet::default_rules(), + ¶ms, + ); + assert!( + !verdict.tripped, + "a lone sub-band straddle must be debounced away: {:?}", + verdict.trips + ); +} + +#[test] +fn sustained_divergence_fires() { + // The SUT sits far over band on every column. The sup fold trips at the first over-band column. + let params = DecisionParams::launch(); + let verdict = evaluate( + &scalar_curve(&[100.0; 8]), + &scalar_curve(&[200.0; 8]), + &RuleSet::default_rules(), + ¶ms, + ); + assert!(verdict.tripped); + assert!( + has_trip(&verdict, "sup_fold"), + "sustained over-band divergence trips the sup: {:?}", + verdict.trips + ); +} + +#[test] +fn sustained_sub_band_offset_fires() { + // The C1 repair: every column carries the same sub-band bias, so the sup never fires, but the + // signed CUSUM accumulates the consistent drift past `h`. A length-normalized accumulator would + // dilute this to nothing; the anytime CUSUM does not. + let params = DecisionParams::launch(); + let base = 1000.0; + let bias = 0.5 * params.band.at(base); + let verdict = evaluate( + &scalar_curve(&[base; 16]), + &scalar_curve(&[base + bias; 16]), + &RuleSet::default_rules(), + ¶ms, + ); + assert!(verdict.tripped, "a sustained sub-band offset must fire the CUSUM"); + assert!( + has_trip(&verdict, "value_cusum"), + "the sub-band offset fires the value CUSUM: {:?}", + verdict.trips + ); + assert!( + !has_trip(&verdict, "sup_fold"), + "no single column is over band, so the sup stays green" + ); +} + +#[test] +fn extent_never_flips_the_verdict() { + // A counter increment that slid one flush over: the reference is zero at the coupled column and + // carries the SUT's value one bucket on. The missing-counter-zero rule suppresses the sup and the + // CUSUM, so the context is green, yet the triage extent still records the large raw gap. Extent is + // metadata; it must never turn the verdict red. + let params = DecisionParams::launch(); + let reference = scalar_curve(&[0.0, 1000.0, 0.0, 0.0]); + let sut = scalar_curve(&[1000.0, 0.0, 0.0, 0.0]); + let verdict = evaluate(&reference, &sut, &RuleSet::default_rules(), ¶ms); + assert!( + !verdict.tripped, + "the phase-shifted increment is benign: {:?}", + verdict.trips + ); + assert!( + verdict.extent.worst_gap >= 1000.0, + "extent still records the raw gap: {:?}", + verdict.extent + ); +} + +#[test] +fn absolute_law_violation_reds() { + // Both lanes carry identical bins and matching means, so the aligner and the CUSUMs stay green. + // The SUT's count field disagrees with its own bins, an internal-consistency break the per-lane + // absolute-law detector reds regardless of cross-lane agreement. + let params = DecisionParams::launch(); + let reference = sketch_curve(5, 50.0, 10.0, 10.0, vec![(10, 5)]); + let sut = sketch_curve(99, 990.0, 10.0, 10.0, vec![(10, 5)]); + let verdict = evaluate(&reference, &sut, &RuleSet::default_rules(), ¶ms); + assert!(verdict.tripped); + assert!( + verdict.trips.iter().any(|t| t.rule.starts_with("absolute_law")), + "the count/bins mismatch reds via an absolute law: {:?}", + verdict.trips + ); +} + +#[test] +fn defect_count_ignores_extent() { + // A green context with a huge triage extent and a red context with none: the count follows only + // `tripped`, never the extent magnitude. + let green = ContextVerdict { + tripped: false, + trips: Vec::new(), + extent: Extent { + run_length: 9, + worst_gap: 1.0e9, + accumulated_gap: 1.0e12, + }, + }; + let red = ContextVerdict { + tripped: true, + trips: vec![Trip { + rule: "value_cusum".to_string(), + }], + extent: Extent::default(), + }; + assert_eq!(defect_count(&[green, red]), 1); +} + +#[test] +fn triage_sample_is_bounded_and_sorted() { + // The triage sample orders by continuous worst_gap descending and bounds to the limit. + let entries: Vec = vec![3.0, 100.0, 7.0, 42.0, 1.0]; + let sample = bounded_triage(entries, |gap| *gap, 3); + assert_eq!(sample.len(), 3, "bounded to the limit"); + assert_eq!(sample, vec![100.0, 42.0, 7.0], "sorted by worst_gap descending"); +} diff --git a/test/antithesis/scenarios/differential/src/frechet.rs b/test/antithesis/scenarios/differential/src/frechet.rs new file mode 100644 index 00000000000..74f3080dc7f --- /dev/null +++ b/test/antithesis/scenarios/differential/src/frechet.rs @@ -0,0 +1,333 @@ +//! The banded discrete-Frechet aligner: sup/bottleneck fold with the J1 refinement. +//! +//! Per context per settled column the aligner computes ONE monotone in-band coupling pi over the +//! Sakoe-Chiba band and folds the sup verdict, applying the [`RuleSet`] to each over-band column. Every +//! downstream mechanism reads this single coupling; nothing runs a second aligner. + +use std::fmt; + +use crate::contexts::{Bucket, BucketValue}; +use crate::ground::{d_v, Band, GroundError, Leash, RuleContext, RuleSet}; + +/// The mechanism name a surviving sup-fold trip carries. +const SUP_FOLD: &str = "sup_fold"; + +/// The mechanism name a present/absent bottom trip carries. +const PRESENT_ABSENT_BOTTOM: &str = "present_absent_bottom"; + +/// A surviving trip on a context's curve: an equivalence violation the guard rules did not suppress. +/// `rule` names the mechanism that fired (the sup fold, a CUSUM arm, or an absolute-law detector). +#[derive(Clone, Debug, PartialEq)] +pub struct Trip { + /// The name of the mechanism that produced the trip, surfaced in triage details. + pub rule: String, +} + +/// The aligner's verdict for one curve: the trips that survived the guard rules, and the worst leash +/// the sup fold reached along the chosen coupling. +/// +/// `worst_leash` is the single coupling reading the decision layer needs: the sup gap and the band that +/// admitted it. It is an anytime quantity, so it stays identical however long the curve grows. The full +/// per-column coupling is deliberately NOT carried here: it would make the verdict length dependent and +/// break the equality the length-independence property rests on. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct CurveVerdict { + /// The surviving trips, empty when the two curves are visually equivalent. + pub trips: Vec, + /// The worst leash the chosen coupling reached: the sup gap and its band. `None` before any column + /// is folded. + pub worst_leash: Option, +} + +impl CurveVerdict { + /// Whether the curve tripped. + #[must_use] + pub fn tripped(&self) -> bool { + !self.trips.is_empty() + } +} + +/// Why the aligner could not produce a [`CurveVerdict`]. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum AlignerError { + /// A curve has no settled buckets to align. + EmptyCurve, + /// A ground-distance computation failed. + Ground(GroundError), + /// The stub has no implementation yet. + Unimplemented, +} + +impl fmt::Display for AlignerError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + AlignerError::EmptyCurve => write!(f, "a curve has no settled buckets to align"), + AlignerError::Ground(error) => write!(f, "ground metric failed: {error}"), + AlignerError::Unimplemented => write!(f, "banded Frechet aligner is not yet implemented"), + } + } +} + +impl std::error::Error for AlignerError {} + +impl From for AlignerError { + fn from(error: GroundError) -> Self { + AlignerError::Ground(error) + } +} + +/// The banded discrete-Frechet frontier: the in-band monotone coupling the aligner computes once per +/// context per settled column. Holds the Sakoe-Chiba band half-width in buckets. +#[derive(Clone, Copy, Debug)] +pub struct Frontier { + band: usize, +} + +impl Frontier { + /// A frontier with the given Sakoe-Chiba band half-width, in buckets. + #[must_use] + pub fn new(band: usize) -> Self { + Self { band } + } + + /// The Sakoe-Chiba band half-width in buckets. + #[must_use] + pub fn band(self) -> usize { + self.band + } + + /// Aligns the reference and SUT curves under the band, folds the sup verdict, and applies the rule + /// set and the J1 envelope guard to each over-band column. Reads the single coupling; never runs a + /// second aligner. + /// + /// The fold is a banded discrete-Frechet sup over the ground metric [`d_v`]. Each coupled column + /// carries the effective normalized excess `|gap|/band`, zeroed green-ward when the column is within + /// band or a guard suppresses it. The DP minimizes the worst such excess over every monotone + /// in-band coupling; the coupling wins green whenever one exists that stays acceptable everywhere. + /// The frontier holds only the `2B+1` band cells of two rows, so the fold is `O((n+m)*B)` in time + /// and `O(B)` in space, never the `O(n*m)` blow-up. + /// + /// A length mismatch wider than the band is a present/absent bottom: a context on one lane the other + /// never carried, past what a benign flush-phase slide could absorb. It trips at an infinite leash. + /// A shorter mismatch is a transient the band absorbs. + /// + /// # Errors + /// + /// Returns [`AlignerError::EmptyCurve`] when a curve is empty and [`AlignerError::Ground`] when the + /// ground metric fails on a coupled column. + pub fn align( + &self, reference: &[Bucket], sut: &[Bucket], value_band: Band, rules: &RuleSet, + ) -> Result { + if reference.is_empty() || sut.is_empty() { + return Err(AlignerError::EmptyCurve); + } + let n = reference.len(); + let m = sut.len(); + + // A length mismatch the band cannot absorb is a present/absent bottom: the corner is off band, + // so no monotone banded coupling reaches it. Trip at an infinite leash before folding. + if n.abs_diff(m) > self.band { + return Ok(CurveVerdict { + trips: vec![Trip { + rule: PRESENT_ABSENT_BOTTOM.to_string(), + }], + worst_leash: Some(Leash { + gap: f64::INFINITY, + band: value_band.floor, + }), + }); + } + + // The scalar tracks the guard rules read. Sketches fold to their mean so the guard has a + // comparable level on every kind. + let reference_track: Vec = reference.iter().map(|b| project(&b.value)).collect(); + let sut_track: Vec = sut.iter().map(|b| project(&b.value)).collect(); + + // The banded min-max frontier: two rows, each holding only its `2B+1` in-band cells. + let mut previous: Vec> = Vec::new(); + let mut previous_low = 0usize; + for (i, reference_bucket) in reference.iter().enumerate() { + let low = i.saturating_sub(self.band); + let high = (i + self.band).min(m - 1); + let mut current: Vec> = vec![None; high - low + 1]; + for (j, sut_bucket) in sut.iter().enumerate().take(high + 1).skip(low) { + let leash = d_v(&reference_bucket.value, &sut_bucket.value, value_band)?; + let excess = self.effective_excess(leash, &reference_track, &sut_track, i, j, value_band, rules); + + // The best predecessor is the reachable neighbor with the smallest worst excess so far. + let mut best: Option = None; + if i > 0 { + best = pick(best, cell_at(&previous, previous_low, j)); + if j > 0 { + best = pick(best, cell_at(&previous, previous_low, j - 1)); + } + } + if j > low { + best = pick(best, current[j - 1 - low]); + } + + current[j - low] = match best { + // The corner (0,0) seeds the fold; any other cell with no in-band predecessor is + // unreachable and stays `None`. + None if i == 0 && j == 0 => Some(Cell { cost: excess, leash }), + None => None, + // min-max: the path cost is the larger of this column's excess and the best + // predecessor's, and the worst leash rides with whichever term won. + Some(predecessor) if excess >= predecessor.cost => Some(Cell { cost: excess, leash }), + Some(predecessor) => Some(predecessor), + }; + } + previous = current; + previous_low = low; + } + + let corner = cell_at(&previous, previous_low, m - 1); + match corner { + // A monotone banded coupling exists; it trips only if its worst column stays over band after + // the guards. The sup fold fires once per curve, so the verdict is length independent. + Some(cell) => { + let mut trips = Vec::new(); + if cell.cost > 1.0 { + trips.push(Trip { + rule: SUP_FOLD.to_string(), + }); + } + Ok(CurveVerdict { + trips, + worst_leash: Some(cell.leash), + }) + } + // No coupling reached the corner within the band: a present/absent bottom the width check + // above did not already catch. + None => Ok(CurveVerdict { + trips: vec![Trip { + rule: PRESENT_ABSENT_BOTTOM.to_string(), + }], + worst_leash: Some(Leash { + gap: f64::INFINITY, + band: value_band.floor, + }), + }), + } + } + + /// The column's effective normalized excess: `|gap|/band` when it is over band and neither the rule + /// set nor the J1 envelope guard suppresses it, and `0` otherwise. Guards are monotone green-ward: + /// they only ever pull a column's excess to zero, never raise it. + fn effective_excess( + &self, leash: Leash, reference_track: &[f64], sut_track: &[f64], reference_index: usize, sut_index: usize, + value_band: Band, rules: &RuleSet, + ) -> f64 { + if !leash.over_band() { + return 0.0; + } + let ctx = RuleContext { + reference: reference_track, + sut: sut_track, + ref_index: reference_index, + sut_index, + band: value_band, + window: self.band, + }; + if rules.suppressed(&ctx) || j1_envelope_suppresses(&ctx) { + return 0.0; + } + leash.gap.abs() / leash.band + } +} + +/// One cell of the banded frontier: the min-max path cost to here and the worst leash that path met. +#[derive(Clone, Copy)] +struct Cell { + cost: f64, + leash: Leash, +} + +/// The lower-cost of two candidate cells, treating an absent cell as unreachable. +fn pick(current: Option, candidate: Option) -> Option { + match (current, candidate) { + (Some(a), Some(b)) => Some(if b.cost < a.cost { b } else { a }), + (Some(a), None) => Some(a), + (None, other) => other, + } +} + +/// The band cell at column `j` of a stored row that begins at `low`, or `None` when `j` is off band. +fn cell_at(row: &[Option], low: usize, j: usize) -> Option { + j.checked_sub(low).and_then(|offset| row.get(offset).copied().flatten()) +} + +/// The scalar level a bucket value presents to the guard: the value itself for a scalar, the sketch +/// mean for a sketch. An empty sketch folds to zero. +fn project(value: &BucketValue) -> f64 { + match value { + BucketValue::Scalar(scalar) => *scalar, + BucketValue::Sketch(sketch) if sketch.count != 0 => sketch.sum / sketch.count as f64, + BucketValue::Sketch(_) => 0.0, + } +} + +/// The J1 matched-jump plus envelope-containment guard, monotone green-ward. It suppresses an over-band +/// trip at a level transition only when both lanes jump the same sign and matched magnitude, the pre and +/// post levels agree across lanes, and the whole `+/-window` on both lanes lies inside the two levels' +/// band-widened envelope. A jump only one lane makes, a mismatched magnitude, or an interior spike that +/// escapes the envelope all leave the trip standing. +fn j1_envelope_suppresses(ctx: &RuleContext<'_>) -> bool { + let (i, j) = (ctx.ref_index, ctx.sut_index); + if i == 0 || j == 0 { + return false; + } + let (Some(&r0), Some(&r1)) = (ctx.reference.get(i - 1), ctx.reference.get(i)) else { + return false; + }; + let (Some(&s0), Some(&s1)) = (ctx.sut.get(j - 1), ctx.sut.get(j)) else { + return false; + }; + if ![r0, r1, s0, s1].iter().all(|level| level.is_finite()) { + return false; + } + + // Both lanes must jump the same nonzero sign. + let delta_reference = r1 - r0; + let delta_sut = s1 - s0; + if delta_reference == 0.0 || delta_sut == 0.0 || (delta_reference > 0.0) != (delta_sut > 0.0) { + return false; + } + + // Matched magnitude: the two jumps agree within the band at the larger jump. + let jump = delta_reference.abs().max(delta_sut.abs()); + if (delta_reference - delta_sut).abs() > ctx.band.at(jump) { + return false; + } + + // Pre and post levels agree across lanes. + if (r0 - s0).abs() > ctx.band.at(r0.abs().max(s0.abs())) { + return false; + } + if (r1 - s1).abs() > ctx.band.at(r1.abs().max(s1.abs())) { + return false; + } + + // Envelope containment: every bucket in the `+/-window` on both lanes lies inside the two levels + // widened by the band. An interior spike breaches the ceiling and defeats the guard. + let low_level = r0.min(r1).min(s0).min(s1); + let high_level = r0.max(r1).max(s0).max(s1); + let margin = ctx.band.at(high_level.abs().max(low_level.abs())); + let (floor, ceiling) = (low_level - margin, high_level + margin); + window_contained(ctx.reference, i, ctx.window, floor, ceiling) + && window_contained(ctx.sut, j, ctx.window, floor, ceiling) +} + +/// Whether every finite bucket in the `+/-window` around `center` on `track` lies in `[floor, ceiling]`. +/// A non-finite bucket is a presence concern handled elsewhere, so it does not defeat containment. +fn window_contained(track: &[f64], center: usize, window: usize, floor: f64, ceiling: f64) -> bool { + let low = center.saturating_sub(window); + let high = (center + window).min(track.len().saturating_sub(1)); + (low..=high).all(|index| { + let value = track[index]; + !value.is_finite() || (floor..=ceiling).contains(&value) + }) +} + +#[cfg(test)] +mod tests; diff --git a/test/antithesis/scenarios/differential/src/frechet/tests.rs b/test/antithesis/scenarios/differential/src/frechet/tests.rs new file mode 100644 index 00000000000..8a98c566399 --- /dev/null +++ b/test/antithesis/scenarios/differential/src/frechet/tests.rs @@ -0,0 +1,171 @@ +//! Tests for the banded discrete-Frechet aligner: the sup fold, the J1 envelope guard, the +//! present/absent bottom trip, and the length-independence of the anytime verdict. + +use harness::SAKOE_CHIBA_BAND; +use proptest::prelude::*; + +use super::{AlignerError, Frontier}; +use crate::contexts::{Bucket, BucketValue}; +use crate::ground::{Band, RuleSet}; + +/// A scalar curve: one bucket per value, ten wire-seconds apart in bucket-start order. +fn scalar_curve(values: &[f64]) -> Vec { + values + .iter() + .enumerate() + .map(|(index, &value)| Bucket { + bucket_start: index as u64 * 10, + value: BucketValue::Scalar(value), + conflict: false, + }) + .collect() +} + +/// The launch aligner: the provisional Sakoe-Chiba band, the ddsketch-derived value band, no operator +/// rules so the sup fold and the baked-in J1 guard alone decide. +fn aligner() -> Frontier { + Frontier::new(SAKOE_CHIBA_BAND) +} + +#[test] +fn an_empty_curve_is_an_error() { + let rules = RuleSet::new(); + let present = scalar_curve(&[1.0]); + assert!(matches!( + aligner().align(&[], &present, Band::ddsketch(), &rules), + Err(AlignerError::EmptyCurve) + )); + assert!(matches!( + aligner().align(&present, &[], Band::ddsketch(), &rules), + Err(AlignerError::EmptyCurve) + )); +} + +#[test] +fn identical_curves_do_not_trip() { + let rules = RuleSet::new(); + let curve = scalar_curve(&[1.0, 2.0, 3.0, 4.0]); + let verdict = aligner() + .align(&curve, &curve, Band::ddsketch(), &rules) + .expect("identical curves align"); + assert!(!verdict.tripped()); +} + +#[test] +fn a_sup_trip_names_the_sup_fold() { + let rules = RuleSet::new(); + let reference = scalar_curve(&[10.0, 10.0, 10.0]); + let sut = scalar_curve(&[10.0, 9000.0, 10.0]); + let verdict = aligner() + .align(&reference, &sut, Band::ddsketch(), &rules) + .expect("aligns"); + assert!(verdict.tripped()); + assert_eq!(verdict.trips[0].rule, "sup_fold"); +} + +proptest! { + /// The worst-gap verdict is LENGTH INDEPENDENT: the same localized spike embedded in a three-bucket + /// curve and in an up-to-three-thousand-bucket curve produces the identical verdict. The sup fold is + /// an anytime quantity, never diluted by panel length the way a length-normalized accumulator would + /// be. + #[test] + fn property_test_worst_gap_length_independence(n in 3usize..=3000, spike in 1_000.0f64..1.0e6) { + let rules = RuleSet::new(); + let band = Band::ddsketch(); + + let small_ref = scalar_curve(&[100.0, 100.0, 100.0]); + let small_sut = scalar_curve(&[100.0, spike, 100.0]); + let small = aligner().align(&small_ref, &small_sut, band, &rules).expect("small aligns"); + + let big_ref = scalar_curve(&vec![100.0; n]); + let mut big_sut_values = vec![100.0; n]; + big_sut_values[n / 2] = spike; + let big_sut = scalar_curve(&big_sut_values); + let big = aligner().align(&big_ref, &big_sut, band, &rules).expect("big aligns"); + + prop_assert!(small.tripped()); + prop_assert_eq!(small, big); + } + + /// A time-slide of at most the band is ZERO COST: the same step-shaped curve shifted by up to `B` + /// buckets aligns green. The band absorbs benign flush-phase skew and nothing else. + #[test] + fn property_test_small_slide_within_band_is_zero_cost( + slide in 0usize..=SAKOE_CHIBA_BAND, + low in -50.0f64..50.0, + delta in 100.0f64..1_000.0, + ) { + let rules = RuleSet::new(); + let high = low + delta; + let length = 8usize; + + let mut reference = vec![low; length]; + for value in reference.iter_mut().skip(4) { + *value = high; + } + let mut sut = vec![low; length]; + for value in sut.iter_mut().skip(4 + slide) { + *value = high; + } + + let verdict = aligner() + .align(&scalar_curve(&reference), &scalar_curve(&sut), Band::ddsketch(), &rules) + .expect("slide aligns"); + prop_assert!(!verdict.tripped(), "a <=B slide must be absorbed, slide={}", slide); + } + + /// A double burst the SUT collapsed to nothing reds AT THE PEAK GAP: the sup reports the burst height + /// itself, not a length-averaged dilution of it. + #[test] + fn property_test_double_burst_collapse_reds_at_the_peak_gap(height in 50.0f64..1.0e5) { + let rules = RuleSet::new(); + let reference = scalar_curve(&[0.0, height, 0.0, height, 0.0]); + let sut = scalar_curve(&[0.0, 0.0, 0.0, 0.0, 0.0]); + + let verdict = aligner() + .align(&reference, &sut, Band::ddsketch(), &rules) + .expect("burst aligns"); + prop_assert!(verdict.tripped()); + let worst = verdict.worst_leash.expect("a tripped curve reports its worst leash"); + prop_assert_eq!(worst.gap.abs(), height); + } + + /// A context present on one lane and absent on the other PAST THE BAND is an infinite-leash trip; a + /// shorter absence the band can absorb is not. + #[test] + fn property_test_present_absent_bottom_trips( + beyond in (SAKOE_CHIBA_BAND + 1)..20, + within in 0usize..=SAKOE_CHIBA_BAND, + ) { + let rules = RuleSet::new(); + let band = Band::ddsketch(); + let reference = scalar_curve(&[10.0, 10.0, 10.0]); + + let long = scalar_curve(&vec![10.0; 3 + beyond]); + let tripped = aligner().align(&reference, &long, band, &rules).expect("long aligns"); + prop_assert!(tripped.tripped()); + prop_assert_eq!(tripped.trips[0].rule.as_str(), "present_absent_bottom"); + prop_assert!(tripped.worst_leash.expect("infinite leash").gap.is_infinite()); + + let absorbed = scalar_curve(&vec![10.0; 3 + within]); + let green = aligner().align(&reference, &absorbed, band, &rules).expect("absorbed aligns"); + prop_assert!(!green.tripped(), "a <=B absence is a transient the band absorbs"); + } + + /// The J1 envelope guard is GREEN-WARD ONLY: it never green-washes an interior spike. One lane + /// spikes for a single bucket while the other stays flat; the guard declines and the spike trips. + #[test] + fn property_test_envelope_guard_rejects_an_interior_spike(pos in 1usize..9, height in 100.0f64..1.0e6) { + let rules = RuleSet::new(); + let length = 10usize; + let reference = scalar_curve(&vec![10.0; length]); + let mut sut_values = vec![10.0; length]; + sut_values[pos] = 10.0 + height; + let sut = scalar_curve(&sut_values); + + let verdict = aligner() + .align(&reference, &sut, Band::ddsketch(), &rules) + .expect("spike aligns"); + prop_assert!(verdict.tripped(), "an interior spike must survive the guard"); + } +} diff --git a/test/antithesis/scenarios/differential/src/ground.rs b/test/antithesis/scenarios/differential/src/ground.rs new file mode 100644 index 00000000000..1227e178db3 --- /dev/null +++ b/test/antithesis/scenarios/differential/src/ground.rs @@ -0,0 +1,343 @@ +//! The per-kind ground metric and the operator-extensible named-rule layer. +//! +//! The ground metric `d_v` reduces a coupled column pair to a signed [`Leash`]: the value gap and the +//! [`Band`] that admits benign difference at that value. The [`RuleSet`] holds the named guard rules +//! that may suppress an over-band trip; rules are monotone green-ward and never introduce a trip. + +use std::fmt; + +use ddsketch::canonical::IndexMapping; +use ddsketch::DDSketch; + +use crate::contexts::{BucketValue, SketchValue}; + +/// The signed vertical gap between the SUT value and its coupled reference value at one column, paired +/// with the band that admits benign difference there. `over_band` when the gap exceeds the band. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Leash { + /// `gap_t = s_t - a_{pi(t)}`, signed. + pub gap: f64, + /// `b_t = max(alpha*|v|, floor)`. + pub band: f64, +} + +impl Leash { + /// Whether the gap exceeds its band, `|gap| > band`. + #[must_use] + pub fn over_band(self) -> bool { + self.gap.abs() > self.band + } +} + +/// A per-value band `max(alpha*|v|, floor)`. `alpha` is the ddsketch relative accuracy and `floor` its +/// smallest representable magnitude; both derive from the ddsketch public API, never hardcoded. +#[derive(Clone, Copy, Debug)] +pub struct Band { + /// The ddsketch relative accuracy. + pub alpha: f64, + /// The ddsketch value floor. + pub floor: f64, +} + +impl Band { + /// The band derived entirely from the ddsketch public API: `alpha` is the agent sketch mapping's + /// relative accuracy and `floor` its smallest representable magnitude, `value_for_key(1)`. Neither + /// is a hardcoded numeric literal. + #[must_use] + pub fn ddsketch() -> Self { + Self { + alpha: DDSketch::remap_mapping().relative_accuracy(), + floor: DDSketch::value_for_key(1), + } + } + + /// The band admitted at `value`: `max(alpha*|value|, floor)`. + #[must_use] + pub fn at(self, value: f64) -> f64 { + (self.alpha * value.abs()).max(self.floor) + } +} + +/// Why a ground-distance computation could not produce a [`Leash`]. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum GroundError { + /// The reference and SUT values are different kinds, so no ground distance is defined. + KindMismatch, + /// The stub has no implementation yet. + Unimplemented, +} + +impl fmt::Display for GroundError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + GroundError::KindMismatch => write!(f, "reference and SUT bucket values are different kinds"), + GroundError::Unimplemented => write!(f, "ground metric is not yet implemented"), + } + } +} + +impl std::error::Error for GroundError {} + +/// The ground distance at one coupled column: the signed value gap and the band that admits benign +/// difference. Per kind the value fold differs -- counter/rate summed, gauge last, sketch mean -- and +/// the caller passes the already-folded value pair, so this stays kind-driven only in how it reads. +/// +/// # Errors +/// +/// Returns [`GroundError::KindMismatch`] when the two values are different kinds, and +/// [`GroundError::Unimplemented`] while the metric is a stub. +pub fn d_v(reference: &BucketValue, sut: &BucketValue, band: Band) -> Result { + match (reference, sut) { + (BucketValue::Scalar(a), BucketValue::Scalar(s)) => Ok(scalar_leash(*a, *s, band)), + (BucketValue::Sketch(a), BucketValue::Sketch(s)) => Ok(sketch_leash(a, s, band)), + _ => Err(GroundError::KindMismatch), + } +} + +/// The displayed quantiles the sketch channel takes a sup over. Reading these off the shared log grid +/// keeps the sketch verdict a bottleneck over the visible curve rather than a probability-axis average. +const DISPLAYED_QUANTILES: [f64; 4] = [0.5, 0.9, 0.95, 0.99]; + +/// The scalar ground distance: the signed value gap and the band admitted at the larger of the two +/// magnitudes, so the leash is symmetric under swapping the lanes. +fn scalar_leash(reference: f64, sut: f64, band: Band) -> Leash { + Leash { + gap: sut - reference, + band: band.at(reference.abs().max(sut.abs())), + } +} + +/// The sketch ground distance: a strict total-count gate then a sup over the displayed quantiles. +/// +/// The total count is summed from the bins on both sides; any mismatch snaps to an infinite leash so a +/// count divergence always trips regardless of quantile agreement. Otherwise each displayed quantile is +/// read off the shared log grid and compared against its own per-value band, and the worst normalized +/// excess wins the sup. Identical bins give a zero gap. +fn sketch_leash(reference: &SketchValue, sut: &SketchValue, band: Band) -> Leash { + if bins_total(&reference.bins) != bins_total(&sut.bins) { + return Leash { + gap: f64::INFINITY, + band: band.at(0.0), + }; + } + let mut worst = Leash { + gap: 0.0, + band: band.at(0.0), + }; + let mut worst_excess = 0.0; + for &q in &DISPLAYED_QUANTILES { + let (Some(reference_q), Some(sut_q)) = + (quantile_from_bins(&reference.bins, q), quantile_from_bins(&sut.bins, q)) + else { + continue; + }; + let gap = sut_q - reference_q; + let quantile_band = band.at(reference_q.abs().max(sut_q.abs())); + let excess = gap.abs() / quantile_band; + if excess > worst_excess { + worst_excess = excess; + worst = Leash { + gap, + band: quantile_band, + }; + } + } + worst +} + +/// The total sample count a sketch's bins carry, summed on the shared grid. +pub(crate) fn bins_total(bins: &[(i32, u32)]) -> u64 { + bins.iter().map(|(_, count)| u64::from(*count)).sum() +} + +/// The representative value the displayed quantile `q` reads off a sketch's log-grid bins. Walks the +/// key-sorted bins accumulating count and returns the crossing bin's representative value from the +/// shared grid via the ddsketch public API. `None` when the bins carry no samples. +pub(crate) fn quantile_from_bins(bins: &[(i32, u32)], q: f64) -> Option { + let total = bins_total(bins); + if total == 0 { + return None; + } + let wanted = q * total as f64; + let mut cumulative = 0u64; + let mut last_key = None; + for &(key, count) in bins { + cumulative += u64::from(count); + last_key = Some(key); + if cumulative as f64 >= wanted { + return Some(value_for_key(key)); + } + } + last_key.map(value_for_key) +} + +/// The representative value the shared log grid assigns a bin key, via the ddsketch public API. Agent +/// sketch keys are `i16`; a key outside that range saturates rather than panicking. The negative bound +/// stops at `i16::MIN + 1` because the ddsketch key space negates the key, and negating `i16::MIN` +/// overflows. +fn value_for_key(key: i32) -> f64 { + let clamped = key.clamp(i32::from(i16::MIN) + 1, i32::from(i16::MAX)); + let key = i16::try_from(clamped).unwrap_or(i16::MAX); + DDSketch::value_for_key(key) +} + +/// The read-only view a [`GroundRule`] inspects to decide suppression at one coupled column: both +/// lanes' scalar value tracks, the coupled indices, the band, and the envelope half-window `B`. +#[derive(Clone, Copy, Debug)] +pub struct RuleContext<'a> { + /// The reference (Agent) curve's per-bucket scalar track. + pub reference: &'a [f64], + /// The SUT (ADP) curve's per-bucket scalar track. + pub sut: &'a [f64], + /// The reference column index the coupling selected. + pub ref_index: usize, + /// The SUT column index the coupling selected. + pub sut_index: usize, + /// The band settings in force. + pub band: Band, + /// The envelope half-window `B`, in buckets. + pub window: usize, +} + +/// A named equivalence rule that may suppress an over-band trip. Rules are monotone green-ward: a rule +/// only ever removes a trip, never adds one. The J1 matched-jump + envelope-containment guard is one. +pub trait GroundRule { + /// The rule's stable name, surfaced in triage details. + fn name(&self) -> &'static str; + + /// Whether this rule suppresses the trip at `ctx`. Must never turn a green column red. + fn suppresses(&self, ctx: &RuleContext<'_>) -> bool; +} + +/// The ordered set of named guard rules applied to each over-band column. A trip survives only if no +/// registered rule suppresses it. +#[derive(Default)] +pub struct RuleSet { + rules: Vec>, +} + +impl RuleSet { + /// An empty rule set. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Registers a rule. + pub fn register(&mut self, rule: Box) { + self.rules.push(rule); + } + + /// Whether any registered rule suppresses the trip at `ctx`. + #[must_use] + pub fn suppressed(&self, ctx: &RuleContext<'_>) -> bool { + self.rules.iter().any(|rule| rule.suppresses(ctx)) + } + + /// The three initial guard rules, in registration order: the missing-counter-zero equivalence, the + /// unknown-kind presence-only pass, and the sketch adjacent-bin quantization tolerance. Each only + /// ever suppresses a trip, so the set only widens `same`. + #[must_use] + pub fn default_rules() -> Self { + let mut set = Self::new(); + set.register(Box::new(MissingCounterZero)); + set.register(Box::new(UnknownKindPresenceOnly)); + set.register(Box::new(SketchAdjacentBinQuantization)); + set + } + + /// The registered rules' names, in registration order. + #[must_use] + pub fn names(&self) -> Vec<&'static str> { + self.rules.iter().map(|rule| rule.name()).collect() + } +} + +/// A counter bucket that never flushed reads as a missing column, which the aligned track carries as a +/// zero. This rule suppresses an over-band trip when one coupled lane is zero and the same value the +/// other lane carries also appears on the zero lane within the `+/-window`: the counter's increment +/// merely landed in an adjacent flush bucket, a benign phase slide, not lost or invented data. +struct MissingCounterZero; + +impl GroundRule for MissingCounterZero { + fn name(&self) -> &'static str { + "missing_counter_zero" + } + + fn suppresses(&self, ctx: &RuleContext<'_>) -> bool { + let (Some(&reference), Some(&sut)) = (ctx.reference.get(ctx.ref_index), ctx.sut.get(ctx.sut_index)) else { + return false; + }; + if !reference.is_finite() || !sut.is_finite() { + return false; + } + let (zero_lane, value, value_index) = if reference == 0.0 && sut != 0.0 { + (ctx.reference, sut, ctx.ref_index) + } else if sut == 0.0 && reference != 0.0 { + (ctx.sut, reference, ctx.sut_index) + } else { + return false; + }; + if zero_lane.is_empty() { + return false; + } + let tolerance = ctx.band.at(value.abs()); + let low = value_index.saturating_sub(ctx.window); + let high = (value_index + ctx.window).min(zero_lane.len() - 1); + (low..=high).any(|i| (zero_lane[i] - value).abs() <= tolerance) + } +} + +/// An unknown-kind (`Other`) context has no defined value fold, so its aligned track carries a +/// non-finite placeholder rather than a comparable scalar. Presence is checked elsewhere, so this rule +/// suppresses any value trip at a column where either lane's value is non-finite. Known-kind tracks are +/// finite, so it never fires on them. +struct UnknownKindPresenceOnly; + +impl GroundRule for UnknownKindPresenceOnly { + fn name(&self) -> &'static str { + "unknown_kind_presence_only" + } + + fn suppresses(&self, ctx: &RuleContext<'_>) -> bool { + match (ctx.reference.get(ctx.ref_index), ctx.sut.get(ctx.sut_index)) { + (Some(&reference), Some(&sut)) => !reference.is_finite() || !sut.is_finite(), + _ => false, + } + } +} + +/// DDSketch quantization can place identical underlying data one bin apart across the two lanes. This +/// rule suppresses a trip when the coupled values are same-signed and at most one adjacent bin apart on +/// the shared log grid, read via the ddsketch public API. A larger gap is a real quantile divergence. +struct SketchAdjacentBinQuantization; + +impl GroundRule for SketchAdjacentBinQuantization { + fn name(&self) -> &'static str { + "sketch_adjacent_bin_quantization" + } + + fn suppresses(&self, ctx: &RuleContext<'_>) -> bool { + let (Some(&reference), Some(&sut)) = (ctx.reference.get(ctx.ref_index), ctx.sut.get(ctx.sut_index)) else { + return false; + }; + if !reference.is_finite() || !sut.is_finite() { + return false; + } + if reference == sut { + return true; + } + let same_sign = (reference >= 0.0) == (sut >= 0.0); + if !same_sign { + return false; + } + let mapping = DDSketch::remap_mapping(); + let floor = ctx.band.floor; + let reference_key = mapping.index(reference.abs().max(floor)); + let sut_key = mapping.index(sut.abs().max(floor)); + (reference_key - sut_key).abs() <= 1 + } +} + +#[cfg(test)] +mod tests; diff --git a/test/antithesis/scenarios/differential/src/ground/tests.rs b/test/antithesis/scenarios/differential/src/ground/tests.rs new file mode 100644 index 00000000000..017b86b06a1 --- /dev/null +++ b/test/antithesis/scenarios/differential/src/ground/tests.rs @@ -0,0 +1,263 @@ +//! Tests for the per-kind ground metric and the named-rule layer. + +use ddsketch::canonical::IndexMapping; +use ddsketch::DDSketch; +use proptest::prelude::*; + +use super::{ + bins_total, d_v, quantile_from_bins, value_for_key, Band, GroundError, GroundRule, MissingCounterZero, RuleContext, + RuleSet, SketchAdjacentBinQuantization, UnknownKindPresenceOnly, +}; +use crate::contexts::{BucketValue, SketchValue}; + +fn scalar(value: f64) -> BucketValue { + BucketValue::Scalar(value) +} + +fn sketch(bins: Vec<(i32, u32)>) -> BucketValue { + let count = bins.iter().map(|(_, n)| i64::from(*n)).sum(); + BucketValue::Sketch(SketchValue { + count, + sum: 0.0, + min: 0.0, + max: 0.0, + bins, + }) +} + +fn ctx<'a>(reference: &'a [f64], sut: &'a [f64], ref_index: usize, sut_index: usize, window: usize) -> RuleContext<'a> { + RuleContext { + reference, + sut, + ref_index, + sut_index, + band: Band::ddsketch(), + window, + } +} + +#[test] +fn band_derives_alpha_and_floor_from_the_ddsketch_api() { + // The band is the ddsketch mapping's relative accuracy and its value_for_key(1) floor, read from + // the public API rather than any baked-in constant. + let band = Band::ddsketch(); + assert_eq!(band.alpha, DDSketch::remap_mapping().relative_accuracy()); + assert_eq!(band.floor, DDSketch::value_for_key(1)); + assert!(band.alpha > 0.0 && band.floor > 0.0); +} + +#[test] +fn ground_source_carries_no_hardcoded_ddsketch_literal() { + // The derivation must not smuggle the mapping's gamma, alpha, or eps in as a numeric literal. Each + // needle is split so this test file itself does not contain the forbidden literal verbatim. + let source = + std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/ground.rs")).expect("read ground.rs source"); + let needles = [["1.015", "625"], ["0.015", "625"], ["0.0077", "519"]]; + for parts in needles { + let literal = parts.concat(); + assert!( + !source.contains(&literal), + "ground.rs hardcodes the ddsketch literal {literal}" + ); + } +} + +#[test] +fn identical_scalars_give_a_zero_leash() -> Result<(), GroundError> { + let leash = d_v(&scalar(42.0), &scalar(42.0), Band::ddsketch())?; + assert_eq!(leash.gap, 0.0); + assert!(!leash.over_band()); + Ok(()) +} + +#[test] +fn identical_sketches_give_a_zero_leash() -> Result<(), GroundError> { + let value = sketch(vec![(5, 3), (7, 2)]); + let leash = d_v(&value, &value, Band::ddsketch())?; + assert_eq!(leash.gap, 0.0); + assert!(!leash.over_band()); + Ok(()) +} + +#[test] +fn different_kinds_are_a_kind_mismatch() { + let error = d_v(&scalar(1.0), &sketch(vec![(1, 1)]), Band::ddsketch()).unwrap_err(); + assert_eq!(error, GroundError::KindMismatch); +} + +#[test] +fn scalar_band_uses_the_larger_magnitude() -> Result<(), GroundError> { + // Per-kind band: a scalar leash bands on max(|reference|,|sut|) so the leash is symmetric. + let band = Band::ddsketch(); + let leash = d_v(&scalar(3.0), &scalar(-5.0), band)?; + assert_eq!(leash.gap, -8.0); + assert_eq!(leash.band, band.at(5.0)); + Ok(()) +} + +#[test] +fn sketch_sup_catches_a_p99_blowup() -> Result<(), GroundError> { + // The reference concentrates at a low key; the SUT pushes a tenth of its mass into a far higher + // key so its upper quantiles blow up. The sup over displayed quantiles trips even though p50/p90 + // agree. + let band = Band::ddsketch(); + let reference = sketch(vec![(10, 100)]); + let sut = sketch(vec![(10, 90), (500, 10)]); + let leash = d_v(&reference, &sut, band)?; + assert!(leash.over_band()); + Ok(()) +} + +#[test] +fn sketch_leash_bands_on_the_winning_quantile_values() -> Result<(), GroundError> { + // Per-kind band: the sketch leash reports the band at the quantile that won the sup, not a summary + // band. p95 is the first quantile to cross into the high bin, so its values set the band. + let band = Band::ddsketch(); + let reference_bins = vec![(10, 100)]; + let sut_bins = vec![(10, 90), (500, 10)]; + let leash = d_v(&sketch(reference_bins.clone()), &sketch(sut_bins.clone()), band)?; + + let reference_q = quantile_from_bins(&reference_bins, 0.95).expect("reference p95"); + let sut_q = quantile_from_bins(&sut_bins, 0.95).expect("sut p95"); + assert_eq!(leash.gap, sut_q - reference_q); + assert_eq!(leash.band, band.at(reference_q.abs().max(sut_q.abs()))); + Ok(()) +} + +#[test] +fn count_mismatch_snaps_to_an_infinite_leash() -> Result<(), GroundError> { + // A strict total-count gate, computed from the bins: one extra sample snaps the leash to infinity + // regardless of quantile agreement. + let band = Band::ddsketch(); + let reference = sketch(vec![(10, 100)]); + let sut = sketch(vec![(10, 101)]); + let leash = d_v(&reference, &sut, band)?; + assert!(leash.gap.is_infinite()); + assert!(leash.over_band()); + Ok(()) +} + +#[test] +fn bins_total_sums_the_bin_counts() { + assert_eq!(bins_total(&[(1, 2), (3, 5)]), 7); + assert_eq!(bins_total(&[]), 0); +} + +#[test] +fn default_rules_registers_the_three_named_rules_in_order() { + assert_eq!( + RuleSet::default_rules().names(), + vec![ + "missing_counter_zero", + "unknown_kind_presence_only", + "sketch_adjacent_bin_quantization" + ] + ); +} + +#[test] +fn empty_ruleset_suppresses_nothing() { + let reference = [0.0, 1000.0]; + let sut = [1000.0, 0.0]; + assert!(!RuleSet::new().suppressed(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn missing_counter_zero_suppresses_a_phase_shifted_increment() { + // The reference is zero at the coupled column but carries the SUT's value one bucket over: the + // counter increment slid into an adjacent flush, a benign phase shift. + let reference = [0.0, 1000.0]; + let sut = [1000.0, 0.0]; + assert!(MissingCounterZero.suppresses(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn missing_counter_zero_keeps_a_real_drop() { + // The reference is zero and never carries the SUT's value in the window: real invented data. + let reference = [0.0, 0.0]; + let sut = [1000.0, 0.0]; + assert!(!MissingCounterZero.suppresses(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn unknown_kind_presence_only_suppresses_a_non_finite_column() { + let reference = [f64::NAN]; + let sut = [5.0]; + assert!(UnknownKindPresenceOnly.suppresses(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn unknown_kind_presence_only_keeps_a_finite_column() { + let reference = [1.0]; + let sut = [2.0]; + assert!(!UnknownKindPresenceOnly.suppresses(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn sketch_adjacent_bin_quantization_suppresses_one_bin_apart() { + // One multiplicative gamma step is exactly one bin up on the log grid: a quantization boundary, + // not a real divergence. + let gamma = DDSketch::remap_mapping().gamma(); + let reference = [100.0]; + let sut = [100.0 * gamma]; + assert!(SketchAdjacentBinQuantization.suppresses(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn sketch_adjacent_bin_quantization_keeps_a_multi_bin_gap() { + let reference = [100.0]; + let sut = [1_000_000.0]; + assert!(!SketchAdjacentBinQuantization.suppresses(&ctx(&reference, &sut, 0, 0, 2))); +} + +#[test] +fn value_for_key_saturates_out_of_range_keys() { + // A key past the i16 range must saturate rather than panic, and must never read back as NaN: a NaN + // magnitude would make every gap comparison false and silently mask a trip. An extreme key may + // legitimately overflow to a saturated infinite magnitude, which only ever trips more. + assert!(!value_for_key(i32::MAX).is_nan()); + assert!(!value_for_key(i32::MIN).is_nan()); +} + +proptest! { + /// Swapping the two lanes negates the gap and leaves the band unchanged: the scalar leash is + /// symmetric. + #[test] + fn property_test_d_v_scalar_symmetric(a in -1.0e6f64..1.0e6, s in -1.0e6f64..1.0e6) { + let band = Band::ddsketch(); + let forward = d_v(&scalar(a), &scalar(s), band).expect("forward scalar leash"); + let reverse = d_v(&scalar(s), &scalar(a), band).expect("reverse scalar leash"); + prop_assert_eq!(forward.gap, -reverse.gap); + prop_assert_eq!(forward.band, reverse.band); + } + + /// An empty rule set suppresses nothing, and registering more rules only ever widens `same`: any + /// single rule that suppresses a column is subsumed by the full default set. + #[test] + fn property_test_rules_only_widen( + reference in prop::collection::vec(-1.0e3f64..1.0e3, 1..8), + sut in prop::collection::vec(-1.0e3f64..1.0e3, 1..8), + ri in 0usize..8, + si in 0usize..8, + ) { + let ref_index = ri % reference.len(); + let sut_index = si % sut.len(); + let context = ctx(&reference, &sut, ref_index, sut_index, 2); + + prop_assert!(!RuleSet::new().suppressed(&context)); + + let full = RuleSet::default_rules(); + for rule in [ + Box::new(MissingCounterZero) as Box, + Box::new(UnknownKindPresenceOnly), + Box::new(SketchAdjacentBinQuantization), + ] { + let mut single = RuleSet::new(); + let name = rule.name(); + single.register(rule); + if single.suppressed(&context) { + prop_assert!(full.suppressed(&context), "rule {} suppressed but the default set did not", name); + } + } + } +} diff --git a/test/antithesis/scenarios/differential/src/lib.rs b/test/antithesis/scenarios/differential/src/lib.rs index 2c1dac8bdf2..ad0ba8e8a4e 100644 --- a/test/antithesis/scenarios/differential/src/lib.rs +++ b/test/antithesis/scenarios/differential/src/lib.rs @@ -1,3 +1,6 @@ //! Shared context-comparison logic for the differential scenario's checks. pub mod contexts; +pub mod decision; +pub mod frechet; +pub mod ground;