# Phase 18: Dead Letter Queue Enhancements

## Goal

Enhance the existing dead letter queue mechanism with structured DLQ message envelopes (preserving provenance metadata), a Nack protocol frame for explicit client-initiated dead-lettering, DLQ-specific metrics, per-topic DLQ overrides, and proper DLQ write failure handling. The core DLQ routing infrastructure already exists — this plan fills the gaps.

## Already Implemented (NOT in scope)

| Capability | Implementation | Location |
|---|---|---|
| `delivery_count` on inflight entries | `InflightEntry.delivery_count: u32`, starts at 1, incremented on redelivery | `src/namespace/state.rs:13-19` |
| `max_redelivery_attempts` config | In both `ManagerConfig` (default 5) and `QueueConfig` (default 5) | `src/namespace/manager.rs:166`, `src/config.rs:347` |
| `dlq_topic_suffix` config | In both `ManagerConfig` (default `"__dlq"`) and `QueueConfig` (default `"__dlq"`) | `src/namespace/manager.rs:167`, `src/config.rs:349` |
| Automatic DLQ routing | `route_to_dlq()` publishes message to `{topic}{suffix}` when `delivery_count > max_redelivery` | `src/namespace/manager.rs:1004-1038` |
| DLQ check in group delivery | Checked before each group delivery attempt in `group_deliver` task | `src/namespace/manager.rs:888-912` |
| Empty suffix disables DLQ | When `dlq_topic_suffix = ""`, messages are dropped with a warning log | `src/namespace/manager.rs:1012-1018` |
| Publish interceptor bypass | DLQ topic publishes skip `publish_interceptors` (checked via `ends_with(dlq_suffix)`) | `src/namespace/manager.rs:1564-1570` |

## Guiding Principles

- **DLQ is a regular topic.** Dead-lettered messages land in a normal fafb-q topic (e.g., `orders.__dlq`) that supports all standard operations — subscribe, fetch, publish, retention. No special storage format or query API.
- **DLQ writes are durable.** A message is only removed from inflight tracking after it is successfully written to the DLQ topic. Crash between "decided to DLQ" and "written to DLQ" must not lose the message.
- **Backpressure applies.** DLQ topics respect the same quotas and backpressure as any other topic. If the DLQ write fails, the message stays inflight for retry.

## Current Gaps

| Gap | Impact | Priority |
|---|---|---|
| DLQ message has no metadata | Messages are republished as-is — no source topic, original offset, delivery count, or reason. Operators can't debug why a message was DLQ'd. | **High** |
| No Nack frame | Clients cannot explicitly reject a message. Only timeout-based redelivery triggers DLQ. | **High** |
| DLQ write failure silently continues | `route_to_dlq()` logs an error but still removes the message from inflight (`group.inflight.remove(&offset)`). Message is lost. | **Critical** |
| No `first_delivered_at` timestamp | Cannot measure how long a message cycled through redelivery before DLQ | **Medium** |
| No DLQ-specific metrics | No counters for DLQ writes, failures, or delivery attempt histograms | **Medium** |
| No per-topic DLQ overrides | All topics in a namespace share the same DLQ suffix and max attempts | **Low** |
| No individual subscriber DLQ routing | DLQ check only runs in group delivery path, not individual subscriber delivery | **Medium** |
| No capability negotiation for Nack | Would need `CAP_NACK` bit for backward compatibility | **Low** |

## Design

### DLQ Message Envelope

**Problem:** Currently, `route_to_dlq()` publishes the original `MessageEnvelope` with only the topic name changed. Consumers of the DLQ topic see a bare payload with no context about why it was dead-lettered.

**Solution:** Wrap the original payload in a structured DLQ envelope:

```rust
/// Metadata prepended to the original payload when routing to DLQ.
/// The DLQ message payload = DlqEnvelope (serialized) + original payload.
pub struct DlqEnvelope {
    pub source_topic: TopicName,
    pub original_offset: TopicOffset,
    pub first_delivery_timestamp: u64,    // millis since epoch
    pub dlq_timestamp: u64,               // millis since epoch (when DLQ'd)
    pub delivery_count: u32,
    pub reason: DlqReason,
    // original_payload follows immediately after the envelope
}

pub enum DlqReason {
    MaxAttemptsExceeded,
    ClientNack { reason: String },   // human-readable reason from Nack frame
    // Future: MessageExpired (TTL integration, plan 22)
}
```

**Wire format (binary, prepended to original payload):**

```
+------------------+-------------------+---------------------------+
| magic (2B: 0xDL) | version (1B: 0x01)| source_topic_len (2B)     |
+------------------+-------------------+---------------------------+
| source_topic (N B)                   | original_offset (8B)      |
+--------------------------------------+---------------------------+
| first_delivery_ts (8B)              | dlq_ts (8B)                |
+--------------------------------------+---------------------------+
| delivery_count (4B)                 | reason_code (1B)           |
+--------------------------------------+---------------------------+
| reason_len (2B) | reason (N B)      | original_payload (rest)    |
+------------------+-------------------+---------------------------+
```

Reason codes: `0x00` = MaxAttemptsExceeded, `0x01` = ClientNack.

**DLQ envelope is the payload.** The consumer of the DLQ topic reads the DLQ envelope as the message payload. The `DlqEnvelope::decode(payload)` function parses the envelope and extracts the original payload. This means DLQ consumers need to understand the envelope format — document this in the SDK spec.

**Module location:** `src/dlq.rs` — new module for `DlqEnvelope`, `DlqReason`, encode/decode functions.

### Fix: DLQ Write Failure Must Retain Message

**Critical bug in current code.** In `src/namespace/manager.rs:888-912`, after calling `route_to_dlq()`, the code unconditionally removes the message from inflight and advances `next_offset`:

```rust
// CURRENT (buggy): always removes, even if DLQ write failed
Self::route_to_dlq(&engine, &ns, &topic, &dlq_suffix, msg.envelope.clone()).await;
// ... inflight.remove(&offset) <-- happens regardless of DLQ result
```

**Fix:** `route_to_dlq()` must return `Result<TopicOffset, Error>`. On failure, the message stays inflight and will be re-attempted on the next delivery cycle:

```rust
async fn route_to_dlq(
    engine: &Q,
    ns: &NamespaceId,
    topic: &TopicName,
    dlq_suffix: &str,
    entry: &InflightEntry,
    payload: Bytes,
    reason: DlqReason,
) -> Result<TopicOffset, Error> {
    if dlq_suffix.is_empty() {
        // DLQ disabled — drop the message (existing behavior, but now explicit)
        tracing::warn!(...);
        return Ok(TopicOffset(0)); // sentinel — not a real offset
    }

    let envelope = DlqEnvelope::encode(topic, entry, reason, &payload);
    let dlq_topic = TopicName(format!("{}{}", topic.0, dlq_suffix));
    let dlq_msg = MessageEnvelope {
        topic: dlq_topic.clone(),
        payload: envelope,
        timestamp: now_millis(),
        ..
    };
    let result = engine.publish(ns, &dlq_topic, dlq_msg).await?;
    Ok(result.offset)
}
```

Then in the caller:

```rust
match Self::route_to_dlq(&engine, &ns, &topic, &dlq_suffix, &inflight_entry, payload, reason).await {
    Ok(_) => {
        // Only now safe to remove from inflight
        group.inflight.remove(&offset);
        group.next_offset = offset.next();
        metrics::counter!("fafb_dlq_messages_total", ...).increment(1);
    }
    Err(e) => {
        // Leave in inflight — will retry on next cycle
        tracing::warn!("DLQ write failed, message stays inflight for retry");
        metrics::counter!("fafb_dlq_write_failures_total", ...).increment(1);
        // Don't continue — let the message be re-attempted
    }
}
```

### `first_delivered_at` Timestamp

Add `first_delivered_at: u64` (millis since epoch) to `InflightEntry`:

```rust
pub struct InflightEntry {
    pub offset: TopicOffset,
    pub message_id: MessageId,
    pub delivered_to: ClientId,
    pub delivered_at: Instant,
    pub delivery_count: u32,
    pub first_delivered_at: u64,    // NEW: set on first delivery, preserved across redeliveries
}
```

Set `first_delivered_at = now_millis()` when `delivery_count == 1` (first delivery). On redelivery, increment `delivery_count` but keep `first_delivered_at` unchanged. This enables the DLQ envelope to report total time-in-flight.

### Nack Frame

New protocol frame for explicit dead-lettering by the client:

```
Frame::Nack (0x16) — client → server
  [u64 message_id]    — the delivered message being rejected
  [u16 reason_length]
  [u8* reason]         — human-readable reason (e.g., "invalid JSON", max 1024 bytes)
```

```
Frame::NackAck (0x17) — server → client
  [u64 message_id]
  [u8  status]         — 0x00 = dead-lettered, 0x01 = DLQ disabled (dropped), 0x02 = DLQ write failed
```

**Nack semantics:**
- If DLQ is enabled: message is written to DLQ topic with `DlqReason::ClientNack { reason }`, removed from inflight, `NackAck` sent with status 0x00
- If DLQ is disabled: message is dropped from inflight (acknowledged-and-discarded), `NackAck` with 0x01
- If DLQ write fails (full/error): message stays inflight for redelivery, `NackAck` with 0x02

**Session handling:**

```rust
Frame::Nack { message_id, reason } => {
    // Validate message_id is in session's inflight_topics
    let Some(topic) = self.inflight_topics.get(&message_id) else {
        return FrameOutcome::Reply(Frame::Error {
            code: ErrorCode::InternalError,
            message: format!("unknown message id: {message_id}"),
        });
    };

    let result = manager.nack(ns, topic, self.client_id, message_id, reason).await;
    let status = match result {
        Ok(NackResult::DeadLettered) => 0x00,
        Ok(NackResult::Dropped) => 0x01,
        Err(_) => 0x02,
    };

    if status != 0x02 {
        self.inflight_topics.remove(&message_id);
    }

    FrameOutcome::Reply(Frame::NackAck { message_id, status })
}
```

**Nack vs. Ack difference:** Ack is fire-and-forget (no response). Nack always gets a response (`NackAck`) because the client needs to know whether the message was successfully dead-lettered, dropped, or needs retry.

**Reason string limits:** Max 1024 bytes. Truncated if longer, not rejected.

### Individual Subscriber DLQ Routing

**Gap:** The current DLQ check only runs in the consumer group delivery path (`group_deliver`). Individual subscriber delivery (`individual_deliver`) has no DLQ routing — messages cycle through inflight timeout → redeliver indefinitely.

**Fix:** Add the same `delivery_count > max_redelivery` check to the individual delivery path, calling the same `route_to_dlq()` function.

### Metrics

- `fafb_dlq_messages_total{namespace, source_topic, reason}` — messages dead-lettered (reason: `max_attempts`, `client_nack`)
- `fafb_dlq_write_failures_total{namespace, source_topic}` — DLQ write failures (message stayed inflight)
- `fafb_delivery_attempts` (histogram) — delivery count at time of ack/nack/dlq. Shows distribution of how many attempts messages need.
- `fafb_nack_total{namespace, topic}` — client Nack frames received
- `fafb_dlq_time_in_flight_seconds` (histogram) — `dlq_timestamp - first_delivery_timestamp`, time a message spent cycling through redelivery

### Per-Topic DLQ Overrides (Deferred)

Per-topic DLQ config overrides (custom DLQ topic name, custom max attempts) are useful but add significant complexity to the config schema. The current global `dlq_topic_suffix` and `max_redelivery_attempts` per-namespace cover most use cases.

**Recommendation:** Defer per-topic overrides to a follow-up. If needed, the config structure would be:

```toml
[namespaces.myapp.topics.orders]
dlq_topic = "orders-failures"      # override default {topic}__dlq naming
max_delivery_attempts = 3          # stricter for this topic
```

This requires adding a `topics: HashMap<TopicName, TopicConfig>` to `NamespaceConfig`, which is a bigger config schema change best done alongside plan 19 (message headers) or plan 22 (TTL) if those also need per-topic config.

### HTTP DLQ Inspection

No new HTTP endpoints required — DLQ topics are regular topics. Operators use existing endpoints:

- `GET /v1/{ns}/{topic}.__dlq/fetch?offset=0&max_count=100` — fetch dead-lettered messages
- `GET /v1/{ns}/{topic}.__dlq/info` — queue depth, offsets

DLQ message payloads contain the `DlqEnvelope` — consumers parse the envelope to extract metadata and original payload. Document the envelope format in the SDK spec.

A future enhancement could add a DLQ-specific replay endpoint (`POST /v1/{ns}/{topic}.__dlq/replay`) that re-publishes messages to the source topic, but this is out of scope for the initial implementation.

### Nack During Drain

Nack frames should be allowed during drain (same as Ack) — they resolve inflight state, which helps the drain complete faster. Update the `Session::Draining` state to accept Nack alongside Ack/Publish/Ping/Pong.

## Implementation Phases

### Phase 18.1 — Fix DLQ Write Failure Handling (Critical)

**Goal:** Fix the critical bug where `route_to_dlq()` failure silently drops messages. After this phase, messages only leave inflight tracking after a successful DLQ write.

- [x] Change `route_to_dlq()` signature to return `Result<TopicOffset, Error>` instead of `()`
  - [x] Propagate `engine.publish()` errors up to the caller
  - [x] When `dlq_suffix` is empty (DLQ disabled), return `Ok(TopicOffset(0))` sentinel with existing warning log
- [x] Fix group delivery caller in `src/namespace/manager.rs` to check `route_to_dlq()` result
  - [x] On `Ok`: remove from inflight and advance `next_offset` (existing behavior)
  - [x] On `Err`: leave message in inflight for retry on next delivery cycle, log warning
- [x] Add `fafb_dlq_write_failures_total{namespace, source_topic}` counter metric
- [x] Add `fafb_dlq_messages_total{namespace, source_topic, reason}` counter metric on successful DLQ routing
- [x] Add `dlq_failures: u32` field to `InflightEntry` for consecutive-failure tracking
- [x] Extract shared DLQ helpers: `check_dlq_needed()` (state check) and `handle_dlq_result()` (result handling with metrics/logging)
  - [x] `DlqCheckResult` struct captures inflight state snapshot under lock for use after lock is dropped
  - [x] `DLQ_FAILURE_ESCALATION_THRESHOLD` (3) — log escalation from warn to error after repeated failures
  - [x] `handle_dlq_result()` increments `dlq_failures` on inflight entry on failure
- [x] Refactor group delivery DLQ block to use shared helpers
- [x] Unit test: engine that rejects publishes (1-byte memory budget) — `route_to_dlq` returns error
- [x] Unit test: engine that succeeds — `route_to_dlq` returns offset
- [x] Unit test: DLQ disabled returns sentinel offset
- [x] Unit test: `check_dlq_needed` — below threshold, at threshold, above threshold
- [x] Unit test: `handle_dlq_result` — success removes from inflight, failure keeps and increments counter, escalation threshold reached

**Verification:** DLQ write failures no longer lose messages. Repeated failures escalate to ERROR. Shared helpers ready for Phase 18.3 (individual subscriber DLQ). `cargo test` passes (684 tests). Existing redelivery tests unchanged.

---

### Phase 18.2 — DLQ Envelope & `first_delivered_at`

**Goal:** Add structured metadata to dead-lettered messages so operators can debug why messages were DLQ'd. Add `first_delivered_at` tracking to measure redelivery duration.

- [x] Create `src/dlq.rs` module
  - [x] Define `DlqEnvelope` struct with fields: `source_topic`, `original_offset`, `first_delivery_timestamp`, `dlq_timestamp`, `delivery_count`, `reason`
  - [x] Define `DlqReason` enum: `MaxAttemptsExceeded`, `ClientNack { reason: String }`
  - [x] Implement `DlqEnvelope::encode()` — binary format with magic bytes `0xDL`, version `0x01`, and original payload appended
  - [x] Implement `DlqEnvelope::decode(payload: &[u8]) -> Result<(DlqEnvelope, &[u8])>` — parse envelope and return remaining bytes as original payload
  - [x] Define reason code constants: `0x00` = MaxAttemptsExceeded, `0x01` = ClientNack
- [x] Add `first_delivered_at: u64` field to `InflightEntry` in `src/namespace/state.rs`
  - [x] Set `first_delivered_at = now_millis()` when `delivery_count == 1` (first delivery)
  - [x] Preserve `first_delivered_at` unchanged on redelivery (increment `delivery_count` only)
- [x] Modify `route_to_dlq()` to build and serialize `DlqEnvelope` wrapping the original payload
  - [x] Pass `InflightEntry` fields (offset, delivery count, first_delivered_at) to envelope constructor
  - [x] Published DLQ message payload = encoded envelope + original payload
- [x] Add `fafb_dlq_messages_total{namespace, source_topic, reason}` counter metric
- [x] Add `fafb_dlq_time_in_flight_seconds` histogram metric (`dlq_timestamp - first_delivery_timestamp`)
- [x] Unit tests for `DlqEnvelope`
  - [x] Encode/decode round-trip preserves all fields and original payload
  - [x] Decode rejects invalid magic bytes
  - [x] Decode handles unknown version gracefully
  - [x] Reason codes round-trip correctly
- [x] Integration test: publish → consume → let inflight timeout N times → fetch from DLQ topic → verify envelope has correct source_topic, original_offset, delivery_count, reason=MaxAttemptsExceeded (covered via unit tests on `route_to_dlq` that verify envelope payload contents)

**Verification:** DLQ messages contain structured metadata. `first_delivered_at` is tracked correctly across redeliveries. DLQ consumers can decode the envelope and extract the original payload. `cargo test` passes (695 tests). `DlqReason::reason_label()` provides metrics-ready strings. Existing redelivery tests unchanged. Post-implementation refactor: `DlqContext` struct bundles all DLQ routing params (replaces 9-param `route_to_dlq` signature with 6 params); `DlqWriteResult` carries `dlq_timestamp` so time-in-flight metric uses the exact envelope timestamp; `handle_dlq_result` takes `&DlqContext` directly (type-safe reason label, no manual `&'static str` threading).

---

### Phase 18.3 — Individual Subscriber DLQ Routing

**Goal:** Extend DLQ routing to individual subscriber delivery path (not just consumer groups). Currently, individual subscriber messages with exceeded delivery count cycle forever.

- [x] Extract shared DLQ routing logic from group delivery into a common helper function
  - [x] Helper checks `delivery_count > max_redelivery_attempts`, calls `route_to_dlq()`, handles result
  - [x] Both `group_deliver` and `individual_deliver` call the same helper
- [x] Add DLQ check to the individual subscriber delivery path (`individual_deliver`)
  - [x] Mirror the check that exists in `group_deliver`: before delivery attempt, check if delivery count exceeded
  - [x] On DLQ routing success: remove from inflight, advance offset
  - [x] On DLQ routing failure: leave in inflight for retry
- [x] Inflight timeout sweep in individual delivery task
  - [x] Periodic sweep (every 1s) scans inflight entries past `inflight_timeout`
  - [x] Below-threshold entries: `delivery_count` incremented, `delivered_at` reset
  - [x] Above-threshold entries: routed to DLQ via `route_to_dlq()` (message re-read from engine)
  - [x] Failed DLQ writes leave entry unmodified for immediate retry on next sweep
  - [x] Stale entries (message unavailable from engine) removed with warning log
  - [x] Caught-up select wakes periodically when inflight entries exist (prevents indefinite sleep)
  - [x] `inflight_timeout_ms` threaded from manager to delivery task via `Arc<AtomicU64>` in `DeliveryTaskContext`
- [x] Unit test: individual subscriber message exceeds max attempts → routed to DLQ with correct envelope
- [x] Integration test: individual subscriber with DLQ enabled — publish → subscribe (individual) → let timeout exceed max attempts → verify message appears in DLQ topic
- [x] Integration test: sweep escalates delivery_count below threshold without DLQ routing
- [x] Integration test: sweep escalates and routes to DLQ with correct envelope metadata

**Verification:** Both individual and group delivery paths route messages to DLQ. No infinite redelivery loops for individual subscribers. `cargo test` passes (703 tests). Shared helpers (`check_dlq_needed`, `handle_dlq_result`, `route_to_dlq`) are called identically by both `spawn_individual_delivery_task` and `spawn_group_delivery_task`. Individual delivery inflight entry creation now increments `delivery_count` and preserves `first_delivered_at` on re-insert (matching group behavior), in both hot-path and `cold_deliver_individual`. Inflight timeout sweep actively escalates unacked messages through delivery_count thresholds and routes to DLQ — the primary mechanism, not just a pre-delivery safety net. `inflight_timeout_ms` shared via `Arc<AtomicU64>` from manager through `DeliveryTaskContext`, so drain-accelerated timeouts are reflected in the sweep.

---

### Phase 18.4 — Nack Protocol Frame

**Goal:** Allow clients to explicitly reject messages and route them to the DLQ immediately, rather than waiting for redelivery timeout cycles.

- [x] Add frame types to codec (`src/codec/frame.rs`)
  - [x] `MessageType::Nack = 0x16`
  - [x] `MessageType::NackAck = 0x17`
  - [x] `Frame::Nack { message_id: u64, reason: String }` — client → server
  - [x] `Frame::NackAck { message_id: u64, status: u8 }` — server → client (status: 0x00=dead-lettered, 0x01=dropped, 0x02=write-failed)
- [x] Implement encode/decode for Nack and NackAck frames
  - [x] Nack: encode/decode message_id (8B) + reason_length (2B) + reason (NB). Encoder writes full reason; decoder truncates to max 1024 bytes (defense in depth against non-standard clients)
  - [x] NackAck: encode/decode message_id (8B) + status (1B)
- [x] Add `NackResult` enum to `src/namespace.rs`: `DeadLettered`, `Dropped`
- [x] Implement `NamespaceManager::nack(ns, topic, client_id, message_id, reason) -> Result<NackResult>`
  - [x] Look up inflight entry by message_id
  - [x] Call `route_to_dlq()` with `DlqReason::ClientNack { reason }`
  - [x] Return `Dropped` when DLQ is disabled (empty suffix)
  - [x] Return `DeadLettered` on successful DLQ write
  - [x] Return `Err` on DLQ write failure
- [x] Handle Nack in `Session::process_frame()`
  - [x] Validate message_id is in session's `inflight_topics`
  - [x] Call `manager.nack()`, map result to NackAck status
  - [x] On success (0x00 or 0x01): remove from `inflight_topics`
  - [x] On failure (0x02): leave in `inflight_topics` for redelivery
- [x] Allow Nack during drain — `Session::Draining` already passes all non-Subscribe frames through to `process_authenticated()`, so Nack is handled naturally alongside Ack/Publish/Ping/Pong
- [x] Add `fafb_nack_total{namespace, topic}` counter metric
- [x] Update protocol documentation with Nack/NackAck frame format and semantics (documented in CLAUDE.md DLQ section)
- [x] Unit tests for codec
  - [x] Nack frame encode/decode round-trip
  - [x] NackAck frame encode/decode round-trip
  - [x] Nack with reason exceeding 1024 bytes is truncated
- [x] Unit tests for session handling
  - [x] Nack on unknown message_id returns error
  - [x] Nack with DLQ enabled → NackAck status 0x00
  - [x] Nack with DLQ disabled → NackAck status 0x01 (via stub manager returning DeadLettered; manager-level test covers Dropped)
- [x] Integration test: publish → subscribe → consume → Nack with reason → fetch DLQ topic → verify envelope has `ClientNack` reason with correct reason string
- [x] TCP integration test (`tests/tcp_integration.rs`): full end-to-end Nack flow over real TCP connection — subscribe → publish → deliver → Nack → NackAck 0x00 → Fetch DLQ → verify ClientNack envelope
- [x] TCP integration test: Nack with unknown message_id returns Error frame

**Verification:** Clients can explicitly dead-letter messages via Nack. NackAck reports the outcome. Nack is allowed during drain. `cargo test` passes (720 tests). `NackResult` enum defined in `src/namespace.rs` (trait-level). `nack()` method on `NamespaceManager` trait + `DefaultNamespaceManager` implementation. Three-phase approach: (1) snapshot inflight state under lock, (2) re-read message + route to DLQ without lock, (3) re-acquire lock to update inflight via `handle_dlq_result`. DLQ-disabled path validates message_id exists in inflight before removing (consistent with DLQ-enabled path). TOCTOU guard: Phase 4 checks inflight entry still exists before calling `handle_dlq_result` — concurrent ack/sweep handled gracefully. Original engine errors propagated on DLQ write failure (not generic message). Session distinguishes "not in inflight" (concurrent ack/sweep → NackAck 0x01) from "DLQ write failed" (NackAck 0x02). Reason truncation is decoder-only (defense in depth); encoder writes full reason. Proptest `arb_frame()` updated with Nack/NackAck variants.

---

### Phase 18.5 — Capability Negotiation for Nack (Optional)

**Goal:** Add `CAP_NACK` capability bit so the server can distinguish clients that understand Nack from legacy clients.

> **Note:** This phase is optional. If all clients are built from the SDK (not third-party), capability negotiation adds complexity without benefit. Consider shipping Nack without a capability gate and relying on protocol versioning instead.

- [x] Add `CAP_NACK` capability constant (bit 2, value `0b100`) to `src/codec/frame.rs`
- [x] Add `CAP_NACK` to `SERVER_CAPABILITIES` bitmask
- [x] Gate Nack frame acceptance: if client lacks `CAP_NACK`, return `ErrorCode::InvalidFrame`
- [x] Legacy clients (no `CAP_NACK`) continue with timeout-based redelivery only
- [x] Integration test: legacy client (capabilities=0) sends Nack → receives InvalidFrame error
- [x] Integration test: capable client sends Nack → receives NackAck
- [x] Unit test: session with capabilities=0 sends Nack → receives InvalidFrame error
- [x] Update existing Nack unit and integration tests to authenticate with `CAP_NACK`
- [x] Unit test: Reconnect with `CAP_NACK` negotiates capability correctly
- [x] Integration test: Reconnect with `CAP_NACK` → Nack after reconnect succeeds

**Verification:** Backward-compatible — legacy clients unaffected. Only capable clients can use Nack. `cargo test` passes (725 tests). `CAP_NACK = 1 << 2` (bit 2, value `0b100`). `SERVER_CAPABILITIES = CAP_GOING_AWAY | CAP_RECONNECT | CAP_NACK`. Capability check in `process_authenticated` before Nack handling — returns `InvalidFrame` error with "Nack requires CAP_NACK capability" message. Session test helpers extended with `connect_frame_with_caps()` and `authenticate_session_with_caps()` for capability-aware test setup. All existing Nack tests updated to use `CAP_NACK`. Reconnect path verified: `CAP_NACK` is negotiated via the same `capabilities & SERVER_CAPABILITIES` logic, and Nack works correctly after reconnect.

---

### Phase 18.6 — Delivery Attempts Histogram

**Goal:** Add observability into how many delivery attempts messages require before being acknowledged, nacked, or dead-lettered.

- [x] Add `fafb_delivery_attempts` histogram metric
- [x] Record `delivery_count` at these points:
  - [x] On Ack (successful processing by consumer)
  - [x] On Nack (explicit rejection by consumer)
  - [x] On DLQ routing (automatic dead-letter after max attempts)
- [x] Unit test: verify histogram is recorded with correct delivery_count values
- [x] Integration test: publish → consume → ack after first attempt → histogram records 1; publish → consume → timeout → redeliver → ack → histogram records 2

**Verification:** `fafb_delivery_attempts` histogram shows distribution of delivery attempts. Useful for alerting on high-retry topics. `cargo test` passes (733 tests). Histogram records `delivery_count` as `f64` with labels `namespace`, `topic`, and `outcome` (`ack`/`nack`/`dlq`). `record_delivery_attempts()` helper centralizes the recording pattern. `handle_dlq_result` takes explicit `delivery_attempts_outcome: Option<&'static str>` parameter — callers pass `Some("dlq")` for auto-DLQ (max attempts exceeded) and `None` for nack-initiated DLQ. This makes the recording contract visible at each call site rather than implicitly matching on `DlqReason`. Five recording points: individual ack, group ack, `handle_dlq_result` (when `delivery_attempts_outcome` is `Some`), nack-dropped (DLQ disabled), nack-evicted (message unavailable), and nack-success (DLQ write succeeded). The concurrent ack/sweep race path in nack does NOT record (the concurrent path that removed the entry already recorded it). Unit tests use `metrics_util::debugging::DebuggingRecorder` with `metrics::with_local_recorder()` for per-test metric assertions (delivery_count=6 recorded for auto-DLQ, no recording when outcome is None, helper records correct value). Integration test verifies delivery_count=1 on immediate ack and delivery_count=2 after inflight sweep escalation (50ms timeout + 1s sweep interval).

## Dependencies

`metrics-util` added as dev-dependency for test metric assertions. `DlqEnvelope` uses manual binary serialization (same pattern as the protocol codec).

## Testing Strategy

- **Unit tests:** DLQ envelope encode/decode round-trip, `first_delivered_at` tracking, delivery count boundary conditions, Nack frame codec, `route_to_dlq` error handling
- **Integration tests:** Full poison message flow (publish → consume → timeout × N → DLQ), Nack flow, DLQ-disabled behavior, DLQ full backpressure (message stays inflight), DLQ topic readable via Fetch with correct envelope metadata
- **Regression:** Existing redelivery tests must pass unchanged — DLQ changes are additive

## Risks and Mitigations

| Risk | Mitigation |
|---|---|
| DLQ envelope changes payload format — breaks existing DLQ consumers | DLQ envelope is a new feature. No existing consumers read DLQ messages with structured metadata. Document the format in the SDK spec. |
| `first_delivered_at` increases `InflightEntry` size by 8 bytes | Negligible — inflight map entries are already 40+ bytes. At 65K max inflight per connection, this is ~512KB total. |
| Nack flood could overwhelm DLQ writes | DLQ writes go through the same `engine.publish()` path with backpressure. If the DLQ topic is full, NackAck returns status 0x02 and the message stays inflight. Rate limiting Nack frames is not needed — backpressure is built in. |
| DLQ write retry loop on persistent failure | If DLQ is full and the source message keeps cycling through inflight timeout → DLQ attempt → failure, it creates a retry loop. Mitigation: after N consecutive DLQ write failures for the same message, log at ERROR level and increment a specific metric. The message continues to cycle, but operators get visibility. |
| Individual subscriber DLQ routing changes existing behavior | Currently, individual subscriber messages with exceeded delivery count cycle forever. Adding DLQ routing is strictly better — no messages are lost that weren't already stuck. |

## Out of Scope

- **DLQ replay endpoint** (`POST /v1/{ns}/{topic}.__dlq/replay`) — re-publishes DLQ messages to source topic. Useful but separate concern.
- **Per-topic DLQ overrides** — custom DLQ topic names or max attempts per topic. Deferred until per-topic config is needed by other plans.
- **DLQ message expiry** — DLQ topics use the same retention policy as regular topics. Plan 22 (TTL) could add message-level expiry.
- **DLQ alerting** — operators use `fafb_dlq_messages_total` metric with their alerting system. No built-in alerting.
