chore(antithesis): dogstatsd generation with byte limit

This commit updates the harness driver to build dogstatsd to a byte
limit, in the manner of datadog/lading. I have been unable to
investigate SMPTNG-7611 well owing to the error log emission by Datadog
Agent. I've added property tests to assert the payload limit is obeyed,
accepting that this means making the dogstatsd generator pure with
regard to Rng and may not use antithesis SDK's random_* directly.

Author: blt

Cargo.lock

@@ -24,6 +24,9 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 
+[dev-dependencies]
+proptest = { workspace = true }
+
 [lints.clippy]
 all = "deny"
 complexity = "deny"

test/antithesis/harness/Cargo.toml

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 377cba92b14295b3fe5c6f2738a398ee042be8e35f529398ecf9e275e6b0a1da # shrinks to seed = 0

test/antithesis/harness/proptest-regressions/payload/dogstatsd.txt

@@ -15,104 +15,69 @@ use std::sync::mpsc::sync_channel;
 use std::thread::{self, sleep};
 use std::time::{Duration, Instant};
 
-use antithesis_sdk::random::{random_choice, AntithesisRng};
+use antithesis_sdk::random::AntithesisRng;
 use anyhow::Context;
-use rand::{rand_core::UnwrapErr, RngExt};
+use rand::seq::IndexedRandom;
+use rand::RngExt;
 
 use crate::payload::dogstatsd;
+pub use crate::payload::dogstatsd::Batch;
 
 const SEND_RETRY_BUDGET: Duration = Duration::from_secs(5);
 const SEND_RETRY_BACKOFF: Duration = Duration::from_millis(1);
 
-/// Per-batch composition: 50% clean, 25% feral, 25% mixed.
-#[derive(Clone, Copy, Debug)]
-pub enum Batch {
-    /// Every line clean.
-    Clean,
-    /// Every line feral.
-    Feral,
-    /// A per-line clean-or-feral mix.
-    Mixed,
-}
-
-impl Batch {
-    /// Sample a batch composition: half clean, a quarter feral, a quarter mixed.
-    #[must_use]
-    pub fn sample() -> Self {
-        match random_choice(&[Batch::Clean, Batch::Clean, Batch::Feral, Batch::Mixed]) {
-            Some(Batch::Feral) => Batch::Feral,
-            Some(Batch::Mixed) => Batch::Mixed,
-            _ => Batch::Clean,
-        }
-    }
-
-    /// The vibe for one line drawn from this batch.
-    fn vibe(self) -> dogstatsd::Vibe {
-        match self {
-            Batch::Clean => dogstatsd::Vibe::Clean,
-            Batch::Feral => dogstatsd::Vibe::Feral,
-            Batch::Mixed => dogstatsd::sample_vibe(),
-        }
+/// Sample a line composition: half clean, a quarter feral, a quarter mixed.
+#[must_use]
+pub fn sample() -> Batch {
+    let mut rng = AntithesisRng;
+    match [Batch::Clean, Batch::Clean, Batch::Feral, Batch::Mixed].choose(&mut rng) {
+        Some(Batch::Feral) => Batch::Feral,
+        Some(Batch::Mixed) => Batch::Mixed,
+        _ => Batch::Clean,
     }
 }
 
-/// A generated dogstatsd line queued for the sockets.
-enum Line {
-    /// A single-value line.
-    Single { bytes: Vec<u8> },
-    /// A multi-value `:`-packed metric.
-    Multi {
-        /// The encoded line.
-        bytes: Vec<u8>,
-        /// The number of values in the packed run.
-        count: usize,
-    },
-}
-
-impl Line {
-    /// The encoded bytes to ship over a socket.
-    fn bytes(&self) -> &[u8] {
-        match self {
-            Line::Single { bytes } | Line::Multi { bytes, .. } => bytes,
-        }
-    }
+/// A generated payload queued for the sockets: the packed bytes and what they hold.
+struct Datagram {
+    /// The `\n`-packed payload bytes to ship over a socket.
+    bytes: Vec<u8>,
+    /// The lines and largest packed run in `bytes`.
+    payload: dogstatsd::Payload,
 }
 
 /// What a driver run shipped, for anchoring assertions.
 #[derive(Clone, Debug)]
 pub struct Stats {
-    /// Lines pulled from the channel, whether or not any send succeeded.
+    /// Payloads pulled from the channel, whether or not any send succeeded.
     pub received: usize,
-    /// Successful sends per socket, indexed as the sockets were passed to [`run`].
+    /// Lines delivered per socket, summed across payloads, indexed as the sockets
+    /// were passed to [`run`].
     pub sent: Vec<usize>,
     /// Largest packed run that reached each socket, indexed likewise. Zero when
     /// no multi-value line reached that socket.
     pub max_packed: Vec<usize>,
 }
 
-/// Drive one batch of sampled `DogStatsD` lines to every socket, blocking through
-/// backpressure so on success `sent[i] == received` for all `i`.
+/// Drive a batch of sampled `DogStatsD` payloads to every socket, blocking
+/// through backpressure so every payload reaches every socket.
 ///
 /// # Errors
 ///
-/// Errors if a line cannot be delivered within the retry budget, or a worker panics.
+/// Errors if a payload cannot be delivered within the retry budget, or a worker panics.
 pub fn run(batch: Batch, sockets: Vec<UnixDatagram>) -> anyhow::Result<Stats> {
     let count = {
-        let mut rng = UnwrapErr(AntithesisRng);
+        let mut rng = AntithesisRng;
         rng.random_range(0..=10_000u64)
     };
 
-    let (tx, rx) = sync_channel::<Line>(2024);
+    let (tx, rx) = sync_channel::<Datagram>(2024);
 
     let producer = thread::spawn(move || {
-        let mut rng = UnwrapErr(AntithesisRng);
+        let mut rng = AntithesisRng;
         for _ in 0..count {
             let mut bytes = Vec::new();
-            let line = match dogstatsd::send(&mut rng, &mut bytes, batch.vibe()) {
-                None => Line::Single { bytes },
-                Some(count) => Line::Multi { bytes, count },
-            };
-            if tx.send(line).is_err() {
+            let payload = dogstatsd::write_payload(&mut rng, &mut bytes, batch, dogstatsd::PAYLOAD_BYTE_LIMIT);
+            if tx.send(Datagram { bytes, payload }).is_err() {
                 break;
             }
         }
@@ -122,14 +87,12 @@ pub fn run(batch: Batch, sockets: Vec<UnixDatagram>) -> anyhow::Result<Stats> {
         let mut received = 0usize;
         let mut sent = vec![0usize; sockets.len()];
         let mut max_packed = vec![0usize; sockets.len()];
-        while let Ok(line) = rx.recv() {
+        while let Ok(datagram) = rx.recv() {
             received += 1;
             for (i, socket) in sockets.iter().enumerate() {
-                deliver(socket, line.bytes())?;
-                sent[i] += 1;
-                if let Line::Multi { count, .. } = &line {
-                    max_packed[i] = max_packed[i].max(*count);
-                }
+                deliver(socket, &datagram.bytes)?;
+                sent[i] += datagram.payload.lines;
+                max_packed[i] = max_packed[i].max(datagram.payload.max_packed);
             }
         }
         Ok(Stats {

test/antithesis/harness/src/driver.rs

@@ -101,13 +101,26 @@ fn choose_message<R: Rng + ?Sized>(rng: &mut R) -> Message {
     }
 }
 
-/// Write one `DogStatsD` message of a sampled type to `buf` at the given vibe.
+/// The `dogstatsd_buffer_size` default, via Datadog Agent.
+pub const PAYLOAD_BYTE_LIMIT: usize = 8_192;
+
+/// What a generated payload holds, for anchoring assertions.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct Payload {
+    /// Lines packed into the buffer.
+    pub lines: usize,
+    /// Largest packed multi-value run among those lines. Zero when none.
+    pub max_packed: usize,
+}
+
+/// Append one `DogStatsD` line of a sampled type to `buf`, capping the appended
+/// bytes at `limit`. A non-empty line is always `\n`-terminated.
 ///
 /// Returns the packed value count when a multi-value metric was emitted, else
 /// `None`.
-pub fn send<R: Rng + ?Sized>(rng: &mut R, buf: &mut Vec<u8>, vibe: Vibe) -> Option<usize> {
-    buf.clear();
-    match choose_message(rng) {
+pub fn write_line<R: Rng + ?Sized>(rng: &mut R, buf: &mut Vec<u8>, vibe: Vibe, limit: usize) -> Option<usize> {
+    let start = buf.len();
+    let packed = match choose_message(rng) {
         Message::Event => {
             events::write(rng, buf, vibe);
             None
@@ -117,5 +130,120 @@ pub fn send<R: Rng + ?Sized>(rng: &mut R, buf: &mut Vec<u8>, vibe: Vibe) -> Opti
             None
         }
         Message::Metric => metrics::write(rng, buf, vibe),
+    };
+    if buf.len() - start > limit {
+        buf.truncate(start + limit);
+        // Restore the terminator the truncation dropped. Lines carry no interior
+        // newline, so the result still holds exactly one.
+        if buf.len() > start {
+            if let Some(last) = buf.last_mut() {
+                *last = b'\n';
+            }
+        }
+    }
+    packed
+}
+
+/// Per-run line composition: every line clean, every line feral, or a per-line
+/// clean-or-feral mix.
+#[derive(Clone, Copy, Debug)]
+pub enum Batch {
+    /// Every line clean.
+    Clean,
+    /// Every line feral.
+    Feral,
+    /// Each line independently clean or feral.
+    Mixed,
+}
+
+impl Batch {
+    /// The vibe for one line of this batch, sampled per call so `Mixed` interleaves.
+    fn vibe<R: Rng + ?Sized>(self, rng: &mut R) -> Vibe {
+        match self {
+            Batch::Clean => Vibe::Clean,
+            Batch::Feral => Vibe::Feral,
+            Batch::Mixed => sample_vibe(rng),
+        }
+    }
+}
+
+/// Fill `buf` with `\n`-terminated lines, packing whole lines straight into it
+/// and drawing the `limit` budget down as each lands. When a line overruns the
+/// remaining budget it is rolled back and packing stops, so `buf` holds only
+/// whole lines and never exceeds `limit`. Each line's vibe is sampled from
+/// `batch`, so a `Mixed` payload interleaves clean and feral lines. Clears `buf`
+/// first.
+pub fn write_payload<R: Rng + ?Sized>(rng: &mut R, buf: &mut Vec<u8>, batch: Batch, limit: usize) -> Payload {
+    buf.clear();
+    let mut payload = Payload::default();
+    loop {
+        let vibe = batch.vibe(rng);
+        let start = buf.len();
+        let packed = write_line(rng, buf, vibe, limit);
+        if buf.len() > limit {
+            // The line overran the remaining budget. Roll it back and stop.
+            buf.truncate(start);
+            break;
+        }
+        if buf.len() == start {
+            // Nothing written — the budget is spent.
+            break;
+        }
+        payload.lines += 1;
+        if let Some(count) = packed {
+            payload.max_packed = payload.max_packed.max(count);
+        }
+    }
+    payload
+}
+
+#[cfg(test)]
+mod test {
+    use proptest::prelude::*;
+    use rand::rngs::SmallRng;
+    use rand::SeedableRng;
+
+    use super::{write_line, write_payload, Batch, Vibe};
+
+    fn any_vibe() -> impl Strategy<Value = Vibe> {
+        prop_oneof![Just(Vibe::Clean), Just(Vibe::Feral)]
+    }
+
+    fn any_batch() -> impl Strategy<Value = Batch> {
+        prop_oneof![Just(Batch::Clean), Just(Batch::Feral), Just(Batch::Mixed)]
+    }
+
+    /// Lines carry no interior newline and each is `\n`-terminated, so the line
+    /// count equals the newline count.
+    #[allow(clippy::naive_bytecount)]
+    fn newline_count(buf: &[u8]) -> usize {
+        buf.iter().filter(|&&b| b == b'\n').count()
+    }
+
+    proptest! {
+        #[test]
+        fn write_line_stays_within_its_limit(seed: u64, limit: u16, vibe in any_vibe()) {
+            let mut rng = SmallRng::seed_from_u64(seed);
+            let limit = usize::from(limit);
+            let mut buf = Vec::new();
+            write_line(&mut rng, &mut buf, vibe, limit);
+
+            prop_assert!(buf.len() <= limit);
+            if !buf.is_empty() {
+                prop_assert_eq!(buf[buf.len() - 1], b'\n');
+                prop_assert_eq!(newline_count(&buf), 1);
+            }
+        }
+
+        #[test]
+        fn write_payload_stays_within_its_limit(seed: u64, limit: u16, batch in any_batch()) {
+            let mut rng = SmallRng::seed_from_u64(seed);
+            let limit = usize::from(limit);
+            let mut buf = Vec::new();
+            let payload = write_payload(&mut rng, &mut buf, batch, limit);
+
+            prop_assert!(buf.len() <= limit);
+            prop_assert_eq!(newline_count(&buf), payload.lines);
+        }
     }
 }