duncanista · Nov 5, 2026 · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026
@@ -27,4 +27,6 @@ bottlecap/proptest-regressions
 integration-tests/cdk.context.json
 
 .gitlab/pipeline*
-/CLAUDE_PERSONAL.md
+/CLAUDE_PERSONAL.md
+# Generated benchmark fixture corpus (see bottlecap/tests/fixtures/README.md)
+bottlecap/tests/fixtures/generated
@@ -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";
 

@@ -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) {

@@ -0,0 +1,93 @@
+# Fixture corpus
+
+Shared inputs for the criterion benches (`bottlecap/benches/`) and the fault-injection harness.
+Everything is generated by the Rust module in this directory, so no large blob is checked in.
+
+## What the fixtures guarantee
+
+- **One invocation.** Every generator takes the identity of one simulated invocation from
+  `invocation.rs`: function name and ARN, region, account, memory, architecture, runtime,
+  `service`/`env`/`version`, request ID, X-Ray root, tracer, and times. The invocation is a cold
+  start: `platform.initStart` at `BASE_NS` (2026-10-27T12:00:00Z), `platform.start` at +50 ms,
+  the handler (the `aws.lambda` root span) from +51 ms to +141 ms, and `platform.runtimeDone` at
+  +150 ms. Log lines fall between `platform.start` and `platform.runtimeDone`. The
+  `fixtures_share_one_invocation` test fails when two fixtures disagree.
+- **Same bytes on every run**, except `lwa_next_headers` (its deadline is 60 s from now) and
+  `make_traces_at(now_ns(), ...)`.
+
+Not covered:
+
+- The 38 trigger events in `tests/payloads/` are checked-in samples with their own IDs and times.
+- The generated SQS and DynamoDB events share the region, account and `BASE_NS`, but the traces
+  do not continue their `_datadog` trace context.
+- `make_traces` repeats the invocation's root in every trace, for volume.
+
+## Use from Rust
+
+```rust
+#[path = "../tests/fixtures/mod.rs"] // from benches/; integration tests use `mod fixtures;`
+mod fixtures;
+use fixtures::{dogstatsd, events, invocation, logs, telemetry, traces};
+```
+
+| Module | Generator | What it returns |
+|---|---|---|
+| `invocation` | `FUNCTION_NAME`, `FUNCTION_ARN`, `REGION`, `ACCOUNT_ID`, `MEMORY_SIZE_MB`, `ARCHITECTURE`, `RUNTIME`, `SERVICE`, `ENV`, `VERSION`, `REQUEST_ID`, `XRAY_ROOT`, `XRAY_TRACE_HEADER`, `TRACER_*` | The identity of the simulated invocation |
+| | `BASE_NS`, `START_MS`, `HANDLER_START_MS`, `HANDLER_MS`, `RUNTIME_DONE_MS`, `time(millis)`, `log_millis(i)` | Its times: offsets in ms from `BASE_NS`, and the time of the `i`th log line |
+| `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>` | The invocation in the POST after its init events: `platform.start`, `n` log lines, `runtimeDone`, the cold-start `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)` (cold), `warm_report(id)` (for another request ID), `function_event(kind, i)`, `extension_event(i)`, `batch(&[String])` | Building blocks |
+| | `REQUEST_ID`, `FUNCTION_ARN`, `XRAY_TRACE_HEADER` | Re-exported from `invocation` |
+| `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 over the handler, SQL and HTTP clients, internal spans |
+| | `make_traces_at(base_ns, ...)`, `now_ns()`, `BASE_NS` | The same traces for an invocation that starts at `base_ns`; `BASE_NS` is re-exported from `invocation` |
+| | `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 Node.js 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()` | The tags bottlecap adds to every metric, for `FUNCTION_ARN` and the current `EXTENSION_VERSION` |
+| | `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, `Id` key plus `attrs` attributes per image; reference 100 x 30 |
+| | `lwa_next_headers(request_id)` | Headers of the `/invocation/next` response LWA intercepts, with `XRAY_TRACE_HEADER` |
+| 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. A new run first removes the files the previous
+`MANIFEST.txt` lists.
+
+```
+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_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
+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/`.
@@ -0,0 +1,84 @@
+//! DogStatsD datagrams as a client library sends them to UDP 8125.
+
+use bottlecap::tags::lambda::tags::EXTENSION_VERSION;
+
+use crate::fixtures::invocation::{
+    ACCOUNT_ID, ARCHITECTURE, ENV, FUNCTION_ARN, FUNCTION_NAME, INIT_TYPE, MEMORY_SIZE_MB, REGION,
+    SERVICE, VERSION,
+};
+
+/// The tags bottlecap adds to every metric of the `$LATEST` version of [`FUNCTION_ARN`]
+/// (`SortedTags::parse(&enrich_tags())`). No `runtime` tag: Lambda does not give
+/// `AWS_EXECUTION_ENV` to extensions.
+pub fn enrich_tags() -> String {
+    format!(
+        "function_arn:{FUNCTION_ARN},region:{REGION},account_id:{ACCOUNT_ID},aws_account:{ACCOUNT_ID},functionname:{FUNCTION_NAME},resource:{FUNCTION_NAME},memorysize:{MEMORY_SIZE_MB},architecture:{ARCHITECTURE},init_type:{INIT_TYPE},env:{ENV},service:{SERVICE},version:{VERSION},dd_extension_version:{EXTENSION_VERSION}"
+    )
+}
+
+/// A distribution with 8 tags, the heaviest common line. Its `env`, `service`, `version` and
+/// `region` tags are those of the shared invocation.
+pub const DIST_8_TAGS: &str = "app.checkout.latency:12.5|d|#env:prod,service:orders,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 `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 / 100) % 50,
+            i % 100
+        ),
+        Kind::Gauge => format!(
+            "app.queue.depth{}:{}|g|#endpoint:/api/v{},status:200,method:GET",
+            (i / 100) % 50,
+            i % 900,
+            i % 100
+        ),
+        Kind::Distribution => format!(
+            "app.request.latency{}:{}|d|#endpoint:/api/v{},status:200,method:GET",
+            (i / 100) % 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:{ENV},service:{SERVICE},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 (at least 1).
+pub fn packets(kind: Kind, total: usize, per_packet: usize) -> Vec<String> {
+    let per_packet = per_packet.max(1);
+    (0..total)
+        .step_by(per_packet)
+        .map(|first| packet(kind, first, per_packet.min(total - first)))
+        .collect()
+}
@@ -0,0 +1,164 @@
+//! Invocation payloads: the 38 trigger events in `tests/payloads/`, checked in with their own
+//! ids and times, plus large generated SQS and DynamoDB events in the region, account and time
+//! of the shared invocation.
+
+use std::path::{Path, PathBuf};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use serde_json::{Value, json};
+
+use crate::fixtures::invocation::{ACCOUNT_ID, BASE_NS, FUNCTION_ARN, REGION, XRAY_TRACE_HEADER};
+
+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")
+        .map(|e| e.expect("read a tests/payloads entry"))
+        .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 from a queue in the function's region and account, each
+/// with a JSON array body of exactly `body_bytes` bytes (at least 2) and a `_datadog` message
+/// attribute carrying the producer's 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.. {
+        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 sent_ms = BASE_NS / 1_000_000;
+    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": sent_ms.to_string(),
+                    "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+                    "ApproximateFirstReceiveTimestamp": (sent_ms + 2).to_string()
+                },
+                "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": format!("arn:aws:sqs:{REGION}:{ACCOUNT_ID}:my-queue"),
+                "awsRegion": REGION
+            })
+        })
+        .collect();
+    json!({ "Records": recs })
+}
+
+/// The reference SQS payload: 10 records of 22 KB, about 250 KB serialized.
+pub fn sqs_220kb() -> Value {
+    sqs_event(10, 22_000)
+}
+
+/// A DynamoDB stream batch of `records` INSERT records from a table in the function's region
+/// and account, created at [`BASE_NS`]. Each `NewImage` holds the `Id` key plus `attrs` other
+/// attributes.
+pub fn dynamodb_event(records: usize, attrs: usize) -> Value {
+    let recs: Vec<Value> = (0..records)
+        .map(|i| {
+            let key = json!({"N": (101 + i).to_string()});
+            let mut img = serde_json::Map::new();
+            img.insert("Id".to_string(), key.clone());
+            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": REGION,
+                "dynamodb": {
+                    "ApproximateCreationDateTime": BASE_NS / 1_000_000_000,
+                    "Keys": {"Id": key},
+                    "NewImage": Value::Object(img),
+                    "SequenceNumber": format!("{i}00000000000000000"),
+                    "SizeBytes": 26,
+                    "StreamViewType": "NEW_AND_OLD_IMAGES"
+                },
+                "eventSourceARN": format!("arn:aws:dynamodb:{REGION}:{ACCOUNT_ID}: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. 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", deadline_ms.to_string()),
+        (
+            "lambda-runtime-invoked-function-arn",
+            FUNCTION_ARN.to_string(),
+        ),
+        (
+            "lambda-runtime-trace-id",
+            format!("{XRAY_TRACE_HEADER};Lineage=1:73f724a8:0"),
+        ),
+    ]
+}