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

test/antithesis/harness/src/driver.rs

@@ -4,18 +4,26 @@
 //! 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;
 use std::thread::{self, sleep};
 use std::time::{Duration, Instant};
 
 use antithesis_sdk::random::{random_choice, AntithesisRng};
+use anyhow::Context;
 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 +90,13 @@ pub struct Stats {
     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.
+/// Drive one batch of sampled `DogStatsD` lines to every socket, blocking through
+/// backpressure so on success `sent[i] == received` for all `i`.
 ///
-/// # Panics
+/// # Errors
 ///
-/// 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 +118,49 @@ 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() {
             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);
-                    }
+                deliver(socket, line.bytes())?;
+                sent[i] += 1;
+                if let Line::Multi { count, .. } = &line {
+                    max_packed[i] = max_packed[i].max(*count);
                 }
             }
         }
-        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"))?
+}
+
+fn deliver(socket: &UnixDatagram, bytes: &[u8]) -> anyhow::Result<()> {
+    let deadline = Instant::now() + SEND_RETRY_BUDGET;
+    loop {
+        match socket.send(bytes) {
+            Ok(_) => return Ok(()),
+            Err(e) if is_transient(&e) && Instant::now() < deadline => sleep(SEND_RETRY_BACKOFF),
+            Err(e) => return Err(e).context("deliver line to socket"),
+        }
+    }
+}
+
+fn is_transient(error: &std::io::Error) -> bool {
+    const ENOBUFS: i32 = 105;
+    matches!(error.kind(), ErrorKind::WouldBlock | ErrorKind::Interrupted) || error.raw_os_error() == Some(ENOBUFS)
 }
 
 /// Wait for the remote process to bind `path`, intentionally naive. Returns

test/antithesis/scenarios/general/src/bin/parallel_driver_send_dogstatsd.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];