# Phase 19: Message Headers / Metadata

## Goal

Allow publishers to attach key-value headers to messages that are preserved through the entire lifecycle — publish, store, deliver, fetch — without being embedded in the payload. Headers enable content-type negotiation, correlation IDs for request-reply patterns, distributed trace propagation (OpenTelemetry), routing metadata, and application-defined filtering — all without forcing structure on the payload bytes.

## Guiding Principles

- **Headers are optional.** A message with zero headers is byte-identical on the wire to the current format (backward compatible). The presence of headers is signaled by a capability flag negotiated during Connect.
- **Headers are opaque key-value pairs.** Both key and value are byte strings. The server does not interpret header contents (no built-in "content-type" semantics). The server stores and forwards headers exactly as received.
- **Headers are stored with the message.** Headers are part of the segment record on disk. They survive restart, are returned by Fetch, and are included in DLQ envelopes. They are not sidecar metadata.
- **Headers have a size budget.** Total header bytes per message are capped (default 8 KB) to prevent abuse. Individual header count is capped (default 64) to prevent hashtable DoS.
- **Reserved prefix for system headers.** Keys starting with `_fafb.` are reserved for server-set headers (e.g., `_fafb.dlq.source_topic`, `_fafb.dlq.reason` in future DLQ integration). Clients cannot set `_fafb.*` headers — the server rejects them.

## Current State

| Capability | Implementation | Gap |
|---|---|---|
| Publish payload | `Bytes` in `Frame::Publish { topic, payload }` | No metadata channel alongside payload |
| Deliver payload | `Frame::Deliver { message_id, offset, timestamp_millis, topic, payload }` | No headers |
| Fetch payload | `FetchedMessage { offset, timestamp_millis, payload }` | No headers |
| Message envelope | `MessageEnvelope { id, namespace, topic, payload: Vec<u8>, timestamp }` — `Serialize`/`Deserialize` via serde | No headers field |
| Storage record | `[4B record_len][8B offset][bincode(MessageEnvelope)]` — uses **bincode** serialization, NOT manual binary | No header field in envelope |
| Wire protocol | Publish frame: `[u16 topic_len][topic][u32 payload_len][payload]` | No header section |
| HTTP publish | Raw body = payload | No header pass-through |
| Capability bits | `CAP_GOING_AWAY` (bit 0), `CAP_RECONNECT` (bit 1), `CAP_NACK` (bit 2) | Bit 3 used for `CAP_HEADERS` |

## Design

### Wire Format Changes

#### Capability Negotiation

- `CAP_HEADERS` = `1 << 3` (bit 3) in `Capabilities` bitmask
  - **Note:** The original design specified bit 2, but Plan 18 (DLQ) landed first and claimed bit 2 for `CAP_NACK`. Bit 3 is used instead.
- Client and server must both advertise `CAP_HEADERS` for headers to be used
- Without `CAP_HEADERS`, wire format is identical to current — zero breaking change

#### Publish Frame (updated)

When `CAP_HEADERS` is negotiated:

```
Frame::Publish (0x07) — with headers:
  [u16 topic_length]
  [u8* topic]
  [u16 header_count]              ← NEW
  for each header:
    [u16 key_length]
    [u8* key]
    [u32 value_length]            ← u32 to support binary values
    [u8* value]
  [u32 payload_length]
  [u8* payload]
```

Without `CAP_HEADERS` (legacy format, unchanged):
```
Frame::Publish (0x07):
  [u16 topic_length]
  [u8* topic]
  [u32 payload_length]
  [u8* payload]
```

**Decoder disambiguation:** When `CAP_HEADERS` is negotiated, the decoder always reads the header section. When not negotiated, the decoder uses the existing format. This is a per-connection decision, not per-frame — no ambiguity.

#### Deliver Frame (updated)

```
Frame::Deliver (0x09) — with headers:
  [u64 message_id]
  [u16 topic_length]
  [u8* topic]
  [u64 offset]
  [u64 timestamp_millis]
  [u16 header_count]              ← NEW
  for each header:
    [u16 key_length]
    [u8* key]
    [u32 value_length]
    [u8* value]
  [u32 payload_length]
  [u8* payload]
```

#### FetchResponse Messages (updated)

```
FetchedMessage — with headers:
  [u64 offset]
  [u64 timestamp_millis]
  [u16 header_count]              ← NEW
  for each header:
    [u16 key_length]
    [u8* key]
    [u32 value_length]
    [u8* value]
  [u32 payload_length]
  [u8* payload]
```

### Storage Record Format

**Current format:** Records are stored as `[4B record_len (u32 BE)][8B offset (u64 BE)][bincode(MessageEnvelope)]`. The `MessageEnvelope` is serialized via **bincode** (serde-based), not manual binary encoding. The `RecordStream` parser in `src/storage/replay.rs` reads the 4-byte length prefix, then the 8-byte offset, then deserializes the rest via `bincode::deserialize::<MessageEnvelope>()`.

**Approach: Add `headers` field to `MessageEnvelope`:**

Since storage uses bincode serialization, the simplest approach is adding a `headers` field to `MessageEnvelope`:

```rust
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageEnvelope {
    pub id: MessageId,
    pub namespace: NamespaceId,
    pub topic: TopicName,
    pub payload: Vec<u8>,
    pub timestamp: u64,
    #[serde(default)]          // ← backward compat: missing field = empty vec
    pub headers: Vec<(Vec<u8>, Vec<u8>)>,  // key-value pairs
}
```

**Backward compatibility (reading old segments):** bincode does NOT support `#[serde(default)]` — it's a fixed-layout format and cannot skip missing fields. Two approaches:

**Option A — Segment format version in the record header (recommended):**

Extend the record layout to include a version byte between the offset and the envelope:

```
Old format (implicit v1):
  [4B record_len][8B offset][bincode(MessageEnvelope_v1)]

New format (v2):
  [4B record_len][8B offset][1B version=0x02][bincode(MessageEnvelope_v2)]
```

The `RecordStream` parser checks the total `record_len` minus 8 (offset bytes). Before reading the bincode payload, it peeks at `record_buf[8]`:
- If `record_len - 8 == bincode_size_of_old_envelope` → v1 format, deserialize without headers
- Better: use a sentinel version byte. The first byte of a bincode-serialized `MessageEnvelope` is the start of the `MessageId` (u64 LE). For version 0x02, this is unambiguous because valid `MessageId` values start at 1 (the LE encoding of 1 is `[0x01, 0x00, ...]`), and 0x02 would correspond to `MessageId(2)`. **This is NOT unambiguous.** Instead:

**Option B — Separate envelope types with explicit version tag (recommended):**

Keep the old `MessageEnvelope` as-is for deserialization. Add a new `MessageEnvelopeV2` with headers. Use a version byte in the record to select which to deserialize:

```
Record layout:
  [4B record_len][8B offset][1B format_version][N bytes bincode envelope]
```

- `format_version = 0x01` → `bincode::deserialize::<MessageEnvelopeV1>()` (no headers)
- `format_version = 0x02` → `bincode::deserialize::<MessageEnvelopeV2>()` (with headers)

**Old segments (no version byte):** Detected by checking `record_buf[8]`. Old segments have the first byte of bincode data here, which is the low byte of `MessageId` (u64 LE). Since `MessageId` starts from 1 and increments, values 0x01+ are expected. We reserve `0x00` as the format version tag indicator — no valid `MessageId` has LE byte 0x00 (that would be `MessageId(0)`, which is never assigned). **However**, this assumption is fragile. Better approach:

**Option C — Write a migration marker at segment open (cleanest):**

When opening a segment for writing in the new version, write a 4-byte magic header at the start of new segments only. Old segments are detected by the absence of the magic. `RecordStream` checks for magic on `open()` and sets a per-stream version flag.

**Recommendation: Option B with a simpler detection heuristic.** Write all new records with a version prefix byte `0x00` followed by the actual version `0x02`:

```
New record: [4B record_len][8B offset][0x00][0x02][bincode(MessageEnvelopeV2)]
Old record: [4B record_len][8B offset][bincode(MessageEnvelopeV1)]
```

Detection: after reading `[8B offset]`, the next byte is either `0x00` (new format — read version byte, then bincode v2) or non-zero (old format — bincode v1 starts here, first byte is LE `MessageId`). This works because `MessageId` is always ≥ 1, so its first LE byte is always ≥ 0x01.

> **Implementation note:** `encode_record()` in `src/storage/fs.rs` must be updated to write the version prefix for new records. `RecordStream::next()` in `src/storage/replay.rs` must detect old vs. new format. Old segments are read-only — they are never rewritten. Compaction of old segments produces new-format output.

> **Index files (.idx):** No changes needed. Index entries store `[offset(8B)][timestamp(8B)][byte_position(8B)]` — they don't contain message data.

### Data Types

```rust
/// A single message header — key-value pair of byte strings.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Header {
    pub key: Vec<u8>,    // max 256 bytes
    pub value: Vec<u8>,  // max 8 KB
}

/// Ordered list of headers on a message.
/// Preserves insertion order. Duplicate keys are allowed
/// (same semantics as HTTP headers).
pub type Headers = Vec<Header>;
```

**MessageEnvelope changes:**

The existing `MessageEnvelope` gains a `headers` field:

```rust
/// Current (src/types.rs):
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageEnvelope {
    pub id: MessageId,
    pub namespace: NamespaceId,
    pub topic: TopicName,
    pub payload: Vec<u8>,
    pub timestamp: u64,
}

/// After this plan:
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageEnvelope {
    pub id: MessageId,
    pub namespace: NamespaceId,
    pub topic: TopicName,
    pub payload: Vec<u8>,
    pub timestamp: u64,
    pub headers: Headers,  // NEW: empty vec = no headers
}
```

> **Note:** `MessageEnvelope` uses `Vec<u8>` for payload (not `Bytes`), and the `Header` type should match. `Bytes` is used in the codec/frame layer; `Vec<u8>` is used in the storage/engine layer. The conversion happens at the TCP session boundary.

### Limits

| Limit | Default | Config Key |
|---|---|---|
| Max headers per message | 64 | `queue.max_headers_per_message` |
| Max header key size | 256 bytes | `queue.max_header_key_bytes` |
| Max header value size | 8,192 bytes | `queue.max_header_value_bytes` |
| Max total header bytes per message | 32,768 bytes | `queue.max_header_bytes_total` |

> **Note:** Config keys use the `[queue]` section (matching the existing `QueueConfig` struct), not `[engine]`.

Exceeding any limit returns `ErrorCode::InvalidFrame` with a descriptive message. Headers are validated at publish time — malformed headers never reach storage.

**Header size accounting:** Header bytes count toward the `max_message_size` frame limit (currently `u32`, default 1MB). With default header limits (32KB max headers), the remaining payload budget is `max_message_size - header_bytes - frame_overhead`. This is checked in the codec decoder before allocating memory.

### HTTP Integration

#### Publish with Headers

Headers are passed via HTTP headers with a `X-Fafb-` prefix:

```
POST /v1/myapp/orders/publish
Authorization: Bearer sk_live_abc123
X-Fafb-Content-Type: application/json
X-Fafb-Correlation-Id: req-12345
X-Fafb-Trace-Parent: 00-abcdef...

{"order_id": 42}
```

The server strips the `X-Fafb-` prefix and stores them as message headers:
- `X-Fafb-Content-Type` → header key `Content-Type`
- `X-Fafb-Correlation-Id` → header key `Correlation-Id`

#### Fetch with Headers

HTTP fetch responses include headers in the JSON:

```json
{
  "messages": [
    {
      "offset": 42,
      "timestamp_millis": 1711234567890,
      "headers": {
        "Content-Type": "application/json",
        "Correlation-Id": "req-12345"
      },
      "payload_base64": "eyJvcmRlcl9pZCI6IDQyfQ=="
    }
  ],
  "next_offset": 43,
  "head_offset": 100
}
```

Note: HTTP headers are text-only. Binary header values are base64-encoded in the JSON response with a `_base64` suffix on the key.

### Memory Accounting

Headers count toward the message's memory footprint in the in-memory buffer. `MessageEnvelope::wire_size()` includes header bytes. This ensures memory budgets and backpressure remain accurate.

### Metrics

- `fafb_message_header_count` (histogram) — headers per published message
- `fafb_message_header_bytes` (histogram) — total header bytes per message
- `fafb_header_rejected_total{reason}` — header validation failures (too_many, key_too_large, value_too_large, total_too_large, reserved_prefix)

## Implementation Phases

### Phase 19.1 — Core Types, Config & Codec

**Goal:** Define header types, add capability negotiation, and update the wire protocol codec to encode/decode headers on Publish, Deliver, and FetchResponse frames.

- [x] Add `Header` struct and `Headers` type alias to `src/types.rs`
  - [x] `Header { key: Vec<u8>, value: Vec<u8> }` with `Serialize`/`Deserialize`
  - [x] `pub type Headers = Vec<Header>`
- [x] Add header limit fields to `QueueConfig` in `src/config.rs`
  - [x] `max_headers_per_message: usize` (default 64)
  - [x] `max_header_key_bytes: usize` (default 256)
  - [x] `max_header_value_bytes: usize` (default 8192)
  - [x] `max_header_bytes_total: usize` (default 32768)
- [x] Add `CAP_HEADERS` capability constant (`1 << 3`, bit 3) to `src/codec/frame.rs`
  - [x] Add to `SERVER_CAPABILITIES` bitmask
- [x] Implement header validation function
  - [x] Check header count ≤ `max_headers_per_message`
  - [x] Check each key size ≤ `max_header_key_bytes`
  - [x] Check each value size ≤ `max_header_value_bytes`
  - [x] Check total header bytes ≤ `max_header_bytes_total`
  - [x] Reject keys starting with `_fafb.` (reserved prefix) — return `ErrorCode::InvalidFrame`
- [x] Update `Frame::Publish` codec (encode/decode)
  - [x] When `CAP_HEADERS` negotiated: encode `[u16 header_count]` + header entries before payload
  - [x] When `CAP_HEADERS` not negotiated: existing format unchanged
  - [x] Validate headers during decode, reject on limit violations
- [x] Update `Frame::Deliver` codec to include headers section
- [x] Update `FetchedMessage` codec to include headers section
- [x] Add `fafb_header_rejected_total{reason}` metric (too_many, key_too_large, value_too_large, total_too_large, reserved_prefix)
- [x] Unit tests for codec
  - [x] Publish encode/decode round-trip with 0, 1, many headers
  - [x] Deliver encode/decode round-trip with headers
  - [x] FetchedMessage encode/decode round-trip with headers
  - [x] Legacy format (no `CAP_HEADERS`) unchanged
  - [x] Validation rejects: too many headers, oversized key, oversized value, total exceeded, reserved prefix
  - [x] Header bytes count toward `max_message_size` frame limit

**Additional work completed beyond original plan:**
- [x] `HeaderLimits` stored on `MessageCodec` — decode-time validation of Publish headers (defense in depth)
- [x] Header limit fields wired into `RuntimeConfig`, `RedactedRuntimeConfig`, `from_server_config()`, and `diff()` for hot-reload support
- [x] Connection handler (`connection.rs`) configures codec with `set_headers_enabled(true)` and `set_header_limits()` from `SharedConfig` after successful Connect/Reconnect when `CAP_HEADERS` is negotiated
- [x] Decode-time validation tests: reject too-many headers, reject reserved prefix, accept valid headers

**Verification:** Codec round-trips work with and without headers. Legacy clients (no `CAP_HEADERS`) see no format change. `cargo test` passes (751 tests, 0 failures).

---

### Phase 19.2 — Storage Format Migration

**Goal:** Update the segment storage format to persist headers. Old segments (v1) remain readable; new writes use v2 format with a version marker byte.

> **Design decision:** Modify `MessageEnvelope` in place (add `headers: Vec<Header>`) rather than creating a separate type. The compiler flags every construction site that needs updating. The field defaults to `Vec::new()` (zero overhead when empty).

- [x] Add `headers: Headers` field to `MessageEnvelope` in `src/types.rs`
  - [x] Default to `Vec::new()` (empty — no headers)
  - [x] Update every `MessageEnvelope` construction site to include `headers: vec![]` or propagate headers (compiler exhaustiveness catches all sites)
- [x] Update `encode_record()` in `src/storage/fs.rs` to write version prefix
  - [x] New record format: `[4B record_len][8B offset][0x00][0x02][bincode(MessageEnvelope)]`
  - [x] `0x00` sentinel byte distinguishes new format from old (old format's first bincode byte is always ≥ 0x01 since `MessageId` ≥ 1)
  - [x] Add assertion in `encode_record()` verifying the invariant that `MessageId` is never 0
- [x] Update `RecordStream::next()` in `src/storage/replay.rs` for v1/v2 detection
  - [x] After reading `[8B offset]`, check next byte: `0x00` = new format, else = old format
  - [x] Old format: `bincode::deserialize::<MessageEnvelopeV1>()` (without headers field), convert to `MessageEnvelope` with empty headers
  - [x] New format: skip `[0x00][0x02]` version bytes, `bincode::deserialize::<MessageEnvelope>()` with headers
  - [x] Keep old `MessageEnvelopeV1` type (without headers) as a deserialization target for backward compat
- [x] Verify `SegmentIndex` needs no changes (indexes by offset/timestamp, not payload)
- [x] Verify compaction handles mixed v1/v2 records — compacted output always uses v2 format
  - [x] `rebuild_index()` handles new byte positions from compacted segments
- [x] Unit tests for storage format
  - [x] Write v2 record, read back — headers intact
  - [x] Read v1 record (old format, no version prefix) — headers default to empty
  - [x] Mixed v1/v2 segment — both records readable
- [x] Integration test: write records with headers → restart server → read back with headers intact
- [x] Integration test: simulate upgrade — write v1 records (construct manually), verify old segments still readable with empty headers

**Additional work completed beyond original plan:**
- [x] `MessageEnvelopeV1` type in `src/storage/replay.rs` — deserialization target for v1 records, with `into_current()` conversion
- [x] `encode_record_v1()` test helper in `src/storage/replay.rs` — writes legacy v1 format for backward-compat testing
- [x] `RECORD_FORMAT_SENTINEL` (`0x00`) and `RECORD_FORMAT_VERSION_2` (`0x02`) constants in `src/storage/fs.rs`
- [x] All test `MessageId` values updated to be ≥ 1 (enforces the sentinel byte invariant)
- [x] `v2_record_empty_headers_roundtrip` test — v2 format with zero headers
- [x] `MessageId::new(id)` constructor in `src/types.rs` — panics on 0, enforcing the sentinel byte invariant at the type level. Production publish paths (TCP session, HTTP routes) use `MessageId::new()`. Wire-protocol decoding and tests use the raw tuple constructor `MessageId(val)`. `debug_assert` in `encode_record()` kept as defense-in-depth.
- [x] `compact_segments_converts_v1_to_v2` test in `src/storage/fs.rs` — writes v1-format segments manually, runs compaction, verifies merged output uses v2 format (0x00 sentinel + 0x02 version byte) with empty headers and all payloads preserved.

**Verification:** Old segments remain readable after upgrade. New records include headers. Compaction converts v1→v2. `cargo test` passes (758 tests, 0 failures).

---

### Phase 19.3 — End-to-End Pipeline

**Goal:** Thread headers through the entire message pipeline: publish → engine → storage → delivery/fetch. Headers survive the full lifecycle.

- [x] Thread `Headers` through `QueueEngine::publish()` path
  - [x] `MessageEnvelope` already carries headers (from Phase 19.2)
  - [x] Verify headers pass through engine to storage unchanged
- [x] Thread `Headers` through delivery path
  - [x] `PublishedMessage` (delivery channel) wraps `MessageEnvelope` — headers flow automatically
  - [x] Session converts `MessageEnvelope.headers` to `Frame::Deliver` headers when `CAP_HEADERS` negotiated
  - [x] Session omits headers for legacy clients (no `CAP_HEADERS`) — encoder gates on `headers_enabled`
- [x] Thread `Headers` through `NamespaceManager::fetch()` → `FetchedMessage`
  - [x] `FetchResult.messages` includes headers from stored messages
  - [x] Session converts to `FetchedMessage` wire format with headers when `CAP_HEADERS` negotiated
- [x] Update memory accounting to include header bytes
  - [x] `MessageEnvelope::mem_size()` returns `payload.len() + header_bytes()` — used in engine and topic
  - [x] `MessageEnvelope::header_bytes()` returns sum of all key+value lengths
  - [x] Engine global memory budget (`total_bytes`) uses `msg.mem_size()` instead of `payload.len()`
  - [x] Topic per-topic `bytes_used` and eviction tracking uses `mem_size()`
  - [x] Backpressure calculations (depth eviction, byte-limit eviction) account for header overhead
  - [x] Fetch response budget in `NamespaceManager::fetch()` includes header wire overhead (`FETCH_MESSAGE_HEADER_OVERHEAD` + per-header `PER_HEADER_OVERHEAD` + key/value bytes)
  - [x] `NamespaceManager::fetch()` takes `headers_enabled: bool` — header wire overhead only counted when client supports `CAP_HEADERS`, so legacy clients get full payload budget
- [x] Add `fafb_message_header_count` histogram metric (headers per published message)
- [x] Add `fafb_message_header_bytes` histogram metric (total header bytes per message)
  - [x] Metrics recorded unconditionally (including 0 for zero-header messages) for accurate distribution data
- [x] Integration test: TCP publish with headers → subscribe → receive `Deliver` frame → verify all headers present
- [x] Integration test: TCP publish with headers → fetch → verify headers in `FetchResponse`
- [x] Integration test: legacy client (no `CAP_HEADERS`) publishes (no headers) → capable client fetches → message has empty headers
- [x] Integration test: TCP publish with headers → consumer group subscribe → verify `Deliver` frame includes all headers (covers group delivery code path)
- [ ] *(Deferred)* Benchmark: publish throughput with 0, 1, 5, 10 headers vs. baseline — not blocking; expected overhead is ~8 bytes for 0 headers (bincode Vec length prefix), ~750 bytes for 5 typical headers

**Additional work completed beyond original plan:**
- [x] `TestClient::enable_headers()` method and automatic headers enablement in `connect_and_auth_with_caps()` when `CAP_HEADERS` is negotiated
- [x] `MessageEnvelope::mem_size()` and `header_bytes()` methods for consistent memory accounting across engine and topic
- [x] `NamespaceManager::fetch()` signature extended with `headers_enabled: bool` for capability-aware fetch budget — TCP session passes `self.capabilities & CAP_HEADERS != 0`, HTTP passes `false`
- [x] Header metrics recorded unconditionally (0 values included) for accurate histogram distributions

**Verification:** Headers survive publish → store → deliver and publish → store → fetch. Memory accounting is accurate. Legacy clients work unchanged. Consumer group delivery path verified. `cargo test` passes (762 tests, 0 failures).

---

### Phase 19.3.5 — DLQ Refactor: System Headers Replace Binary Envelope

**Goal:** Replace Plan 18's `DlqEnvelope` binary wire format with `_fafb.*` system headers. DLQ messages carry the original payload directly, with provenance metadata in headers. This is a clean replacement — no migration from the old format is needed.

**What gets removed:**
- `DlqEnvelope` struct and its `encode()`/`decode()` methods in `src/dlq.rs`
- Binary wire format constants (`DLQ_MAGIC`, `DLQ_VERSION`, reason code bytes)
- All `DlqEnvelope` unit tests (replaced by new header-based tests)

**What gets added/changed:**
- `route_to_dlq()` sets `_fafb.*` system headers on the `MessageEnvelope` instead of wrapping the payload
- DLQ consumers read headers instead of decoding a binary envelope

#### Tasks

- [x] Add DLQ header key constants to `src/dlq.rs`
  - [x] `DLQ_HEADER_SOURCE_TOPIC = b"_fafb.dlq.source_topic"`
  - [x] `DLQ_HEADER_ORIGINAL_OFFSET = b"_fafb.dlq.original_offset"`
  - [x] `DLQ_HEADER_DELIVERY_COUNT = b"_fafb.dlq.delivery_count"`
  - [x] `DLQ_HEADER_REASON = b"_fafb.dlq.reason"`
  - [x] `DLQ_HEADER_FIRST_DELIVERY_TS = b"_fafb.dlq.first_delivery_ts"`
  - [x] `DLQ_HEADER_TIMESTAMP = b"_fafb.dlq.timestamp"`
  - [x] `DLQ_HEADER_NACK_REASON = b"_fafb.dlq.nack_reason"` (only for `ClientNack`)
- [x] Add `fn build_dlq_headers(source_topic, original_offset, delivery_count, reason, first_delivered_at, dlq_timestamp, original_headers) -> Headers` helper in `src/dlq.rs`
  - [x] Creates `Header` entries from DLQ metadata (all values as UTF-8 byte strings)
  - [x] Includes `nack_reason` header only when reason is `ClientNack`
  - [x] `DlqReason::reason_label()` provides the `_fafb.dlq.reason` value (`"max_attempts"`, `"client_nack"`)
- [x] Update `route_to_dlq()` in `src/namespace/manager.rs`
  - [x] Remove `DlqEnvelope::new()` + `envelope.encode(&msg.payload)` call
  - [x] Instead: clone original `MessageEnvelope`, set `headers = build_dlq_headers(...)`, publish to DLQ topic with original payload and system headers
  - [x] The published message's `payload` is the **original payload** (not wrapped)
  - [x] The published message's `topic` is `{source_topic}{dlq_suffix}` (unchanged)
- [x] Remove `DlqEnvelope` struct, `encode()`, `decode()`, binary format constants from `src/dlq.rs`
  - [x] Keep `DlqReason` enum and `reason_label()` — still needed
  - [x] `DlqContext` stays private in `namespace/manager.rs` — `build_dlq_headers()` takes individual params, not the full context struct
  - [x] Remove `DLQ_MAGIC`, `DLQ_VERSION`, `REASON_MAX_ATTEMPTS`, `REASON_CLIENT_NACK` binary constants
- [x] Add `fn parse_dlq_headers(headers: &Headers) -> Option<DlqMetadata>` helper in `src/dlq.rs`
  - [x] `DlqMetadata { source_topic, original_offset, delivery_count, reason, first_delivery_ts, dlq_timestamp, nack_reason }` — parsed from `_fafb.dlq.*` headers
  - [x] Returns `None` if required DLQ headers are missing (message is not a DLQ message)
  - [x] Convenience for DLQ consumers — not used server-side (server just sets headers, doesn't read them back)
- [x] Update existing DLQ tests in `src/dlq.rs`
  - [x] Replace `encode_decode_round_trip_*` tests with `build_dlq_headers` + `parse_dlq_headers` round-trip tests
  - [x] Test `MaxAttemptsExceeded` reason — no `nack_reason` header present
  - [x] Test `ClientNack` reason — `nack_reason` header present with reason string
  - [x] Test `parse_dlq_headers` returns `None` for non-DLQ messages (no `_fafb.dlq.*` headers)
  - [x] Remove binary format tests (`decode_rejects_invalid_magic`, `decode_rejects_unknown_version`, etc.)
- [x] Update DLQ integration tests in `src/namespace/manager.rs`
  - [x] `dlq_routing_on_max_redelivery` — verify DLQ message has original payload (not wrapped) and `_fafb.dlq.*` headers
  - [x] `dlq_write_success_returns_offset` — verify headers on DLQ message
  - [x] `nack_routes_to_dlq_with_client_nack_reason` — verify `_fafb.dlq.nack_reason` header present
- [x] Integration test: publish → DLQ routing → fetch from DLQ topic → verify original payload + system headers
- [x] Verify reserved prefix validation does NOT reject server-set `_fafb.*` headers
  - [x] The reserved prefix check in header validation applies to **client-submitted** headers only (Publish frame decode)
  - [x] `route_to_dlq()` sets `_fafb.*` headers server-side — bypasses client validation (headers are set on `MessageEnvelope` directly, not through the codec)

**Additional work completed beyond original plan:**
- [x] `RESERVED_HEADER_PREFIX` constant (`b"_fafb."`) in `src/dlq.rs` — single source of truth shared by codec validation (`frame.rs` imports it) and DLQ header construction. Test verifies all DLQ header keys start with the prefix.
- [x] `build_dlq_headers()` takes individual params (not `&DlqContext`) — `DlqContext` stays private in `manager.rs` where `dlq_failures` is used. Clean module boundary: `dlq.rs` has no knowledge of manager bookkeeping.
- [x] `build_dlq_headers()` takes `original_headers: &Headers` parameter — merges application headers with system headers so DLQ messages preserve the original message's headers
- [x] `parse_dlq_headers()` returns `None` for empty headers (additional edge case test)
- [x] `original_headers_preserved` test in `dlq.rs` — verifies application headers appear before system headers in DLQ messages
- [x] `dlq_preserves_original_application_headers` test in `manager.rs` — end-to-end through `route_to_dlq()`: publishes message with 2 app headers, routes to DLQ, verifies both app headers and system headers survive
- [x] TCP integration test `nack_routes_message_to_dlq_topic` uses `CAP_NACK` only — verifies payload is original (not wrapped), keeps test focused on nack behavior
- [x] New TCP integration test `nack_dlq_message_has_system_headers_via_fetch` uses `CAP_NACK | CAP_HEADERS` — publishes with app header, nacks, fetches from DLQ, verifies both app header and `_fafb.dlq.*` system headers visible in FetchResponse

**Verification:** DLQ messages carry original payload with `_fafb.*` system headers. No binary envelope encoding/decoding remains. All DLQ tests pass with header-based implementation. `cargo test` passes (761 tests, 0 failures).

---

### Phase 19.4 — HTTP Support

**Goal:** HTTP publish accepts headers via `X-Fafb-*` HTTP headers. HTTP fetch returns headers in JSON responses.

- [x] Parse `X-Fafb-*` HTTP headers on publish endpoint
  - [x] Strip `X-Fafb-` prefix to get message header key
  - [x] Apply same validation limits as TCP path (count, key size, value size, total, reserved prefix)
  - [x] Reject `X-Fafb-_fafb.*` (reserved prefix after stripping)
- [x] Include headers in JSON fetch response
  - [x] Add `headers` array to each message in fetch JSON output — array of `{"key", "value"/"value_base64"}` objects preserving order and duplicates
  - [x] Text-safe header values: include as `"value"` string field
  - [x] Binary header values: base64-encode as `"value_base64"` field (no `"value"` field)
- [x] Integration test: HTTP publish with `X-Fafb-*` headers → engine verification → headers stored
- [x] Integration test: HTTP publish with `X-Fafb-*` headers → HTTP fetch → headers in JSON response
- [x] Integration test: HTTP publish without headers → HTTP fetch → no headers field in JSON
- [x] Integration test: validation — HTTP publish with reserved prefix `X-Fafb-_fafb.*` → 400 error
- [x] Integration test: validation — HTTP publish with too many `X-Fafb-*` headers → 400 error
- [x] Integration test: binary header values → base64-encoded with `value_base64` field in fetch JSON
- [x] Integration test: duplicate header keys preserved in fetch JSON array

**Additional work completed beyond original plan:**
- [x] `extract_message_headers()` helper in `routes.rs` — iterates HTTP `HeaderMap`, strips `x-fafb-` prefix (case-insensitive via hyper's lowercasing), collects into `Vec<Header>`
- [x] `headers_to_json()` helper in `routes.rs` — converts `&[Header]` to JSON array of `{"key", "value"/"value_base64"}` objects. Array format chosen over flat JSON object to preserve insertion order and allow duplicate keys (matching `Vec<Header>` semantics)
- [x] Header metrics (`fafb_message_header_count`, `fafb_message_header_bytes`) recorded on HTTP publish path (matching TCP behavior)
- [x] Header limits read from `SharedConfig` at request time (hot-reloadable, consistent with TCP path)
- [x] Empty headers omitted from fetch JSON response (no `"headers": []` for messages without headers)

**Known limitation:** HTTP header names are lowercased by hyper, so `X-Fafb-Content-Type` becomes key `content-type`. TCP clients preserve original key casing. This is inherent to HTTP transport.

**Verification:** HTTP publish and fetch support headers. Binary values handled via base64. Duplicate keys preserved. Validation consistent with TCP path. `cargo test` passes (768 tests, 0 failures).

---

### Phase 19.5 — Documentation & CLAUDE.md Updates

**Goal:** Update all protocol and API documentation to reflect header support.

- [x] Update protocol documentation (`docs/protocol.md` or equivalent) with header wire format for Publish, Deliver, FetchResponse frames
  - [x] Added `CAP_NACK` (bit 2) and `CAP_HEADERS` (bit 3) to capability bits table
  - [x] Updated `SERVER_CAPABILITIES` constant
  - [x] Added Nack (0x16) and NackAck (0x17) frame definitions
  - [x] Updated Publish frame with `CAP_HEADERS`-conditional header section
  - [x] Updated Deliver frame with `CAP_HEADERS`-conditional header section
  - [x] Updated FetchResponse message entry with `CAP_HEADERS`-conditional header section
  - [x] Added Header Encoding section with wire format, validation limits, and reserved prefix
  - [x] Added Nack to frame dispatch table and drain lifecycle
- [x] Update HTTP API documentation (`docs/api.md` or equivalent) with `X-Fafb-*` header conventions
  - [x] Documented `X-Fafb-*` headers on publish endpoint with validation limits and reserved prefix
  - [x] Updated fetch response JSON to include `headers` array with `value`/`value_base64` fields
- [x] Update client SDK spec (`docs/client_sdk_spec.md` or equivalent) with header types and publish/consume API
  - [x] Added `CAP_NACK`, `CAP_HEADERS` to Capabilities scalar type
  - [x] Added Nack/NackAck to MessageType enum and Frame composite type
  - [x] Added `Header` type, `NackAckStatus` enum
  - [x] Updated Publish, Deliver, FetchResponse wire formats with header sections
  - [x] Added Nack/NackAck wire format sections
  - [x] Updated `DeliveredMessage` with `headers` and `nack()` method
  - [x] Updated `FetchedMessage` with `headers` field
  - [x] Updated `publish()` API with `headers` parameter
  - [x] Added header and nack compat test scenarios (TCP #32-39, HTTP #16-20)
  - [x] Added Message Headers and Nack sections to implementation checklist
  - [x] Added version 1.3 to version history
- [x] Update `CLAUDE.md` with header-related constants, types, and storage format version
  - [x] `CAP_HEADERS` capability bit
  - [x] `Header` type, `Headers` alias
  - [x] Storage record format v2 detection logic
  - [x] Header limits config keys
  - [x] `_fafb.*` reserved prefix
  - [x] DLQ system headers (`_fafb.dlq.*`) replacing `DlqEnvelope` binary format
  - [x] `build_dlq_headers()` and `parse_dlq_headers()` helpers
  - [x] Update DLQ subsystem description to reflect header-based approach (remove `DlqEnvelope` references)
  - [x] Added Message Headers subsystem row to Key Subsystems table
  - [x] Added header metrics to Observability Metrics section
  - [x] Updated `MessageEnvelope`, `FetchedMessage`, `NamespaceManager::fetch()` constructor patterns

**Additional work completed beyond original plan:**
- [x] Nack (0x16) and NackAck (0x17) frames documented in protocol.md (from Plan 18, not previously documented)
- [x] `CAP_NACK` (bit 2) added to capability bits table across all docs (from Plan 18, not previously documented)
- [x] Nack added to connection state machine valid sends (Subscribed, Draining states) in client_sdk_spec.md
- [x] `NackAckStatus` enum added to client_sdk_spec.md

- [x] Cross-verified all documentation claims against source code: `SERVER_CAPABILITIES` (`0x0F`), header limits (64/256/8192/32768), frame structures, metric names, storage format constants, `headers_to_json()` output format, `MessageId::new()` invariant — all confirmed matching

**Verification:** Documentation accurately reflects the implementation. All doc claims cross-verified against source code — no discrepancies. New developers can understand the header system from docs alone. DLQ documentation reflects the header-based approach with no references to the removed binary envelope format. `cargo build` passes.

### DLQ Refactor: Replace `DlqEnvelope` with System Headers

Plan 18 implemented DLQ routing using a custom binary `DlqEnvelope` format (magic bytes `0x44 0x4C`, version `0x01`) that wraps the original payload with provenance metadata. Now that Plan 19 provides first-class message headers, the `DlqEnvelope` binary format is replaced entirely with `_fafb.*` system headers. This is a clean replacement (not a migration) — the project is not yet published, so no backward compatibility with the old DLQ wire format is needed.

**DLQ messages become regular messages with system headers:**

```
_fafb.dlq.source_topic = "orders"
_fafb.dlq.original_offset = "42"
_fafb.dlq.delivery_count = "7"
_fafb.dlq.reason = "max_attempts"
_fafb.dlq.first_delivery_ts = "1711234567890"
_fafb.dlq.timestamp = "1711234568000"
_fafb.dlq.nack_reason = "invalid schema"     ← only present for ClientNack
```

**Benefits over the binary envelope:**
- DLQ consumers receive the **original payload directly** — no need to decode a binary envelope wrapper
- DLQ metadata is inspectable via standard header APIs (including HTTP fetch JSON responses)
- Eliminates `DlqEnvelope::encode()`/`decode()` and the custom binary wire format
- DLQ messages are first-class messages that work with all header-aware tooling

**Implementation:** See Phase 19.3.5 below for detailed tasks.

### Codec Impact on `max_message_size`

Headers are part of the frame body, so they consume `max_message_size` budget. The codec decoder must account for this:

1. Read topic
2. Read header_count → validate ≤ 64
3. Read each header key+value → accumulate total header bytes, validate per-header limits
4. Check `accumulated_bytes + remaining` ≤ `max_frame_size` before reading payload
5. Read payload

If header validation fails mid-parse, the frame is rejected with `ErrorCode::InvalidFrame` and a descriptive message. The remaining bytes in the frame are skipped (the outer 4-byte length prefix tells the decoder how much to skip).

### `PublishedMessage` and Delivery Channel

`PublishedMessage` (used in the delivery channel `mpsc::channel<PublishedMessage>`) wraps `MessageEnvelope`. Adding `headers` to `MessageEnvelope` means headers flow through the delivery channel automatically — no separate change needed for push-based delivery.

### Benchmarks

Adding headers increases the per-message overhead (even when empty — bincode still serializes the `Vec` length). Benchmark:
- Publish throughput with 0, 1, 5, 10 headers
- Storage write throughput with headers
- Memory accounting accuracy with header bytes

Expected impact with 0 headers: ~8 bytes overhead (bincode `Vec<T>` length prefix). With 5 typical headers (50B keys, 100B values): ~750B per message.

## Risks and Mitigations

| Risk | Mitigation |
|---|---|
| Storage format version detection is fragile | Use `0x00` sentinel byte which is guaranteed not to appear as the first byte of a bincode `MessageEnvelope` (because `MessageId` ≥ 1 and bincode uses LE encoding). Add an assertion in `encode_record()` to verify this invariant. |
| Adding `headers` to `MessageEnvelope` breaks all struct literal construction sites | Intentional — compiler exhaustiveness checks catch every site. Use `..Default::default()` or named field `headers: vec![]` at each site. This is a mechanical change. |
| Bincode `Vec<Header>` adds overhead even with 0 headers | ~8 bytes (length prefix). Negligible compared to existing envelope overhead (id, namespace, topic strings). |
| `CAP_HEADERS` negotiation adds complexity to the codec | Per-connection flag, not per-frame. The codec checks once on connection setup, then uses the fixed format for all frames on that connection. |
| Old segments with v1 records after upgrade | v1 records are read-only. Compaction converts them to v2. Retention eventually deletes them. No manual migration needed. |
| HTTP headers are text-only but message headers support binary values | Binary header values are base64-encoded in JSON responses. The `_base64` suffix convention (or a `Content-Transfer-Encoding` meta-header) tells clients to decode. |

## Testing Strategy

- **Unit tests:** Codec round-trip (0/1/N headers, edge cases), validation (limits, reserved prefix), v1/v2 record detection
- **Integration tests:** TCP publish+headers → deliver, TCP publish+headers → fetch, HTTP publish+headers → TCP fetch, restart persistence, compaction of mixed records
- **DLQ header tests:** `build_dlq_headers`/`parse_dlq_headers` round-trip, DLQ routing produces original payload + `_fafb.dlq.*` system headers, DLQ fetch returns headers in response
- **Compatibility tests:** Legacy client (no `CAP_HEADERS`) connects — no headers in wire format, messages delivered without headers
- **Fuzz testing:** Random header counts/sizes against validation limits
- **Benchmark:** Publish throughput with 0/1/5/10 headers vs. baseline (no headers)
- **Storage migration:** Write v1 records (old binary), upgrade, read back — headers default to empty

## Dependencies

No new crate dependencies. `Header` uses `Serialize`/`Deserialize` from serde (already in tree). Bincode (already in tree) handles the storage serialization.

## Out of Scope

- **Server-side header filtering/routing** — headers are opaque to the server. Content-based routing using headers is a separate feature.
- **Header-based subscription filtering** — "only deliver messages where header X = Y". Powerful but adds query complexity. Separate plan.
- **Header compression** — repeated header keys across messages could be dictionary-compressed. Not needed at current scale.
- **Header schema validation** — server does not enforce that specific headers exist or have specific types. Application-level concern.
