chore(antithesis): Expand cheap-win antithesis SDK assertions

This commit is a child to #1928. It's meant to demonstrate how the new
saluki-antithesis works in a bulk change and also, incidentally, flag
things to Antithesis that demonstrate bugs we've previously found.

See for instance PR #1923.

Author: blt

lib/ddsketch/src/agent/sketch.rs

@@ -198,6 +198,8 @@ impl DDSketch {
             self.max = v;
         }
 
+        saluki_antithesis::always_le!(self.min, self.max, "DDSketch min does not exceed max after insert");
+
         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

lib/saluki-components/src/encoders/datadog/metrics/mod.rs

@@ -977,7 +977,9 @@ fn encode_series_v2_metric(
             .map(|interval| value / interval.as_secs_f64())
             .unwrap_or(value);
 
-        if !emittable(value) {
+        let dropped = !emittable(value);
+        saluki_antithesis::sometimes!(dropped, "non-finite series point dropped before wire");
+        if dropped {
             continue;
         }
 
@@ -1016,7 +1018,9 @@ fn encode_series_v1_metric(
             .map(|interval| value / interval.as_secs_f64())
             .unwrap_or(value);
 
-        if !emittable(value) {
+        let dropped = !emittable(value);
+        saluki_antithesis::sometimes!(dropped, "non-finite series point dropped before wire");
+        if dropped {
             continue;
         }
 

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

@@ -670,7 +670,18 @@ impl DogStatsDConfiguration {
     fn effective_context_string_interner_bytes(&self) -> ByteSize {
         match self.context_string_interner_size_bytes {
             Some(explicit_bytes) => explicit_bytes,
-            None => ByteSize::b(self.context_string_interner_entry_count * INTERNER_BASELINE_BYTES_PER_ENTRY),
+            None => {
+                saluki_antithesis::always_le!(
+                    self.context_string_interner_entry_count,
+                    u64::MAX / INTERNER_BASELINE_BYTES_PER_ENTRY,
+                    "dogstatsd interner byte-size multiply does not overflow",
+                    { "entry_count": self.context_string_interner_entry_count }
+                );
+                ByteSize::b(
+                    self.context_string_interner_entry_count
+                        .saturating_mul(INTERNER_BASELINE_BYTES_PER_ENTRY),
+                )
+            }
         }
     }
 
@@ -1133,6 +1144,12 @@ async fn process_io_buffer_pool_shrinker(
 fn build_io_buffer_pool(
     min_buffers: usize, max_buffers: usize, buffer_size: usize,
 ) -> (ElasticObjectPool<BytesBuffer>, impl Future<Output = ()> + Send) {
+    saluki_antithesis::always_le!(
+        buffer_size,
+        usize::MAX - 4,
+        "dogstatsd buffer size add does not overflow",
+        { "buffer_size": buffer_size }
+    );
     let adjusted_buffer_size = get_adjusted_buffer_size(buffer_size);
     ElasticObjectPool::with_builder("dsd_packet_bufs", min_buffers, max_buffers, move || {
         FixedSizeVec::with_capacity(adjusted_buffer_size)

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

@@ -64,7 +64,11 @@ impl HistogramStatistic {
             HistogramStatistic::Maximum => summary.max().unwrap_or(0.0),
             HistogramStatistic::Average => summary.avg(),
             HistogramStatistic::Median => summary.median().unwrap_or(0.0),
-            HistogramStatistic::Percentile { q, .. } => summary.quantile(*q).unwrap_or(0.0),
+            HistogramStatistic::Percentile { q, .. } => {
+                saluki_antithesis::always_ge!(*q, 0.0, "histogram percentile quantile at or above zero");
+                saluki_antithesis::always_le!(*q, 1.0, "histogram percentile quantile at or below one");
+                summary.quantile(*q).unwrap_or(0.0)
+            }
         }
     }
 }

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

@@ -700,7 +700,13 @@ impl AggregationState {
             // This is useful for sparsely-updated counters.
             let should_expire_if_empty = match &am.values {
                 MetricValues::Counter(..) => {
-                    counter_expire_secs != 0 && am.last_seen + counter_expire_secs < current_time
+                    saluki_antithesis::always_le!(
+                        am.last_seen,
+                        u64::MAX - counter_expire_secs,
+                        "aggregate counter expiry add does not overflow",
+                        { "last_seen": am.last_seen, "counter_expire_secs": counter_expire_secs }
+                    );
+                    counter_expire_secs != 0 && am.last_seen.saturating_add(counter_expire_secs) < current_time
                 }
                 _ => true,
             };
@@ -711,7 +717,7 @@ impl AggregationState {
             // This is also safe to do even when there are real values in those buckets since adding zero to anything is
             // a no-op from the perspective of what we end up flushing, and it doesn't mess with the "last seen" time.
             if let MetricValues::Counter(..) = &mut am.values {
-                let expires_at = am.last_seen + counter_expire_secs;
+                let expires_at = am.last_seen.saturating_add(counter_expire_secs);
                 for (zv_bucket_start, zero_value) in &zero_value_buckets {
                     if expires_at > *zv_bucket_start {
                         am.values.merge(zero_value.clone());

lib/saluki-core/src/topology/interconnect/dispatcher.rs

@@ -102,6 +102,7 @@ where
         let item_count = item.item_count();
 
         // Send the item to all senders except the last one by cloning the item.
+        saluki_antithesis::always_gt!(self.senders.len(), 0, "dispatcher fanout has at least one sender");
         let cloned_sends = self.senders.len() - 1;
         for sender in &self.senders[0..cloned_sends] {
             sender

lib/saluki-io/src/buf/vec.rs

@@ -93,6 +93,7 @@ impl Buf for BytesBuffer {
 
     fn advance(&mut self, cnt: usize) {
         let data = self.data_mut();
+        saluki_antithesis::always_le!(data.read_idx + cnt, data.data.len(), "buffer advance within bounds");
         assert!(data.read_idx + cnt <= data.data.len());
         data.read_idx += cnt;
     }
@@ -193,6 +194,11 @@ impl Buf for FrozenBytesBuffer {
     }
 
     fn advance(&mut self, cnt: usize) {
+        saluki_antithesis::always_le!(
+            self.data.read_idx + cnt,
+            self.data.data.len(),
+            "buffer advance within bounds"
+        );
         assert!(self.data.read_idx + cnt <= self.data.data.len());
         self.data.read_idx += cnt;
     }