chore(antithesis): is_malformed based load generation

The essential property we want to determine is whether ADP-on/ADP-off emits
payloads to intake API that the intake API rejects. This is especially of
interest for inputs that we know are ultimately rejected by intake API --
non-utf8 bytes in the wrong spot -- but are _not_ rejected by the SUT. The
old mechanism had a feral/clean 'vibe' which served for a while but was
confusing to debug. Is feral malformed? Is feral well-formed but wild?

Anyway I got tired of it. There's now a predicate which defines whether
a payload is well-formed or not -- that is, accepted by ADP-off Datadog Agent --
independent of whether intake API ultimately accepts the payloads that ingress
inspires. Later I will build a generator that only emits malformed ingress
but that is only hinted at in this work.

Author: blt

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

@@ -33,6 +33,10 @@ pub struct Payload {
     pub max_packed: usize,
 }
 
+/// One in this many datagrams gets a single non-UTF-8 byte spliced onto a metric name, so a
+/// controlled fraction of agent-lane v3 payloads carry a non-UTF-8 strict field and whole-reject.
+const NON_UTF8_PAYLOAD_RATE: u32 = 100;
+
 /// Pack whole `\n`-terminated lines into `buf`, each a fresh render of a sampled context, until the
 /// next line does not fit the space left under `limit_bytes`. That overflowing line ends the payload,
 /// so the packed datagram never exceeds `limit_bytes` and holds only whole lines the Agent forwards.
@@ -46,6 +50,7 @@ pub fn write_payload(
         return payload;
     }
     let mut line = Vec::new();
+    let mut line_starts = Vec::new();
     loop {
         line.clear();
         let context = &contexts[rng.random_range(0..contexts.len())];
@@ -54,11 +59,29 @@ pub fn write_payload(
         if buf.len() + line.len() + 1 > limit_bytes {
             break;
         }
+        line_starts.push(buf.len());
         buf.extend_from_slice(&line);
         buf.push(b'\n');
         payload.lines += 1;
         payload.max_packed = payload.max_packed.max(packed);
     }
+    // A budgeted fraction of datagrams carry one non-UTF-8 byte on a metric name -- the dictNameStr
+    // strict field that drives the v3 whole-payload reject. Events and service checks route to other
+    // endpoints, so target a metric line. The byte leaves the line `!is_malformed`, the Agent forwards
+    // a non-UTF-8 name, so the green-by-construction anchor holds while the intake still sees the
+    // reject at the budgeted rate. Rolling once per datagram, not per minted context, keeps the blast
+    // radius budgeted rather than compounding across every render of a bad-named context.
+    if rng.random_range(0..NON_UTF8_PAYLOAD_RATE) == 0 && buf.len() < limit_bytes {
+        let metric_starts: Vec<usize> = line_starts
+            .iter()
+            .copied()
+            .filter(|&i| !buf[i..].starts_with(b"_e{") && !buf[i..].starts_with(b"_sc"))
+            .collect();
+        if !metric_starts.is_empty() {
+            let at = metric_starts[rng.random_range(0..metric_starts.len())];
+            buf.insert(at, common::invalid_utf8_byte(rng));
+        }
+    }
     payload
 }
 
@@ -84,6 +107,35 @@ mod test {
         buf.iter().filter(|&&b| b == b'\n').count()
     }
 
+    /// Routine content and minted contexts are all valid UTF-8, so an invalid byte in a datagram means
+    /// the per-payload injection fired. It must fire at roughly the budgeted rate and never break
+    /// well-formedness -- a non-UTF-8 name is still a line the Agent forwards.
+    #[test]
+    fn non_utf8_injection_is_rare_and_stays_well_formed() {
+        let mut rng = SmallRng::seed_from_u64(0x00C0_FFEE);
+        let contexts = pool(&mut rng, 8);
+        let mut buf = Vec::new();
+        let total = 2000usize;
+        let mut injected = 0usize;
+        for _ in 0..total {
+            write_payload(&mut rng, &contexts, &mut buf, 1024);
+            assert_eq!(
+                is_malformed(&buf),
+                Ok(()),
+                "injection broke well-formedness: {:?}",
+                String::from_utf8_lossy(&buf)
+            );
+            if simdutf8::basic::from_utf8(&buf).is_err() {
+                injected += 1;
+            }
+        }
+        // ~1% expected; assert it is neither ~0 nor rampant.
+        assert!(
+            (3..120).contains(&injected),
+            "non-UTF-8 datagrams {injected}/{total}, expected ~1%"
+        );
+    }
+
     proptest! {
         /// The whole point: every datagram the driver packs from pooled contexts is one the Agent
         /// forwards. A packed datagram must be entirely well-formed.

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

@@ -28,18 +28,16 @@ const COMPLIANT_WORD: &[&[u8]] = &[
     b"workers",
 ];
 
-/// Aberrant identifier segments: whitespace, NUL, ill-formed and non-conforming UTF-8, and exotic
-/// Unicode. Omits `\n` and `\r`, which are datagram framing rather than content.
+/// Aberrant identifier segments: whitespace, NUL, and exotic but valid UTF-8. Invalid-UTF-8 bytes are
+/// deliberately absent. The per-datagram injection in `write_payload` adds one at a fixed rate rather
+/// than sprinkling them per segment, so their blast radius stays a budgeted fraction of datagrams
+/// instead of compounding across every minted context into a near-certain whole-payload v3 reject.
+/// Omits `\n` and `\r`, which are datagram framing rather than content.
 const ABERRANT_WORD: &[&[u8]] = &[
     b" ",
     b"\t",
     b"\0",
-    b"\x80",                // lone continuation byte
-    b"\xc3",                // truncated two-byte lead
-    b"\xed\xa0\x80",        // UTF-16 surrogate, ill-formed UTF-8
-    b"\xc0\x80",            // overlong NUL
-    b"\xff\xfe",            // non-character bytes
-    "café".as_bytes(),      // non-conforming but valid UTF-8
+    "café".as_bytes(),      // non-ASCII but valid UTF-8
     "Ωμέγα".as_bytes(),     // Greek
     "日本語".as_bytes(),    // CJK
     "🦆".as_bytes(),        // emoji, non-ASCII multi-byte
@@ -49,6 +47,14 @@ const ABERRANT_WORD: &[&[u8]] = &[
     "\u{feff}".as_bytes(),  // byte-order mark
 ];
 
+/// Bytes that are never valid UTF-8 in any position, for the per-datagram non-UTF-8 injection.
+const INVALID_UTF8: &[u8] = &[0x80, 0xC0, 0xC1, 0xF5, 0xFE, 0xFF];
+
+/// One byte that makes any surrounding ASCII field invalid UTF-8.
+pub(crate) fn invalid_utf8_byte(rng: &mut (impl Rng + ?Sized)) -> u8 {
+    INVALID_UTF8[rng.random_range(0..INVALID_UTF8.len())]
+}
+
 /// Message delimiters, mixed into content so the generator explores delimiter-bearing fields. Most
 /// land the message in the drop set and are repaired away; the survivors are the forwarded oddities.
 const DELIMITERS: &[&[u8]] = &[b":", b"|", b",", b"#", b"@"];