test: process-level fault-injection harness and the contract tests that pass today

AGENTS.md

@@ -8,7 +8,18 @@ The Datadog Lambda Extension ("Bottlecap") is an AWS Lambda Extension written in
 
 ## Common Commands
 
-All Rust code lives under `bottlecap/`. Run these from that directory unless noted.
+All Rust code lives under `bottlecap/`. Run these from that directory unless noted. CI runs the same commands (`.github/workflows/rs_ci.yml`).
+
+- Build: `cargo build`
+- Test: `cargo nextest run --workspace`
+- Lint: `cargo clippy --workspace --all-targets --features default`. CI also runs it with `--features default,test-mode` and with `--no-default-features --features fips`.
+- Format: `cargo fmt --all -- --check`
+
+## Testing
+
+- Unit tests live next to the code. Integration tests are in `bottlecap/tests/`.
+- Process-level tests run the real binary against a fake Lambda API and an intake that fails on demand. Read `bottlecap/tests/HARNESS.md` before you add or change one. Each reliability fix should come with one of these tests.
+- Shared test inputs (telemetry batches, logs, traces, metrics, trigger events) are in `bottlecap/tests/fixtures/`. See `bottlecap/tests/fixtures/README.md`.
 
 ## Where New Code Goes
 

bottlecap/.config/nextest.toml

@@ -0,0 +1,20 @@
+# Nextest configuration for the bottlecap workspace (CI runs `cargo nextest run
+# --workspace` from this directory, see .github/workflows/rs_ci.yml).
+#
+# Nextest runs every test in its own process, so `#[serial]` cannot protect a
+# fixed port across test binaries. The process-level contract tests spawn the
+# real `bottlecap` binary, which binds 8124, 8125/udp, 8126, 8999 and 9000.
+# Those binaries go in the `fixed-ports` group: one test at a time, across all
+# of them. See tests/HARNESS.md.
+
+[test-groups]
+fixed-ports = { max-threads = 1 }
+
+# A regex so the config stays valid when one of these binaries is absent
+# (nextest rejects an exact `binary(name)` that matches nothing).
+[[profile.default.overrides]]
+filter = 'binary(/(^|_)(lifecycle_contract|fault_injection|config_robustness|port_conflict)_test$/)'
+test-group = "fixed-ports"
+# A debug build of the extension, a blocking flush and a shutdown deadline add
+# up; the longest test (a storm with a memory window) stays under this.
+slow-timeout = { period = "60s", terminate-after = 3 }

bottlecap/Cargo.toml

@@ -122,6 +122,8 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["loggin
 # fake-intake test harness: decode msgpack+gzip stats payloads on arrival
 rmp-serde = { version = "1.3.1", default-features = false }
 flate2 = { version = "1.1", default-features = false, features = ["rust_backend"] }
+# fault-intake test harness: decode the `datadog-protos` metrics payloads
+protobuf = { version = "3.7", default-features = false }
 
 [build-dependencies]
 # No external dependencies needed for the build script

bottlecap/tests/HARNESS.md

@@ -0,0 +1,187 @@
+# Fault-injection test harness
+
+Process-level contract tests run the real `bottlecap` binary against a fake
+Lambda control plane and a Datadog intake that misbehaves on demand. Each test
+asserts what Lambda or the customer would see: `/next` is called before the
+deadline, the process exits within the SHUTDOWN deadline, data reaches the
+intake, and nothing panics. No test reads internal state.
+
+Every reliability fix should come with one of these tests. A test for a bug that
+is not fixed yet is committed `#[ignore = "<ID>: <what the extension does today>"]`,
+so CI stays green and the test turns into a regression test when the fix lands
+(remove the attribute in the same PR). The ID names the Jira ticket that tracks
+the fix (for example `SVLS-10061`). A reason that starts with `needs owner decision` marks
+a test whose expected behavior is a design choice, not a plain bug.
+
+## Layout
+
+| File | Role |
+|---|---|
+| `tests/common/harness.rs` | Umbrella module and `prelude`. |
+| `tests/common/fake_extensions_api.rs` | `FakeExtensionsApi`: register, telemetry subscribe (status knob), test-driven `/next`, request log, Telemetry API sender. |
+| `tests/common/fault_intake.rs` | `FaultIntake` with `FaultMode`, per-path modes, body-size rejection, request log; `SynBlackhole`. |
+| `tests/common/bottlecap_process.rs` | `BottlecapProcess`: spawn the binary with the base env, capture stdout/stderr, RSS/CPU, exit with a deadline, kill on drop. |
+| `tests/common/port_guard.rs` | Hold or probe the fixed ports (8124, 8125/udp, 8126, 8999, 9000). |
+| `tests/common/telemetry_events.rs` | Telemetry API records as `Value`, thin wrappers over `tests/fixtures/telemetry.rs`. |
+| `tests/fixtures/` | The shared corpus for benches and these tests (see its `README.md`). Each test crate declares `mod fixtures;`. |
+| `tests/lifecycle_contract_test.rs` | Shutdown and `/next` contracts. |
+| `tests/fault_injection_test.rs` | Intake faults: a 503 storm, an oversized trace, a response cut off mid-body. |
+| `tests/config_robustness_test.rs` | Wrong-typed config values. |
+| `tests/port_conflict_test.rs` | Another process holds port 8124 or 8126. |
+| `.config/nextest.toml` | `fixed-ports` test group (`max-threads = 1`) for these binaries. |
+
+## Write a new fault test in 10 lines
+
+```rust
+mod fixtures;
+#[path = "common/harness.rs"]
+mod harness;
+use harness::prelude::*;
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial_test::serial(bottlecap_ports)]
+#[ignore = "SVLS-xxxxx: <what the extension does today>"]
+async fn my_fault() {
+    let s = Scenario::start(FaultMode::Ok).await;          // fake API + intake, fixed ports checked
+    let mut bc = s.bottlecap().spawn();                     // real binary, base env points at the fakes
+    let startup = bc.startup_deadline();                    // /next #1 must arrive by it
+    let deadline = s.invoke("req-1", secs(10), 1, startup).await.unwrap(); // answer it with an INVOKE
+    s.intake.set_mode(FaultMode::Status(503));              // the fault
+    deliver_start(&s.api, "req-1", true).await.unwrap();   // initStart (cold start), then platform.start
+    deliver_runtime_done(&s.api, "req-1", true).await.unwrap();
+    // /next #2 must arrive by the INVOKE's deadline, counted from before it was queued
+    assert!(s.api.wait_for_next_calls(2, deadline).await, "{}", bc.log_tail(40));
+    s.shutdown_cleanly(&mut bc).await;                      // SHUTDOWN, exit 0 within 2 s, no panic
+}
+```
+
+## Rules
+
+- Never sleep a fixed time. Wait on a recorded event with a deadline:
+  `api.wait_for_next_calls`, `intake.wait_for(...)`, `bc.wait_for_exit`.
+- Assert the customer-visible contract (a `/next` call, an exit, an intake
+  request), never a log line. Print `bc.log_tail(40)` and `s.intake.summary()`
+  in the message. One exception: a log line may prove that the fault was really
+  injected (the port-conflict tests check that the held port conflicted), never
+  that the contract holds.
+- Bound a flush by the invocation or shutdown deadline, not by a number from
+  an implementation.
+- Count every deadline from the instant its event was queued: use the `Instant` that `invoke`, `push_invoke` or `push_shutdown` returns, and `bc.startup_deadline()` for the first `/next`.
+- Compare a deadline with the instant the fakes recorded (`next_call_at(n)`, `IntakeRequest::at`, `bc.seen_running()`), never with `Instant::now()` after a wait: a slow test thread delays the wait, not the event.
+- Match decoded content in every data assertion (the log text in `body_text()`, a name in `metric_names()`, an ID in `trace_ids()`), not only a path, status, size or API key; a check that only counts requests says why in a comment.
+- Require `api_key() == Some(TEST_API_KEY)` in every positive data assertion, a faulted request too: the fake intake accepts a request without the key, the real one does not.
+- Always check `bc.assert_no_panic()` (or use `shutdown_cleanly`). Dev builds
+  unwind and keep running, release builds abort; both print `panicked at`.
+- Check memory growth only on Linux (`BottlecapProcess::can_measure_memory()`).
+- Send telemetry in Lambda's order: `deliver_start` (with `platform.initStart` on the cold
+  start), then the function logs, then `deliver_runtime_done`. Lambda sends `platform.report`
+  only after the extension's next `/next`. Until `platform.initStart` arrives the extension
+  holds every log and the cold start's `aws.lambda.enhanced.invocations`.
+- When the extension could break a Lambda phase rule, make the fake refuse it (a subscribe
+  after the first `/next` gets `403`) and list the rule in the API summary.
+- Every spawning test carries `#[serial_test::serial(bottlecap_ports)]`: the
+  binary binds fixed ports, and plain `cargo test` runs tests in parallel. New
+  test binaries must also match the regex in `.config/nextest.toml`.
+- Run with a per-worktree `CARGO_TARGET_DIR`: the binary under test is the
+  uplifted `target/debug/bottlecap`, which a shared target dir overwrites.
+  `log_tail` prints the binary path and age.
+
+## API summary
+
+- `FakeExtensionsApi`: `push_invoke(id, deadline_in)`, `push_shutdown(reason, deadline_in)` (each
+  returns the event's deadline as an `Instant`, taken before the event is queued), `next_calls()`, `next_call_at(n)`, `wait_for_next_calls(n, deadline)` (true only when `/next` call `n` arrived by
+  the `Instant`), `set_subscribe_status(StatusCode)`, `request_lines()`,
+  `send_telemetry(&[Value], budget)`, `send_telemetry_raw(Vec<u8>, budget)`, `telemetry_port()`
+  (from the last accepted subscribe body, `None` before one; `send_telemetry` fails without it,
+  as Lambda delivers telemetry only to subscribers). Both post to the port and path of the
+  subscribed URI. Like Lambda, `send_telemetry` withholds
+  events whose type (`platform`, `function`, `extension`) the subscribe body did not ask for;
+  `send_telemetry_raw` posts its body as is. Like Lambda, `/next` serves only the event types from
+  the register body and drops the rest, a `/next` while another is pending gets `403`, `/next`
+  and subscribe get `403` without the `Lambda-Extension-Identifier` issued at register, subscribe
+  gets `403` after the first `/next` (Lambda takes subscriptions only during init), register
+  gets `403` without `Lambda-Extension-Name: datadog-agent` or with an invalid body, a subscribe
+  body outside the Telemetry API reference rules gets `400` (also an `https://` URI, port 9001,
+  or a missing `Content-Type: application/json`), and
+  `accountId` needs `Lambda-Extension-Accept-Feature`. `set_managed_instance()` applies the Lambda
+  Managed Instances rules: register gets `403` for `INVOKE`, and subscribe gets `400` for any
+  `schemaVersion` but `2025-01-29`. The SHUTDOWN reason is lower case (`spindown`), as the Runtime
+  Interface Emulator sends it.
+  Any other path gets `202`.
+- `FaultIntake` (HTTP/1.1): modes `Ok`, `Status(code)`, `Slow(d)`, `Blackhole`,
+  `RetryAfter{status, secs}`, `CloseMidBody`; string forms `ok`, `status:<c>`, `slow:<ms>`,
+  `blackhole`, `retry_after:<c>:<s>`, `close_mid_body`.
+  `Blackhole` never reads a body: with a path mode set, it reads only the request head to pick
+  the mode, and records the request with `body_len` 0. An upload that breaks off is not
+  recorded and gets no answer.
+  `set_mode`, `set_path_mode(path, mode)`, `clear_path_mode`, `set_reject_over(Some(bytes))`,
+  `set_ok_body(path, body)`; `requests()`, `requests_to`, `count`, `connections()`, `summary()`,
+  `wait_for(n, pred, deadline)` (true only when `n` matching requests arrived by the `Instant`;
+  `Slow` records a request before its delay),
+  `wait_for_path`; `IntakeRequest::at` (arrival), `api_key()`, `decoded_body()`, `body_text()`,
+  `trace_ids()` (every span's trace ID in an `AgentPayload`) and `metric_names()` (a sketches or
+  series payload), each `Err` when the body does not decode. `SynBlackhole::start()` (backlog 1,
+  queue pre-filled), `is_saturated()`.
+- `BottlecapProcess`: `scenario.bottlecap()` then `.intake(url)`, `.env`, `.envs`, `.remove_env`,
+  `.managed_instance()` (also switches the fake API to its rules), `.flush_strategy(s)`,
+  `.secret_from(credentials_url)`, `.spawn()`; `wait_for_exit(budget)` (kills on timeout),
+  `is_running()`, `kill()`, `stdout()`, `stderr()`, `log_tail(n)`, `log_contains`, `panic_lines()`,
+  `assert_no_panic()`, `rss_kb()`, `hwm_kb()`, `cpu_ms()`, `uptime()`, `startup_deadline()`
+  (`STARTUP_BUDGET` from before the spawn), `seen_running()` (start of the last poll that found
+  the process running); `STARTUP_BUDGET`,
+  `EXIT_BUDGET`, `TEST_API_KEY`.
+- `Scenario`: `start(mode)`, `bottlecap()`, `invoke(id, deadline_in, poll, poll_by)` (waits for
+  the poll by `poll_by`, then answers it; returns the INVOKE's deadline, `None` when the poll
+  missed `poll_by`), `wait_next_or_exit(bc, n, deadline)`, `shutdown(&mut bc)` (asserts the process is still
+  running when SHUTDOWN is queued and no poll saw it running after the 2 s SHUTDOWN deadline),
+  `shutdown_cleanly(&mut bc)` (also asserts exit 0 and no panic).
+  Helpers: `deliver_start(api, id, cold_start)` (`platform.initStart` on a cold start, then
+  `platform.start`), `deliver_runtime_done(api, id, with_metrics)` (`platform.runtimeDone` only),
+  `msgpack_trace(trace_id, meta_bytes)`, `post_trace` (to 8126),
+  `INVOCATIONS_METRIC`,
+  `incompressible_text`, `secs`, `ms`. Records: `platform_init_start`, `platform_start`,
+  `platform_runtime_done(id, status, with_metrics)`, `platform_report`, `function_log`,
+  `platform_logs_dropped`, `unknown_event`; larger bodies come from `fixtures::telemetry`.
+- `PortGuard::tcp(port)` / `udp(port)` / `*_wait` bind the address the extension uses.
+
+## Base environment
+
+`AWS_LAMBDA_RUNTIME_API=<fake api>`, `AWS_LAMBDA_FUNCTION_NAME=test`,
+`AWS_LAMBDA_FUNCTION_MEMORY_SIZE=128`, `AWS_LAMBDA_FUNCTION_VERSION=$LATEST`,
+`AWS_REGION`/`AWS_DEFAULT_REGION=us-east-1`, `LAMBDA_TASK_ROOT=<tempdir>`,
+`DD_API_KEY=<TEST_API_KEY>`, `DD_SITE=datadoghq.com`, `DD_LOG_LEVEL=debug`,
+`DD_SERVERLESS_FLUSH_STRATEGY=end`,
+`DD_DD_URL`/`DD_URL`/`DD_APM_DD_URL`/`DD_LOGS_CONFIG_LOGS_DD_URL=http://<intake>`,
+`DD_HTTP_PROTOCOL=http1` (the fakes speak HTTP/1.1; the default client uses h2
+prior knowledge), `DD_PROXY_HTTPS=http://127.0.0.1:1` (trace stats follow
+`DD_SITE` over https; an https-only proxy on a closed port fails them at once
+while the `http://` intake URLs bypass it). Managed Instance mode adds
+`AWS_LAMBDA_INITIALIZATION_TYPE=lambda-managed-instances`.
+
+## Running
+
+```sh
+cd bottlecap
+cargo nextest run --test lifecycle_contract_test --test fault_injection_test \
+    --test config_robustness_test --test port_conflict_test            # CI set, ignored skipped
+cargo nextest run --test fault_injection_test --run-ignored only          # not-yet-fixed tests (none yet)
+cargo test --test fault_injection_test                                    # also serialized
+```
+
+`.config/nextest.toml` puts these binaries in the `fixed-ports` group
+(`max-threads = 1`); nextest runs each test in its own process, so the
+`serial` attribute alone would not protect the ports across binaries. A test
+fails fast with `fixed ports still busy before spawn` when a leaked process
+holds one.
+
+## Hand-offs
+
+- The fault modes belong in the shared `datadog-mock-intake` crate once
+  bottlecap consumes it: a `FaultMode` enum with the string forms above,
+  `set_mode`/`set_path_mode`, body-size rejection, and the recorded status and
+  mode per request. `blackhole` needs the accept loop, so the crate must expose
+  it. Until then `fault_intake.rs` stays here next to the shared crate's payload
+  decoding.
+- A secret-recovery test (the secret fetch fails, then recovers) needs the extension to honor
+  `AWS_ENDPOINT_URL_<SERVICE>` / `AWS_ENDPOINT_URL` (the AWS SDK convention) for
+  its Secrets Manager call; that hook is a separate small change.