enhancement(antithesis): Expand SDK assertions, hammer sketch

This commit expands the SDK asserts in the SUT, allowing us to flag more
invariant flaws in runs. Of particular interest this commit introduces a
new driver, the 'sketchburst'. The goal of this driver is to transmit
sketch load in a way that 'bursts' load that will fall into a small
number of ddsketch bins.

This commit is also a demonstration of what antithesis sdk assertions
without some manner of shim or other way of tidying them up looks like
in a codebase, per recent review discussions.

Author: blt

Cargo.lock

@@ -50,6 +50,30 @@ async fn main() -> Result<(), GenericError> {
     #[cfg(feature = "antithesis")]
     antithesis_sdk::antithesis_init();
 
+    // Report any panic to Antithesis before the process aborts. The antithesis build sets
+    // `panic = "abort"`, and the captured container logs do not surface ADP's stderr, so the default
+    // panic message is invisible in triage. Route the panic payload and site through the SDK, which
+    // writes to sdk.jsonl -- a channel that survives the abort -- then chain to the default hook so a
+    // local build still prints normally.
+    #[cfg(feature = "antithesis")]
+    {
+        let default_hook = std::panic::take_hook();
+        std::panic::set_hook(Box::new(move |info| {
+            let location = info.location().map_or_else(String::new, |l| l.to_string());
+            let payload = info.payload();
+            let message = payload
+                .downcast_ref::<&str>()
+                .map(|s| (*s).to_string())
+                .or_else(|| payload.downcast_ref::<String>().cloned())
+                .unwrap_or_else(|| "<non-string panic payload>".to_string());
+            antithesis_sdk::assert_unreachable!(
+                "agent-data-plane panicked",
+                &serde_json::json!({ "message": message, "location": location })
+            );
+            default_hook(info);
+        }));
+    }
+
     let cli: Cli = argh::from_env();
 
     // Print version and exit early without requiring config.
@@ -61,14 +85,27 @@ async fn main() -> Result<(), GenericError> {
     // Load our "bootstrap" configuration -- static configuration on disk or from environment variables -- so we can
     // initialize basic subsystems before executing the given subcommand.
     let bootstrap_config_path = cli.config_file.unwrap_or_else(PlatformSettings::get_config_file_path);
-    let bootstrap_config = ConfigurationLoader::default()
+    let loaded = ConfigurationLoader::default()
         .with_key_aliases(KEY_ALIASES)
         .from_yaml(&bootstrap_config_path)
-        .error_context("Failed to load Datadog Agent configuration file during bootstrap.")?
-        .add_providers([DatadogRemapper::new()])
-        .from_environment(PlatformSettings::get_env_var_prefix())
-        .error_context("Environment variable prefix should not be empty.")?
-        .bootstrap_generic();
+        .error_context("Failed to load Datadog Agent configuration file during bootstrap.")
+        .and_then(|loader| {
+            loader
+                .add_providers([DatadogRemapper::new()])
+                .from_environment(PlatformSettings::get_env_var_prefix())
+                .error_context("Environment variable prefix should not be empty.")
+        });
+
+    // Classify a graceful config rejection (process exit 1) versus a clean boot. A false evaluation
+    // carries the rejection reason, so triage shows which sampled `datadog.yaml` ADP refused and why.
+    #[cfg(feature = "antithesis")]
+    antithesis_sdk::assert_always_or_unreachable!(
+        loaded.is_ok(),
+        "agent-data-plane boots under sampled config",
+        &serde_json::json!({ "phase": "config_load", "error": loaded.as_ref().err().map(|e| format!("{e:?}")) })
+    );
+
+    let bootstrap_config = loaded?.bootstrap_generic();
 
     // Translate the bootstrap configuration into ADP's logging configuration, applying ADP-specific rules
     // (per-subagent log file key, never sharing a file with the Core Agent).
@@ -157,6 +194,14 @@ async fn run_inner(
                     }
                     Err(e) => {
                         error!("{:?}", e);
+                        // Same property as the config-load gate: a run-setup error is also a graceful
+                        // exit-1 under this sampled config, distinguished by `phase` in the details.
+                        #[cfg(feature = "antithesis")]
+                        antithesis_sdk::assert_always_or_unreachable!(
+                            false,
+                            "agent-data-plane boots under sampled config",
+                            &serde_json::json!({ "phase": "run_setup", "error": format!("{e:?}") })
+                        );
                         Some(1)
                     }
                 };

bin/agent-data-plane/src/main.rs

@@ -9,10 +9,12 @@ repository = { workspace = true }
 workspace = true
 
 [features]
+antithesis = ["dep:antithesis_sdk", "antithesis_sdk/full"]
 ddsketch_extended = []
 serde = ["dep:serde", "smallvec/serde"]
 
 [dependencies]
+antithesis_sdk = { workspace = true, optional = true }
 datadog-protos = { workspace = true }
 float-cmp = { workspace = true, features = ["ratio"] }
 ordered-float = { workspace = true }

lib/ddsketch/Cargo.toml

@@ -186,6 +186,9 @@ impl DDSketch {
     }
 
     fn adjust_basic_stats(&mut self, v: f64, n: u64) {
+        #[cfg(feature = "antithesis")]
+        antithesis_sdk::assert_always!(v.is_finite(), "DDSketch sample is finite at insert");
+
         if v < self.min {
             self.min = v;
         }
@@ -197,6 +200,9 @@ impl DDSketch {
         self.count += n;
         self.sum += v * n as f64;
 
+        #[cfg(feature = "antithesis")]
+        antithesis_sdk::assert_always!(self.sum.is_finite(), "sketch sum stays finite after insert");
+
         if n == 1 {
             self.avg += (v - self.avg) / self.count as f64;
         } else {
@@ -711,6 +717,20 @@ fn trim_left(bins: &mut SmallVec<[Bin; 4]>, bin_limit: u16) {
 
     // Drop the removed prefix, leaving exactly bin_limit bins.
     bins.drain(0..num_to_remove);
+
+    // We only reach here when the sketch exceeded `bin_limit` and collapsed; the drain leaves exactly `bin_limit`
+    // bins. This is the one place every mutating method routes through, so asserting here guards the bin-count bound
+    // for insert / insert_n / merge / interpolation alike. The reachability anchor proves a real corpus actually hits
+    // the collapse, else the bound passes vacuously.
+    #[cfg(feature = "antithesis")]
+    {
+        antithesis_sdk::assert_reachable!("trim_left collapsed bins");
+        antithesis_sdk::assert_always_less_than_or_equal_to!(
+            bins.len(),
+            bin_limit,
+            "sketch bin count within bin_limit"
+        );
+    }
 }
 
 #[allow(clippy::cast_possible_truncation)]

lib/ddsketch/src/agent/sketch.rs

@@ -12,7 +12,16 @@ workspace = true
 default = []
 config-test-support = []
 fips = ["saluki-io/fips"]
-antithesis = ["dep:antithesis_sdk", "antithesis_sdk/full"]
+antithesis = [
+  "dep:antithesis_sdk",
+  "antithesis_sdk/full",
+  "ddsketch/antithesis",
+  "saluki-config/antithesis",
+  "saluki-context/antithesis",
+  "saluki-core/antithesis",
+  "saluki-io/antithesis",
+  "stringtheory/antithesis",
+]
 
 [dependencies]
 antithesis_sdk = { workspace = true, optional = true }

lib/saluki-components/Cargo.toml

@@ -1677,28 +1677,59 @@ async fn dispatch_events(mut event_buffer: EventsBuffer, source_context: &Source
     // Dispatch any eventd events, if present.
     if event_buffer.has_event_type(EventType::EventD) {
         let eventd_events = event_buffer.extract(Event::is_eventd);
-        if let Err(e) = source_context
-            .dispatcher()
-            .buffered_named("events")
+        let events_output = source_context.dispatcher().buffered_named("events");
+
+        // The `events` output is always wired in the DSD topology, so a missing output is a structural invariant
+        // violation that crashes this fail-stop component. Surface it to Antithesis too.
+        #[cfg(feature = "antithesis")]
+        if events_output.is_err() {
+            antithesis_sdk::assert_unreachable!("dsd 'events' output missing at dispatch", &serde_json::json!({}));
+        }
+
+        if let Err(e) = events_output
             .expect("events output should always exist")
             .send_all(eventd_events)
             .await
         {
             error!(%listen_addr, error = %e, "Failed to dispatch eventd events.");
+
+            // Silent-loss anchor: dispatch failure increments no counter, so this is the only in-SUT signal that the
+            // failure path ran. Keeps the no-silent-loss properties non-vacuous.
+            #[cfg(feature = "antithesis")]
+            antithesis_sdk::assert_sometimes!(
+                true,
+                "dsd dispatch failed mid-buffer",
+                &serde_json::json!({ "stream": "events" })
+            );
         }
     }
 
     // Dispatch any service check events, if present.
     if event_buffer.has_event_type(EventType::ServiceCheck) {
         let service_check_events = event_buffer.extract(Event::is_service_check);
-        if let Err(e) = source_context
-            .dispatcher()
-            .buffered_named("service_checks")
+        let service_checks_output = source_context.dispatcher().buffered_named("service_checks");
+
+        #[cfg(feature = "antithesis")]
+        if service_checks_output.is_err() {
+            antithesis_sdk::assert_unreachable!(
+                "dsd 'service_checks' output missing at dispatch",
+                &serde_json::json!({})
+            );
+        }
+
+        if let Err(e) = service_checks_output
             .expect("service checks output should always exist")
             .send_all(service_check_events)
             .await
         {
             error!(%listen_addr, error = %e, "Failed to dispatch service check events.");
+
+            #[cfg(feature = "antithesis")]
+            antithesis_sdk::assert_sometimes!(
+                true,
+                "dsd dispatch failed mid-buffer",
+                &serde_json::json!({ "stream": "service_checks" })
+            );
         }
     }
 
@@ -1710,6 +1741,13 @@ async fn dispatch_events(mut event_buffer: EventsBuffer, source_context: &Source
             .await
         {
             error!(%listen_addr, error = %e, "Failed to dispatch metric events.");
+
+            #[cfg(feature = "antithesis")]
+            antithesis_sdk::assert_sometimes!(
+                true,
+                "dsd dispatch failed mid-buffer",
+                &serde_json::json!({ "stream": "metrics" })
+            );
         }
     }
 }

lib/saluki-components/src/sources/dogstatsd/mod.rs

@@ -93,6 +93,16 @@ impl TrafficCaptureReader {
         // The writer emits a zero-length prefix to mark the start of the tagger state trailer; treat
         // that (and any size that would overrun the buffer) as the end of the record stream.
         if size == 0 || self.offset + size > self.contents.len() {
+            // A zero-length prefix is the legitimate trailer marker. A non-zero `size` that overruns the buffer is a
+            // corrupt/oversized length prefix being silently read as clean EOF, which drops every following
+            // well-formed record. Surface the corrupt case as distinct from a real trailer.
+            #[cfg(feature = "antithesis")]
+            antithesis_sdk::assert_always_or_unreachable!(
+                size == 0,
+                "replay read_next stopped at the real trailer, not on a corrupt length prefix",
+                &serde_json::json!({ "size": size, "offset": self.offset, "len": self.contents.len() })
+            );
+
             return Ok(None);
         }
 

lib/saluki-components/src/sources/dogstatsd/replay/reader.rs

@@ -569,8 +569,27 @@ impl AggregationState {
     }
 
     fn insert(&mut self, timestamp: u64, metric: Metric) -> bool {
+        // The context map is hard-capped at `context_limit` and no path grows it past the cap. This is the one
+        // non-advisory runtime memory bound, so we assert it as an invariant under Antithesis. The numeric form hands
+        // the search the margin to the limit as a gradient.
+        #[cfg(feature = "antithesis")]
+        antithesis_sdk::assert_always_less_than_or_equal_to!(
+            self.contexts.len(),
+            self.context_limit,
+            "aggregate context map within context_limit",
+            &serde_json::json!({ "len": self.contexts.len(), "limit": self.context_limit })
+        );
+
         // If we haven't seen this context yet, and it would put us over the limit to insert it, then return early.
         if !self.contexts.contains_key(metric.context()) && self.contexts.len() >= self.context_limit {
+            // Anti-vacuity anchor: prove a run actually reaches the cap, else the invariant above passes trivially.
+            #[cfg(feature = "antithesis")]
+            antithesis_sdk::assert_sometimes!(
+                true,
+                "aggregate context limit breached",
+                &serde_json::json!({ "limit": self.context_limit })
+            );
+
             self.context_limit_breached = true;
             return false;
         }
@@ -632,11 +651,45 @@ impl AggregationState {
         if self.last_flush != 0 {
             let start = align_to_bucket_start(self.last_flush, bucket_width_secs);
 
+            // Clock-skew guards. Bucketing reads the wall clock while the flush cadence is monotonic, so a wall-clock
+            // jump is not bounded by the flush interval. A backward jump empties the zero-value range (a silent counter
+            // gap); a forward jump makes the loop below run once per bucket across the whole jumped span — O(jump) work
+            // and allocation. Assert before the loop so a flood fails fast rather than after the damage is done.
+            #[cfg(feature = "antithesis")]
+            {
+                // Generous versus the normal cadence (default 15s flush over a 10s bucket yields 1-2 buckets); a bound
+                // this large trips only on a multi-hour wall-clock jump, never on a slow-but-sane flush.
+                const MAX_ZERO_VALUE_BUCKETS_PER_FLUSH: u64 = 10_000;
+                antithesis_sdk::assert_always!(
+                    current_time >= self.last_flush,
+                    "aggregate flush wall-clock did not move backward",
+                    &serde_json::json!({ "current_time": current_time, "last_flush": self.last_flush })
+                );
+                antithesis_sdk::assert_always_less_than_or_equal_to!(
+                    current_time.saturating_sub(self.last_flush) / bucket_width_secs.get(),
+                    MAX_ZERO_VALUE_BUCKETS_PER_FLUSH,
+                    "aggregate zero-value bucket span bounded across a flush",
+                    &serde_json::json!({
+                        "current_time": current_time,
+                        "last_flush": self.last_flush,
+                        "bucket_width_secs": bucket_width_secs.get()
+                    })
+                );
+            }
+
             for bucket_start in (start..current_time).step_by(bucket_width_secs.get() as usize) {
                 if is_bucket_closed(current_time, bucket_start, bucket_width_secs, flush_open_buckets) {
                     zero_value_buckets.push((bucket_start, MetricValues::counter((bucket_start, 0.0))));
                 }
             }
+
+            // Anti-vacuity anchor: prove the idle-counter zero-value path actually runs in some timeline.
+            #[cfg(feature = "antithesis")]
+            antithesis_sdk::assert_sometimes!(
+                !zero_value_buckets.is_empty(),
+                "aggregate flush generated zero-value counter buckets",
+                &serde_json::json!({ "count": zero_value_buckets.len() })
+            );
         }
 
         // Iterate over each context we're tracking, and flush any values that are in buckets which are now closed.

lib/saluki-components/src/transforms/aggregate/mod.rs

@@ -8,7 +8,11 @@ repository = { workspace = true }
 [lints]
 workspace = true
 
+[features]
+antithesis = ["dep:antithesis_sdk", "antithesis_sdk/full"]
+
 [dependencies]
+antithesis_sdk = { workspace = true, optional = true }
 figment = { workspace = true, features = ["env"] }
 saluki-error = { workspace = true }
 serde = { workspace = true }

lib/saluki-config/Cargo.toml

@@ -59,6 +59,13 @@ impl FieldUpdateWatcher {
                 // Ignore other key changes.
                 Ok(_) => continue,
                 Err(broadcast::error::RecvError::Lagged(_)) => {
+                    // A Lagged drop on the bounded broadcast means a config update was lost with no re-read, so live
+                    // filtering can stay permanently stale. Surface it.
+                    #[cfg(feature = "antithesis")]
+                    antithesis_sdk::assert_unreachable!(
+                        "filter config update Lagged-dropped with no re-read",
+                        &serde_json::json!({ "key": self.key.to_string() })
+                    );
                     warn!(
                         "FieldUpdateWatcher dropped events for key: {}. Continuing to wait for the next event.",
                         self.key