enhancement(antithesis): Expand SDK assertions, hammer sketch

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

@@ -5,6 +5,7 @@
 
 #![deny(warnings)]
 #![deny(missing_docs)]
+use std::path::Path;
 use std::time::Instant;
 
 // Pull in the Antithesis coverage-instrumentation runtime shim only when
@@ -45,10 +46,8 @@ static ALLOC: resource_accounting::TrackingAllocator<std::alloc::System> =
 async fn main() -> Result<(), GenericError> {
     let started = Instant::now();
 
-    // Initialize the Antithesis SDK as early as possible so assertions and lifecycle hooks register
-    // their catalog before any are evaluated. No-op outside Antithesis and absent in production builds.
     #[cfg(feature = "antithesis")]
-    antithesis_sdk::antithesis_init();
+    initialize_antithesis();
 
     let cli: Cli = argh::from_env();
 
@@ -61,14 +60,7 @@ 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()
-        .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();
+    let bootstrap_config = load_bootstrap_config(&bootstrap_config_path)?.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).
@@ -122,6 +114,52 @@ async fn main() -> Result<(), GenericError> {
     Ok(())
 }
 
+/// Initializes the Antithesis SDK and installs a panic-reporting hook. Set
+/// ideally before any panics are possible.
+#[cfg(feature = "antithesis")]
+fn initialize_antithesis() {
+    antithesis_sdk::antithesis_init();
+
+    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);
+    }));
+}
+
+/// Loads bootstrap configuration from the on-disk file and environment
+/// variables.
+fn load_bootstrap_config(bootstrap_config_path: &Path) -> Result<ConfigurationLoader, GenericError> {
+    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.")
+        .and_then(|loader| {
+            loader
+                .add_providers([DatadogRemapper::new()])
+                .from_environment(PlatformSettings::get_env_var_prefix())
+                .error_context("Environment variable prefix should not be empty.")
+        });
+    // A graceful config rejection exits 1 rather than crashing; classify that against a clean boot.
+    #[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:?}")) })
+    );
+    loaded
+}
+
 fn parse_metrics_level(config: &GenericConfiguration) -> Result<Level, GenericError> {
     let raw = config
         .try_get_typed::<String>("metrics_level")
@@ -157,6 +195,13 @@ async fn run_inner(
                     }
                     Err(e) => {
                         error!("{:?}", e);
+                        // Same boot property as the config-load gate, 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)
                     }
                 };

lib/ddsketch/Cargo.toml

@@ -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/src/agent/sketch.rs

@@ -186,6 +186,11 @@ impl DDSketch {
     }
 
     fn adjust_basic_stats(&mut self, v: f64, n: u64) {
+        // Every insert path funnels through here, so this is where we guard that the incoming sample is finite. If
+        // this is not true something has gone wrong with the DogStatsD codec.
+        #[cfg(feature = "antithesis")]
+        antithesis_sdk::assert_always!(v.is_finite(), "DDSketch sample is finite at insert");
+
         if v < self.min {
             self.min = v;
         }
@@ -195,6 +200,9 @@ impl DDSketch {
         }
 
         self.count += n;
+        // It's possible that self.sum will be INF after this multiplication, even though we've demonstrated that `v`
+        // is finite. The Datadog Agent sketch sum behaves the same way, so we do not assert that self.sum is itself
+        // finite.
         self.sum += v * n as f64;
 
         if n == 1 {
@@ -711,6 +719,18 @@ 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);
+
+    // This is the one place every mutating method routes through, so asserting here guards the bin-count bound for
+    // all of them.
+    #[cfg(feature = "antithesis")]
+    {
+        antithesis_sdk::assert_reachable!("DDSketch bin collapse reached");
+        antithesis_sdk::assert_always_less_than_or_equal_to!(
+            bins.len(),
+            bin_limit,
+            "DDSketch bin count within bin_limit"
+        );
+    }
 }
 
 #[allow(clippy::cast_possible_truncation)]

lib/saluki-components/Cargo.toml

@@ -11,7 +11,16 @@ workspace = true
 [features]
 default = []
 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/src/sources/dogstatsd/mod.rs

@@ -1703,28 +1703,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 an invariant violation that
+        // crashes this component.
+        #[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.");
+
+            // Dispatch failure increments no counter, so this assertion is the only in-SUT signal that the failure
+            // path ran.
+            #[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" })
+            );
         }
     }
 
@@ -1736,6 +1767,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/replay/reader.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/transforms/aggregate/mod.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.