chore(antithesis): Parallel drivers retry sends

We modify parallel driver to tolerate both backpressure and network
faults. The goal of this work is to ensure we do not delver to one SUT
and not the other. All payloads rip through or the driver fails loudly.

Author: blt

Cargo.lock

@@ -17,6 +17,7 @@ clap = { workspace = true, features = [
   "usage",
 ] }
 itoa = { workspace = true }
+libc = { workspace = true }
 num-traits = { workspace = true }
 rand = { workspace = true }
 ryu = { workspace = true }

test/antithesis/harness/Cargo.toml

@@ -4,7 +4,11 @@
 //! 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.
+//!
+//! NOTE: this driver intentionally blocks on backpressure from the SUT. Retry
+//! and backoff timers are meant to endure transient errors.
 
+use std::io::ErrorKind;
 use std::os::unix::net::UnixDatagram;
 use std::path::Path;
 use std::sync::mpsc::sync_channel;
@@ -16,6 +20,9 @@ use rand::{rand_core::UnwrapErr, RngExt};
 
 use crate::payload::dogstatsd;
 
+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 {
@@ -82,17 +89,13 @@ pub struct Stats {
     pub max_packed: Vec<usize>,
 }
 
-/// Drive one batch of sampled `DogStatsD` lines to every socket.
+/// Drive one batch of sampled `DogStatsD` lines to every socket, blocking through
+/// backpressure so on success `sent[i] == received` for all `i`.
 ///
-/// 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.
+/// # Errors
 ///
-/// # Panics
-///
-/// Panics if the producer or consumer thread panics.
-#[must_use]
-pub fn run(batch: Batch, sockets: Vec<UnixDatagram>) -> Stats {
+/// Errors if a line 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);
         rng.random_range(0..=10_000u64)
@@ -114,30 +117,63 @@ pub fn run(batch: Batch, sockets: Vec<UnixDatagram>) -> Stats {
         }
     });
 
-    let consumer = thread::spawn(move || {
+    let consumer = thread::spawn(move || -> 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() {
+        'recv: 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);
+                match deliver(socket, line.bytes()) {
+                    Delivery::Sent => {
+                        sent[i] += 1;
+                        if let Line::Multi { count, .. } = &line {
+                            max_packed[i] = max_packed[i].max(*count);
+                        }
                     }
+                    // Peer left mid-batch after Antithesis killed the SUT. Stop and
+                    // report the partial batch rather than failing the run.
+                    Delivery::Unavailable => break 'recv,
                 }
             }
         }
-        Stats {
+        Ok(Stats {
             received,
             sent,
             max_packed,
-        }
+        })
     });
 
-    producer.join().expect("producer thread panicked");
-    consumer.join().expect("consumer thread panicked")
+    producer
+        .join()
+        .map_err(|_| anyhow::anyhow!("producer thread panicked"))?;
+    consumer
+        .join()
+        .map_err(|_| anyhow::anyhow!("consumer thread panicked"))?
+}
+
+/// Outcome of delivering one line to a socket.
+enum Delivery {
+    /// The line reached the socket.
+    Sent,
+    /// The peer is gone or backpressure outlasted the retry budget. Stop the batch.
+    Unavailable,
+}
+
+fn deliver(socket: &UnixDatagram, bytes: &[u8]) -> Delivery {
+    let deadline = Instant::now() + SEND_RETRY_BUDGET;
+    loop {
+        match socket.send(bytes) {
+            Ok(_) => return Delivery::Sent,
+            Err(e) if is_transient(&e) && Instant::now() < deadline => sleep(SEND_RETRY_BACKOFF),
+            Err(_) => return Delivery::Unavailable,
+        }
+    }
+}
+
+fn is_transient(error: &std::io::Error) -> bool {
+    matches!(error.kind(), ErrorKind::WouldBlock | ErrorKind::Interrupted)
+        || error.raw_os_error() == Some(libc::ENOBUFS)
 }
 
 /// Wait for the remote process to bind `path`, intentionally naive. Returns

test/antithesis/harness/src/driver.rs

@@ -33,7 +33,7 @@ mod unix_driver {
         };
 
         let batch = Batch::sample();
-        let stats = driver::run(batch, vec![socket]);
+        let stats = driver::run(batch, vec![socket])?;
         let sent = stats.sent[0];
         let max_packed = stats.max_packed[0];