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

@@ -25,6 +25,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,74 +15,42 @@ use std::sync::mpsc::sync_channel;
 use std::thread::{self, sleep};
 use std::time::{Duration, Instant};
 
-use antithesis_sdk::random::{random_choice, AntithesisRng};
-use rand::{rand_core::UnwrapErr, RngExt};
+use antithesis_sdk::random::AntithesisRng;
+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.
@@ -92,30 +60,30 @@ pub struct Stats {
     pub timed_out: bool,
 }
 
-/// 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 transient backpressure so every payload reaches every socket.
+///
+/// A peer that leaves mid-batch, or backpressure that outlasts the retry budget,
+/// ends the run early with a partial [`Stats`] rather than an error.
 ///
 /// # Errors
 ///
-/// Errors if a worker thread panics. Sustained backpressure past the retry budget
-/// is reported via [`Stats::timed_out`], not as an error.
+/// Errors if a worker thread panics. Sustained backpressure is reported via
+/// [`Stats::timed_out`], not as an error.
 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;
             }
         }
@@ -126,15 +94,13 @@ pub fn run(batch: Batch, sockets: Vec<UnixDatagram>) -> anyhow::Result<Stats> {
         let mut sent = vec![0usize; sockets.len()];
         let mut max_packed = vec![0usize; sockets.len()];
         let mut timed_out = false;
-        'recv: while let Ok(line) = rx.recv() {
+        'recv: while let Ok(datagram) = rx.recv() {
             received += 1;
             for (i, socket) in sockets.iter().enumerate() {
-                match deliver(socket, line.bytes()) {
+                match deliver(socket, &datagram.bytes) {
                     Delivery::Sent => {
-                        sent[i] += 1;
-                        if let Line::Multi { count, .. } = &line {
-                            max_packed[i] = max_packed[i].max(*count);
-                        }
+                        sent[i] += datagram.payload.lines;
+                        max_packed[i] = max_packed[i].max(datagram.payload.max_packed);
                     }
                     // Peer left mid-batch after Antithesis killed the SUT. Stop and
                     // report the partial batch rather than failing the run.

test/antithesis/harness/src/driver.rs

@@ -101,13 +101,27 @@ 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`. When a line would
+/// exceed `limit`, drop it whole rather than shear it, leaving `buf` unchanged.
+/// A non-empty line always ends in `\n`.
 ///
 /// 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 +131,126 @@ 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 {
+        // Drop a whole line that exceeds the limit rather than shear it mid-token.
+        // A sheared fragment is exactly the parse-error spew this generator avoids.
+        buf.truncate(start);
+        return None;
+    }
+    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
+/// until the next would exceed `limit` total bytes. A line that overruns the
+/// remaining budget rolls back and ends the payload. A single line too large to
+/// fit at all gets skipped so a later, smaller line can still pack. `buf` holds
+/// only whole lines and never exceeds `limit`. Each line takes its vibe 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();
+    // Consecutive oversized lines write_line dropped. Bounded so a run of them
+    // cannot spin when `limit` is smaller than any line.
+    let mut skipped = 0u8;
+    loop {
+        let vibe = batch.vibe(rng);
+        let start = buf.len();
+        let packed = write_line(rng, buf, vibe, limit);
+        if buf.len() > limit {
+            // This whole line overruns the remaining budget. Roll it back and stop.
+            buf.truncate(start);
+            break;
+        }
+        if buf.len() == start {
+            // write_line dropped a line too large to fit at all. Skip it and try
+            // another rather than end the payload early.
+            skipped += 1;
+            if skipped >= 16 {
+                break;
+            }
+            continue;
+        }
+        skipped = 0;
+        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);
+        }
     }
 }