From fa7a94d02a715705c489da62eb04bb2d275adf0d Mon Sep 17 00:00:00 2001
From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com>
Date: Wed, 28 Oct 2026 14:49:03 -0400
Subject: [PATCH 1/4] [SVLS-9977] test: shared fixture corpus for tests and
 benches

---
 .gitignore                                  |   4 +-
 bottlecap/src/extension/telemetry/mod.rs    |   3 +-
 bottlecap/src/extension/telemetry/stitch.rs |   6 +-
 bottlecap/tests/fixtures/README.md          |  70 +++++
 bottlecap/tests/fixtures/dogstatsd.rs       |  69 +++++
 bottlecap/tests/fixtures/events.rs          | 137 ++++++++++
 bottlecap/tests/fixtures/logs.rs            |  73 +++++
 bottlecap/tests/fixtures/mod.rs             | 156 +++++++++++
 bottlecap/tests/fixtures/telemetry.rs       | 278 ++++++++++++++++++++
 bottlecap/tests/fixtures/traces.rs          | 161 ++++++++++++
 bottlecap/tests/fixtures_corpus.rs          | 101 +++++++
 11 files changed, 1053 insertions(+), 5 deletions(-)
 create mode 100644 bottlecap/tests/fixtures/README.md
 create mode 100644 bottlecap/tests/fixtures/dogstatsd.rs
 create mode 100644 bottlecap/tests/fixtures/events.rs
 create mode 100644 bottlecap/tests/fixtures/logs.rs
 create mode 100644 bottlecap/tests/fixtures/mod.rs
 create mode 100644 bottlecap/tests/fixtures/telemetry.rs
 create mode 100644 bottlecap/tests/fixtures/traces.rs
 create mode 100644 bottlecap/tests/fixtures_corpus.rs

diff --git a/.gitignore b/.gitignore
index 78015339d..06fcc1e10 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,4 +27,6 @@ bottlecap/proptest-regressions
 integration-tests/cdk.context.json
 
 .gitlab/pipeline*
-/CLAUDE_PERSONAL.md
\ No newline at end of file
+/CLAUDE_PERSONAL.md
+# Generated benchmark fixture corpus (see bottlecap/tests/fixtures/README.md)
+bottlecap/tests/fixtures/generated
diff --git a/bottlecap/src/extension/telemetry/mod.rs b/bottlecap/src/extension/telemetry/mod.rs
index 93b362917..a7b6d2ebb 100644
--- a/bottlecap/src/extension/telemetry/mod.rs
+++ b/bottlecap/src/extension/telemetry/mod.rs
@@ -5,7 +5,8 @@ use crate::extension::{EXTENSION_ID_HEADER, base_url};
 
 pub mod events;
 pub mod listener;
-mod stitch;
+#[doc(hidden)]
+pub mod stitch;
 
 pub const TELEMETRY_SUBSCRIPTION_ROUTE: &str = "2022-07-01/telemetry";
 
diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs
index 0807ff5fc..bdf2eed3f 100644
--- a/bottlecap/src/extension/telemetry/stitch.rs
+++ b/bottlecap/src/extension/telemetry/stitch.rs
@@ -67,7 +67,7 @@ const MAX_SEAMS: usize = 64;
 
 /// What came of pairing an unparseable body with a held fragment.
 #[derive(Debug)]
-pub(crate) enum Stitch {
+pub enum Stitch {
     /// The body completed a split payload.
     Complete(Vec<TelemetryEvent>),
     /// The body opens a split payload, and is held for its continuation.
@@ -78,13 +78,13 @@ pub(crate) enum Stitch {
 
 /// Holds the leading half of a split payload until its continuation arrives.
 #[derive(Clone, Default)]
-pub(crate) struct FragmentBuffer {
+pub struct FragmentBuffer {
     held: Arc<Mutex<Option<Fragment>>>,
 }
 
 impl FragmentBuffer {
     /// Joins `body` onto the held fragment, or holds `body` if it opens a split payload.
-    pub(crate) fn stitch(&self, body: &[u8]) -> Stitch {
+    pub fn stitch(&self, body: &[u8]) -> Stitch {
         let mut slot = self.held.lock().unwrap_or_else(PoisonError::into_inner);
 
         if let Some(stale) = slot.take_if(|held| held.received.elapsed() > FRAGMENT_TTL) {
diff --git a/bottlecap/tests/fixtures/README.md b/bottlecap/tests/fixtures/README.md
new file mode 100644
index 000000000..d5c1efe52
--- /dev/null
+++ b/bottlecap/tests/fixtures/README.md
@@ -0,0 +1,70 @@
+# Fixture corpus
+
+Shared inputs for the criterion benches (`bottlecap/benches/`), the fault-injection harness and
+the Lambda simulator. Everything is generated by the Rust module in this directory, so no large
+blob is checked in.
+
+## Use from Rust
+
+```rust
+#[path = "../tests/fixtures/mod.rs"] // from benches/; integration tests use `mod fixtures;`
+mod fixtures;
+use fixtures::{dogstatsd, events, logs, telemetry, traces};
+```
+
+| Module | Generator | What it returns |
+|---|---|---|
+| `telemetry` | `platform_batch(request_id) -> Vec<u8>` | One cold-start invocation: `initStart`, `initRuntimeDone`, `initReport`, `start`, `runtimeDone`, `report` |
+| | `log_batch(LogKind, n) -> Vec<u8>` | One POST body of `n` function log events (`LogKind::{Text, JsonObject, JsonString}`) |
+| | `log_batches(LogKind, n, per_post) -> Vec<Vec<u8>>` | The same, split into POSTs of `per_post` events (Lambda sends at most 1000) |
+| | `invocation_batch(LogKind, n) -> Vec<u8>` | `platform.start`, `n` log lines, `runtimeDone`, `report` |
+| | `split_record_posts(total, piece, per_post) -> Vec<Vec<u8>>` | A record of `total` bytes cut into `piece`-byte pieces, spread over POSTs; the last carries `runtimeDone`. Feed to `FragmentBuffer::stitch` in order |
+| | `split_record_single_post(total, piece) -> Vec<u8>` | All pieces of one split record in one body (seams only) |
+| | `runtime_done_with(id, status, with_metrics)`, `logs_dropped(n)`, `unknown_event()` | A `runtimeDone` of any status with or without `metrics` (Lambda omits it on a timeout), a `platform.logsDropped`, an event of an unknown type |
+| | `event(millis, type, record_json)`, `init_start()`, `platform_start(id)`, `runtime_done(id)`, `report(id)`, `function_event(kind, i)`, `extension_event(i)`, `batch(&[String])` | Building blocks |
+| | `REQUEST_ID`, `FUNCTION_ARN` | Constants used everywhere |
+| `logs` | `text_line(i)`, `json_object(i) -> Value`, `json_line(i)`, `text_line_of_len(i, len)`, `OOM_LINES` | Raw log lines: Lambda text format (~200 B), Powertools-like JSON (~2 KB), one OOM line per runtime |
+| `traces` | `make_traces(n_traces, spans_per_trace, with_sql) -> Vec<Vec<pb::Span>>` | Synthetic traces: `aws.lambda` root, SQL and HTTP clients, internal spans |
+| | `trace_2000_spans()`, `v04_2000_spans() -> Vec<u8>` | The reference payload: 100 x 20 spans, ~1 MB v0.4 msgpack |
+| | `v04_msgpack(&traces)`, `header_tags()`, `TRACE_HEADERS` | Encoding and the matching tracer headers |
+| `dogstatsd` | `line(Kind, i)`, `packet(Kind, first, n)`, `packets(Kind, total, per_packet)`, `unique_tag_line(i)` | Datagrams: 50 metric names x 100 endpoints; `Kind::{Count, Gauge, Distribution}` |
+| | `ENRICH_TAGS`, `DIST_8_TAGS`, `COUNT_2_TAGS`, `SKIPPED_LINES` | Constants |
+| `events` | `payload(name)`, `payload_names()`, `all_payloads()` | The 38 trigger events in `tests/payloads/` |
+| | `sqs_event(records, body_bytes)`, `sqs_220kb()` | SQS batch with `_datadog` trace context; reference 10 x 22 KB |
+| | `dynamodb_event(records, attrs)`, `dynamodb_100_records()` | DynamoDB stream batch; reference 100 x 30 attributes |
+| | `lwa_next_headers(request_id)` | Headers of the `/invocation/next` response LWA intercepts |
+| root | `write_corpus(&dir) -> Vec<(PathBuf, usize)>`, `default_corpus_dir()` | Writes the files below |
+
+## Files on disk
+
+Generated on demand. Run from `bottlecap/`:
+
+```sh
+cargo test --test fixtures_corpus write_fixture_corpus -- --ignored
+```
+
+Output goes to `bottlecap/tests/fixtures/generated/` (gitignored), or to `$BOTTLECAP_FIXTURES_DIR`.
+`MANIFEST.txt` lists every file with its size.
+
+```
+telemetry/platform_batch.json                 platform events of one cold start
+telemetry/function_text_1000.json             1000 text log events, one POST
+telemetry/function_json_object_1000.json      1000 JSON-object log events, one POST
+telemetry/function_json_string_1000.json      1000 JSON-in-string log events, one POST
+telemetry/invocation_batch_text_200.json      start + 200 lines + runtimeDone + report
+telemetry/split_record_1mib_post_{0..1}.json  a 1 MiB record in 256 KiB pieces, 3 per POST
+telemetry/split_record_1mib_single_post.json  the same pieces in one body
+traces/v04_2000_spans.msgpack                 100 traces x 20 spans
+traces/v04_100_spans.msgpack                  5 traces x 20 spans
+logs/text_lines_1000.txt                      raw text lines
+logs/json_lines_1000.jsonl                    raw JSON lines
+logs/oom_lines.txt                            one OOM line per runtime
+dogstatsd/count_packet_{0..9}.txt             20 count lines per datagram
+dogstatsd/distribution_packet_{0..9}.txt      20 distribution lines per datagram
+dogstatsd/enrich_tags.txt                     the tags bottlecap adds to every metric
+events/sqs_220kb.json                         10 SQS records x 22 KB
+events/dynamodb_100_records.json              100 DynamoDB records x 30 attributes
+events/payloads/<name>.json                   a copy of the 38 trigger events in tests/payloads/
+```
+
+The checked-in trigger events stay in `bottlecap/tests/payloads/`.
diff --git a/bottlecap/tests/fixtures/dogstatsd.rs b/bottlecap/tests/fixtures/dogstatsd.rs
new file mode 100644
index 000000000..aa6698fba
--- /dev/null
+++ b/bottlecap/tests/fixtures/dogstatsd.rs
@@ -0,0 +1,69 @@
+//! DogStatsD datagrams as a client library sends them to UDP 8125.
+
+/// The tags bottlecap adds to every metric (`SortedTags::parse(ENRICH_TAGS)`).
+pub const ENRICH_TAGS: &str = "function_arn:arn:aws:lambda:us-east-1:123456789012:function:my-checkout-fn,region:us-east-1,account_id:123456789012,aws_account:123456789012,functionname:my-checkout-fn,resource:my-checkout-fn,executedversion:$LATEST,memorysize:128,architecture:arm64,runtime:nodejs22.x,_dd.runtime_family:nodejs,env:prod,service:checkout,version:1.2.3,dd_extension_version:90,_dd.compute_stats:1";
+
+/// A distribution with 8 tags, the heaviest common line.
+pub const DIST_8_TAGS: &str = "app.checkout.latency:12.5|d|#env:prod,service:checkout,version:1.2.3,region:us-east-1,team:payments,endpoint:/api/v1/pay,status:200,method:POST";
+
+/// A count with 2 tags, the lightest common line.
+pub const COUNT_2_TAGS: &str = "my.metric:1|c|#tag1:a,tag2:b";
+
+/// Lines bottlecap drops before parsing: service checks, events, and its own enhanced metric.
+pub const SKIPPED_LINES: [&str; 3] = [
+    "_sc|my.check|0|#env:prod",
+    "_e{5,4}:title|text",
+    "aws.lambda.enhanced.invocations:1|c",
+];
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Kind {
+    Count,
+    Gauge,
+    Distribution,
+}
+
+/// One metric line. The name cycles over 50 values and the `endpoint` tag over 100, so a
+/// stream of `n` lines has at most 5000 contexts.
+pub fn line(kind: Kind, i: usize) -> String {
+    match kind {
+        Kind::Count => format!(
+            "app.request.count{}:1|c|#endpoint:/api/v{},status:200,method:GET",
+            i % 50,
+            i % 100
+        ),
+        Kind::Gauge => format!(
+            "app.queue.depth{}:{}|g|#endpoint:/api/v{},status:200,method:GET",
+            i % 50,
+            i % 900,
+            i % 100
+        ),
+        Kind::Distribution => format!(
+            "app.request.latency{}:{}|d|#endpoint:/api/v{},status:200,method:GET",
+            i % 50,
+            (i % 900) + 1,
+            i % 100
+        ),
+    }
+}
+
+/// A line whose tag value is unique per `i` (one new interned string per line).
+pub fn unique_tag_line(i: usize) -> String {
+    format!("my.metric:1|c|#env:prod,service:api,request_id:{i:08x}-0000-4000-8000-{i:012x}")
+}
+
+/// `n` lines joined by `\n`, one UDP datagram (keep it under 8 KiB).
+pub fn packet(kind: Kind, first: usize, n: usize) -> String {
+    (first..first + n)
+        .map(|i| line(kind, i))
+        .collect::<Vec<_>>()
+        .join("\n")
+}
+
+/// `total` lines split into datagrams of `per_packet` lines.
+pub fn packets(kind: Kind, total: usize, per_packet: usize) -> Vec<String> {
+    (0..total)
+        .step_by(per_packet.max(1))
+        .map(|first| packet(kind, first, per_packet.min(total - first)))
+        .collect()
+}
diff --git a/bottlecap/tests/fixtures/events.rs b/bottlecap/tests/fixtures/events.rs
new file mode 100644
index 000000000..3741ffbe7
--- /dev/null
+++ b/bottlecap/tests/fixtures/events.rs
@@ -0,0 +1,137 @@
+//! Invocation payloads: the 38 trigger events in `tests/payloads/`, plus large generated
+//! SQS and DynamoDB events.
+
+use std::path::{Path, PathBuf};
+
+use serde_json::{Value, json};
+
+fn payloads_dir() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("payloads")
+}
+
+/// Reads `tests/payloads/<name>.json`.
+pub fn payload(name: &str) -> String {
+    let path = payloads_dir().join(format!("{name}.json"));
+    std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {}: {e}", path.display()))
+}
+
+/// Names of every file in `tests/payloads/`, sorted, without the `.json` suffix.
+pub fn payload_names() -> Vec<String> {
+    let mut names: Vec<String> = std::fs::read_dir(payloads_dir())
+        .expect("list tests/payloads")
+        .filter_map(Result::ok)
+        .filter_map(|e| {
+            let name = e.file_name().to_str()?.to_string();
+            name.strip_suffix(".json").map(str::to_string)
+        })
+        .collect();
+    names.sort();
+    names
+}
+
+/// Every payload, parsed, with its name.
+pub fn all_payloads() -> Vec<(String, Value)> {
+    payload_names()
+        .into_iter()
+        .map(|name| {
+            let value: Value = serde_json::from_str(&payload(&name))
+                .unwrap_or_else(|e| panic!("parse payload {name}: {e}"));
+            (name, value)
+        })
+        .collect()
+}
+
+/// An SQS batch of `records` messages, each with a `body_bytes` body and a `_datadog`
+/// message attribute carrying trace context.
+pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
+    let body: String = (0..body_bytes / 40)
+        .map(|i| format!("{{\"k{i}\":\"v{i}xxxxxxxxxxxxxxxxxxxxxx\"}},"))
+        .collect();
+    let body = format!("[{}]", body.trim_end_matches(','));
+    let recs: Vec<Value> = (0..records)
+        .map(|i| {
+            json!({
+                "messageId": format!("059f36b4-87a3-44ab-83d2-66197{i:07}"),
+                "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
+                "body": body,
+                "attributes": {
+                    "ApproximateReceiveCount": "1",
+                    "SentTimestamp": "1545082649183",
+                    "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+                    "ApproximateFirstReceiveTimestamp": "1545082649185"
+                },
+                "messageAttributes": {
+                    "_datadog": {
+                        "stringValue": "{\"x-datadog-trace-id\":\"2684756524522091840\",\"x-datadog-parent-id\":\"7431398482019833808\",\"x-datadog-sampling-priority\":\"1\"}",
+                        "stringListValues": [],
+                        "binaryListValues": [],
+                        "dataType": "String"
+                    }
+                },
+                "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
+                "eventSource": "aws:sqs",
+                "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:my-queue",
+                "awsRegion": "us-east-1"
+            })
+        })
+        .collect();
+    json!({ "Records": recs })
+}
+
+/// The reference SQS payload: 10 records of 22 KB, about 220 KB serialized.
+pub fn sqs_220kb() -> Value {
+    sqs_event(10, 22_000)
+}
+
+/// A DynamoDB stream batch of `records` INSERT records with `attrs` attributes each.
+pub fn dynamodb_event(records: usize, attrs: usize) -> Value {
+    let recs: Vec<Value> = (0..records)
+        .map(|i| {
+            let mut img = serde_json::Map::new();
+            for a in 0..attrs {
+                img.insert(format!("attr{a}"), json!({"S": format!("value-{i}-{a}")}));
+            }
+            json!({
+                "eventID": format!("{i}"),
+                "eventName": "INSERT",
+                "eventVersion": "1.1",
+                "eventSource": "aws:dynamodb",
+                "awsRegion": "us-east-1",
+                "dynamodb": {
+                    "ApproximateCreationDateTime": 1_428_537_600,
+                    "Keys": {"Id": {"N": "101"}},
+                    "NewImage": Value::Object(img),
+                    "SequenceNumber": format!("{i}00000000000000000"),
+                    "SizeBytes": 26,
+                    "StreamViewType": "NEW_AND_OLD_IMAGES"
+                },
+                "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTableWithStream/stream/2015-06-27T00:48:05.899"
+            })
+        })
+        .collect();
+    json!({ "Records": recs })
+}
+
+/// The reference DynamoDB payload: 100 records of 30 attributes, about 100 KB serialized.
+pub fn dynamodb_100_records() -> Value {
+    dynamodb_event(100, 30)
+}
+
+/// Headers of the `GET /runtime/invocation/next` response that LWA intercepts.
+pub fn lwa_next_headers(request_id: &str) -> Vec<(&'static str, String)> {
+    vec![
+        ("content-type", "application/json".to_string()),
+        ("lambda-runtime-aws-request-id", request_id.to_string()),
+        ("lambda-runtime-deadline-ms", "1741965555094".to_string()),
+        (
+            "lambda-runtime-invoked-function-arn",
+            crate::fixtures::telemetry::FUNCTION_ARN.to_string(),
+        ),
+        (
+            "lambda-runtime-trace-id",
+            "Root=1-67d448e8-3ae320be0e2cdf2f53ccdaba;Lineage=1:73f724a8:0".to_string(),
+        ),
+    ]
+}
diff --git a/bottlecap/tests/fixtures/logs.rs b/bottlecap/tests/fixtures/logs.rs
new file mode 100644
index 000000000..162c37d59
--- /dev/null
+++ b/bottlecap/tests/fixtures/logs.rs
@@ -0,0 +1,73 @@
+//! Raw log lines, as a function writes them to stdout.
+
+use serde_json::Value;
+
+use crate::fixtures::telemetry::REQUEST_ID;
+
+/// Lambda text-format line (about 200 bytes): `<time>\t<request id>\t<level>\t<message>`.
+pub fn text_line(i: usize) -> String {
+    let mut s = format!(
+        "2026-10-27T12:00:00.{:03}Z\t{REQUEST_ID}\tINFO\tUser {i} performed checkout action on cart {} with status=ok latency_ms={} region=us-east-1 path=/api/v1/orders",
+        i % 1000,
+        i * 7,
+        i % 300
+    );
+    while s.len() < 200 {
+        s.push('x');
+    }
+    s
+}
+
+/// Structured line in the Lambda JSON log format (Powertools-like, about 2 KB).
+pub fn json_object(i: usize) -> Value {
+    let items: Vec<Value> = (0..12)
+        .map(|k| {
+            serde_json::json!({
+                "sku": format!("SKU-{i}-{k}"),
+                "qty": k,
+                "price": 19.99,
+                "name": "Widget deluxe edition",
+            })
+        })
+        .collect();
+    serde_json::json!({
+        "timestamp": "2026-10-27T12:00:00.123Z",
+        "level": "INFO",
+        "requestId": REQUEST_ID,
+        "message": format!("Processed order {i} for customer cust-{} with \"quoted\" note", i * 3),
+        "logger": "orders.handler",
+        "service": "orders",
+        "cold_start": false,
+        "function_name": "orders-prod-handler",
+        "function_memory_size": "128",
+        "xray_trace_id": "1-5759e988-bd862e3fe1be46a994272793",
+        "items": items,
+    })
+}
+
+/// [`json_object`] serialized on one line.
+pub fn json_line(i: usize) -> String {
+    serde_json::to_string(&json_object(i)).expect("serialize json log line")
+}
+
+/// One runtime OOM line per supported runtime, in the order of `OOM_ERRORS` in
+/// `src/logs/lambda/processor.rs`.
+pub const OOM_LINES: [&str; 7] = [
+    "fatal error: runtime: out of memory",
+    "Exception in thread \"main\" java.lang.OutOfMemoryError: Java heap space",
+    "FATAL ERROR: Reached heap limit Allocation failed - JavaScript heap out of memory",
+    "Runtime exited with error: signal: killed Runtime.ExitError",
+    "MemoryError",
+    "failed to allocate memory (NoMemoryError)",
+    "Unhandled exception. System.OutOfMemoryException: Insufficient memory to continue the execution of the program.",
+];
+
+/// A text line padded to `len` bytes, for size-sensitive scans such as the OOM check.
+pub fn text_line_of_len(i: usize, len: usize) -> String {
+    let mut s = text_line(i);
+    while s.len() < len {
+        s.push_str(" some more log text here");
+    }
+    s.truncate(len);
+    s
+}
diff --git a/bottlecap/tests/fixtures/mod.rs b/bottlecap/tests/fixtures/mod.rs
new file mode 100644
index 000000000..588dedb6b
--- /dev/null
+++ b/bottlecap/tests/fixtures/mod.rs
@@ -0,0 +1,156 @@
+//! Shared fixture corpus for benchmarks, the fault-injection harness and the Lambda simulator.
+//!
+//! Everything is generated in-process, so nothing large is checked in. Use the generators
+//! directly from a bench or a test:
+//!
+//! ```ignore
+//! #[path = "../tests/fixtures/mod.rs"]
+//! mod fixtures;
+//! let body = fixtures::telemetry::log_batch(fixtures::telemetry::LogKind::Text, 1000);
+//! ```
+//!
+//! To write the corpus to disk once (for tools that read files, such as a simulator written in
+//! another language), run:
+//!
+//! ```text
+//! cargo test --test fixtures_corpus write_fixture_corpus -- --ignored
+//! ```
+//!
+//! That writes `bottlecap/tests/fixtures/generated/` (gitignored), or the directory named by
+//! `BOTTLECAP_FIXTURES_DIR`. See `README.md` in this directory for the file list.
+#![allow(dead_code, clippy::missing_panics_doc, clippy::must_use_candidate)]
+
+pub mod dogstatsd;
+pub mod events;
+pub mod logs;
+pub mod telemetry;
+pub mod traces;
+
+use std::fs;
+use std::io::Write as _;
+use std::path::{Path, PathBuf};
+
+/// Default output directory for [`write_corpus`]: `bottlecap/tests/fixtures/generated`.
+pub fn default_corpus_dir() -> PathBuf {
+    if let Ok(dir) = std::env::var("BOTTLECAP_FIXTURES_DIR") {
+        return PathBuf::from(dir);
+    }
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("fixtures")
+        .join("generated")
+}
+
+/// Writes every fixture to `dir` and returns the written files with their sizes.
+pub fn write_corpus(dir: &Path) -> Vec<(PathBuf, usize)> {
+    let mut written = Vec::new();
+    let mut put = |rel: &str, bytes: &[u8]| {
+        let path = dir.join(rel);
+        fs::create_dir_all(path.parent().expect("fixture path has a parent"))
+            .expect("create fixture directory");
+        fs::write(&path, bytes).expect("write fixture");
+        written.push((path, bytes.len()));
+    };
+
+    // Telemetry API POST bodies.
+    put(
+        "telemetry/platform_batch.json",
+        &telemetry::platform_batch(telemetry::REQUEST_ID),
+    );
+    for (kind, name) in [
+        (telemetry::LogKind::Text, "function_text_1000.json"),
+        (
+            telemetry::LogKind::JsonObject,
+            "function_json_object_1000.json",
+        ),
+        (
+            telemetry::LogKind::JsonString,
+            "function_json_string_1000.json",
+        ),
+    ] {
+        put(
+            &format!("telemetry/{name}"),
+            &telemetry::log_batch(kind, 1000),
+        );
+    }
+    put(
+        "telemetry/invocation_batch_text_200.json",
+        &telemetry::invocation_batch(telemetry::LogKind::Text, 200),
+    );
+    for (i, post) in telemetry::split_record_posts(1 << 20, 256 << 10, 3)
+        .iter()
+        .enumerate()
+    {
+        put(&format!("telemetry/split_record_1mib_post_{i}.json"), post);
+    }
+    put(
+        "telemetry/split_record_1mib_single_post.json",
+        &telemetry::split_record_single_post(1 << 20, 256 << 10),
+    );
+
+    // Traces: one v0.4 msgpack body of 2000 spans.
+    put("traces/v04_2000_spans.msgpack", &traces::v04_2000_spans());
+    put(
+        "traces/v04_100_spans.msgpack",
+        &traces::v04_msgpack(&traces::make_traces(5, 20, true)),
+    );
+
+    // Raw log lines.
+    let text: Vec<String> = (0..1000).map(logs::text_line).collect();
+    put("logs/text_lines_1000.txt", text.join("\n").as_bytes());
+    let json: Vec<String> = (0..1000).map(logs::json_line).collect();
+    put("logs/json_lines_1000.jsonl", json.join("\n").as_bytes());
+    put("logs/oom_lines.txt", logs::OOM_LINES.join("\n").as_bytes());
+
+    // DogStatsD datagrams, one file per packet.
+    for (i, packet) in dogstatsd::packets(dogstatsd::Kind::Count, 200, 20)
+        .iter()
+        .enumerate()
+    {
+        put(
+            &format!("dogstatsd/count_packet_{i}.txt"),
+            packet.as_bytes(),
+        );
+    }
+    for (i, packet) in dogstatsd::packets(dogstatsd::Kind::Distribution, 200, 20)
+        .iter()
+        .enumerate()
+    {
+        put(
+            &format!("dogstatsd/distribution_packet_{i}.txt"),
+            packet.as_bytes(),
+        );
+    }
+    put(
+        "dogstatsd/enrich_tags.txt",
+        dogstatsd::ENRICH_TAGS.as_bytes(),
+    );
+
+    // Large invocation payloads, plus a copy of the checked-in trigger events so a tool that
+    // reads this directory finds every payload in one place.
+    put(
+        "events/sqs_220kb.json",
+        &serde_json::to_vec(&events::sqs_220kb()).expect("serialize sqs event"),
+    );
+    put(
+        "events/dynamodb_100_records.json",
+        &serde_json::to_vec(&events::dynamodb_100_records()).expect("serialize dynamodb event"),
+    );
+    for name in events::payload_names() {
+        put(
+            &format!("events/payloads/{name}.json"),
+            events::payload(&name).as_bytes(),
+        );
+    }
+
+    let mut manifest = String::new();
+    for (path, len) in &written {
+        let rel = path.strip_prefix(dir).unwrap_or(path);
+        manifest.push_str(&format!("{len:>10}  {}\n", rel.display()));
+    }
+    let manifest_path = dir.join("MANIFEST.txt");
+    let mut f = fs::File::create(&manifest_path).expect("create manifest");
+    f.write_all(manifest.as_bytes()).expect("write manifest");
+    written.push((manifest_path, manifest.len()));
+    written
+}
diff --git a/bottlecap/tests/fixtures/telemetry.rs b/bottlecap/tests/fixtures/telemetry.rs
new file mode 100644
index 000000000..296224262
--- /dev/null
+++ b/bottlecap/tests/fixtures/telemetry.rs
@@ -0,0 +1,278 @@
+//! Telemetry API POST bodies: platform events and function log lines, whole and split.
+//!
+//! Bottlecap subscribes with `maxItems: 1000`, `maxBytes: 1 MiB`, `timeoutMs: 25`. Lambda
+//! cuts a log record longer than about 256 KiB into pieces, each an event of its own, and a
+//! POST can end on a cut piece. `split_record_posts` reproduces that shape.
+
+use serde_json::Value;
+
+use crate::fixtures::logs;
+
+/// Request id used by every fixture.
+pub const REQUEST_ID: &str = "8f5c1b9e-3a2d-4c6f-9e1b-7d2a5c8e4f10";
+
+/// Function ARN used by every fixture.
+pub const FUNCTION_ARN: &str = "arn:aws:lambda:us-east-1:123456789012:function:orders-prod-handler";
+
+/// Which shape a function log record takes in the Telemetry API.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LogKind {
+    /// `"record": "<text line>"` (text log format, about 200 B).
+    Text,
+    /// `"record": { ... }` (JSON log format, the record is already an object, about 2 KB).
+    JsonObject,
+    /// `"record": "{ ... }"` (text log format carrying a JSON document, about 2 KB).
+    JsonString,
+}
+
+impl LogKind {
+    pub const ALL: [LogKind; 3] = [LogKind::Text, LogKind::JsonObject, LogKind::JsonString];
+
+    pub fn name(self) -> &'static str {
+        match self {
+            LogKind::Text => "text",
+            LogKind::JsonObject => "json_object",
+            LogKind::JsonString => "json_string",
+        }
+    }
+}
+
+fn time(millis: usize) -> String {
+    format!(
+        "2026-10-27T12:00:{:02}.{:03}Z",
+        (millis / 1000) % 60,
+        millis % 1000
+    )
+}
+
+/// One telemetry event: `{"time":"...","type":<kind>,"record":<record_json>}`.
+pub fn event(millis: usize, kind: &str, record_json: &str) -> String {
+    format!(
+        r#"{{"time":"{}","type":"{kind}","record":{record_json}}}"#,
+        time(millis)
+    )
+}
+
+pub fn init_start() -> String {
+    event(
+        0,
+        "platform.initStart",
+        r#"{"initializationType":"on-demand","phase":"init","runtimeVersion":"nodejs:22.v10","runtimeVersionArn":"arn:aws:lambda:us-east-1::runtime:abc"}"#,
+    )
+}
+
+pub fn init_runtime_done() -> String {
+    event(
+        40,
+        "platform.initRuntimeDone",
+        r#"{"initializationType":"on-demand","phase":"init","status":"success"}"#,
+    )
+}
+
+pub fn init_report() -> String {
+    event(
+        41,
+        "platform.initReport",
+        r#"{"initializationType":"on-demand","phase":"init","status":"success","metrics":{"durationMs":38.12}}"#,
+    )
+}
+
+pub fn platform_start(request_id: &str) -> String {
+    event(
+        50,
+        "platform.start",
+        &format!(
+            r#"{{"requestId":"{request_id}","version":"$LATEST","tracing":{{"spanId":"54565fb41ac79632","type":"X-Amzn-Trace-Id","value":"Root=1-62e900b2-710d76f009d6e7785905449a;Parent=0efbd19962d95b05;Sampled=1"}}}}"#
+        ),
+    )
+}
+
+pub fn runtime_done(request_id: &str) -> String {
+    event(
+        150,
+        "platform.runtimeDone",
+        &format!(
+            r#"{{"requestId":"{request_id}","status":"success","spans":[{{"name":"responseLatency","start":"2026-10-27T12:00:00.051Z","durationMs":90.0}},{{"name":"responseDuration","start":"2026-10-27T12:00:00.141Z","durationMs":1.0}}],"metrics":{{"durationMs":98.5,"producedBytes":1024}}}}"#
+        ),
+    )
+}
+
+/// `platform.runtimeDone` for `request_id` with `status`; `with_metrics = false`
+/// omits the `metrics` object, as Lambda does on some failed invocations.
+pub fn runtime_done_with(request_id: &str, status: &str, with_metrics: bool) -> String {
+    let metrics = if with_metrics {
+        r#","metrics":{"durationMs":98.5,"producedBytes":1024}"#
+    } else {
+        ""
+    };
+    let error = if status == "success" {
+        String::new()
+    } else {
+        r#","errorType":"Runtime.ExitError""#.to_string()
+    };
+    event(
+        150,
+        "platform.runtimeDone",
+        &format!(
+            r#"{{"requestId":"{request_id}","status":"{status}","spans":[{{"name":"responseLatency","start":"2026-10-27T12:00:00.051Z","durationMs":90.0}}]{metrics}{error}}}"#
+        ),
+    )
+}
+
+/// `platform.logsDropped` for `n` records, as Lambda sends when its buffer overflows.
+pub fn logs_dropped(n: u64) -> String {
+    event(
+        149,
+        "platform.logsDropped",
+        &format!(
+            r#"{{"droppedBytes":{},"droppedRecords":{n},"reason":"Consumer seems to have fallen behind"}}"#,
+            n * 1024
+        ),
+    )
+}
+
+/// An event of a type the extension does not know. Must never break a batch.
+pub fn unknown_event() -> String {
+    event(70, "platform.somethingNew", r#"{"field":"value"}"#)
+}
+
+pub fn report(request_id: &str) -> String {
+    event(
+        151,
+        "platform.report",
+        &format!(
+            r#"{{"requestId":"{request_id}","status":"success","metrics":{{"durationMs":98.5,"billedDurationMs":99,"memorySizeMB":128,"maxMemoryUsedMB":64,"initDurationMs":38.12}}}}"#
+        ),
+    )
+}
+
+/// One function log event of the given shape.
+pub fn function_event(kind: LogKind, i: usize) -> String {
+    let record = match kind {
+        LogKind::Text => serde_json::to_string(&Value::String(logs::text_line(i))),
+        LogKind::JsonObject => serde_json::to_string(&logs::json_object(i)),
+        LogKind::JsonString => serde_json::to_string(&Value::String(logs::json_line(i))),
+    }
+    .expect("serialize function record");
+    event(60 + i % 80, "function", &record)
+}
+
+/// An `extension` log event (bottlecap's own stdout as Lambda re-ingests it).
+pub fn extension_event(i: usize) -> String {
+    let line = format!("[{}Z] DEBUG LOGS | processed batch {i}", time(60 + i % 80));
+    event(
+        60 + i % 80,
+        "extension",
+        &serde_json::to_string(&Value::String(line)).expect("serialize extension record"),
+    )
+}
+
+/// Joins events into one POST body.
+pub fn batch(events: &[String]) -> Vec<u8> {
+    let mut body =
+        Vec::with_capacity(events.iter().map(String::len).sum::<usize>() + events.len() + 2);
+    body.push(b'[');
+    for (i, e) in events.iter().enumerate() {
+        if i > 0 {
+            body.push(b',');
+        }
+        body.extend_from_slice(e.as_bytes());
+    }
+    body.push(b']');
+    body
+}
+
+/// The platform events of one cold-start invocation, with no function logs.
+pub fn platform_batch(request_id: &str) -> Vec<u8> {
+    batch(&[
+        init_start(),
+        init_runtime_done(),
+        init_report(),
+        platform_start(request_id),
+        runtime_done(request_id),
+        report(request_id),
+    ])
+}
+
+/// `n` function log events of one shape, as one POST body (`n` is at most 1000 in practice).
+pub fn log_batch(kind: LogKind, n: usize) -> Vec<u8> {
+    let events: Vec<String> = (0..n).map(|i| function_event(kind, i)).collect();
+    batch(&events)
+}
+
+/// `n` function log events split into POST bodies of `per_post` events each.
+pub fn log_batches(kind: LogKind, n: usize, per_post: usize) -> Vec<Vec<u8>> {
+    let events: Vec<String> = (0..n).map(|i| function_event(kind, i)).collect();
+    events.chunks(per_post.max(1)).map(batch).collect()
+}
+
+/// One warm invocation as the Telemetry API delivers it: `platform.start`, `n` function
+/// lines, `platform.runtimeDone`, `platform.report`.
+pub fn invocation_batch(kind: LogKind, n: usize) -> Vec<u8> {
+    let mut events = Vec::with_capacity(n + 3);
+    events.push(platform_start(REQUEST_ID));
+    events.extend((0..n).map(|i| function_event(kind, i)));
+    events.push(runtime_done(REQUEST_ID));
+    events.push(report(REQUEST_ID));
+    batch(&events)
+}
+
+/// A JSON record of short key/value pairs, about `total_bytes` long.
+pub fn large_json_record(total_bytes: usize) -> String {
+    let mut rec = String::with_capacity(total_bytes + 64);
+    rec.push('{');
+    let mut i = 0;
+    while rec.len() < total_bytes.saturating_sub(64) {
+        if i > 0 {
+            rec.push(',');
+        }
+        rec.push_str(&format!("\"k{i}\":\"value {i}\""));
+        i += 1;
+    }
+    rec.push('}');
+    rec
+}
+
+/// Cuts `record` into `function` events of `piece_bytes` each, the way Lambda does: every
+/// piece has its own envelope and carries a raw slice of the record's bytes.
+pub fn record_pieces(record: &str, piece_bytes: usize) -> Vec<String> {
+    record
+        .as_bytes()
+        .chunks(piece_bytes.max(1))
+        .enumerate()
+        .map(|(n, chunk)| {
+            event(
+                n,
+                "function",
+                std::str::from_utf8(chunk).expect("fixture record is ASCII"),
+            )
+        })
+        .collect()
+}
+
+/// All pieces of one split record plus a `platform.runtimeDone` in a single POST body. The
+/// body does not parse as a whole; the stitcher must drop the seams between the pieces.
+pub fn split_record_single_post(total_bytes: usize, piece_bytes: usize) -> Vec<u8> {
+    let mut events = record_pieces(&large_json_record(total_bytes), piece_bytes);
+    events.push(runtime_done(REQUEST_ID));
+    batch(&events)
+}
+
+/// A split record spread over several POST bodies of `pieces_per_post` pieces each. Every
+/// body but the last ends on a cut piece; the last one carries the rest of the record and a
+/// `platform.runtimeDone`. Feed them to `FragmentBuffer::stitch` in order.
+pub fn split_record_posts(
+    total_bytes: usize,
+    piece_bytes: usize,
+    pieces_per_post: usize,
+) -> Vec<Vec<u8>> {
+    let pieces = record_pieces(&large_json_record(total_bytes), piece_bytes);
+    let mut posts: Vec<Vec<String>> = pieces
+        .chunks(pieces_per_post.max(1))
+        .map(<[String]>::to_vec)
+        .collect();
+    if let Some(last) = posts.last_mut() {
+        last.push(runtime_done(REQUEST_ID));
+    }
+    posts.iter().map(|p| batch(p)).collect()
+}
diff --git a/bottlecap/tests/fixtures/traces.rs b/bottlecap/tests/fixtures/traces.rs
new file mode 100644
index 000000000..974bfcd37
--- /dev/null
+++ b/bottlecap/tests/fixtures/traces.rs
@@ -0,0 +1,161 @@
+//! Trace payloads as a tracer library POSTs them to `/v0.4/traces`.
+
+use std::collections::HashMap;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use libdd_trace_protobuf::pb;
+use libdd_trace_utils::tracer_header_tags::TracerHeaderTags;
+
+/// Headers a Python tracer sends with a v0.4 payload.
+pub const TRACE_HEADERS: &[(&str, &str)] = &[
+    ("Content-Type", "application/msgpack"),
+    ("Datadog-Meta-Lang", "python"),
+    ("Datadog-Meta-Lang-Version", "3.12"),
+    ("Datadog-Meta-Tracer-Version", "2.10.0"),
+    ("X-Datadog-Trace-Count", "100"),
+];
+
+/// Fixed start time of the generated spans: 2026-10-27T12:00:00Z, the instant the log
+/// fixtures use. Fixed, so the encoded payload is the same on every run.
+pub const BASE_NS: i64 = 1_761_566_400_000_000_000;
+
+/// The current time, for consumers that drop old spans (the stats concentrator keeps a time
+/// window): `make_traces_at(now_ns(), ...)`.
+pub fn now_ns() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("clock after epoch")
+        .as_nanos()
+        .try_into()
+        .unwrap_or(i64::MAX)
+}
+
+/// The tracer header tags that match [`TRACE_HEADERS`].
+pub fn header_tags() -> TracerHeaderTags<'static> {
+    TracerHeaderTags {
+        lang: "python",
+        lang_version: "3.12",
+        tracer_version: "2.10.0",
+        ..Default::default()
+    }
+}
+
+/// `n_traces` traces of `spans_per_trace` spans each, starting at `base_ns`. Span 0 is the
+/// `aws.lambda` root; spans 1 to 4 are SQL clients when `with_sql` is set, else HTTP clients;
+/// spans 5 to 9 are HTTP clients; the rest are internal work. Trace ids start at 1_000_000.
+pub fn make_traces_at(
+    base_ns: i64,
+    n_traces: usize,
+    spans_per_trace: usize,
+    with_sql: bool,
+) -> Vec<Vec<pb::Span>> {
+    (0..n_traces)
+        .map(|t| {
+            let trace_id = 1_000_000 + t as u64;
+            (0..spans_per_trace)
+                .map(|s| {
+                    let span_id = trace_id * 1000 + s as u64 + 1;
+                    let parent_id = if s == 0 { 0 } else { trace_id * 1000 + 1 };
+                    let (name, resource, typ, extra): (&str, &str, &str, Vec<(&str, &str)>) =
+                        match s {
+                            0 => (
+                                "aws.lambda",
+                                "my-function",
+                                "serverless",
+                                vec![
+                                    ("span.kind", "server"),
+                                    ("request_id", "8f5a7e2c-1b2c-4d5e-9f00-0123456789ab"),
+                                    ("function_trigger.event_source", "api-gateway"),
+                                    ("http.method", "GET"),
+                                    ("http.status_code", "200"),
+                                ],
+                            ),
+                            1..=4 if with_sql => (
+                                "postgres.query",
+                                "SELECT id, name FROM users WHERE id = 42 AND email = 'bob@example.com'",
+                                "sql",
+                                vec![
+                                    ("db.type", "postgres"),
+                                    ("db.name", "app"),
+                                    ("out.host", "db.internal"),
+                                    ("span.kind", "client"),
+                                ],
+                            ),
+                            1..=9 => (
+                                "http.request",
+                                "GET /v1/items",
+                                "http",
+                                vec![
+                                    ("http.method", "GET"),
+                                    ("http.url", "https://svc.internal/v1/items/987?token=abc"),
+                                    ("http.status_code", "200"),
+                                    ("span.kind", "client"),
+                                    ("out.host", "svc.internal"),
+                                ],
+                            ),
+                            _ => ("internal.work", "compute", "custom", vec![("component", "app")]),
+                        };
+                    let mut meta: HashMap<String, String> = [
+                        ("runtime-id", "f45568ad09d5480b99087d86ebda26e6"),
+                        ("language", "python"),
+                        ("_dd.p.dm", "-0"),
+                        ("_dd.p.tid", "66f1a2b300000000"),
+                        ("env", "prod"),
+                        ("version", "1.2.3"),
+                        ("component", "requests"),
+                    ]
+                    .into_iter()
+                    .chain(extra)
+                    .map(|(k, v)| (k.to_string(), v.to_string()))
+                    .collect();
+                    meta.shrink_to_fit();
+                    let mut metrics: HashMap<String, f64> = HashMap::new();
+                    metrics.insert("process_id".to_string(), 1234.0);
+                    metrics.insert("_sampling_priority_v1".to_string(), 1.0);
+                    if s == 0 {
+                        metrics.insert("_dd.top_level".to_string(), 1.0);
+                        metrics.insert("_dd.agent_psr".to_string(), 1.0);
+                    }
+                    if typ == "sql" || typ == "http" {
+                        metrics.insert("_dd.measured".to_string(), 1.0);
+                    }
+                    pb::Span {
+                        service: "my-service".to_string(),
+                        name: name.to_string(),
+                        resource: resource.to_string(),
+                        trace_id,
+                        span_id,
+                        parent_id,
+                        start: base_ns + (s as i64) * 1000,
+                        duration: 1_000_000,
+                        error: 0,
+                        meta,
+                        metrics,
+                        r#type: typ.to_string(),
+                        ..Default::default()
+                    }
+                })
+                .collect()
+        })
+        .collect()
+}
+
+/// [`make_traces_at`] starting at [`BASE_NS`].
+pub fn make_traces(n_traces: usize, spans_per_trace: usize, with_sql: bool) -> Vec<Vec<pb::Span>> {
+    make_traces_at(BASE_NS, n_traces, spans_per_trace, with_sql)
+}
+
+/// The reference payload: 100 traces of 20 spans, with SQL spans.
+pub fn trace_2000_spans() -> Vec<Vec<pb::Span>> {
+    make_traces(100, 20, true)
+}
+
+/// Encodes traces as a v0.4 msgpack body (`rmp_serde::to_vec_named`, as tracers do).
+pub fn v04_msgpack(traces: &[Vec<pb::Span>]) -> Vec<u8> {
+    rmp_serde::to_vec_named(traces).expect("encode v0.4 msgpack")
+}
+
+/// The reference v0.4 body: 2000 spans, about 1 MB.
+pub fn v04_2000_spans() -> Vec<u8> {
+    v04_msgpack(&trace_2000_spans())
+}
diff --git a/bottlecap/tests/fixtures_corpus.rs b/bottlecap/tests/fixtures_corpus.rs
new file mode 100644
index 000000000..4b81555ae
--- /dev/null
+++ b/bottlecap/tests/fixtures_corpus.rs
@@ -0,0 +1,101 @@
+//! Checks the fixture generators, and writes the corpus to disk on request.
+//!
+//! Write the corpus with:
+//! `cargo test --test fixtures_corpus write_fixture_corpus -- --ignored`
+
+mod fixtures;
+
+use bottlecap::extension::telemetry::events::TelemetryEvent;
+use bottlecap::extension::telemetry::stitch::{FragmentBuffer, Stitch};
+use bottlecap::lifecycle::invocation::triggers::IdentifiedTrigger;
+use fixtures::{dogstatsd, events, logs, telemetry, traces};
+
+#[test]
+fn telemetry_batches_parse() {
+    let events: Vec<TelemetryEvent> =
+        serde_json::from_slice(&telemetry::platform_batch(telemetry::REQUEST_ID)).unwrap();
+    assert_eq!(events.len(), 6);
+    for kind in telemetry::LogKind::ALL {
+        let events: Vec<TelemetryEvent> =
+            serde_json::from_slice(&telemetry::log_batch(kind, 10)).unwrap();
+        assert_eq!(events.len(), 10, "{}", kind.name());
+    }
+    let events: Vec<TelemetryEvent> =
+        serde_json::from_slice(&telemetry::invocation_batch(telemetry::LogKind::Text, 5)).unwrap();
+    assert_eq!(events.len(), 8);
+}
+
+#[test]
+fn split_record_stitches_back_together() {
+    let posts = telemetry::split_record_posts(1 << 20, 256 << 10, 3);
+    assert_eq!(posts.len(), 2);
+    let fragments = FragmentBuffer::default();
+    assert!(serde_json::from_slice::<Vec<TelemetryEvent>>(&posts[0]).is_err());
+    assert!(matches!(fragments.stitch(&posts[0]), Stitch::Pending));
+    match fragments.stitch(&posts[1]) {
+        Stitch::Complete(events) => assert_eq!(events.len(), 2),
+        other => panic!("expected a complete stitch, got {other:?}"),
+    }
+
+    let single = telemetry::split_record_single_post(1 << 20, 256 << 10);
+    assert!(serde_json::from_slice::<Vec<TelemetryEvent>>(&single).is_err());
+    match FragmentBuffer::default().stitch(&single) {
+        Stitch::Complete(events) => assert_eq!(events.len(), 2),
+        other => panic!("expected a complete stitch, got {other:?}"),
+    }
+}
+
+#[test]
+fn trace_payload_round_trips() {
+    let traces = traces::trace_2000_spans();
+    assert_eq!(traces.iter().map(Vec::len).sum::<usize>(), 2000);
+    let body = traces::v04_msgpack(&traces);
+    let decoded: Vec<Vec<libdd_trace_protobuf::pb::Span>> = rmp_serde::from_slice(&body).unwrap();
+    assert_eq!(decoded.len(), 100);
+}
+
+#[test]
+fn log_lines_have_the_expected_shape() {
+    assert!(logs::text_line(0).len() >= 200);
+    assert!(logs::json_object(0).is_object());
+    assert_eq!(logs::text_line_of_len(1, 2000).len(), 2000);
+    assert_eq!(logs::OOM_LINES.len(), 7);
+}
+
+#[test]
+fn dogstatsd_packets_split_by_lines() {
+    let packets = dogstatsd::packets(dogstatsd::Kind::Count, 200, 20);
+    assert_eq!(packets.len(), 10);
+    assert_eq!(packets[0].lines().count(), 20);
+    assert!(packets.iter().all(|p| p.len() < 8192));
+}
+
+#[test]
+fn large_events_identify_as_their_trigger() {
+    assert_eq!(events::all_payloads().len(), 38);
+    let sqs = events::sqs_220kb();
+    let sqs_len = serde_json::to_vec(&sqs).unwrap().len();
+    assert!((200_000..260_000).contains(&sqs_len), "sqs {sqs_len} B");
+    assert!(matches!(
+        IdentifiedTrigger::from_value(&sqs),
+        IdentifiedTrigger::SqsRecord(_)
+    ));
+    let ddb = events::dynamodb_100_records();
+    let ddb_len = serde_json::to_vec(&ddb).unwrap().len();
+    assert!((80_000..140_000).contains(&ddb_len), "dynamodb {ddb_len} B");
+    assert!(matches!(
+        IdentifiedTrigger::from_value(&ddb),
+        IdentifiedTrigger::DynamoDbRecord(_)
+    ));
+}
+
+#[test]
+#[ignore = "writes bottlecap/tests/fixtures/generated/; run on demand"]
+fn write_fixture_corpus() {
+    let dir = fixtures::default_corpus_dir();
+    let written = fixtures::write_corpus(&dir);
+    for (path, len) in &written {
+        println!("{len:>10}  {}", path.display());
+    }
+    assert!(written.len() > 20);
+}

From 3e499bbbed5ef0c1e9446616597ca511a196a8cb Mon Sep 17 00:00:00 2001
From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com>
Date: Wed, 28 Oct 2026 17:22:31 -0400
Subject: [PATCH 2/4] test: address review feedback on the fixture corpus

- dogstatsd: advance the metric name every 100 lines so n lines give min(n, 5000) contexts
- traces: set BASE_NS to 2026-10-27T12:00:00Z and encode meta/metrics in key order
- events: pad the SQS body to the exact size and put the LWA deadline 60 s from now
- telemetry: emit extension logs in the logger::Formatter JSON shape
---
 bottlecap/tests/fixtures/dogstatsd.rs | 10 ++---
 bottlecap/tests/fixtures/events.rs    | 36 ++++++++++++-----
 bottlecap/tests/fixtures/telemetry.rs |  6 ++-
 bottlecap/tests/fixtures/traces.rs    | 58 +++++++++++++++++++++++++--
 bottlecap/tests/fixtures_corpus.rs    |  6 +++
 5 files changed, 97 insertions(+), 19 deletions(-)

diff --git a/bottlecap/tests/fixtures/dogstatsd.rs b/bottlecap/tests/fixtures/dogstatsd.rs
index aa6698fba..a85372af7 100644
--- a/bottlecap/tests/fixtures/dogstatsd.rs
+++ b/bottlecap/tests/fixtures/dogstatsd.rs
@@ -23,24 +23,24 @@ pub enum Kind {
     Distribution,
 }
 
-/// One metric line. The name cycles over 50 values and the `endpoint` tag over 100, so a
-/// stream of `n` lines has at most 5000 contexts.
+/// One metric line. The `endpoint` tag cycles over 100 values and the name moves to the next of
+/// 50 values every 100 lines, so a stream of `n` lines has `min(n, 5000)` contexts.
 pub fn line(kind: Kind, i: usize) -> String {
     match kind {
         Kind::Count => format!(
             "app.request.count{}:1|c|#endpoint:/api/v{},status:200,method:GET",
-            i % 50,
+            (i / 100) % 50,
             i % 100
         ),
         Kind::Gauge => format!(
             "app.queue.depth{}:{}|g|#endpoint:/api/v{},status:200,method:GET",
-            i % 50,
+            (i / 100) % 50,
             i % 900,
             i % 100
         ),
         Kind::Distribution => format!(
             "app.request.latency{}:{}|d|#endpoint:/api/v{},status:200,method:GET",
-            i % 50,
+            (i / 100) % 50,
             (i % 900) + 1,
             i % 100
         ),
diff --git a/bottlecap/tests/fixtures/events.rs b/bottlecap/tests/fixtures/events.rs
index 3741ffbe7..bd4b7a8b3 100644
--- a/bottlecap/tests/fixtures/events.rs
+++ b/bottlecap/tests/fixtures/events.rs
@@ -2,6 +2,7 @@
 //! SQS and DynamoDB events.
 
 use std::path::{Path, PathBuf};
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use serde_json::{Value, json};
 
@@ -43,13 +44,24 @@ pub fn all_payloads() -> Vec<(String, Value)> {
         .collect()
 }
 
-/// An SQS batch of `records` messages, each with a `body_bytes` body and a `_datadog`
-/// message attribute carrying trace context.
+/// An SQS batch of `records` messages, each with a JSON array body of exactly `body_bytes`
+/// bytes (at least 2) and a `_datadog` message attribute carrying trace context.
 pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
-    let body: String = (0..body_bytes / 40)
-        .map(|i| format!("{{\"k{i}\":\"v{i}xxxxxxxxxxxxxxxxxxxxxx\"}},"))
-        .collect();
-    let body = format!("[{}]", body.trim_end_matches(','));
+    let mut body = String::from("[");
+    for i in 0.. {
+        let item = format!("{{\"k{i}\":\"v{i}xxxxxxxxxxxxxxxxxxxxxx\"}},");
+        // Keep one byte for the closing `]`.
+        if body.len() + item.len() >= body_bytes {
+            break;
+        }
+        body.push_str(&item);
+    }
+    // Pad with spaces before the `]` to reach the exact size.
+    let body = format!(
+        "{:<width$}]",
+        body.trim_end_matches(','),
+        width = body_bytes.saturating_sub(1)
+    );
     let recs: Vec<Value> = (0..records)
         .map(|i| {
             json!({
@@ -80,7 +92,7 @@ pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
     json!({ "Records": recs })
 }
 
-/// The reference SQS payload: 10 records of 22 KB, about 220 KB serialized.
+/// The reference SQS payload: 10 records of 22 KB, about 250 KB serialized.
 pub fn sqs_220kb() -> Value {
     sqs_event(10, 22_000)
 }
@@ -119,12 +131,18 @@ pub fn dynamodb_100_records() -> Value {
     dynamodb_event(100, 30)
 }
 
-/// Headers of the `GET /runtime/invocation/next` response that LWA intercepts.
+/// Headers of the `GET /runtime/invocation/next` response that LWA intercepts. The deadline is
+/// 60 s from now, so a runtime client that honors it does not time out at once.
 pub fn lwa_next_headers(request_id: &str) -> Vec<(&'static str, String)> {
+    let deadline_ms = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("clock after epoch")
+        .as_millis()
+        + 60_000;
     vec![
         ("content-type", "application/json".to_string()),
         ("lambda-runtime-aws-request-id", request_id.to_string()),
-        ("lambda-runtime-deadline-ms", "1741965555094".to_string()),
+        ("lambda-runtime-deadline-ms", deadline_ms.to_string()),
         (
             "lambda-runtime-invoked-function-arn",
             crate::fixtures::telemetry::FUNCTION_ARN.to_string(),
diff --git a/bottlecap/tests/fixtures/telemetry.rs b/bottlecap/tests/fixtures/telemetry.rs
index 296224262..0f6c5a5cc 100644
--- a/bottlecap/tests/fixtures/telemetry.rs
+++ b/bottlecap/tests/fixtures/telemetry.rs
@@ -157,9 +157,11 @@ pub fn function_event(kind: LogKind, i: usize) -> String {
     event(60 + i % 80, "function", &record)
 }
 
-/// An `extension` log event (bottlecap's own stdout as Lambda re-ingests it).
+/// An `extension` log event (bottlecap's own stdout as Lambda re-ingests it), in the JSON
+/// shape `logger::Formatter` writes.
 pub fn extension_event(i: usize) -> String {
-    let line = format!("[{}Z] DEBUG LOGS | processed batch {i}", time(60 + i % 80));
+    let line =
+        format!(r#"{{"status":"DEBUG","message":"DD_EXTENSION | DEBUG | processed batch {i}"}}"#);
     event(
         60 + i % 80,
         "extension",
diff --git a/bottlecap/tests/fixtures/traces.rs b/bottlecap/tests/fixtures/traces.rs
index 974bfcd37..f85d9ed20 100644
--- a/bottlecap/tests/fixtures/traces.rs
+++ b/bottlecap/tests/fixtures/traces.rs
@@ -1,10 +1,11 @@
 //! Trace payloads as a tracer library POSTs them to `/v0.4/traces`.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use libdd_trace_protobuf::pb;
 use libdd_trace_utils::tracer_header_tags::TracerHeaderTags;
+use serde::Serialize;
 
 /// Headers a Python tracer sends with a v0.4 payload.
 pub const TRACE_HEADERS: &[(&str, &str)] = &[
@@ -17,7 +18,7 @@ pub const TRACE_HEADERS: &[(&str, &str)] = &[
 
 /// Fixed start time of the generated spans: 2026-10-27T12:00:00Z, the instant the log
 /// fixtures use. Fixed, so the encoded payload is the same on every run.
-pub const BASE_NS: i64 = 1_761_566_400_000_000_000;
+pub const BASE_NS: i64 = 1_793_102_400_000_000_000;
 
 /// The current time, for consumers that drop old spans (the stats concentrator keeps a time
 /// window): `make_traces_at(now_ns(), ...)`.
@@ -150,9 +151,60 @@ pub fn trace_2000_spans() -> Vec<Vec<pb::Span>> {
     make_traces(100, 20, true)
 }
 
+/// A [`pb::Span`] with `meta` and `metrics` in key order, so the encoded bytes are the same on
+/// every run (a `HashMap` iterates in a random order). Same fields and order as `pb::Span`
+/// encodes; the generated spans have no `meta_struct`, `span_links` or `span_events`.
+#[derive(Serialize)]
+struct V04Span<'a> {
+    service: &'a str,
+    name: &'a str,
+    resource: &'a str,
+    trace_id: u64,
+    span_id: u64,
+    parent_id: u64,
+    start: i64,
+    duration: i64,
+    #[serde(skip_serializing_if = "is_zero")]
+    error: i32,
+    meta: BTreeMap<&'a str, &'a str>,
+    metrics: BTreeMap<&'a str, f64>,
+    r#type: &'a str,
+}
+
+fn is_zero(n: &i32) -> bool {
+    *n == 0
+}
+
+impl<'a> From<&'a pb::Span> for V04Span<'a> {
+    fn from(s: &'a pb::Span) -> Self {
+        Self {
+            service: &s.service,
+            name: &s.name,
+            resource: &s.resource,
+            trace_id: s.trace_id,
+            span_id: s.span_id,
+            parent_id: s.parent_id,
+            start: s.start,
+            duration: s.duration,
+            error: s.error,
+            meta: s
+                .meta
+                .iter()
+                .map(|(k, v)| (k.as_str(), v.as_str()))
+                .collect(),
+            metrics: s.metrics.iter().map(|(k, v)| (k.as_str(), *v)).collect(),
+            r#type: &s.r#type,
+        }
+    }
+}
+
 /// Encodes traces as a v0.4 msgpack body (`rmp_serde::to_vec_named`, as tracers do).
 pub fn v04_msgpack(traces: &[Vec<pb::Span>]) -> Vec<u8> {
-    rmp_serde::to_vec_named(traces).expect("encode v0.4 msgpack")
+    let traces: Vec<Vec<V04Span<'_>>> = traces
+        .iter()
+        .map(|t| t.iter().map(V04Span::from).collect())
+        .collect();
+    rmp_serde::to_vec_named(&traces).expect("encode v0.4 msgpack")
 }
 
 /// The reference v0.4 body: 2000 spans, about 1 MB.
diff --git a/bottlecap/tests/fixtures_corpus.rs b/bottlecap/tests/fixtures_corpus.rs
index 4b81555ae..ef5a20e72 100644
--- a/bottlecap/tests/fixtures_corpus.rs
+++ b/bottlecap/tests/fixtures_corpus.rs
@@ -52,6 +52,11 @@ fn trace_payload_round_trips() {
     let body = traces::v04_msgpack(&traces);
     let decoded: Vec<Vec<libdd_trace_protobuf::pb::Span>> = rmp_serde::from_slice(&body).unwrap();
     assert_eq!(decoded.len(), 100);
+    assert_eq!(
+        body,
+        traces::v04_2000_spans(),
+        "encoding is not deterministic"
+    );
 }
 
 #[test]
@@ -74,6 +79,7 @@ fn dogstatsd_packets_split_by_lines() {
 fn large_events_identify_as_their_trigger() {
     assert_eq!(events::all_payloads().len(), 38);
     let sqs = events::sqs_220kb();
+    assert_eq!(sqs["Records"][0]["body"].as_str().unwrap().len(), 22_000);
     let sqs_len = serde_json::to_vec(&sqs).unwrap().len();
     assert!((200_000..260_000).contains(&sqs_len), "sqs {sqs_len} B");
     assert!(matches!(

From e895b770b21cf5d0df956f317daa84c48ca82968 Mon Sep 17 00:00:00 2001
From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com>
Date: Wed, 28 Oct 2026 20:40:34 -0400
Subject: [PATCH 3/4] test: keep corpus telemetry POSTs under 1 MiB, scope the
 trace headers

- write_corpus: 600 events for the JSON shapes so each POST fits the 1 MiB maxBytes
- traces: TRACE_HEADERS matches the reference payload's 100 traces
---
 bottlecap/tests/fixtures/README.md    |  4 ++--
 bottlecap/tests/fixtures/mod.rs       | 20 ++++++++------------
 bottlecap/tests/fixtures/telemetry.rs |  3 ++-
 bottlecap/tests/fixtures/traces.rs    |  3 ++-
 bottlecap/tests/fixtures_corpus.rs    |  2 ++
 5 files changed, 16 insertions(+), 16 deletions(-)

diff --git a/bottlecap/tests/fixtures/README.md b/bottlecap/tests/fixtures/README.md
index d5c1efe52..5460a0b8e 100644
--- a/bottlecap/tests/fixtures/README.md
+++ b/bottlecap/tests/fixtures/README.md
@@ -49,8 +49,8 @@ Output goes to `bottlecap/tests/fixtures/generated/` (gitignored), or to `$BOTTL
 ```
 telemetry/platform_batch.json                 platform events of one cold start
 telemetry/function_text_1000.json             1000 text log events, one POST
-telemetry/function_json_object_1000.json      1000 JSON-object log events, one POST
-telemetry/function_json_string_1000.json      1000 JSON-in-string log events, one POST
+telemetry/function_json_object_600.json       600 JSON-object log events, one POST under 1 MiB
+telemetry/function_json_string_600.json       600 JSON-in-string log events, one POST under 1 MiB
 telemetry/invocation_batch_text_200.json      start + 200 lines + runtimeDone + report
 telemetry/split_record_1mib_post_{0..1}.json  a 1 MiB record in 256 KiB pieces, 3 per POST
 telemetry/split_record_1mib_single_post.json  the same pieces in one body
diff --git a/bottlecap/tests/fixtures/mod.rs b/bottlecap/tests/fixtures/mod.rs
index 588dedb6b..0ab05da84 100644
--- a/bottlecap/tests/fixtures/mod.rs
+++ b/bottlecap/tests/fixtures/mod.rs
@@ -57,20 +57,16 @@ pub fn write_corpus(dir: &Path) -> Vec<(PathBuf, usize)> {
         "telemetry/platform_batch.json",
         &telemetry::platform_batch(telemetry::REQUEST_ID),
     );
-    for (kind, name) in [
-        (telemetry::LogKind::Text, "function_text_1000.json"),
-        (
-            telemetry::LogKind::JsonObject,
-            "function_json_object_1000.json",
-        ),
-        (
-            telemetry::LogKind::JsonString,
-            "function_json_string_1000.json",
-        ),
+    // As many events as one POST holds: 1000 text events, or 600 JSON events (1.3 to 1.5 KB
+    // each) to stay under the 1 MiB `maxBytes`.
+    for (kind, n) in [
+        (telemetry::LogKind::Text, 1000),
+        (telemetry::LogKind::JsonObject, 600),
+        (telemetry::LogKind::JsonString, 600),
     ] {
         put(
-            &format!("telemetry/{name}"),
-            &telemetry::log_batch(kind, 1000),
+            &format!("telemetry/function_{}_{n}.json", kind.name()),
+            &telemetry::log_batch(kind, n),
         );
     }
     put(
diff --git a/bottlecap/tests/fixtures/telemetry.rs b/bottlecap/tests/fixtures/telemetry.rs
index 0f6c5a5cc..20695a1b2 100644
--- a/bottlecap/tests/fixtures/telemetry.rs
+++ b/bottlecap/tests/fixtures/telemetry.rs
@@ -196,7 +196,8 @@ pub fn platform_batch(request_id: &str) -> Vec<u8> {
     ])
 }
 
-/// `n` function log events of one shape, as one POST body (`n` is at most 1000 in practice).
+/// `n` function log events of one shape, as one POST body. Lambda caps a POST at 1000 events
+/// and 1 MiB, so keep `n` at most 600 for the JSON shapes.
 pub fn log_batch(kind: LogKind, n: usize) -> Vec<u8> {
     let events: Vec<String> = (0..n).map(|i| function_event(kind, i)).collect();
     batch(&events)
diff --git a/bottlecap/tests/fixtures/traces.rs b/bottlecap/tests/fixtures/traces.rs
index f85d9ed20..0ea2f72f7 100644
--- a/bottlecap/tests/fixtures/traces.rs
+++ b/bottlecap/tests/fixtures/traces.rs
@@ -7,7 +7,8 @@ use libdd_trace_protobuf::pb;
 use libdd_trace_utils::tracer_header_tags::TracerHeaderTags;
 use serde::Serialize;
 
-/// Headers a Python tracer sends with a v0.4 payload.
+/// Headers a Python tracer sends with the reference payload ([`v04_2000_spans`], 100 traces).
+/// Set `X-Datadog-Trace-Count` to the trace count when sending another payload.
 pub const TRACE_HEADERS: &[(&str, &str)] = &[
     ("Content-Type", "application/msgpack"),
     ("Datadog-Meta-Lang", "python"),
diff --git a/bottlecap/tests/fixtures_corpus.rs b/bottlecap/tests/fixtures_corpus.rs
index ef5a20e72..bc88be3e6 100644
--- a/bottlecap/tests/fixtures_corpus.rs
+++ b/bottlecap/tests/fixtures_corpus.rs
@@ -20,6 +20,8 @@ fn telemetry_batches_parse() {
             serde_json::from_slice(&telemetry::log_batch(kind, 10)).unwrap();
         assert_eq!(events.len(), 10, "{}", kind.name());
     }
+    // The largest shape, as `write_corpus` writes it, fits the 1 MiB `maxBytes`.
+    assert!(telemetry::log_batch(telemetry::LogKind::JsonString, 600).len() <= 1 << 20);
     let events: Vec<TelemetryEvent> =
         serde_json::from_slice(&telemetry::invocation_batch(telemetry::LogKind::Text, 5)).unwrap();
     assert_eq!(events.len(), 8);

From 4c972dc1657d298b27a1f10a1b28237c73cc59de Mon Sep 17 00:00:00 2001
From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com>
Date: Fri, 30 Oct 2026 16:27:37 -0400
Subject: [PATCH 4/4] test: align fixture request ids and timestamps, drop the
 span-only metric tag

---
 bottlecap/tests/fixtures/dogstatsd.rs |  2 +-
 bottlecap/tests/fixtures/events.rs    | 15 ++++++++++-----
 bottlecap/tests/fixtures/traces.rs    |  5 ++++-
 3 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/bottlecap/tests/fixtures/dogstatsd.rs b/bottlecap/tests/fixtures/dogstatsd.rs
index a85372af7..ac65502a2 100644
--- a/bottlecap/tests/fixtures/dogstatsd.rs
+++ b/bottlecap/tests/fixtures/dogstatsd.rs
@@ -1,7 +1,7 @@
 //! DogStatsD datagrams as a client library sends them to UDP 8125.
 
 /// The tags bottlecap adds to every metric (`SortedTags::parse(ENRICH_TAGS)`).
-pub const ENRICH_TAGS: &str = "function_arn:arn:aws:lambda:us-east-1:123456789012:function:my-checkout-fn,region:us-east-1,account_id:123456789012,aws_account:123456789012,functionname:my-checkout-fn,resource:my-checkout-fn,executedversion:$LATEST,memorysize:128,architecture:arm64,runtime:nodejs22.x,_dd.runtime_family:nodejs,env:prod,service:checkout,version:1.2.3,dd_extension_version:90,_dd.compute_stats:1";
+pub const ENRICH_TAGS: &str = "function_arn:arn:aws:lambda:us-east-1:123456789012:function:my-checkout-fn,region:us-east-1,account_id:123456789012,aws_account:123456789012,functionname:my-checkout-fn,resource:my-checkout-fn,executedversion:$LATEST,memorysize:128,architecture:arm64,runtime:nodejs22.x,_dd.runtime_family:nodejs,env:prod,service:checkout,version:1.2.3,dd_extension_version:90";
 
 /// A distribution with 8 tags, the heaviest common line.
 pub const DIST_8_TAGS: &str = "app.checkout.latency:12.5|d|#env:prod,service:checkout,version:1.2.3,region:us-east-1,team:payments,endpoint:/api/v1/pay,status:200,method:POST";
diff --git a/bottlecap/tests/fixtures/events.rs b/bottlecap/tests/fixtures/events.rs
index bd4b7a8b3..ff1f2ee6c 100644
--- a/bottlecap/tests/fixtures/events.rs
+++ b/bottlecap/tests/fixtures/events.rs
@@ -6,6 +6,8 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use serde_json::{Value, json};
 
+use crate::fixtures::traces::BASE_NS;
+
 fn payloads_dir() -> PathBuf {
     Path::new(env!("CARGO_MANIFEST_DIR"))
         .join("tests")
@@ -45,7 +47,8 @@ pub fn all_payloads() -> Vec<(String, Value)> {
 }
 
 /// An SQS batch of `records` messages, each with a JSON array body of exactly `body_bytes`
-/// bytes (at least 2) and a `_datadog` message attribute carrying trace context.
+/// bytes (at least 2) and a `_datadog` message attribute carrying trace context. Sent at
+/// [`BASE_NS`].
 pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
     let mut body = String::from("[");
     for i in 0.. {
@@ -62,6 +65,7 @@ pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
         body.trim_end_matches(','),
         width = body_bytes.saturating_sub(1)
     );
+    let sent_ms = BASE_NS / 1_000_000;
     let recs: Vec<Value> = (0..records)
         .map(|i| {
             json!({
@@ -70,9 +74,9 @@ pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
                 "body": body,
                 "attributes": {
                     "ApproximateReceiveCount": "1",
-                    "SentTimestamp": "1545082649183",
+                    "SentTimestamp": sent_ms.to_string(),
                     "SenderId": "AIDAIENQZJOLO23YVJ4VO",
-                    "ApproximateFirstReceiveTimestamp": "1545082649185"
+                    "ApproximateFirstReceiveTimestamp": (sent_ms + 2).to_string()
                 },
                 "messageAttributes": {
                     "_datadog": {
@@ -97,7 +101,8 @@ pub fn sqs_220kb() -> Value {
     sqs_event(10, 22_000)
 }
 
-/// A DynamoDB stream batch of `records` INSERT records with `attrs` attributes each.
+/// A DynamoDB stream batch of `records` INSERT records with `attrs` attributes each, created
+/// at [`BASE_NS`].
 pub fn dynamodb_event(records: usize, attrs: usize) -> Value {
     let recs: Vec<Value> = (0..records)
         .map(|i| {
@@ -112,7 +117,7 @@ pub fn dynamodb_event(records: usize, attrs: usize) -> Value {
                 "eventSource": "aws:dynamodb",
                 "awsRegion": "us-east-1",
                 "dynamodb": {
-                    "ApproximateCreationDateTime": 1_428_537_600,
+                    "ApproximateCreationDateTime": BASE_NS / 1_000_000_000,
                     "Keys": {"Id": {"N": "101"}},
                     "NewImage": Value::Object(img),
                     "SequenceNumber": format!("{i}00000000000000000"),
diff --git a/bottlecap/tests/fixtures/traces.rs b/bottlecap/tests/fixtures/traces.rs
index 0ea2f72f7..5fb0a34dc 100644
--- a/bottlecap/tests/fixtures/traces.rs
+++ b/bottlecap/tests/fixtures/traces.rs
@@ -7,6 +7,8 @@ use libdd_trace_protobuf::pb;
 use libdd_trace_utils::tracer_header_tags::TracerHeaderTags;
 use serde::Serialize;
 
+use crate::fixtures::telemetry::REQUEST_ID;
+
 /// Headers a Python tracer sends with the reference payload ([`v04_2000_spans`], 100 traces).
 /// Set `X-Datadog-Trace-Count` to the trace count when sending another payload.
 pub const TRACE_HEADERS: &[(&str, &str)] = &[
@@ -66,7 +68,7 @@ pub fn make_traces_at(
                                 "serverless",
                                 vec![
                                     ("span.kind", "server"),
-                                    ("request_id", "8f5a7e2c-1b2c-4d5e-9f00-0123456789ab"),
+                                    ("request_id", REQUEST_ID),
                                     ("function_trigger.event_source", "api-gateway"),
                                     ("http.method", "GET"),
                                     ("http.status_code", "200"),
@@ -200,6 +202,7 @@ impl<'a> From<&'a pb::Span> for V04Span<'a> {
 }
 
 /// Encodes traces as a v0.4 msgpack body (`rmp_serde::to_vec_named`, as tracers do).
+/// For generated spans only: `meta_struct`, `span_links` and `span_events` are not encoded.
 pub fn v04_msgpack(traces: &[Vec<pb::Span>]) -> Vec<u8> {
     let traces: Vec<Vec<V04Span<'_>>> = traces
         .iter()