# Phase 22: Message TTL (Per-Message Expiry)

## Goal

Allow publishers to set a time-to-live (TTL) on individual messages so that expired messages are never delivered to subscribers or returned by fetch. This is essential for time-sensitive events (price quotes, session tokens, request-reply patterns, rate-limited notifications) where delivering stale data is worse than delivering nothing. TTL complements the existing segment-level retention (which cleans up old segments in bulk) by adding per-message expiry semantics at the delivery and fetch layers.

## Guiding Principles

- **TTL is a delivery filter, not a storage operation.** Expired messages are skipped during delivery and fetch but are NOT immediately deleted from storage. They are cleaned up naturally by segment retention. This keeps the storage layer simple — no per-message deletion, no tombstones, no fragmentation.
- **TTL is set by the publisher.** The publisher knows the business relevance window. The server does not guess. TTL is optional — messages without TTL never expire (infinite TTL).
- **Expiry is checked at delivery time.** When the server is about to deliver or return a message, it checks `timestamp + ttl_ms > now`. If expired, the message is skipped and the next message is tried. This means expired messages consume storage until retention cleans them but do NOT consume delivery bandwidth.
- **TTL is stored with the message.** The TTL value is part of the segment record so it survives restarts. It is included in DLQ envelopes if a TTL'd message is dead-lettered before expiry.
- **Namespace-level default TTL.** If a namespace has a `default_ttl_secs` set, all messages without an explicit TTL inherit it. Explicit TTL on a message overrides the namespace default. Set to 0 for "no default TTL" (current behavior).

## Important: No Storage Version Migration Needed

This project has **not been deployed yet** — there are no existing segment files on disk. Therefore, adding `ttl_ms` to `MessageEnvelope` does not require a storage format version bump (no v3, no `MessageEnvelopeV2` shim). The field is added directly to the existing struct and serialized as part of v2 records. All future phases should assume this — no backward-compat shims for TTL in the storage layer.

## Current State

| Capability | Implementation | Status |
|---|---|---|
| Message timestamp | `MessageEnvelope.timestamp: u64` millis since epoch, set at publish time | **Done** — TTL computed as `timestamp + ttl_ms` |
| Message envelope | `MessageEnvelope` with `ttl_ms: u32` field, `is_expired()`, `remaining_ttl()` methods, `::new()` constructor | **Done (Phase 22.1)** |
| TTL config | `NamespaceConfig.default_ttl_secs/max_ttl_secs/min_ttl_secs`, wired into `NamespaceQuota` for hot-reload | **Done (Phase 22.1)** |
| DLQ expiry reason | `DlqReason::MessageExpired` variant in `src/dlq.rs` | **Done (Phase 22.1)** |
| Storage format | `ttl_ms` included in v2 bincode serialization (no version bump — project not deployed) | **Done (Phase 22.1)** |
| Publish wire format | `Frame::Publish { topic, headers, ttl_ms, payload }` with `CAP_MESSAGE_TTL` (bit 6). `MessageCodec.ttl_enabled` controls encode/decode. HTTP: `?ttl=<secs>` query param. Both paths resolve TTL (explicit > default > 0), clamp/reject per namespace limits. `fafb_message_ttl_seconds` histogram. | **Done (Phase 22.2)** |
| TTL hot-reload | `Session` holds `SharedConfig`, publish handler reads TTL limits per-publish via `ArcSwap::load()` | **Done (Phase 22.2)** |
| Push delivery | `spawn_individual_delivery_task` / `spawn_group_delivery_task` check `is_expired()` after interceptors, before DLQ routing. Cold helpers check before `try_send()`. Inflight sweep Phase 1 checks `InflightEntry::is_expired()` for direct DLQ routing. | **Done (Phase 22.3)** |
| Fetch path | `NamespaceManager::fetch()` skips expired messages in consume loop. Scan budget (`max_count * 10`) prevents unbounded scanning. | **Done (Phase 22.3)** |
| Deliver/Fetch wire format | `Frame::Deliver` and `FetchedMessage` include `ttl_ms` when `CAP_MESSAGE_TTL` negotiated. HTTP fetch JSON includes `ttl_ms`. All docs updated. | **Done (Phase 22.4)** |

## Design

### Wire Format

#### Publish Frame (updated)

TTL is gated on `CAP_HEADERS` (Plan 19) since it uses the same capability for extended message metadata. TTL is encoded as a dedicated field after headers (not as a header itself) because the server must interpret it — unlike headers which are opaque.

However, TTL can be implemented independently of headers. If Plan 19 is not yet implemented, TTL uses its own capability flag:

- `CAP_MESSAGE_TTL` = `1 << 5` (bit 5)

```
Frame::Publish (with CAP_MESSAGE_TTL):
  [u16 topic_length]
  [u8* topic]
  [u32 ttl_ms]                  ← NEW: 0 = no TTL (never expires)
  [u32 payload_length]
  [u8* payload]
```

Decoder behavior: When `CAP_MESSAGE_TTL` is negotiated, the decoder reads `ttl_ms` after topic and before payload. When not negotiated, the decoder uses the current format.

#### Deliver Frame (updated)

Include TTL in delivery so clients can see remaining lifetime:

```
Frame::Deliver (with CAP_MESSAGE_TTL):
  [u64 message_id]
  [u16 topic_length]
  [u8* topic]
  [u64 offset]
  [u64 timestamp_millis]
  [u32 ttl_ms]                  ← NEW: original TTL (0 = no expiry)
  [u32 payload_length]
  [u8* payload]
```

Client can compute remaining TTL: `remaining = (timestamp + ttl_ms) - now_ms`.

#### FetchResponse (updated)

```
FetchedMessage (with CAP_MESSAGE_TTL):
  [u64 offset]
  [u64 timestamp_millis]
  [u32 ttl_ms]                  ← NEW
  [u32 payload_length]
  [u8* payload]
```

### Storage Record Format

**No version bump needed.** This project has not been deployed yet — there are no existing segment files to migrate. Plan 19 (headers) already established v2 format with `[0x00][0x02]` sentinel/version. Adding `ttl_ms` to `MessageEnvelope` simply extends the bincode-serialized struct within v2. The `MessageEnvelopeV1` shim (for pre-headers records) sets `ttl_ms: 0` in `into_current()`.

If this project were deployed with existing data, a v3 format and `MessageEnvelopeV2` shim would be needed since bincode cannot handle missing trailing fields. But since there are no existing segments, this complexity is avoided.

### Message Envelope Changes

```rust
/// Current struct (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 — add ttl_ms:
#[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 ttl_ms: u32,          // NEW: 0 = no expiry
}

impl MessageEnvelope {
    /// Check if this message has expired relative to the given timestamp.
    pub fn is_expired(&self, now_ms: u64) -> bool {
        self.ttl_ms > 0 && self.timestamp + (self.ttl_ms as u64) <= now_ms
    }

    /// Remaining TTL in milliseconds. Returns None if no TTL or already expired.
    pub fn remaining_ttl(&self, now_ms: u64) -> Option<u64> {
        if self.ttl_ms == 0 { return None; }
        let expiry = self.timestamp + (self.ttl_ms as u64);
        if expiry <= now_ms { None } else { Some(expiry - now_ms) }
    }
}
```

> **Note:** `MessageEnvelope` uses `Vec<u8>` for payload, not `Bytes`. The plan's original struct was incorrect.

### Delivery Path Changes

#### Push Delivery (Subscribe/Deliver)

The delivery forwarders (`spawn_individual_delivery_task` and `spawn_group_delivery_task`) read messages from the engine buffer via `consume_from()` or cold storage batches. The TTL check is inserted after reading and before sending to the subscriber's `mpsc::Sender`:

```rust
// In delivery task (both individual and group), after reading msg from engine:
let msg = engine.consume_from(ns, topic, offset).await?;

// Skip expired messages
if msg.envelope.is_expired(now_millis()) {
    metrics::counter!("fafb_expired_skipped_total",
        "namespace" => ns.0.clone(), "topic" => topic.0.clone()
    ).increment(1);
    // Advance offset past expired message (individual: sub.next_offset, group: group.next_offset)
    // Do NOT add to inflight — just skip
    continue;
}

// Also applies to cold-path delivery (cold_deliver_individual / cold_deliver_group):
// Check is_expired() before try_send() to avoid delivering stale cold-path messages
```

**Important:** The offset must advance past expired messages even though they are not delivered. Otherwise, restarting the consumer would re-encounter the same expired messages and skip them again — wasting I/O and cold reads.

**Both delivery paths need the check:** The hot path (in-memory buffer via `consume_from()`) and the cold path (`cold_deliver_individual` / `cold_deliver_group` helper functions) both need TTL filtering. The cold path is especially important — cold-path messages are often older and more likely to have expired.

#### Pull Delivery (Fetch/FetchResponse)

In the fetch handler, after reading messages and before building the response:

```rust
// In fetch handler
let messages: Vec<FetchedMessage> = raw_messages
    .into_iter()
    .filter(|m| !m.is_expired(now_millis()))
    .collect();
```

The `next_offset` in `FetchResponse` is still based on the last message read (including expired ones), not the last non-expired message. This ensures the client's offset progresses past expired messages.

### Configuration

TTL config lives on `NamespaceConfig` (per-namespace, in the `[namespaces.NAME]` TOML section):

```toml
[namespaces.myapp]
api_key = "..."
permissions = ["Subscribe", "Publish", "Query", "Fetch"]
# TTL settings
default_ttl_secs = 0              # Default TTL for messages without explicit TTL. 0 = no default (never expires).
max_ttl_secs = 0                  # Maximum TTL a publisher can set. 0 = unlimited.
min_ttl_secs = 0                  # Minimum TTL a publisher can set. Prevents accidentally-tiny TTLs.
```

> **Note:** TTL fields are added directly to `NamespaceConfig` (same struct that has `api_key`, `permissions`, `max_total_messages`, etc.), not in a nested `[namespace.ttl]` sub-table. This matches the existing flat config pattern for namespace settings.

Validation (in `ServerConfig::validate()`):
- If `default_ttl_secs > max_ttl_secs` (when both are non-zero): startup error
- If a publisher sends `ttl_ms` exceeding `max_ttl_secs * 1000`: clamped to max, warning logged
- If a publisher sends `ttl_ms` below `min_ttl_secs * 1000` (when min is non-zero): rejected with `ErrorCode::InvalidFrame`
- `u32` TTL in milliseconds supports up to ~49.7 days. If `default_ttl_secs` or `max_ttl_secs` exceeds 49 days, startup warning.

### HTTP Integration

#### Publish with TTL

```
POST /v1/myapp/orders/publish?ttl=30
Authorization: Bearer sk_live_abc123

{"order_id": 42}
```

The `ttl` query parameter is in seconds (not milliseconds) for HTTP ergonomics. Converted to milliseconds internally.

#### Fetch — Expired Message Filtering

HTTP fetch automatically filters expired messages (same as TCP). The JSON response includes `ttl_ms` on each message:

```json
{
  "messages": [
    {
      "offset": 42,
      "timestamp_millis": 1711234567890,
      "ttl_ms": 30000,
      "payload_base64": "..."
    }
  ]
}
```

### Metrics

- `fafb_expired_skipped_total{namespace, topic}` (counter) — messages skipped due to expiry during delivery
- `fafb_expired_skipped_fetch_total{namespace, topic}` (counter) — messages skipped due to expiry during fetch
- `fafb_message_ttl_seconds` (histogram) — TTL values set by publishers (distribution of TTLs in use)
- `fafb_message_remaining_ttl_seconds` (histogram) — remaining TTL at delivery time (how close to expiry messages are when consumed — if median is near 0, consumers are too slow)

### Interaction with Other Features

| Feature | Interaction |
|---|---|
| **DLQ (Plan 18)** | If a message expires while inflight (timeout + redelivery cycles), it is dead-lettered with `DlqReason::MessageExpired` rather than redelivered. The DLQ envelope preserves the original TTL. |
| **Headers (Plan 19)** | TTL is NOT a header — it's a first-class field because the server must interpret it. But `_fafb.ttl_ms` could be added as a system header for clients that want to inspect TTL without protocol awareness. |
| **Retention** | Segment retention and message TTL are independent. Retention deletes whole segments by age/size. TTL skips individual messages at delivery time. A message can be "expired" (TTL) but still on disk (retention hasn't cleaned the segment yet). |
| **Compaction** | Future optimization: compaction could skip expired messages when rewriting segments, reducing storage. Out of scope for this plan. |

### Inflight Expiry Interaction

A message can expire while it is inflight (delivered to a consumer but not yet acked). Two approaches:

**Option A — Ignore TTL for inflight messages:** Once delivered, the consumer has the message. If the inflight timeout fires and the message has expired, redeliver normally (the consumer may have already processed it). This is simpler and avoids the edge case where a consumer receives a message, starts processing, and then the server decides it's expired.

**Option B — Check TTL on redelivery:** When the inflight timeout fires, check `is_expired()`. If expired, route to DLQ with `DlqReason::MessageExpired` (Plan 18) or drop. Don't redeliver stale messages. This saves network bandwidth for poison messages that are also stale.

**Recommendation: Option B.** The delivery task already checks `delivery_count > max_redelivery` before redelivery — adding an `is_expired()` check alongside is natural and consistent. The DLQ envelope preserves the TTL for debugging.

### Session-Level TTL Application

The TTL check in the session layer is needed for the TCP Publish handler — when a namespace has `default_ttl_secs` configured, the session must set `ttl_ms` on the envelope before passing to the manager:

```rust
// In session.rs, Publish handler:
let ttl_ms = if frame_ttl_ms > 0 {
    frame_ttl_ms  // explicit TTL from client
} else if namespace_default_ttl_ms > 0 {
    namespace_default_ttl_ms  // namespace default
} else {
    0  // no expiry
};
// Clamp to max, reject if below min
```

This requires the session to have access to namespace TTL config. Currently, `AuthenticatedState` holds the namespace ID — it would need to either query the manager for TTL config or hold a copy of the TTL settings.

### Memory Accounting

TTL'd messages that expire while in the in-memory buffer still consume memory until they are evicted by normal buffer management (DropOldest, capacity limits) or consumed past by all subscribers. TTL does NOT trigger early eviction from the in-memory buffer — that would add per-message cleanup overhead to the buffer data structure.

This is acceptable because:
1. The in-memory buffer has its own capacity limits (`max_depth`, `global_max_memory_bytes`)
2. Expired messages are a small fraction of buffered messages in typical usage
3. Segment retention handles bulk cleanup

## Implementation Phases

### Phase 22.1 — Core Types, Config & Storage

**Goal:** Add `ttl_ms` to `MessageEnvelope`, implement expiry checking methods, update the storage format, and add namespace-level TTL configuration. No delivery filtering yet — just data plumbing.

- [x] Add `ttl_ms: u32` field to `MessageEnvelope` in `src/types.rs`
  - [x] Default to `0` (no expiry) for backward compatibility
  - [x] Update every `MessageEnvelope` construction site to include `ttl_ms: 0` (compiler catches all sites)
- [x] Implement expiry methods on `MessageEnvelope`
  - [x] `is_expired(&self, now_ms: u64) -> bool` — returns `true` when `ttl_ms > 0 && timestamp + ttl_ms <= now_ms`
  - [x] `remaining_ttl(&self, now_ms: u64) -> Option<u64>` — returns `None` if no TTL or expired, otherwise remaining millis
- [x] Update storage format for TTL
  - [x] No version bump needed — project not yet deployed, `ttl_ms` added directly to v2 `MessageEnvelope`. `MessageEnvelopeV1::into_current()` sets `ttl_ms: 0`.
- [x] Add TTL configuration to `NamespaceConfig` in `src/config.rs`
  - [x] `default_ttl_secs: u64` (default 0 = no default TTL)
  - [x] `max_ttl_secs: u64` (default 0 = unlimited)
  - [x] `min_ttl_secs: u64` (default 0 = no minimum)
- [x] TTL fields (`default_ttl_secs`, `max_ttl_secs`, `min_ttl_secs`) added to `NamespaceQuota` (no separate struct — co-located with other per-namespace limits for simpler hot-reload)
- [x] Add config validation in `ServerConfig::validate()`
  - [x] If `default_ttl_secs > max_ttl_secs` (when both non-zero): startup error
  - [x] If `min_ttl_secs > max_ttl_secs` (when both non-zero): startup error
  - [x] If `default_ttl_secs` or `max_ttl_secs` exceeds 49 days (`u32::MAX` millis): startup warning
- [x] Unit tests for expiry logic
  - [x] `is_expired()`: exactly at expiry time → true, 1ms before → false, 1ms after → true, TTL=0 → false
  - [x] `remaining_ttl()`: returns correct remaining millis, `None` when expired or TTL=0
- [x] Storage format: no new tests needed — existing v2 roundtrip tests cover `ttl_ms` implicitly via bincode serialization. v1 compat tests verify `ttl_ms` defaults to 0.

**Additional improvements (post-Phase 22.1 cleanup):**
- [x] Added `MessageEnvelope::new(id, namespace, topic, payload, timestamp)` constructor defaulting `headers: vec![]` and `ttl_ms: 0`. Added `.with_headers()` and `.with_ttl()` builder methods. Converted all ~40 construction sites — future field additions only touch `new()`.
- [x] Consolidated TTL config into `NamespaceQuota` (removed standalone `NamespaceTtlConfig` struct and `namespace_ttl` HashMap). TTL limits co-located with other per-namespace quotas for simpler hot-reload and diff logic.
- [x] Added `DlqReason::MessageExpired` variant to `src/dlq.rs` with `reason_label() -> "message_expired"`. Prepares for Phase 22.3 inflight expiry → DLQ routing.

**Verification:** `MessageEnvelope` carries `ttl_ms`. Old segments remain readable with `ttl_ms = 0`. Config validates TTL constraints. All 908 tests pass. No delivery behavior changes yet.

---

### Phase 22.2 — Publish Path & Wire Protocol

**Goal:** Publishers can set TTL on messages via TCP and HTTP. The server applies namespace defaults, validates against min/max limits, and stores the TTL with the message.

- [x] Add `CAP_MESSAGE_TTL` capability constant (`1 << 6`, bit 6) to `src/codec/frame.rs`
  - [x] Add to `SERVER_CAPABILITIES` bitmask
  - [x] Note: bit 5 was already taken by `CAP_BATCH_ACK`, so used bit 6 instead of the plan's `1 << 5`
- [x] Add `ttl_enabled: bool` to `MessageCodec` with `set_ttl_enabled()` / `ttl_enabled()` methods
- [x] Update `Frame::Publish` codec for TTL
  - [x] Added `ttl_ms: u32` field to `Frame::Publish` enum variant
  - [x] When `CAP_MESSAGE_TTL` negotiated: encode/decode `[u32 ttl_ms]` after topic, before headers/payload
  - [x] When not negotiated: existing format unchanged (no TTL field), decoder produces `ttl_ms: 0`
- [x] Apply namespace default TTL in session publish handler
  - [x] If `frame_ttl_ms > 0`: use explicit TTL from client
  - [x] Else if `namespace.default_ttl_secs > 0`: use namespace default (converted to millis)
  - [x] Else: `ttl_ms = 0` (no expiry)
  - [x] `Session` holds `SharedConfig` — publish handler reads TTL config per-publish via `ArcSwap::load()` (~2ns), making TTL limits hot-reloadable for live connections
- [x] Validate TTL against namespace limits
  - [x] If `ttl_ms > max_ttl_secs * 1000` (when max is non-zero): clamp to max, log warning
  - [x] If `ttl_ms < min_ttl_secs * 1000` (when min is non-zero and ttl_ms > 0): reject with `ErrorCode::InvalidFrame`
- [x] Wire `ttl_ms` through `QueueEngine::publish()` to storage
  - [x] `MessageEnvelope` already carries `ttl_ms` (from Phase 22.1) — set via `.with_ttl()` before publish
- [x] HTTP publish: parse `ttl` query parameter (in seconds, converted to millis internally)
  - [x] Added `PublishParams` query struct with `ttl: Option<u64>` field
  - [x] Same default/min/max logic as TCP path, reads from `SharedConfig.namespace_quotas`
- [x] Add `fafb_message_ttl_seconds` histogram metric — TTL values set by publishers (both TCP and HTTP paths)
- [x] Unit tests for codec: Publish encode/decode round-trip with TTL, without TTL (legacy), TTL + headers combined, wire size verification
- [x] Integration test: TCP publish with TTL → verify stored `ttl_ms = 30000` via engine
- [x] Integration test: TCP publish without CAP_MESSAGE_TTL → verify stored `ttl_ms = 0` via engine
- [x] Integration test: TCP publish without TTL + namespace default → verify stored `ttl_ms = 60000` via engine
- [x] Integration test: TCP publish with TTL exceeding max → verify stored `ttl_ms` clamped to max via engine
- [x] Integration test: TCP publish with TTL below min → rejected with `ErrorCode::InvalidFrame`
- [x] Integration test: HTTP publish with `?ttl=30` → verify stored with `ttl_ms = 30000`
- [x] Integration test: HTTP publish without TTL → verify stored with `ttl_ms = 0`
- [x] Integration test: HTTP publish with namespace default TTL → verify stored with correct `ttl_ms`
- [x] Integration test: HTTP publish TTL overrides default → verify stored with explicit `ttl_ms`
- [x] Integration test: HTTP publish TTL clamped to max → verify stored with max `ttl_ms`
- [x] Integration test: HTTP publish TTL below min → 400 error

**Additional improvements (post-Phase 22.2 cleanup):**
- [x] **Hot-reloadable TTL config:** `Session` holds `SharedConfig` directly. Publish handler reads TTL limits per-publish via `ArcSwap::load()` (~2ns), so config hot-reload takes effect on live connections without reconnection. Removed stale `set_ttl_config()` / per-field caching approach.
- [x] **Fixed `start_test_tcp_server_multi_ns` shared_config bug:** Was passing `test_shared_config()` (empty defaults) to `TcpServer` instead of the `shared_cfg` built with namespace quotas. Connection handler now correctly sees namespace-specific quotas.
- [x] **Exposed engine on `TestServer`:** Added `engine: Option<Arc<InMemoryQueueEngine<MockStorage>>>` so TCP integration tests can verify stored `ttl_ms` directly (matching HTTP test coverage quality).

**Verification:** Publishers can set TTL via TCP and HTTP. Namespace defaults and limits applied correctly. TTL limits are hot-reloadable. Legacy clients (no `CAP_MESSAGE_TTL`) unaffected. All 924 tests pass.

---

### Phase 22.3 — Delivery & Fetch Filtering

**Goal:** Expired messages are skipped during push delivery and fetch. Offsets advance past expired messages. Inflight expiry check prevents redelivering stale messages.

- [x] Add expiry check to push delivery paths
  - [x] In `spawn_individual_delivery_task`: after interceptors/slow-consumer check, check `msg.envelope.is_expired(now_millis())` — if expired, advance offset, remove from inflight, and `continue` (skip delivery)
  - [x] In `spawn_group_delivery_task`: same check before adding to inflight and sending to consumer
  - [x] In `cold_deliver_individual` / `cold_deliver_group` helpers: check before `try_send()` — if expired, advance offset and return `Sent`
  - [x] Important: advance `next_offset` past expired messages even though they're not delivered
- [x] Add expiry check to fetch path
  - [x] In `NamespaceManager::fetch()`: skip expired messages in consume loop — advance `current_offset` past them without adding to response
  - [x] `next_offset` in `FetchResponse` still based on last message read (including expired), so client offset progresses past expired messages
- [x] Add inflight expiry check on redelivery (Option B from design)
  - [x] In inflight sweep Phase 1 (under lock): `InflightEntry::is_expired()` checks TTL using stored `msg_timestamp`/`msg_ttl_ms` — expired entries go directly to DLQ candidates without incrementing `delivery_count`
  - [x] In inflight sweep Phase 2 (DLQ routing): expired flag from Phase 1 determines `DlqReason::MessageExpired` vs `MaxAttemptsExceeded`
  - [x] Both individual and group sweep paths updated
  - [x] Do NOT redeliver stale messages
- [x] Add `fafb_expired_skipped_total{namespace, topic}` counter metric (push delivery — hot path + cold path, individual + group)
- [x] Add `fafb_expired_skipped_fetch_total{namespace, topic}` counter metric (fetch)
- [x] Add `fafb_message_remaining_ttl_seconds` histogram metric — remaining TTL at delivery time (all paths: hot, cold, fetch)
- [x] Integration test: publish with 100ms TTL → wait 200ms → subscribe → verify message NOT delivered, only permanent message received
- [x] Integration test: publish 5 messages (3 with short TTL, 2 without) → wait for short TTL to expire → subscribe → receive only 2 non-expired messages
- [x] Integration test: fetch with mix of expired and live messages → only live messages in response, `next_offset` advances past all (TCP + HTTP)
- [x] Integration test: consumer group delivery also skips expired messages
- [x] Integration test: messages without TTL (ttl_ms=0) always delivered regardless of age
- [x] Integration test: HTTP fetch filters expired messages, returns only live ones

**Additional notes (Phase 22.3):**
- Expiry check in hot-path delivery is placed after interceptors and slow-consumer policy but before DLQ/inflight routing. This ensures interceptors still run for expired messages (they may log or emit metrics), while preventing any delivery or inflight tracking.
- Cold-path helpers (`cold_deliver_individual`, `cold_deliver_group`) check expiry before `try_send()` and return `ColdDeliverResult::Sent` to signal the caller to advance — no new enum variant needed.
- Inflight entries for expired messages are removed (via `inflight.remove(&offset)`) to prevent stale entries from blocking offset advancement or DLQ routing.

**Additional improvements (post-Phase 22.3 hardening):**
- [x] **Cached `now_millis()` in hot paths:** Both individual and group hot-path delivery now call `now_millis()` once per message, reusing the result for both `is_expired()` and `remaining_ttl()` checks.
- [x] **Remaining TTL metric on all paths:** `fafb_message_remaining_ttl_seconds` now recorded on cold-path delivery (individual + group) and fetch, not just hot-path delivery.
- [x] **Fetch scan limit:** Added a scan budget (`max_count * 10`, minimum 100) to `NamespaceManager::fetch()` to prevent unbounded scanning when large ranges of messages have expired. Returns partial results with correct `next_offset`.
- [x] **TTL fields on InflightEntry:** Added `msg_timestamp: u64` and `msg_ttl_ms: u32` to `InflightEntry`. Phase 1 of the inflight sweep now checks `entry.is_expired()` directly — expired messages go to DLQ candidates immediately without wasting a redelivery cycle. `InflightEntry::is_expired()` method mirrors `MessageEnvelope::is_expired()`.
- [x] **DLQ messages never expire:** `route_to_dlq()` now sets `ttl_ms: 0` on DLQ envelopes. Previously, DLQ messages inherited the original TTL and could expire before consumers processed them — a critical bug caught by the inflight→DLQ integration test.
- [x] **Inflight→DLQ integration test:** Full end-to-end test: publish with 500ms TTL → subscribe → receive → don't ack → wait for TTL + inflight timeout → verify DLQ message with `message_expired` reason in system headers.

**Verification:** Expired messages never delivered to subscribers or returned by fetch. Offsets advance correctly. Inflight expiry uses `DlqReason::MessageExpired`. DLQ messages are TTL-free. All 940 tests pass.

---

### Phase 22.4 — Deliver/Fetch Wire Format & Documentation

**Goal:** Include `ttl_ms` in Deliver and FetchResponse frames so clients can see message TTL. Update all documentation.

- [x] Update `Frame::Deliver` codec to include `ttl_ms` when `CAP_MESSAGE_TTL` negotiated
  - [x] `[u32 ttl_ms]` after `timestamp_millis`, before topic
  - [x] Client can compute remaining: `(timestamp + ttl_ms) - now_ms`
  - [x] Legacy clients (no `CAP_MESSAGE_TTL`) see existing format unchanged
- [x] Update `FetchedMessage` codec to include `ttl_ms` when `CAP_MESSAGE_TTL` negotiated
- [x] Update HTTP fetch JSON response to include `ttl_ms` field on each message
- [x] Unit tests: Deliver and FetchedMessage codec round-trip with TTL (10 new codec tests)
  - [x] `deliver_roundtrip_with_ttl`, `deliver_roundtrip_zero_ttl`, `deliver_roundtrip_ttl_and_headers`
  - [x] `deliver_ttl_not_in_wire_when_disabled`, `deliver_ttl_in_wire_increases_size`
  - [x] `fetch_response_roundtrip_with_ttl`, `fetch_response_ttl_not_in_wire_when_disabled`, `fetch_response_roundtrip_ttl_and_headers`
- [x] Integration test: publish with TTL → subscribe → Deliver frame includes correct `ttl_ms` (`deliver_frame_includes_ttl_ms`)
- [x] Integration test: legacy client without `CAP_MESSAGE_TTL` → Deliver frame has `ttl_ms=0` (`deliver_frame_ttl_zero_without_cap`)
- [x] Integration test: fetch with `CAP_MESSAGE_TTL` → FetchResponse includes `ttl_ms` (`fetch_response_includes_ttl_ms`)
- [x] Integration test: HTTP fetch → JSON response includes `ttl_ms` field (`http_fetch_includes_ttl_ms_in_response`)
- [x] Update protocol documentation with TTL wire format for Deliver, FetchResponse (`docs/protocol.md`)
  - [x] Added `CAP_BATCH_ACK` (bit 5) and `CAP_MESSAGE_TTL` (bit 6) to capability bits table
  - [x] Updated Deliver (0x09) section with `CAP_MESSAGE_TTL` wire variants
  - [x] Updated FetchResponse (0x14) message entry with `CAP_MESSAGE_TTL` wire variants
- [x] Update HTTP API documentation with `ttl` query parameter on publish and `ttl_ms` in fetch response (`docs/api.md`)
- [x] Update client SDK spec with TTL publish/consume API (`docs/client_sdk_spec.md`)
  - [x] Updated Publish and Deliver frame field tables with `ttl_ms`
  - [x] Updated `FetchedMessage` type with `ttl_ms` field
  - [x] Updated Deliver wire format section with combined capability tables
- [x] Add TTL row to delivery guarantees summary table (`docs/delivery_guarantees.md`)
- [x] Update CLAUDE.md with TTL-related `Frame::Deliver` and `FetchedMessage` field descriptions

**Additional notes (Phase 22.4):**
- Wire field ordering for Deliver: `message_id | offset | timestamp_millis | [ttl_ms] | topic | [headers] | payload`. TTL is after the fixed-size fields and before the variable-length topic string, matching the Publish frame pattern.
- Wire field ordering for FetchedMessage: `offset | timestamp_millis | [ttl_ms] | [headers] | payload`.
- `build_deliver_frame()` in session.rs passes `msg.envelope.ttl_ms` through to the frame.
- Fetch response construction in session.rs passes `m.envelope.ttl_ms` through to `FetchedMessage`.
- HTTP fetch JSON includes `ttl_ms` on every message (always present, 0 for no-TTL messages).

**Additional improvements (post-Phase 22.4 hardening):**
- [x] **`FETCH_MESSAGE_OVERHEAD` bumped from 20 to 24 bytes:** Accounts for TTL's 4 bytes in fetch response budget calculations unconditionally. Non-TTL clients use 20 bytes on wire, but the 4-byte conservative over-estimate prevents edge-case frame oversize and avoids threading `ttl_enabled` through the fetch API.
- [x] **Publish (0x07) in `docs/protocol.md` updated:** Added `CAP_MESSAGE_TTL` wire variants (previously only Deliver and FetchResponse had been updated). Now all three TTL-carrying frames are fully documented with combined capability tables.
- [x] **CLAUDE.md `FETCH_MESSAGE_OVERHEAD` updated:** Reflects new 24-byte value and rationale.

**Verification:** Clients see `ttl_ms` in Deliver and Fetch responses. Documentation covers TTL semantics end-to-end for all three frame types (Publish, Deliver, FetchResponse). Legacy clients unaffected. All 944 tests pass.

## Testing Strategy

- **Unit tests:** `is_expired()` boundary conditions (exactly at expiry, 1ms before, 1ms after, TTL=0), record format round-trip, config validation
- **Integration tests:** Publish with TTL → subscribe immediately → receive (not yet expired), publish with short TTL → wait → subscribe → skip, fetch with mix of expired and live messages, restart with TTL'd messages on disk
- **Edge cases:** TTL=1ms (minimum), TTL=u32::MAX (max ~49 days), clock skew considerations (documented, not solved — use monotonic checks where possible)

## Dependencies

No new crate dependencies. TTL is a `u32` field on the envelope, checked via simple arithmetic.

**Plan dependency:** If Plan 19 (Headers) is implemented first, the storage format migration is already handled and TTL is simply another field. If TTL lands first, the storage migration approach from Plan 19 applies here (record version byte).

## Risks and Mitigations

| Risk | Mitigation |
|---|---|
| Storage format change (same as Plan 19) | Coordinate with Plan 19 — single version bump for both headers and TTL. If TTL lands alone, use the same `[0x00][version]` prefix scheme. |
| Clock skew between publisher and server | `timestamp` is set server-side at publish time (in `session.rs`), not by the client. TTL is relative to server time. No cross-machine clock dependency. |
| Expired messages waste cold-storage reads | Cold-path delivery checks `is_expired()` before sending. If a large range of messages has expired, the cold read still happens but delivery is skipped. Optimization: use `.idx` timestamp field to skip entire segments where `max_timestamp + max_ttl < now`. Out of scope for initial implementation. |
| `u32` TTL limits expiry to ~49 days | Sufficient for most use cases. Messages needing longer lifetimes should use TTL=0 (no expiry) and rely on segment retention. Document the limit. |
| TTL per message increases envelope size by 4 bytes | Negligible. At 1M messages, that's 4MB additional storage. |
| Delivery of nearly-expired messages | `fafb_message_remaining_ttl_seconds` histogram alerts operators when consumers are too slow (median remaining TTL near 0). |

## Out of Scope

- **Per-message deletion from storage** — Expired messages are not deleted from segment files. Segment retention handles bulk cleanup. Per-message deletion would require tombstones or rewriting segments.
- **TTL-aware compaction** — Compaction could skip expired messages when rewriting segments, reducing storage. Useful optimization for a follow-up.
- **Per-topic TTL** — This plan covers namespace-level default TTL and per-message TTL. Per-topic TTL config (all messages in a topic inherit the same TTL) could be added alongside Plan 20's per-topic config.
- **Scheduled delivery** — "Deliver this message at time T" (future delivery). Different from TTL — TTL is about expiry, not delayed delivery.
