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/proptest-regressions/payload/dogstatsd.txt

@@ -6,3 +6,7 @@
 # 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
+cc 89407cb915422a73621ab39b63ac3bb99e1a12ebb33825d9f623da074b805f9f # shrinks to seed = 17633913455663304866, limit_bytes = 3

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

@@ -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/config.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,32 +258,33 @@ 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.
     ///
     /// Sampled boundary-biased log-uniform in `1..=1_024`. Valid values run `1..=65_536`, the intake's
     /// per-request ceiling. Outside that the intake rejects every `/contexts` request, the driver waits
     /// out its fetch budget and ships nothing, so [`Self::read`] rejects such a config rather than
-    /// letting a timeline generate no load. A larger working set spreads load over more identities and
-    /// so puts fewer points in each, which is the trade against recurrence.
+    /// letting a timeline generate no load. Every context in a pull gets a line in every datagram that
+    /// has room, so a larger pull makes fatter datagrams over more identities rather than thinner
+    /// series, and the trade is against how often any one identity recurs.
     pub context_count: usize,
 }
 
 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 +305,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()))?;
@@ -328,19 +329,21 @@ const MAX_CONTEXTS_TOTAL: u64 = 1_000_000;
 /// The per-kind caps a timeline's shared context pool fills to before it recurs existing contexts.
 /// `first_sample_config` samples this beside `datadog.yaml` so cardinality varies per timeline. Each
 /// cap is drawn against the budget the earlier draws left, so a kind's cardinality still varies at
-/// random while the three together stay under [`MAX_CONTEXTS_TOTAL`].
+/// random while the three together stay under `MAX_CONTEXTS_TOTAL`.
 #[derive(Clone, Copy, Debug, Serialize, Deserialize)]
 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 +357,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),
@@ -384,18 +387,38 @@ impl ContextSourceConfig {
     ///
     /// # Errors
     ///
-    /// Returns an error if the config is unreadable or is not valid YAML.
+    /// Returns an error if the config is unreadable, is not valid YAML, or caps a kind below two. The
+    /// pool seeds every kind with one context carrying an invalid UTF-8 byte and one without, so a cap of
+    /// one cannot hold both and the pool would exceed its own cap assertion. [`Self::sample`] never draws
+    /// below two, and this rejects a config that reached the pool by another route rather than letting it
+    /// redden a run.
     pub fn read(config_dir: &Path) -> anyhow::Result<Self> {
         let path = config_dir.join("context_source.yaml");
         let yaml = fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
-        serde_yaml::from_str(&yaml).with_context(|| format!("parse context source config from {}", path.display()))
+        let config: Self = serde_yaml::from_str(&yaml)
+            .with_context(|| format!("parse context source config from {}", path.display()))?;
+        for (kind, cap) in [
+            ("metric_contexts", config.metric_contexts),
+            ("event_contexts", config.event_contexts),
+            ("service_check_contexts", config.service_check_contexts),
+        ] {
+            anyhow::ensure!(
+                cap >= MIN_CONTEXTS_PER_KIND,
+                "{kind} is {cap}, which is below the {MIN_CONTEXTS_PER_KIND} the pool seeds"
+            );
+        }
+        Ok(config)
     }
 }
 
-/// A single per-kind cap in `1..=ceiling`. The ceiling is well within `usize` on every supported
+/// The fewest contexts a kind may be capped at. The pool seeds every kind with one context carrying an
+/// invalid UTF-8 byte and one without, and both count against the cap.
+pub const MIN_CONTEXTS_PER_KIND: usize = 2;
+
+/// 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)]
@@ -450,14 +473,29 @@ mod tests {
         yaml.lines().any(|line| line.starts_with(&format!("{key}:")))
     }
 
+    // A cap below what the pool seeds would make the pool exceed its own cap assertion and redden the
+    // run on a config value. Rejected at the boundary, as the sibling driver config is.
+    #[test]
+    fn context_source_read_rejects_a_cap_below_the_seeded_minimum() {
+        let dir = std::env::temp_dir().join(format!("ctxcfg-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).expect("create temp dir");
+        std::fs::write(
+            dir.join("context_source.yaml"),
+            "datagram_byte_limit: 8192\nmetric_contexts: 1\nevent_contexts: 4\nservice_check_contexts: 4\n",
+        )
+        .expect("write config");
+        let err = ContextSourceConfig::read(&dir).expect_err("a cap of 1 must be rejected");
+        assert!(err.to_string().contains("metric_contexts"), "{err}");
+    }
+
     #[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]
@@ -511,8 +549,14 @@ mod tests {
             let caps = ContextSourceConfig::sample(&mut SeqRng(seed), 8_192);
             let total = (caps.metric_contexts + caps.event_contexts + caps.service_check_contexts) as u64;
             assert!(total <= MAX_CONTEXTS_TOTAL, "seed {seed} sampled {total}");
-            // A kind of zero panics the pool's `random_range(0..0)`, so every kind keeps at least one.
-            assert!(caps.metric_contexts >= 1 && caps.event_contexts >= 1 && caps.service_check_contexts >= 1);
+            // The pool seeds every kind with one context carrying an invalid UTF-8 byte and one without,
+            // and both count against the cap, so a kind capped below two makes the pool exceed its own
+            // cap assertion.
+            assert!(
+                caps.metric_contexts >= MIN_CONTEXTS_PER_KIND
+                    && caps.event_contexts >= MIN_CONTEXTS_PER_KIND
+                    && caps.service_check_contexts >= MIN_CONTEXTS_PER_KIND
+            );
         }
     }