enhancement(antithesis): floor buffer size, multi-value packed metrics

test/antithesis/harness/src/bin/first_sample_config/config.rs

@@ -163,13 +163,26 @@ pub(crate) struct DogStatsdConfig {
     dogstatsd_stats_enable: bool,
 }
 
+/// Receive-buffer size in bytes. Usually realistic so lines actually arrive,
+/// rarely tiny or wild to probe the truncation edge. A sampled `0` leaves ADP
+/// no room past the 4-byte length prefix, so it drops every packet before
+/// decode and `finally_verify_delivery` sees nothing delivered end-to-end.
+/// Keep `0` and sub-128 values rare.
+fn sample_buffer_size<R: Rng + ?Sized>(rng: &mut R) -> u64 {
+    if rng.random_ratio(1, 16) {
+        Probe.sample(rng)
+    } else {
+        rng.random_range(128..=65_536)
+    }
+}
+
 impl DogStatsdConfig {
     /// Sample the `DogStatsD` options from `rng`, taking the socket from the
     /// environment.
     fn sample<R: Rng + ?Sized>(rng: &mut R, dogstatsd_socket: &Path) -> Self {
         Self {
             dogstatsd_socket: dogstatsd_socket.to_path_buf(),
-            dogstatsd_buffer_size: Probe.sample(rng),
+            dogstatsd_buffer_size: sample_buffer_size(rng),
             dogstatsd_so_rcvbuf: Probe.sample(rng),
             dogstatsd_packet_buffer_size: Probe.sample(rng),
             dogstatsd_packet_buffer_flush_timeout: Probe.sample(rng),

test/antithesis/harness/src/bin/parallel_driver_send_dogstatsd.rs

@@ -1,20 +1,14 @@
-//! Feral `DogStatsD` load generator. A producer thread writes sampled lines into
-//! a bounded channel; a consumer thread ships them over the socket. Antithesis
-//! runs many of these in parallel to drive concurrency and push context limits.
+//! Feral `DogStatsD` load generator. Drives one batch of sampled lines to the
+//! dogstatsd socket via the shared `harness::driver` engine. Antithesis runs
+//! many of these in parallel to drive concurrency and push context limits.
 
 #[cfg(unix)]
 mod unix_driver {
-    use std::os::unix::net::UnixDatagram;
-    use std::path::{Path, PathBuf};
-    use std::sync::mpsc::sync_channel;
-    use std::thread::{self, sleep};
-    use std::time::{Duration, Instant};
+    use std::path::PathBuf;
 
     use antithesis_sdk::prelude::*;
-    use antithesis_sdk::random::{random_choice, AntithesisRng};
     use clap::Parser;
-    use harness::payload::dogstatsd;
-    use rand::{rand_core::UnwrapErr, RngExt};
+    use harness::driver::{self, Batch};
     use serde_json::json;
 
     #[derive(Debug, Parser)]
@@ -28,108 +22,49 @@ mod unix_driver {
         dogstatsd_socket: PathBuf,
     }
 
-    /// Per-batch composition: 50% clean, 25% feral, 25% mixed.
-    #[derive(Clone, Copy)]
-    enum Batch {
-        Clean,
-        Feral,
-        Mixed,
-    }
-
     pub(super) fn run() -> anyhow::Result<()> {
         antithesis_init();
 
         let config = Config::try_parse()?;
 
         // Socket unavailable (ADP booting, or a fault). No-op exit, not a failure.
-        let Some(socket) = connect_with_retry(&config.dogstatsd_socket) else {
+        let Some(socket) = driver::connect_with_retry(&config.dogstatsd_socket) else {
             return Ok(());
         };
 
-        let batch = match random_choice(&[Batch::Clean, Batch::Clean, Batch::Feral, Batch::Mixed]) {
-            Some(Batch::Feral) => Batch::Feral,
-            Some(Batch::Mixed) => Batch::Mixed,
-            _ => Batch::Clean,
-        };
-        let count = {
-            let mut rng = UnwrapErr(AntithesisRng);
-            rng.random_range(0..=10_000u64)
-        };
-
-        let (tx, rx) = sync_channel::<Vec<u8>>(2024);
-
-        let producer = thread::spawn(move || {
-            let mut rng = UnwrapErr(AntithesisRng);
-            for _ in 0..count {
-                let vibe = match batch {
-                    Batch::Clean => dogstatsd::Vibe::Clean,
-                    Batch::Feral => dogstatsd::Vibe::Feral,
-                    Batch::Mixed => dogstatsd::sample_vibe(),
-                };
-                let mut line = Vec::new();
-                dogstatsd::send(&mut rng, &mut line, vibe);
-                if tx.send(line).is_err() {
-                    break;
-                }
-            }
-        });
-
-        let consumer = thread::spawn(move || {
-            let mut attempted = 0usize;
-            while let Ok(line) = rx.recv() {
-                if socket.send(&line).is_ok() {
-                    attempted += 1;
-                }
-            }
-            attempted
-        });
-
-        producer.join().expect("producer thread panicked");
-        let attempted = consumer.join().expect("consumer thread panicked");
+        let batch = Batch::sample();
+        let stats = driver::run(batch, vec![socket]);
+        let sent = stats.sent[0];
+        let max_packed = stats.max_packed[0];
 
         assert_reachable!(
             "workload ran a dogstatsd batch",
-            &json!({ "attempted": attempted, "dogstatsd_socket": config.dogstatsd_socket.display().to_string() })
+            &json!({ "sent": sent, "dogstatsd_socket": config.dogstatsd_socket.display().to_string() })
         );
+        assert_sometimes!(sent > 0, "workload sent a dogstatsd line", &json!({ "sent": sent }));
         assert_sometimes!(
-            attempted > 0,
-            "workload delivered a dogstatsd line",
-            &json!({ "attempted": attempted })
+            max_packed > 0,
+            "workload emitted a multi-value metric",
+            &json!({ "sent": sent, "max_packed_values": max_packed })
         );
         assert_sometimes!(
-            attempted > 0 && matches!(batch, Batch::Clean),
+            sent > 0 && matches!(batch, Batch::Clean),
             "workload ran a fully clean batch",
-            &json!({ "attempted": attempted })
+            &json!({ "sent": sent })
         );
         assert_sometimes!(
-            attempted > 0 && matches!(batch, Batch::Feral),
+            sent > 0 && matches!(batch, Batch::Feral),
             "workload ran a fully feral batch",
-            &json!({ "attempted": attempted })
+            &json!({ "sent": sent })
         );
         assert_sometimes!(
-            attempted > 0 && matches!(batch, Batch::Mixed),
+            sent > 0 && matches!(batch, Batch::Mixed),
             "workload ran a mixed batch",
-            &json!({ "attempted": attempted })
+            &json!({ "sent": sent })
         );
 
         Ok(())
     }
-
-    /// Wait for ADP to bind the socket, intentionally naive.
-    fn connect_with_retry(path: &Path) -> Option<UnixDatagram> {
-        let deadline = Instant::now() + Duration::from_secs(30);
-        loop {
-            if let Ok(socket) = UnixDatagram::unbound() {
-                if socket.connect(path).is_ok() {
-                    return Some(socket);
-                }
-            }
-            if Instant::now() >= deadline {
-                return None;
-            }
-            sleep(Duration::from_millis(250));
-        }
-    }
 }
 
 #[cfg(unix)]

test/antithesis/harness/src/driver.rs

@@ -0,0 +1,159 @@
+//! Shared `DogStatsD` load-driver engine.
+//!
+//! A producer thread samples lines into a bounded channel; a consumer thread
+//! fans each line out to every socket and tallies per-socket sends. Drivers
+//! differ only in how many sockets they target and which anchors they fire, so
+//! both the single-socket and differential drivers run on this one engine.
+
+use std::os::unix::net::UnixDatagram;
+use std::path::Path;
+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 crate::payload::dogstatsd;
+
+/// 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(),
+        }
+    }
+}
+
+/// 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,
+        }
+    }
+}
+
+/// 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.
+    pub received: usize,
+    /// Successful sends per socket, 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.
+///
+/// A producer samples up to ~10k lines at the batch's vibe and queues them; a
+/// consumer ships each line to every socket and tallies per-socket sends. The
+/// returned [`Stats`] anchor the caller's assertions.
+///
+/// # Panics
+///
+/// Panics if the producer or consumer thread panics.
+#[must_use]
+pub fn run(batch: Batch, sockets: Vec<UnixDatagram>) -> Stats {
+    let count = {
+        let mut rng = UnwrapErr(AntithesisRng);
+        rng.random_range(0..=10_000u64)
+    };
+
+    let (tx, rx) = sync_channel::<Line>(2024);
+
+    let producer = thread::spawn(move || {
+        let mut rng = UnwrapErr(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() {
+                break;
+            }
+        }
+    });
+
+    let consumer = thread::spawn(move || {
+        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() {
+            received += 1;
+            for (i, socket) in sockets.iter().enumerate() {
+                if socket.send(line.bytes()).is_ok() {
+                    sent[i] += 1;
+                    if let Line::Multi { count, .. } = &line {
+                        max_packed[i] = max_packed[i].max(*count);
+                    }
+                }
+            }
+        }
+        Stats {
+            received,
+            sent,
+            max_packed,
+        }
+    });
+
+    producer.join().expect("producer thread panicked");
+    consumer.join().expect("consumer thread panicked")
+}
+
+/// Wait for the remote process to bind `path`, intentionally naive. Returns
+/// `None` if the socket is still unavailable after 30 seconds.
+#[must_use]
+pub fn connect_with_retry(path: &Path) -> Option<UnixDatagram> {
+    let deadline = Instant::now() + Duration::from_secs(30);
+    loop {
+        if let Ok(socket) = UnixDatagram::unbound() {
+            if socket.connect(path).is_ok() {
+                return Some(socket);
+            }
+        }
+        if Instant::now() >= deadline {
+            return None;
+        }
+        sleep(Duration::from_millis(250));
+    }
+}

test/antithesis/harness/src/lib.rs

@@ -1,5 +1,7 @@
 //! Shared helpers for the Antithesis harness, used by the `src/bin/*` test
 //! commands.
 
+#[cfg(unix)]
+pub mod driver;
 pub mod payload;
 pub mod rand;

test/antithesis/harness/src/payload/dogstatsd.rs

@@ -102,11 +102,19 @@ fn choose_message<R: Rng + ?Sized>(rng: &mut R) -> Message {
 }
 
 /// Write one `DogStatsD` message of a sampled type to `buf` at the given vibe.
-pub fn send<R: Rng + ?Sized>(rng: &mut R, buf: &mut Vec<u8>, vibe: Vibe) {
+/// 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) {
-        Message::Event => events::write(rng, buf, vibe),
-        Message::ServiceCheck => service_checks::write(rng, buf, vibe),
+        Message::Event => {
+            events::write(rng, buf, vibe);
+            None
+        }
+        Message::ServiceCheck => {
+            service_checks::write(rng, buf, vibe);
+            None
+        }
         Message::Metric => metrics::write(rng, buf, vibe),
     }
 }