chore(antithesis): banded-Frechet differential visual-equivalence oracle

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

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 }

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.

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"

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<String>,
+    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<Bucket> = Vec::new();
+
+    let contexts: BTreeSet<&Context> = reference.keys().chain(sut.keys()).collect();
+    let mut verdicts: Vec<ContextVerdict> = Vec::with_capacity(contexts.len());
+    let mut triage: Vec<Triage> = 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, &params);
+        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<f64> {
+    value.is_finite().then_some(value)
+}