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

Cargo.lock

@@ -6,3 +6,6 @@
 # everyone who runs the test benefits from these saved cases.
 cc 377cba92b14295b3fe5c6f2738a398ee042be8e35f529398ecf9e275e6b0a1da # shrinks to seed = 0
 cc 1f807f27fdbf0b983fc773b815b21d6023e1bce87691d07133b438b862edcc5b # shrinks to seed = 6079028945602138863, limit_bytes = 21
+cc 966dfab50a6a0af70bafa9c882cd664e62586012ec563187766e7b07773b588e # shrinks to seed = 816969264511539406, limit_bytes = 9
+cc 4470fd2185844af9954fc8e973eee2ac8f6c939167fb7d950e448a5ca2d2b52a # shrinks to seed = 881271281136835980, limit_bytes = 8
+cc c084e24d26194bedbf39e3055ace5c28a2d3ce2f2b071be7277a18f49e568849 # shrinks to seed = 6133403856440011550

test/antithesis/harness/proptest-regressions/payload/dogstatsd.txt

@@ -73,7 +73,7 @@ fn main() -> anyhow::Result<()> {
     let context_source_path = cli.config_dir.join("context_source.yaml");
     fs::write(
         &context_source_path,
-        ContextSourceConfig::sample(&mut rng, driver.payload_byte_limit)
+        ContextSourceConfig::sample(&mut rng, driver.datagram_byte_limit)
             .to_yaml()?
             .as_bytes(),
     )

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

@@ -15,7 +15,7 @@ use rand::distr::{Distribution, StandardUniform};
 use rand::{Rng, RngExt};
 use serde::{Deserialize, Serialize};
 
-use crate::payload::dogstatsd::PAYLOAD_BYTE_LIMIT;
+use crate::payload::dogstatsd::DATAGRAM_BYTE_LIMIT;
 use crate::rand::Probe;
 
 /// Agent log level.
@@ -258,9 +258,9 @@ const MAX_CONTEXTS_PER_REQUEST: usize = 65_536;
 #[derive(Clone, Copy, Debug, Serialize, Deserialize)]
 pub struct DriverConfig {
     /// Max bytes a generator packs into one datagram, the smaller of the SUT's
-    /// sampled receive buffer and [`PAYLOAD_BYTE_LIMIT`]. A datagram this size
+    /// sampled receive buffer and [`DATAGRAM_BYTE_LIMIT`]. A datagram this size
     /// fits one read, so the SUT never truncates a line mid-token.
-    pub payload_byte_limit: usize,
+    pub datagram_byte_limit: usize,
     /// Datagrams a driver invocation ships this timeline.
     pub datagram_count: usize,
     /// Distinct contexts a driver invocation fetches from the shared pool as its working set.
@@ -276,14 +276,14 @@ pub struct DriverConfig {
 impl DriverConfig {
     /// Sample the driver knobs for a SUT whose receive buffer is `buffer_size`.
     fn sample<R: Rng + ?Sized>(rng: &mut R, buffer_size: u64) -> Self {
-        // The min is at most PAYLOAD_BYTE_LIMIT, so a buffer wider than usize
+        // The min is at most DATAGRAM_BYTE_LIMIT, so a buffer wider than usize
         // caps to the ceiling like any other oversized buffer.
-        let payload_byte_limit = match usize::try_from(buffer_size.min(PAYLOAD_BYTE_LIMIT as u64)) {
+        let datagram_byte_limit = match usize::try_from(buffer_size.min(DATAGRAM_BYTE_LIMIT as u64)) {
             Ok(bytes) => bytes,
-            Err(_) => PAYLOAD_BYTE_LIMIT,
+            Err(_) => DATAGRAM_BYTE_LIMIT,
         };
         Self {
-            payload_byte_limit,
+            datagram_byte_limit,
             datagram_count: rng.random_range(0..=MAX_DATAGRAMS),
             context_count: usize::try_from(Probe::new(1, MAX_WORKING_SET).sample(rng)).unwrap_or(usize::MAX),
         }
@@ -304,7 +304,7 @@ impl DriverConfig {
     /// # Errors
     ///
     /// Returns an error if the config is unreadable or is not valid YAML with an
-    /// integer `payload_byte_limit`.
+    /// integer `datagram_byte_limit`.
     pub fn read(config_dir: &Path) -> anyhow::Result<Self> {
         let path = config_dir.join("driver.yaml");
         let yaml = fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
@@ -333,14 +333,16 @@ const MAX_CONTEXTS_TOTAL: u64 = 1_000_000;
 pub struct ContextSourceConfig {
     /// Bytes a rendered line of any pooled context must fit, this timeline's real datagram budget
     /// rather than the protocol ceiling. Mint builds identities against it, so every served context
-    /// has a rendering the driver can pack. Defaults to the sampled `payload_byte_limit`, which is the
-    /// smaller of the SUT's receive buffer and [`PAYLOAD_BYTE_LIMIT`].
-    pub payload_byte_limit: usize,
+    /// has a rendering the driver can pack. Defaults to the sampled `datagram_byte_limit`, which is the
+    /// smaller of the SUT's receive buffer and [`DATAGRAM_BYTE_LIMIT`].
+    pub datagram_byte_limit: usize,
     /// Distinct metric contexts the pool holds before it recurs the ones it has.
     ///
-    /// Sampled in `1..=MAX_CONTEXTS_TOTAL` minus what the other kinds took. A larger cap explores more
+    /// Sampled in `2..=MAX_CONTEXTS_TOTAL` minus what the other kinds took. A larger cap explores more
     /// identities and puts fewer points in each, and costs memory: the pool retains every context it
-    /// mints for the life of the run. Zero is not sampled and serves nothing for the kind.
+    /// mints for the life of the run. Two is the floor because the pool holds one context carrying an
+    /// invalid UTF-8 byte per kind alongside the rest, and a kind capped at one could hold only one of
+    /// the two.
     pub metric_contexts: usize,
     /// Distinct event contexts the pool holds. Same range and trade as [`Self::metric_contexts`].
     pub event_contexts: usize,
@@ -354,16 +356,16 @@ impl ContextSourceConfig {
     ///
     /// The draws run in order and each spends from one shared ceiling, so the total is bounded by
     /// construction rather than by scaling three independent draws afterwards. Every kind keeps at
-    /// least one context, since a kind capped at zero panics the pool's draw.
+    /// least two contexts, one of which carries an invalid UTF-8 byte.
     #[must_use]
-    pub fn sample<R: Rng + ?Sized>(rng: &mut R, payload_byte_limit: usize) -> Self {
-        // Two contexts held back so the later kinds can each keep their one.
-        let metric_contexts = sample_cap(rng, MAX_CONTEXTS_TOTAL - 2);
+    pub fn sample<R: Rng + ?Sized>(rng: &mut R, datagram_byte_limit: usize) -> Self {
+        // Four contexts held back so the two later kinds can each keep their two.
+        let metric_contexts = sample_cap(rng, MAX_CONTEXTS_TOTAL - 4);
         let free = MAX_CONTEXTS_TOTAL - metric_contexts as u64;
-        let event_contexts = sample_cap(rng, free - 1);
+        let event_contexts = sample_cap(rng, free - 2);
         let free = free - event_contexts as u64;
         Self {
-            payload_byte_limit,
+            datagram_byte_limit,
             metric_contexts,
             event_contexts,
             service_check_contexts: sample_cap(rng, free),
@@ -392,10 +394,10 @@ impl ContextSourceConfig {
     }
 }
 
-/// A single per-kind cap in `1..=ceiling`. The ceiling is well within `usize` on every supported
+/// A single per-kind cap in `2..=ceiling`. The ceiling is well within `usize` on every supported
 /// target, so the saturating conversion is unreachable in practice.
 fn sample_cap<R: Rng + ?Sized>(rng: &mut R, ceiling: u64) -> usize {
-    usize::try_from(Probe::new(1, ceiling.max(1)).sample(rng)).unwrap_or(usize::MAX)
+    usize::try_from(Probe::new(2, ceiling.max(2)).sample(rng)).unwrap_or(usize::MAX)
 }
 
 #[cfg(test)]
@@ -452,12 +454,12 @@ mod tests {
 
     #[test]
     fn driver_config_caps_payload_to_the_smaller_bound() {
-        assert_eq!(DriverConfig::sample(&mut SeqRng(0), 512).payload_byte_limit, 512);
+        assert_eq!(DriverConfig::sample(&mut SeqRng(0), 512).datagram_byte_limit, 512);
         assert_eq!(
-            DriverConfig::sample(&mut SeqRng(0), 1 << 30).payload_byte_limit,
-            PAYLOAD_BYTE_LIMIT
+            DriverConfig::sample(&mut SeqRng(0), 1 << 30).datagram_byte_limit,
+            DATAGRAM_BYTE_LIMIT
         );
-        assert_eq!(DriverConfig::sample(&mut SeqRng(0), 0).payload_byte_limit, 0);
+        assert_eq!(DriverConfig::sample(&mut SeqRng(0), 0).datagram_byte_limit, 0);
     }
 
     #[test]

test/antithesis/harness/src/config.rs

@@ -14,17 +14,18 @@
 use rand::{Rng, RngExt};
 
 use crate::dogstatsd::is_malformed;
+use crate::payload::dogstatsd::common;
 
 pub mod event;
 pub mod metric;
 pub mod service_check;
 
-/// How many times to re-mint an identity whose probe render the Agent would drop before falling back
-/// to a guaranteed-sound identity. Mint is mostly-valid, so the fallback is rare.
+/// How many times to re-mint an identity whose probe render the Agent would drop before yielding
+/// nothing. Mint is mostly-valid, so an exhausted loop is rare.
 const REMINT_TRIES: usize = 16;
 
 /// How many times to re-render a context whose per-occurrence payload the Agent would drop before
-/// falling back to a guaranteed-well-formed line.
+/// yielding nothing. 2.3% of single renders need a retry and none has yet exhausted the loop.
 const RENDER_TRIES: usize = 8;
 
 /// Digits allowed for a rendered length field, generous so a floor is never an underestimate.
@@ -105,6 +106,82 @@ impl Context {
         }
     }
 
+    /// Replace one byte of this identity with an invalid UTF-8 byte, so the identity itself is the
+    /// corrupt one rather than a datagram being edited after the fact.
+    ///
+    /// The Agent does no charset validation, so the line still forwards and the criterion stays "does
+    /// the Agent discard it". Which identity field takes the byte is what the intake distinguishes: a
+    /// v3 name dictionary rejects the whole payload, a tag dictionary coerces. A delimiter is never
+    /// overwritten, since removing one reshapes the line. Returns whether a byte was replaced.
+    fn poison(&mut self, rng: &mut (impl Rng + ?Sized)) -> bool {
+        let fields: Vec<&mut Vec<u8>> = match self {
+            Context::Metric(c) => std::iter::once(&mut c.name).chain(c.tags.iter_mut()).collect(),
+            Context::Event(c) => std::iter::once(&mut c.title).chain(c.tags.iter_mut()).collect(),
+            Context::ServiceCheck(c) => std::iter::once(&mut c.name).chain(c.tags.iter_mut()).collect(),
+        };
+        let targets: Vec<(usize, usize)> = fields
+            .iter()
+            .enumerate()
+            .flat_map(|(f, bytes)| {
+                bytes
+                    .iter()
+                    .enumerate()
+                    .filter(|(_, &b)| !matches!(b, b':' | b'|' | b',' | b'#' | b'@'))
+                    .map(move |(i, _)| (f, i))
+            })
+            .collect();
+        if targets.is_empty() {
+            return false;
+        }
+        let (field, at) = targets[rng.random_range(0..targets.len())];
+        let mut fields = fields;
+        fields[field][at] = common::invalid_utf8_byte(rng);
+        true
+    }
+
+    /// Mint an identity of `kind` that carries an invalid UTF-8 byte, or `None` when none can be built
+    /// within `budget`. Corrupt identities live in the pool like any other, so they recur across
+    /// datagrams and count against the kind's cap instead of appearing as fresh one-offs.
+    #[must_use]
+    pub fn mint_non_utf8_within(kind: Kind, rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<Context> {
+        for _ in 0..REMINT_TRIES {
+            let mut context = Context::mint_within(kind, rng, budget)?;
+            if !context.poison(rng) {
+                continue;
+            }
+            // A replaced byte does not always invalidate the field. `0x80` is a valid continuation byte,
+            // so poisoning the trailing byte of `café` turns `C3 A9` into the valid `C3 80`. Verify the
+            // field instead of trimming `0x80` from the pool, which would drop it from the byte space the
+            // SUT ever sees. Unverified, 3.4% of corrupt mints carried no invalid byte at all.
+            if !context.has_non_utf8() {
+                continue;
+            }
+            let mut probe = Vec::new();
+            context.render(rng, &mut probe);
+            if is_malformed(&probe).is_ok() {
+                return Some(context);
+            }
+        }
+        None
+    }
+
+    /// Whether this identity carries an invalid UTF-8 byte.
+    #[must_use]
+    pub fn has_non_utf8(&self) -> bool {
+        let fields: Vec<&[u8]> = match self {
+            Context::Metric(c) => std::iter::once(c.name.as_slice())
+                .chain(c.tags.iter().map(Vec::as_slice))
+                .collect(),
+            Context::Event(c) => std::iter::once(c.title.as_slice())
+                .chain(c.tags.iter().map(Vec::as_slice))
+                .collect(),
+            Context::ServiceCheck(c) => std::iter::once(c.name.as_slice())
+                .chain(c.tags.iter().map(Vec::as_slice))
+                .collect(),
+        };
+        fields.iter().any(|f| simdutf8::basic::from_utf8(f).is_err())
+    }
+
     /// Bytes every render of this identity must spend, whatever the per-occurrence payload.
     #[must_use]
     pub fn floor(&self) -> usize {
@@ -131,9 +208,17 @@ impl Context {
     pub fn render_wellformed_within(
         &self, rng: &mut (impl Rng + ?Sized), out: &mut Vec<u8>, budget: usize,
     ) -> Option<usize> {
-        for _ in 0..RENDER_TRIES {
+        for try_index in 0..RENDER_TRIES {
             let start = out.len();
-            let packed = self.render_within(rng, out, budget)?;
+            // The last try renders at the identity's floor, where the occurrence is the shortest the
+            // identity admits and no extension chunk has room. Sampling a wide occurrence one more time
+            // would be hoping again, and a caller has nowhere to go when the tries run out.
+            let attempt = if try_index + 1 == RENDER_TRIES {
+                self.floor().min(budget)
+            } else {
+                budget
+            };
+            let packed = self.render_within(rng, out, attempt)?;
             if is_malformed(&out[start..]).is_ok() {
                 return Some(packed);
             }
@@ -286,6 +371,19 @@ mod tests {
     }
 
     proptest! {
+        /// A corrupt mint always carries an invalid byte. The replacement byte does not guarantee it on
+        /// its own, and an identity that looks corrupt but is not lands in the clean half of the working
+        /// set and quietly thins the non-UTF-8 rate.
+        #[test]
+        fn property_test_a_corrupt_mint_is_corrupt(seed: u64) {
+            let mut rng = SmallRng::seed_from_u64(seed);
+            for _ in 0..16 {
+                if let Some(context) = Context::mint_non_utf8_within(Kind::sample(&mut rng), &mut rng, 8_191) {
+                    prop_assert!(context.has_non_utf8(), "a corrupt mint carried no invalid byte: {context:?}");
+                }
+            }
+        }
+
         /// A minted context conforms to is_malformed. Content carries delimiters, so a raw render may
         /// land on the drop side. The repair loop is the sorter, and any line it does yield forwards.
         #[test]
@@ -294,7 +392,10 @@ mod tests {
             let Some(context) = Context::mint_within(kind, &mut rng, 8_192) else { return Ok(()) };
             for _ in 0..8 {
                 let mut line = Vec::new();
-                if context.render_wellformed_within(&mut rng, &mut line, 8_192).is_some() {
+                if context
+                    .render_wellformed_within(&mut rng, &mut line, 8_192)
+                    .is_some()
+                {
                     prop_assert_eq!(is_malformed(&line), Ok(()), "a rendered line was droppable");
                 }
             }

test/antithesis/harness/src/contexts.rs

@@ -93,7 +93,8 @@ impl ServiceCheckContext {
         out.extend_from_slice(itoa.format(fresh_timestamp(rng)).as_bytes());
         common::serialize_tags(&self.tags, out);
         out.extend_from_slice(b"|m:");
-        out.extend_from_slice(&common::optional_text_within(rng, message_room));
+        let message = common::optional_text_within(rng, message_room);
+        out.extend_from_slice(&message);
         Some(0)
     }