chore(antithesis): Parallel drivers retry sends (#1950)

## Summary
<!-- Please provide a brief summary about what this PR does.
This should help the reviewers give feedback faster and with higher quality. -->

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 exits with notice.

## Change Type
- [ ] Bug fix
- [ ] New feature
- [x] Non-functional (chore, refactoring, docs)
- [ ] Performance


## How did you test this PR?
<!-- Please how you tested these changes here -->

## References

<!-- Please list any issues closed by this PR. -->

<!--
- Closes: <issue link>
-->

<!-- Any other issues or PRs relevant to this PR? Feel free to list them here. -->

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 {
@@ -80,19 +87,19 @@ pub struct Stats {
     /// Largest packed run that reached each socket, indexed likewise. Zero when
     /// no multi-value line reached that socket.
     pub max_packed: Vec<usize>,
+    /// Whether a send exhausted the retry budget under sustained backpressure.
+    /// Distinguishes a wedged or paused peer from a clean partial batch.
+    pub timed_out: bool,
 }
 
-/// 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 worker thread panics. Sustained backpressure past the retry budget
+/// 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);
         rng.random_range(0..=10_000u64)
@@ -114,30 +121,78 @@ 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() {
+        let mut timed_out = false;
+        '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,
+                    // Backpressure outlasted the retry budget. A legit Antithesis
+                    // pause reaches here, so record it and stop rather than fail.
+                    Delivery::Timeout => {
+                        timed_out = true;
+                        break 'recv;
                     }
                 }
             }
         }
-        Stats {
+        Ok(Stats {
             received,
             sent,
             max_packed,
-        }
+            timed_out,
+        })
     });
 
-    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. Stop the batch and report the partial result.
+    Unavailable,
+    /// Backpressure outlasted the retry budget. Fail the run.
+    Timeout,
+}
+
+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) => {
+                if Instant::now() >= deadline {
+                    return Delivery::Timeout;
+                }
+                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,13 +33,17 @@ 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];
 
         assert_reachable!(
             "workload ran a dogstatsd batch",
-            &json!({ "sent": sent, "dogstatsd_socket": config.dogstatsd_socket.display().to_string() })
+            &json!({
+                "sent": sent,
+                "timed_out": stats.timed_out,
+                "dogstatsd_socket": config.dogstatsd_socket.display().to_string()
+            })
         );
         assert_sometimes!(sent > 0, "workload sent a dogstatsd line", &json!({ "sent": sent }));
         assert_sometimes!(