# Plan 15: Pull-Based Fetch Protocol

## Goal

Add a stateless, pull-based message retrieval mechanism to the TCP and HTTP interfaces so that clients can request batches of messages on-demand instead of relying exclusively on the push-based Subscribe/Deliver flow. This eliminates the slow-consumer problem for workloads with large messages or slow processing, gives SDK developers a simpler consumption model, and makes the system usable for batch/ETL-style consumers that poll periodically.

## Motivation

The current consumption model is push-only: the server spawns a delivery task per subscription that continuously pushes `Deliver` frames as fast as the client can receive them. This works well for low-latency streaming consumers but creates fundamental problems for:

- **Large messages (100KB–1MB+)**: TCP backpressure stalls the delivery loop quickly (OS buffers hold ~256KB), and the 256-message server channel fills up. The client is flagged as a slow consumer after ~100 consecutive `try_send` failures, then either has messages dropped (`DropForSlow`) or gets disconnected (`DisconnectSlow`).
- **Slow processing (seconds per message)**: Clients that need 1–5 seconds per message will always fall behind a push model. Client-side buffering shifts the problem to unbounded memory growth in the SDK process.
- **Batch/ETL consumers**: Jobs that wake up every N minutes to drain a topic don't fit the persistent-subscription model at all. They'd need to subscribe, drain, unsubscribe — and manage offset tracking externally.
- **Polyglot SDK simplicity**: A pull model with explicit offset tracking is dramatically simpler to implement in an SDK than managing push-based subscriptions with flow control, buffering, and reconnection-aware resubscription.

## Prerequisites

- Plans 1–9 complete (core server functional)
- Plan 12 complete (timestamp retention — `u64` timestamps on `MessageEnvelope`)
- Plan 13 Phase A complete (`.idx` index files for O(log n) seek)
- Plan 13 Phase C complete (SubscribeAt/SubscribeFrom — timestamp-to-offset resolution)

## Dependency Graph

```
Phase A (protocol & codec) ──────► Phase B (server implementation)
                                        │
                                        ├──► Phase C (HTTP fetch endpoint)
                                        │         │
                                        │         ▼
                                        └──► Phase D (SDK documentation)
                                              (D.2 depends on C for HTTP docs)

Phase C ──────► Phase E (relative/negative offsets — HTTP server-side + client-side resolution)
```

**Recommended order**: A → B → C → D (linear — each phase depends on the prior; D.2 specifically depends on C's HTTP endpoint design). Phase E depends on C being complete (it is). No TCP protocol changes — client-side resolution in `FafbClient::fetch_tail()`.

---

## Design Decisions

### Pull vs. push: independent mechanisms

Fetch is **not** a replacement for Subscribe/Deliver. They are independent consumption models that coexist:

| Aspect | Subscribe (push) | Fetch (pull) |
|--------|------------------|--------------|
| Server state | Delivery task + channel per subscription | None — stateless RPC |
| Offset tracking | Server-managed (per subscriber/group) | Client-managed |
| Flow control | Server-side slow consumer detection | Client controls pace |
| Ack required | Yes (`Frame::Ack`) | No — offset advances are client's responsibility |
| Consumer groups | Yes (competing consumers) | No — client coordinates externally if needed |
| Reconnection | Must re-subscribe | Just resume fetching from last offset |
| Best for | Low-latency streaming, fan-out | Batch processing, large messages, ETL, simple SDKs |

### Frame size and batch truncation

The `FetchResponse` frame must fit within `tcp.max_message_size` (default 1MB). The server fills messages into the response until either `max_count` is reached or the next message would cause the frame to exceed the size limit. This means:

- A fetch of `max_count=100` for 500KB messages will return **1 message** (the second would exceed 1MB)
- A fetch of `max_count=100` for 1KB messages will return up to 100 messages (~100KB frame)
- The client detects truncation reliably by checking `next_offset < head_offset` (both fields are in the response)

> **Note:** SDK developers must document this: "You may receive fewer messages than `max_count` even when more are available, due to the server's frame size limit. Always check `next_offset < head_offset` to determine if more data exists."

### Oversized message handling

When a single message at the requested offset exceeds `max_response_bytes` (i.e., it cannot fit in a FetchResponse frame at all), the server returns `Error(MessageTooLarge)` (0x000E) instead of a silent empty response. This prevents clients from getting stuck in an infinite poll loop on an unfetchable message.

The error message includes the stuck offset and the message size so the SDK can surface actionable information: "message at offset 42 is 1,572,864 bytes but server frame limit is 1,048,576 bytes". The client can then decide to:
- Skip the message: `offset + 1`
- Request the server operator increase `tcp.max_message_size`
- Use the HTTP fetch endpoint which has a separate, larger size budget (`http.max_fetch_response_bytes`, default 4MB)

### Permission model

A new `Permission::Fetch` variant provides granular access control independent of Subscribe. This enables use cases like:

- Read-only batch consumers that cannot subscribe (no push delivery, no server-side state)
- Admin dashboards that sample topic data without creating subscriptions
- Audit/compliance tools that replay message ranges

If a namespace grants only `Fetch` (not `Subscribe`), the client can pull messages but cannot receive push deliveries. If it grants both, the client can use either model.

### Offset semantics

Fetch uses the same `TopicOffset` semantics as Subscribe:

- `offset` in the request means "give me the message **at** this offset and subsequent messages"
- `next_offset` in the response is the offset to use in the next Fetch call (one past the last returned message)
- `head_offset` in the response is the topic's current head (next offset to be assigned) — enables `has_more = next_offset < head_offset`
- If the requested offset has been deleted by retention, the server returns messages starting from the oldest available offset and includes an `OffsetNotAvailable` error frame before the `FetchResponse` (same pattern as `SubscribeAt` clamping)
- If the requested offset is at or past `head_offset` (no data yet), the response has `count = 0` and `next_offset = head_offset`

### No server-side cursor state

Unlike Kafka's consumer groups, Fetch does not create any server-side state. The server does not track which offsets a fetch consumer has read. This is deliberate:

- Simplifies the server (no fetch-consumer bookkeeping)
- Gives SDK developers full control over offset persistence (local file, database, in-memory)
- Eliminates cleanup/timeout logic for abandoned fetch consumers
- Matches the design philosophy: pull consumers manage their own state

### Concurrency / pipelining semantics

Fetch requests on a single TCP connection are processed **sequentially**. The `process_frame` function in the session layer handles one frame at a time — if a client sends 5 Fetch frames without waiting for responses, they are queued and processed in order (FIFO). The server will not execute them in parallel.

For parallel fetches (e.g., fetching from multiple topics simultaneously), clients should use **multiple TCP connections**. This is the same constraint that applies to all TCP frame types.

> **Note for SDK developers:** Do not design SDK APIs around pipelining Fetch requests on a single connection for throughput. Use multiple connections instead, or use the HTTP endpoint which naturally supports concurrent requests.

### Rate limiting and abuse protection

Subscribe is self-regulating — the server controls the push pace. Fetch is not — a client can fire many Fetch requests per second, each potentially hitting cold storage. This plan does not implement server-side fetch rate limiting, but the risk should be understood:

- **Hot-path fetches** (in-memory) are cheap — bounded by CPU/serialization, similar to `consume_from`
- **Cold-path fetches** (disk) are expensive — each may open segment files, seek through indexes, and read from disk
- **Mitigation without rate limiting**: the existing `max_open_fds` LRU cache on `FsStorage` bounds file descriptor pressure; `max_connections` per namespace bounds total connection count

If fetch abuse becomes a problem in practice, a per-connection or per-namespace rate limit (`tcp.max_fetch_requests_per_second`) can be added as a follow-up. The `fafb_fetch_requests_total` and `fafb_fetch_cold_reads_total` metrics (Phase B.5) provide the observability needed to detect abuse patterns.

---

## Phase A: Protocol & Codec

### Problem

The protocol has no request/response frame for pulling messages. The only way to consume is Subscribe (push). We need new frame types for pull-based consumption.

### Solution

Add `Fetch`, `FetchFrom`, and `FetchResponse` to the protocol, codec encoder, and decoder. Also add `ErrorCode::MessageTooLarge`.

> **Implementation note (resolved collision):** The original plan specified `Fetch=0x10`, `FetchResponse=0x11`, `FetchFrom=0x12`, `ErrorCode::MessageTooLarge=0x000D`. These collided with Plan 14's `GoingAway=0x10`, `Reconnect=0x11`, `ReconnectAck=0x12`, `ServerDraining=0x000D`. Actual codes used: **`Fetch=0x13`, `FetchResponse=0x14`, `FetchFrom=0x15`, `MessageTooLarge=0x000E`**.

### Wire format

#### Fetch (0x13) — client → server

```
┌────────────────────┬───────────────┬──────────────────────────┐
│ Frame Length (4B)   │ 0x13 (1B)     │ Frame Body               │
└────────────────────┴───────────────┴──────────────────────────┘

Frame Body:
┌──────────────────────┬──────────────────┬──────────────────────┐
│ topic (u16 len + str)│ offset (u64 BE)  │ max_count (u32 BE)   │
└──────────────────────┴──────────────────┴──────────────────────┘
```

| Field | Encoding | Description |
|-------|----------|-------------|
| `topic` | length-prefixed string | Topic to fetch from (non-empty) |
| `offset` | `u64` BE | Starting offset (inclusive) |
| `max_count` | `u32` BE | Maximum number of messages to return. Server may return fewer (frame size limit or end of topic). `0` is invalid — server returns `Error(InvalidFrame)`. |

#### FetchResponse (0x14) — server → client

```
┌────────────────────┬───────────────┬────────────────────────────────────────────────────────────┐
│ Frame Length (4B)   │ 0x14 (1B)     │ Frame Body                                                 │
└────────────────────┴───────────────┴────────────────────────────────────────────────────────────┘

Frame Body:
┌──────────────────────┬───────────────────┬───────────────────┬──────────────────┬──────────────┐
│ topic (u16 len + str)│ next_offset (u64) │ head_offset (u64) │ count (u32 BE)   │ messages ... │
└──────────────────────┴───────────────────┴───────────────────┴──────────────────┴──────────────┘

Each message entry:
┌──────────────────┬───────────────────────┬──────────────────────────┬──────────────────┐
│ offset (u64 BE)  │ timestamp_millis (u64)│ payload_len (u32 BE)     │ payload (N bytes)│
└──────────────────┴───────────────────────┴──────────────────────────┴──────────────────┘
```

| Field | Encoding | Description |
|-------|----------|-------------|
| `topic` | length-prefixed string | Echoes the requested topic |
| `next_offset` | `u64` BE | Offset to use in the next Fetch call. If `count > 0`, this is `last_message_offset + 1`. If `count == 0` (topic empty or caught up), this equals `head_offset`. |
| `head_offset` | `u64` BE | Topic's current head offset (next offset to be assigned). Enables `has_more = next_offset < head_offset` without an extra round-trip. |
| `count` | `u32` BE | Number of message entries that follow |
| `messages` | repeated entries | `count` × `(offset: u64, timestamp_millis: u64, payload_len: u32, payload: bytes)` |

> **Note:** The `FetchResponse` does not include `message_id` fields. Message IDs are for ack correlation in the push model — fetch consumers don't ack, so IDs are unnecessary overhead. This is a deliberate simplification.

> **Why `head_offset`?** Without it, the client cannot distinguish "caught up, no more messages" from "there are more but they were truncated by frame size" without making another round-trip that returns 0 messages. With `head_offset`, SDK developers can compute `has_more = next_offset < head_offset` and `remaining = head_offset - next_offset` reliably, enabling smarter poll loops and progress reporting.

#### FetchFrom (0x15) — client → server

```
┌────────────────────┬───────────────┬──────────────────────────┐
│ Frame Length (4B)   │ 0x15 (1B)     │ Frame Body               │
└────────────────────┴───────────────┴──────────────────────────┘

Frame Body:
┌──────────────────────┬───────────────────────┬──────────────────────┐
│ topic (u16 len + str)│ timestamp_millis (u64) │ max_count (u32 BE)   │
└──────────────────────┴───────────────────────┴──────────────────────┘
```

| Field | Encoding | Description |
|-------|----------|-------------|
| `topic` | length-prefixed string | Topic to fetch from (non-empty) |
| `timestamp_millis` | `u64` BE | Fetch messages starting from the first message at or after this timestamp (milliseconds since epoch) |
| `max_count` | `u32` BE | Maximum number of messages to return. `0` is invalid — server returns `Error(InvalidFrame)`. |

The server resolves the timestamp to an offset using `engine.offset_at_timestamp()` (same mechanism as `SubscribeFrom`), then performs a normal fetch from that offset. If no message exists at or after the timestamp, the response has `count = 0` and `next_offset = head_offset`. The server responds with a `FetchResponse` (0x14) — same response frame as `Fetch`.

> **Why a dedicated frame instead of SDK-side workaround?** The alternative (send `SubscribeFrom` to resolve timestamp, `Unsubscribe`, then `Fetch`) is broken: during the brief subscription window, the server spawns a delivery task and may push messages to a channel that is never consumed. It also requires `Subscribe` permission in addition to `Fetch`, defeating the purpose of granular permission control. `FetchFrom` keeps timestamp resolution server-side with no subscription side effects.

#### ErrorCode::MessageTooLarge (0x000E)

New error code added to the ErrorCode enum:

| Value | Name | Fatal | Description |
|-------|------|-------|-------------|
| `0x000E` | `MessageTooLarge` | No | The message at the requested offset exceeds the frame size limit and cannot be included in a FetchResponse. The error message includes the offset and message size. Client should skip the offset (`offset + 1`) or request a larger frame size from the server operator. |

This is non-fatal — the connection remains open. This only fires when the **first** message at the requested offset exceeds the budget. If subsequent messages (after the first) would exceed the budget, the server simply returns fewer messages (normal truncation behavior).

### A.1: Frame enum variants

**File: `src/codec/frame.rs`**

```rust
// Add to MessageType constants
pub const FETCH: u8 = 0x13;
pub const FETCH_RESPONSE: u8 = 0x14;
pub const FETCH_FROM: u8 = 0x15;

// Add to ErrorCode
pub const MESSAGE_TOO_LARGE: u16 = 0x000E;

// Add to Frame enum
Fetch {
    topic: TopicName,
    offset: TopicOffset,
    max_count: u32,
},
FetchFrom {
    topic: TopicName,
    timestamp_millis: u64,
    max_count: u32,
},
FetchResponse {
    topic: TopicName,
    next_offset: TopicOffset,
    head_offset: TopicOffset,
    messages: Vec<FetchedMessage>,
},
```

New struct for decoded fetch messages:

```rust
/// A single message within a FetchResponse frame.
/// Intentionally lighter than PublishedMessage — no MessageId (not needed
/// for pull consumers who don't ack).
#[derive(Debug, Clone)]
pub struct FetchedMessage {
    pub offset: TopicOffset,
    pub timestamp_millis: u64,
    pub payload: Bytes,
}
```

Update `MessageType::is_server_only()` to include `FETCH_RESPONSE`. Update `MessageType::is_client_only()` to include `FETCH` and `FETCH_FROM`. Update `ErrorCode::is_fatal()` to return `false` for `MESSAGE_TOO_LARGE`. Add display/debug formatting for all new variants.

### A.2: Encoder

**File: `src/codec/encoder.rs`**

Add encoding for `Frame::Fetch`:
- Write `FETCH` type byte
- Write topic as length-prefixed string
- Write offset as `u64` BE
- Write max_count as `u32` BE

Add encoding for `Frame::FetchFrom`:
- Write `FETCH_FROM` type byte
- Write topic as length-prefixed string
- Write timestamp_millis as `u64` BE
- Write max_count as `u32` BE

Add encoding for `Frame::FetchResponse`:
- Write `FETCH_RESPONSE` type byte
- Write topic as length-prefixed string
- Write next_offset as `u64` BE
- Write head_offset as `u64` BE
- Write `messages.len() as u32` BE
- For each message: write offset (`u64` BE), timestamp_millis (`u64` BE), payload length (`u32` BE), payload bytes

### A.3: Decoder

**File: `src/codec/decoder.rs`**

Add decoding for `FETCH`:
- Read topic string
- Read offset as `u64` BE
- Read max_count as `u32` BE
- Validate: topic non-empty, max_count > 0 (return `InvalidFrame` if zero)
- Validate: no trailing bytes

Add decoding for `FETCH_FROM`:
- Read topic string
- Read timestamp_millis as `u64` BE
- Read max_count as `u32` BE
- Validate: topic non-empty, max_count > 0 (return `InvalidFrame` if zero)
- Validate: no trailing bytes

Add decoding for `FETCH_RESPONSE`:
- Read topic string
- Read next_offset as `u64` BE
- Read head_offset as `u64` BE
- Read count as `u32` BE
- Read `count` message entries, each: offset (`u64`), timestamp_millis (`u64`), payload_len (`u32`), payload bytes
- Validate: no trailing bytes after all entries consumed

### A.4: Codec tests

**File: `src/codec/decoder.rs` (inline `#[cfg(test)]`)**

- `encode_decode_fetch_roundtrip` — encode Fetch frame, decode back, verify fields match
- `encode_decode_fetch_from_roundtrip` — encode FetchFrom frame, decode back, verify fields match
- `encode_decode_fetch_response_empty` — FetchResponse with 0 messages, verify next_offset and head_offset
- `encode_decode_fetch_response_single` — FetchResponse with 1 message
- `encode_decode_fetch_response_multi` — FetchResponse with multiple messages, verify all offsets/timestamps/payloads
- `encode_decode_fetch_response_head_offset` — verify head_offset is encoded/decoded correctly and independently from next_offset
- `decode_fetch_zero_max_count` — verify InvalidFrame error
- `decode_fetch_from_zero_max_count` — verify InvalidFrame error
- `decode_fetch_empty_topic` — verify InvalidFrame error
- `decode_fetch_from_empty_topic` — verify InvalidFrame error
- `decode_fetch_trailing_bytes` — verify InvalidFrame error
- `decode_fetch_response_truncated` — partial message entry, verify decode error
- `error_code_message_too_large_non_fatal` — verify `MessageTooLarge.is_fatal() == false`

---

## Phase B: Server Implementation

### Problem

The codec can encode/decode Fetch frames, but the server has no handler for them. We need permission checks, a stateless read path through the namespace manager, and session integration.

### Solution

Add `Permission::Fetch`, a `NamespaceManager::fetch()` trait method that reuses the existing `consume_from` / `read_cold_batch` engine methods, and session handlers that dispatch Fetch and FetchFrom frames.

### B.1: Permission variant

**File: `src/types.rs`**

Add `Fetch` to the `Permission` enum:

```rust
pub enum Permission {
    Subscribe,
    Publish,
    Query,
    Fetch,   // NEW
}
```

**File: `src/auth/permissions.rs` (or wherever PermissionSet is)**

Add `FETCH_BIT = 0b1000` to the bitmask. Update `from_str`/`Display`/serde to handle `"Fetch"`.

**File: `src/config.rs`**

Update config validation to accept `"Fetch"` in namespace permission lists.

**File: `config/config.sdk-test.toml`**

Add `Fetch` to the `test` namespace permissions:

```toml
[[namespaces]]
name = "test"
api_key = "sdk-test-key-000000000000"
permissions = ["Subscribe", "Publish", "Query", "Fetch"]
```

**File: `config/config.example.toml`**

Add examples showing different Fetch permission patterns:

```toml
# Full access — push, pull, publish, query
[[namespaces]]
name = "production"
api_key = "your-api-key-here-min-24-chars"
permissions = ["Subscribe", "Publish", "Query", "Fetch"]

# Pull-only consumer — can fetch but not subscribe (no server-side state)
[[namespaces]]
name = "batch-consumer"
api_key = "batch-key-here-min-24-chars-0"
permissions = ["Fetch"]

# Legacy pattern — push subscribe + publish, no pull
[[namespaces]]
name = "streaming-only"
api_key = "stream-key-here-min-24-chars"
permissions = ["Subscribe", "Publish", "Query"]
```

### B.2: FetchResult type

**File: `src/types.rs`**

```rust
/// Result of a stateless Fetch operation.
pub struct FetchResult {
    pub messages: Vec<PublishedMessage>,
    pub next_offset: TopicOffset,
    pub head_offset: TopicOffset,
    /// Set when the requested offset was below the oldest available and was
    /// clamped forward. The session layer uses this to send an
    /// `OffsetNotAvailable` warning before the `FetchResponse`.
    pub clamped_from: Option<TopicOffset>,
}
```

> **Design note:** `clamped_from` encapsulates the clamping decision inside the manager. The session handler does not need to call `base_offset()` separately — it just checks `result.clamped_from.is_some()` to decide whether to send an `OffsetNotAvailable` warning.

### B.3: NamespaceManager::fetch() trait method

**File: `src/namespace.rs`**

Add to the `NamespaceManager` trait:

```rust
/// Fetch up to `max_count` messages starting at `offset`.
///
/// Stateless read — does not create subscriptions, delivery tasks,
/// or inflight tracking. The caller manages offset progression.
///
/// Returns messages available in [offset, head_offset) up to `max_count`,
/// respecting the frame size budget. If `offset` is below `base_offset`
/// (evicted from memory), reads from storage via cold batch.
///
/// Returns `Error::MessageTooLarge` if the first message at the requested
/// offset exceeds `max_response_bytes` — the client cannot make progress
/// at this offset without a larger frame size budget.
fn fetch(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
    offset: TopicOffset,
    max_count: u32,
    max_response_bytes: usize,
) -> impl Future<Output = Result<FetchResult>> + Send;
```

> **Note:** `max_response_bytes` is passed from the session layer (derived from `max_frame_size` minus frame overhead) so the manager can stop adding messages before the response would exceed the frame limit.

**File: `src/namespace/manager.rs`**

Implementation in `DefaultNamespaceManager`:

> **Prep note:** `head_offset()` has already been added to the `NamespaceManager` trait and implemented on `DefaultNamespaceManager` (delegates to `engine.head_offset()`). This was done in Phase A as prep work.

1. Validate namespace exists (return `Error::NamespaceNotFound` if not)
2. Get `head_offset` — if `offset >= head` return empty result with `next_offset = head`, `head_offset = head`, `clamped_from = None`
3. Get `base_offset` — if `offset < base`, record `clamped_from = Some(offset)` and clamp to base
4. Determine read path:
   - If `offset >= base_offset` (in-memory window): loop calling `engine.consume_from(offset)` up to `max_count` times, accumulating messages and tracking total payload bytes. Stop if `consume_from` returns `None` or total bytes would exceed `max_response_bytes`.
   - If `offset < base_offset` (cold path): call `engine.read_cold_batch(ns, topic, offset, max_count)`, then trim to fit `max_response_bytes`.
   - Mixed path: if cold batch returns fewer than `max_count` and reaches the in-memory window, continue with `consume_from` for remaining count.
5. **Oversized message check:** If step 4 collected 0 messages AND the requested offset is below `head_offset`, check the first message size. If its payload size exceeds `max_response_bytes`, return `Error::MessageTooLarge` with the offset and message size in the error message.
6. Compute `next_offset`: if messages returned, `last_message.offset + 1`. If empty (caught up), `head_offset`.
7. Return `FetchResult { messages, next_offset, head_offset, clamped_from }`.

> **Note:** The `fetch()` method does NOT acquire any delivery state locks, does not create `SubscriberDeliveryState`, and does not spawn tasks. It is a pure read operation that reuses the engine's existing read path. This is the key architectural difference from `subscribe()`.

> **Compaction race note:** If `compact()` runs between the `base_offset` check (step 3) and the `consume_from` loop (step 4), messages may disappear mid-read. This is benign: `consume_from` returns `None` for the compacted offsets, and the loop stops. The returned `next_offset` may point past the gap — the client's next fetch will resume from the correct position. No special handling is needed.

### B.4: Session handler

**File: `src/server/tcp/session.rs`**

In `process_authenticated()`, add match arms for `Frame::Fetch` and `Frame::FetchFrom`:

```rust
Frame::Fetch { topic, offset, max_count } => {
    // 1. Permission check
    if self.auth.permissions.check(Permission::Fetch).is_err() {
        tracing::warn!(
            namespace = %self.auth.namespace, topic = %topic,
            client = %self.client_id,
            "fetch permission denied"
        );
        return FrameOutcome::Reply(Frame::Error {
            code: ErrorCode::PermissionDenied,
            message: "fetch permission denied".into(),
        });
    }

    // 2. Compute max response bytes budget
    // Frame overhead: 4 (len) + 1 (type) + 2+topic_len (topic)
    //                 + 8 (next_offset) + 8 (head_offset) + 4 (count)
    let frame_overhead = 4 + 1 + 2 + topic.0.len() + 8 + 8 + 4;
    let max_response_bytes = (max_frame_size as usize).saturating_sub(frame_overhead);

    // 3. Call manager.fetch() — clamping is handled internally
    self.handle_fetch_at_offset(topic, offset, max_count, max_response_bytes, manager).await
}

Frame::FetchFrom { topic, timestamp_millis, max_count } => {
    // 1. Permission check (same as Fetch)
    if self.auth.permissions.check(Permission::Fetch).is_err() {
        return FrameOutcome::Reply(Frame::Error {
            code: ErrorCode::PermissionDenied,
            message: "fetch permission denied".into(),
        });
    }

    // 2. Compute max response bytes budget
    let frame_overhead = 4 + 1 + 2 + topic.0.len() + 8 + 8 + 4;
    let max_response_bytes = (max_frame_size as usize).saturating_sub(frame_overhead);

    // 3. Resolve timestamp to offset
    let resolved_offset = match manager.offset_at_timestamp(&ns, &topic, timestamp_millis).await {
        Ok(Some(offset)) => offset,
        Ok(None) => {
            // No message at or after timestamp — return empty response at head
            let head = manager.head_offset(&ns, &topic).await.unwrap_or(TopicOffset::ZERO);
            return FrameOutcome::Reply(Frame::FetchResponse {
                topic,
                next_offset: head,
                head_offset: head,
                messages: vec![],
            });
        }
        Err(e) => {
            let code = ErrorCode::from_error(&e);
            return FrameOutcome::Reply(Frame::Error { code, message: e.to_string() });
        }
    };

    // 4. Delegate to shared fetch handler
    self.handle_fetch_at_offset(topic, resolved_offset, max_count, max_response_bytes, manager).await
}
```

> **Implementation note:** Extract `handle_fetch_at_offset()` as a helper on `AuthenticatedState`:

```rust
async fn handle_fetch_at_offset<M: NamespaceManager>(
    &self, topic: TopicName, offset: TopicOffset, max_count: u32,
    max_response_bytes: usize, manager: &M,
) -> FrameOutcome {
    let ns = &self.auth.namespace;
    match manager.fetch(ns, &topic, offset, max_count, max_response_bytes).await {
        Ok(result) => {
            let response = Frame::FetchResponse {
                topic: topic.clone(),
                next_offset: result.next_offset,
                head_offset: result.head_offset,
                messages: result.messages.into_iter().map(|m| FetchedMessage {
                    offset: m.offset,
                    timestamp_millis: m.envelope.timestamp,
                    payload: m.envelope.payload.into(),
                }).collect(),
            };
            // clamped_from is set by the manager when offset was below base
            if let Some(original) = result.clamped_from {
                FrameOutcome::ReplyMany(vec![
                    Frame::Error {
                        code: ErrorCode::OffsetNotAvailable,
                        message: format!("requested offset {} not available, clamped", original),
                    },
                    response,
                ])
            } else {
                FrameOutcome::Reply(response)
            }
        }
        Err(e) if e.is_message_too_large() => {
            FrameOutcome::Reply(Frame::Error {
                code: ErrorCode::MessageTooLarge,
                message: e.to_string(),
            })
        }
        Err(e) => {
            let code = ErrorCode::from_error(&e);
            FrameOutcome::Reply(Frame::Error { code, message: e.to_string() })
        }
    }
}
```

> **Implementation note:** The Fetch and FetchFrom handlers share the same logic after offset resolution. Extract the common path into a helper method `handle_fetch_at_offset(topic, offset, max_count, manager, max_frame_size)` to avoid code duplication.

### B.5: Error variant

**File: `src/types.rs` (or error module)**

Add `Error::MessageTooLarge { offset: TopicOffset, message_bytes: usize, budget_bytes: usize }` variant with a helper method `is_message_too_large() -> bool`.

### B.6: Metrics

Add counter: `fafb_fetch_requests_total{namespace, topic}` — incremented on each Fetch or FetchFrom frame processed.
Add histogram: `fafb_fetch_messages_returned` — number of messages per FetchResponse, for observing batch utilization.
Add counter: `fafb_fetch_cold_reads_total{namespace, topic}` — when fetch reads from disk (offset below in-memory window).
Add counter: `fafb_fetch_message_too_large_total{namespace, topic}` — when MessageTooLarge error is returned.

### B.7: Unit & integration tests

**File: `src/namespace/manager.rs` (inline `#[cfg(test)]`)**

- `fetch_returns_messages_in_range` — publish 10 messages, fetch offset 3 max_count 5, verify 5 messages returned with correct offsets
- `fetch_empty_topic` — fetch from nonexistent topic, verify empty result with next_offset 0
- `fetch_at_head` — fetch at head offset, verify empty result with next_offset = head_offset = head
- `fetch_past_head` — fetch beyond head, verify empty result
- `fetch_respects_max_response_bytes` — publish 10 large messages, set small max_response_bytes, verify fewer than max_count returned
- `fetch_cold_path` — publish messages, compact in-memory buffer so offset is below base, verify fetch reads from storage
- `fetch_clamp_below_base` — fetch at offset below base_offset, verify messages start from base
- `fetch_does_not_create_subscription_state` — fetch, then verify no entries in TopicDeliveryState
- `fetch_head_offset_in_result` — publish 10 messages, fetch first 5, verify `head_offset == 10` and `next_offset == 5`
- `fetch_message_too_large` — publish a message larger than the budget, verify `Error::MessageTooLarge` is returned with correct offset and sizes
- `fetch_message_too_large_only_on_first` — publish a small message then a large one, set budget to fit only the small one, verify the small message is returned (truncation, not error)

**File: `tests/tcp_integration.rs`**

- `tcp_fetch_basic` — connect, publish 5 messages, send Fetch frame, verify FetchResponse with 5 messages and correct head_offset
- `tcp_fetch_from_timestamp` — publish messages with known timestamps, send FetchFrom, verify response starts from correct offset
- `tcp_fetch_from_no_match` — send FetchFrom with timestamp in the future, verify empty response with head_offset
- `tcp_fetch_permission_denied` — connect with subscribe-only key, send Fetch, verify PermissionDenied error
- `tcp_fetch_offset_clamping` — publish messages, trigger retention, fetch old offset, verify OffsetNotAvailable + FetchResponse
- `tcp_fetch_large_batch_truncated` — publish many large messages, fetch with high max_count, verify response truncated by frame size and `next_offset < head_offset`
- `tcp_fetch_interleaved_with_subscribe` — verify fetch and subscribe work independently on same connection
- `tcp_fetch_zero_max_count` — send Fetch with max_count=0, verify InvalidFrame error
- `tcp_fetch_message_too_large` — publish a message exceeding frame size, fetch, verify MessageTooLarge error with offset info

**File: `src/queue/engine.rs` or `tests/common/mod.rs`**

- Update `MockStorage` / mock `NamespaceManager` to implement the `fetch()` trait method

---

## Phase C: HTTP Fetch Endpoint

### Problem

The HTTP API currently supports Publish and Query but not message retrieval. Adding an HTTP Fetch endpoint lets clients pull messages without maintaining a TCP connection — ideal for batch jobs, serverless functions, and simple scripts.

### Solution

Add `GET /v1/{namespace}/{topic}/fetch` that maps to the same `NamespaceManager::fetch()` method.

### C.1: HTTP route

**File: `src/server/http/routes.rs`**

```
GET /v1/{namespace}/{topic}/fetch?offset={u64}&max_count={u32}
GET /v1/{namespace}/{topic}/fetch?timestamp={u64}&max_count={u32}
```

Either `offset` or `timestamp` must be provided (not both). If `timestamp` is provided, the server resolves it to an offset using `engine.offset_at_timestamp()` before fetching.

- **Permission required**: `Fetch`
- **Query parameters**:
  - `offset` (u64): starting offset (inclusive) — mutually exclusive with `timestamp`
  - `timestamp` (u64): milliseconds since epoch, fetch from first message at or after — mutually exclusive with `offset`
  - `max_count` (optional, default 100, max 10,000): maximum messages to return
- **Response** (`200`):

```json
{
  "topic": "my-topic",
  "next_offset": 15,
  "head_offset": 42,
  "messages": [
    {
      "offset": 10,
      "timestamp_millis": 1711036800000,
      "payload_base64": "aGVsbG8="
    },
    {
      "offset": 11,
      "timestamp_millis": 1711036800100,
      "payload_base64": "d29ybGQ="
    }
  ]
}
```

> **Note:** Payloads are base64-encoded in JSON. This is the standard approach for binary data in JSON APIs. SDK developers should decode the base64 payload before returning to users.

- **Response** (`200`, clamped offset): includes `"warning": "requested offset not available, clamped to oldest retained"` field
- **Errors**: `400` (missing/invalid params), `401`, `403`, `404`, `413` (message too large)

### C.2: Response size limit

For HTTP, use a configurable `http.max_fetch_response_bytes` (default 4MB, separate from TCP frame size) as the budget passed to `manager.fetch()`. This allows larger batches over HTTP where frame framing isn't a constraint, and provides a fallback path for messages that exceed the TCP frame size.

**File: `src/config.rs`**

Add to `HttpConfig`:

```rust
#[serde(default = "default_max_fetch_response_bytes")]
pub max_fetch_response_bytes: usize,  // default: 4_194_304 (4MB)
```

### C.3: HTTP error mapping

| Condition | HTTP Status | ErrorCode |
|-----------|-------------|-----------|
| Missing both `offset` and `timestamp` | `400 Bad Request` | `InvalidFrame` |
| Both `offset` and `timestamp` provided | `400 Bad Request` | `InvalidFrame` |
| Invalid `offset` / `max_count` / `timestamp` | `400 Bad Request` | `InvalidFrame` |
| `max_count = 0` | `400 Bad Request` | `InvalidFrame` |
| `max_count > 10000` | `400 Bad Request` | `InvalidFrame` |
| Missing/invalid auth | `401 Unauthorized` | `AuthFailed` |
| No Fetch permission | `403 Forbidden` | `PermissionDenied` |
| Namespace not found | `404 Not Found` | `NamespaceNotFound` |
| Message too large for response | `413 Payload Too Large` | `MessageTooLarge` |
| Topic not found (never published to) | `200` with empty messages | — |

> **Note:** An empty topic returns `200` with `count: 0`, not `404`. This matches the TCP behavior where fetching from a nonexistent topic returns an empty result — topics auto-create on publish, so "not found" and "empty" are indistinguishable by design.

### C.4: Integration tests

**File: `tests/http_integration.rs` (or existing HTTP test file)**

- `http_fetch_basic` — publish via HTTP, fetch via HTTP, verify response including head_offset
- `http_fetch_by_timestamp` — publish messages, fetch by timestamp, verify correct starting offset
- `http_fetch_pagination` — publish 50 messages, fetch in batches of 10 using `next_offset` chaining, verify all 50 retrieved
- `http_fetch_permission_denied` — use query-only key, verify 403
- `http_fetch_missing_offset` — omit both offset and timestamp, verify 400
- `http_fetch_both_offset_and_timestamp` — provide both, verify 400
- `http_fetch_base64_payload` — publish binary payload, fetch, verify base64 decoding roundtrip
- `http_fetch_message_too_large` — publish oversized message, fetch, verify 413

### C.5: Implementation notes from Phase B

> These notes capture lessons from Phase B that inform the Phase C implementation.

**Reuse `NamespaceManager::fetch()` directly.** The HTTP handler should call `manager.fetch(ns, topic, offset, max_count, max_response_bytes)` — the same method used by the TCP session. No new engine read path is needed. The only difference is the `max_response_bytes` budget: TCP derives it from `max_frame_size - frame_overhead`, HTTP uses `http.max_fetch_response_bytes` directly.

**Shared wire overhead constant.** `crate::codec::frame::FETCH_MESSAGE_OVERHEAD` (20 bytes) defines the per-message overhead used for budget tracking. The HTTP handler doesn't need this directly — it's used inside `manager.fetch()` — but the HTTP response body serialization should be aware that the manager's budget math is based on wire overhead, not JSON overhead. The HTTP response may be slightly larger than `max_fetch_response_bytes` due to base64 expansion (~33%) and JSON field names. If exact HTTP body size control is needed, reduce the budget passed to `manager.fetch()` by a safety factor, but this is likely unnecessary for the 4MB default.

**Timestamp resolution.** For `?timestamp=` queries, the HTTP handler should call `manager.offset_at_timestamp()` to resolve to an offset, then call `manager.fetch()` — the same two-step pattern used by `FetchFrom` in the TCP session. When `offset_at_timestamp()` returns `None`, return a 200 with empty messages and `next_offset == head_offset`.

**`max_count` validation.** The TCP protocol validates `max_count > 0` in the decoder (returns `InvalidFrame`). The HTTP handler must validate both `max_count > 0` and `max_count <= 10000` at the handler level, returning 400.

**`clamped_from` → `warning` field.** When `result.clamped_from.is_some()`, include a `"warning"` field in the JSON response. The TCP protocol sends a separate `OffsetNotAvailable` error frame before the `FetchResponse`; HTTP flattens this into a single response with the warning inlined.

**Auth pattern.** Existing HTTP handlers (Publish, Query) already extract the API key from the `Authorization` header and check permissions via `PermissionSet::check()`. Fetch should follow the same pattern, checking `Permission::Fetch`.

**Error mapping for `MessageTooLarge`.** `ErrorCode::MessageTooLarge` already maps to `"MESSAGE_TOO_LARGE"` in `error_code_name()` (added in Phase A). The HTTP handler maps `Error::MessageTooLarge` → `413 Payload Too Large` with the error message containing the stuck offset and sizes. The existing `src/server/http/error.rs` module handles error-to-status mapping — add a match arm there.

**Fetch is allowed during server drain.** The `/v1/*` routes are gated by `require_initialized` middleware, but not by drain state. Since Fetch is stateless and doesn't create server-side state, it should remain accessible during drain (same as Publish). No special drain handling is needed in the HTTP handler.

---

## Phase D: Client SDK Documentation

### Problem

The existing `docs/client_sdk_spec.md` is the single reference for SDK implementers. It must be updated to cover the Fetch protocol so that SDK developers have a clear, complete, and unambiguous specification. The documentation must explain not just the wire format but the behavioral semantics, edge cases, and recommended SDK API patterns for pull consumption.

### Solution

Update `docs/client_sdk_spec.md`, `docs/api.md`, and create a focused `docs/fetch_consumer_guide.md` that explains the pull consumption model in practical terms with code examples.

> **Implementation notes from Phase C (important for documentation accuracy):**
>
> 1. **Base64 encoding must be documented prominently.** HTTP payloads are base64-encoded in JSON (`payload_base64` field). This is easy to miss and will cause silent data corruption if SDK developers treat it as raw text. The `client_sdk_spec.md` and `api.md` must both call this out clearly, ideally with a code example showing encode/decode.
>
> 2. **HTTP vs TCP budget difference.** The HTTP fetch endpoint has a 4MB default response budget (`http.max_fetch_response_bytes`) vs the TCP frame limit of 1MB (`tcp.max_message_size`). This means messages that return `MessageTooLarge` over TCP may be fetchable via HTTP. Document this as an explicit fallback strategy: "If you receive `MessageTooLarge` on TCP, retry via the HTTP endpoint."
>
> 3. **The `warning` field is HTTP-only.** TCP sends a separate `OffsetNotAvailable` error frame before the `FetchResponse`. HTTP flattens this into a `"warning"` string field on the JSON response body. SDK documentation should explain both patterns clearly and note the asymmetry.
>
> 4. **`max_count` upper bound of 10,000 is HTTP-specific.** The TCP protocol validates only `max_count > 0` (in the decoder). The HTTP endpoint additionally enforces `max_count <= 10,000` at the handler level, returning 400. Document this difference — TCP SDKs can request any count (server truncates by frame size), HTTP SDKs are hard-capped at 10,000.

### D.1: Update `docs/client_sdk_spec.md`

#### D.1.1: Type definitions section

Add to MessageType enum table:

```
Fetch           = 0x13    client → server
FetchResponse   = 0x14    server → client
FetchFrom       = 0x15    client → server
```

Add `FetchedMessage` composite type:

```
FetchedMessage:
  offset:           u64        // position within topic log
  timestamp_millis: u64        // publish timestamp (milliseconds since epoch)
  payload:          bytes      // raw message payload
```

Add `FetchResponseData` composite type:

```
FetchResponseData:
  topic:       string
  next_offset: u64             // offset for next fetch call
  head_offset: u64             // topic's current head offset (next to be assigned)
  messages:    FetchedMessage[] // 0 to max_count messages
```

Add `Permission::Fetch` to the Permission enum table.

Add `ErrorCode::MessageTooLarge` (0x000E, non-fatal) to the ErrorCode table:

```
| `0x000E` | `MessageTooLarge` | No | Message at requested offset exceeds frame size limit. Skip with offset+1 or use HTTP endpoint with larger budget. |
```

#### D.1.2: Connection state machine

Update the state machine diagram — `Fetch` and `FetchFrom` are valid in both `Authenticated` and `Subscribed` states (they are stateless, don't require a subscription):

```
Authenticated state valid sends:
  - Subscribe, SubscribeAt, SubscribeFrom
  - Publish
  - Fetch, FetchFrom   ← NEW
  - Ping

Subscribed state valid sends:
  - Subscribe, SubscribeAt, SubscribeFrom
  - Unsubscribe
  - Publish
  - Fetch, FetchFrom   ← NEW
  - Ack
  - Ping
```

Update the "Valid messages per state" table:

| State | Client may send | Client may receive |
|-------|----------------|--------------------|
| `Authenticated` | ..., `Fetch`, `FetchFrom` | ..., `FetchResponse` |
| `Subscribed` | ..., `Fetch`, `FetchFrom` | ..., `FetchResponse` |

> **Note for SDK developers:** `Fetch` and `FetchFrom` do NOT change the connection state. They are stateless request/response operations within the current state. A client can freely interleave Fetch requests with Subscribe/Publish operations on the same connection. However, Fetch requests on a single connection are processed sequentially — for parallel fetches from multiple topics, use multiple connections or the HTTP endpoint.

#### D.1.3: Wire protocol section

Add message field layouts:

**Fetch (0x13) — client → server**

| Field | Encoding |
|-------|----------|
| `topic` | length-prefixed string (non-empty) |
| `offset` | `u64` BE — starting offset (inclusive). If below oldest retained offset, server clamps and sends `OffsetNotAvailable` warning before `FetchResponse`. |
| `max_count` | `u32` BE — maximum messages to return. Must be > 0. Server may return fewer due to frame size limits or end of topic. |

**FetchFrom (0x15) — client → server**

| Field | Encoding |
|-------|----------|
| `topic` | length-prefixed string (non-empty) |
| `timestamp_millis` | `u64` BE — fetch from first message at or after this timestamp (milliseconds since epoch). If no message exists at or after this timestamp, returns empty response at head. |
| `max_count` | `u32` BE — maximum messages to return. Must be > 0. |

The server resolves the timestamp to an offset internally and responds with a `FetchResponse` (same frame as `Fetch`). This requires only `Fetch` permission — no `Subscribe` permission needed.

**FetchResponse (0x14) — server → client**

Sent in response to both `Fetch` and `FetchFrom`.

| Field | Encoding |
|-------|----------|
| `topic` | length-prefixed string — echoes requested topic |
| `next_offset` | `u64` BE — offset for next Fetch call. If `count > 0`: last returned offset + 1. If `count == 0`: topic's head offset. |
| `head_offset` | `u64` BE — topic's current head offset (next offset to be assigned). Use `next_offset < head_offset` to check if more messages are available. |
| `count` | `u32` BE — number of message entries following |
| `messages` | `count` × `(offset: u64 BE, timestamp_millis: u64 BE, payload_len: u32 BE, payload: bytes)` |

Add notes on frame size interaction and oversized messages:

> **Frame size and batch truncation:** The entire `FetchResponse` frame must fit within the server's `max_message_size` (default 1MB). If the requested `max_count` would produce a response exceeding this limit, the server returns fewer messages. Use `next_offset < head_offset` to detect that more messages are available, then issue another Fetch with `offset = next_offset`. For topics with large messages (e.g., 500KB payloads), a single Fetch may return only 1 message.

> **Oversized messages (`MessageTooLarge`):** If the message at the requested offset is individually larger than the frame size limit, the server cannot include it in any FetchResponse. Instead of returning an empty response (which would cause an infinite poll loop), the server returns `Error(MessageTooLarge, 0x000E)` with the stuck offset and message size. The client should either skip the message (`offset + 1`), request the server operator increase `max_message_size`, or use the HTTP fetch endpoint which has a separate, larger size budget (default 4MB).

#### D.1.4: Recommended client API surface

Add a new "Fetch (pull consumption)" subsection after the existing "Message delivery" section:

```
### Fetch (TCP or HTTP)

// Pull a batch of messages starting from an offset
client.fetch(topic: TopicName, offset: TopicOffset, options?: FetchOptions) → Promise<FetchBatch>

// Pull a batch starting from a timestamp
client.fetchFrom(topic: TopicName, timestamp: u64, options?: FetchOptions) → Promise<FetchBatch>

FetchOptions:
  max_count: number             // max messages to return (default: 100)

FetchBatch:
  topic:       TopicName
  next_offset: TopicOffset      // use this as `offset` in next fetch() call
  head_offset: TopicOffset      // topic's current head — for progress calculation
  messages:    FetchedMessage[]  // 0 to max_count messages
  has_more:    boolean           // convenience: next_offset < head_offset

FetchedMessage:
  offset:          TopicOffset
  timestamp_millis: u64
  payload:         bytes
```

> **Implementation note for SDK developers:** Both `fetch()` and `fetchFrom()` map directly to dedicated server frames (`Fetch` 0x13 and `FetchFrom` 0x15 respectively). Both require only `Fetch` permission. The `has_more` field is computed client-side as `next_offset < head_offset` — this is reliable because `head_offset` is included in every `FetchResponse`.

#### D.1.5: Error handling

Add fetch-specific error handling guidance:

```
### Fetch error handling

- **PermissionDenied**: Namespace does not grant Fetch permission. Surface as a typed error.
- **OffsetNotAvailable (0x000C)**: Informational warning sent BEFORE a successful FetchResponse
  when the requested offset was deleted by retention. The FetchResponse that follows contains
  data starting from the oldest available offset. SDKs should:
  1. Log the warning
  2. Update their stored offset to FetchResponse.next_offset
  3. NOT treat this as a failure
- **MessageTooLarge (0x000E)**: The message at the requested offset exceeds the server's frame
  size limit and cannot be fetched over TCP. SDKs should:
  1. Surface the error with the offset and message/limit sizes from the error message
  2. Offer a `skip()` convenience that fetches from `offset + 1`
  3. Document the HTTP fallback (larger default budget)
  4. NOT retry the same offset — it will never succeed without config change
- **InvalidFrame**: max_count was 0, or topic was empty. Client-side bug — fix the request.
- **ResourceExhausted**: Server under pressure. Retry with backoff.
- **NamespaceNotFound**: Configuration error. Do not retry.
```

#### D.1.6: Compatibility test scenarios

Add to the TCP test table:

| # | Scenario | Asserts |
|---|----------|---------|
| 16 | Fetch messages from a topic | Receives `FetchResponse` with correct messages, offsets, and `head_offset` |
| 17 | Fetch with pagination (multiple fetches using `next_offset`) | All messages retrieved across batches, no duplicates, `has_more` transitions from true to false |
| 18 | Fetch empty topic | `FetchResponse` with `count=0`, `next_offset == head_offset` |
| 19 | Fetch without Fetch permission | Receives `Error(PermissionDenied)` |
| 20 | Fetch at clamped offset | Receives `Error(OffsetNotAvailable)` followed by `FetchResponse` starting from oldest available |
| 21 | Fetch with max_count=0 | Receives `Error(InvalidFrame)` |
| 22 | Fetch interleaved with Subscribe on same connection | Both work independently; Fetch does not affect subscription delivery |
| 23 | FetchFrom by timestamp | Receives `FetchResponse` starting from first message at or after the timestamp |
| 24 | FetchFrom with future timestamp | Receives empty `FetchResponse` with `next_offset == head_offset` |
| 25 | Fetch oversized message | Receives `Error(MessageTooLarge)` with offset and size info |

Add to the HTTP test table:

| # | Scenario | Asserts |
|---|----------|---------|
| 10 | Fetch via HTTP | `200` with `topic`, `next_offset`, `head_offset`, and `messages` array |
| 11 | Fetch pagination via HTTP | Chain multiple requests using `next_offset`, verify all messages |
| 12 | Fetch by timestamp via HTTP | `200` with messages starting from resolved offset |
| 13 | Fetch without permission | `403` with error JSON |
| 14 | Fetch missing offset parameter | `400` with error JSON |
| 15 | Fetch oversized message via HTTP | `413` with error JSON |

#### D.1.7: Implementation checklist

Add to the checklist:

```
### Fetch (pull consumption)

- [ ] Encode/decode Fetch frame (0x13): topic + offset + max_count
- [ ] Encode/decode FetchFrom frame (0x15): topic + timestamp_millis + max_count
- [ ] Encode/decode FetchResponse frame (0x14): topic + next_offset + head_offset + count + repeated messages
- [ ] Compute `has_more` from `next_offset < head_offset`
- [ ] Handle fewer-than-requested messages (frame size truncation)
- [ ] Handle `OffsetNotAvailable` warning before FetchResponse (offset clamping)
- [ ] Handle `MessageTooLarge` error — surface offset and sizes, offer skip
- [ ] Handle empty FetchResponse (count=0) — topic empty or caught up
- [ ] Expose `fetch(topic, offset, options)` in public API
- [ ] Expose `fetchFrom(topic, timestamp, options)` in public API
- [ ] Track next_offset from response for pagination
- [ ] HTTP fetch: parse base64-encoded payloads
- [ ] HTTP fetch: handle pagination with next_offset
- [ ] HTTP fetch: support both `offset` and `timestamp` query params
```

### D.2: Update `docs/api.md`

Add the HTTP Fetch endpoint documentation:

```
#### `GET /v1/{namespace}/{topic}/fetch`

Pull a batch of messages from a topic. Requires `Fetch` permission.

**Query parameters:**
  - `offset` (u64): starting offset (inclusive) — mutually exclusive with `timestamp`
  - `timestamp` (u64): milliseconds since epoch, fetch from first message at or after — mutually exclusive with `offset`
  - `max_count` (optional, u32, default 100, max 10000): maximum messages

One of `offset` or `timestamp` must be provided.

**Response** (`200`):
  - `topic` (string): echoed topic name
  - `next_offset` (u64): offset for next fetch call
  - `head_offset` (u64): topic's current head offset
  - `messages` (array): each with `offset`, `timestamp_millis`, `payload_base64`
  - `warning` (string, optional): present when offset was clamped

**Errors**: `400` (invalid params), `401`, `403`, `404`, `413` (message too large)

**Response size limit**: configurable via `http.max_fetch_response_bytes` (default 4MB).
Messages that exceed the TCP frame limit (1MB) may still be fetchable via HTTP.
```

### D.3: Create `docs/fetch_consumer_guide.md`

A practical guide for SDK developers building pull/fetch consumer support. This is NOT a wire protocol reference (that's in `client_sdk_spec.md`) — it's a behavioral guide with patterns, gotchas, and code examples.

#### Contents outline:

**1. When to use Fetch vs. Subscribe**

Decision matrix for SDK users:

| Your situation | Use |
|----------------|-----|
| Real-time message processing, low latency needed | Subscribe (push) |
| Processing takes seconds per message | Fetch (pull) |
| Messages are > 100KB | Fetch (pull) |
| Batch job that runs every N minutes | Fetch (pull) |
| Fan-out to many consumers | Subscribe (push) |
| Competing consumers (load balance) | Subscribe with consumer group (push) |
| Simple SDK, don't want to manage flow control | Fetch (pull) |
| Need server-side offset tracking | Subscribe (push) |
| Serverless function / AWS Lambda | Fetch via HTTP (pull) |

**2. Offset management patterns**

Detailed explanation of how SDK developers should manage offsets for pull consumers:

```
Pattern A: In-memory offset (simplest, at-most-once)
  - Store next_offset in a variable
  - On restart, start from offset 0 or head
  - Messages may be reprocessed or skipped on restart

Pattern B: Persistent offset (at-least-once)
  - After processing a batch, persist next_offset to file/database
  - On restart, read persisted offset
  - Messages may be reprocessed (idempotent processing required)

Pattern C: Per-message checkpointing (exactly-once with external store)
  - Process each message, write (message_offset, result) to database in same transaction
  - On restart, query database for max processed offset
```

Code examples in pseudocode and TypeScript/Python showing each pattern.

**3. Pagination and batch draining**

```
// Drain an entire topic
let offset = 0;
loop {
    let batch = await client.fetch("my-topic", offset, { max_count: 100 });
    if (batch.messages.length === 0) break;  // caught up
    for (const msg of batch.messages) {
        await process(msg);
    }
    offset = batch.next_offset;
    await persistOffset(offset);  // checkpoint
}
```

**4. Using `head_offset` and `has_more`**

Explain how to use the `head_offset` field for efficient poll loops:

```
// Efficient drain with progress reporting
const batch = await client.fetch("my-topic", offset, { max_count: 100 });
const remaining = batch.head_offset - batch.next_offset;
console.log(`Fetched ${batch.messages.length}, ${remaining} remaining`);

if (batch.has_more) {
    // More messages available — fetch immediately, don't sleep
    offset = batch.next_offset;
} else {
    // Caught up — sleep before polling again
    await sleep(pollInterval);
}
```

> **Note:** `head_offset` is a snapshot at the time the server processed the Fetch request. New messages may have been published between the response and the client reading it. This is expected — the next Fetch will pick them up.

**5. Frame size truncation behavior**

Explain with concrete numbers:
- 1MB frame limit, 1KB messages → ~900 messages per fetch (accounting for per-message overhead of 20 bytes)
- 1MB frame limit, 500KB messages → 1 message per fetch
- 1MB frame limit, 1.5MB messages → `MessageTooLarge` error (not a silent empty response)

```
// Handling MessageTooLarge
try {
    const batch = await client.fetch("my-topic", offset);
} catch (err) {
    if (err.code === ErrorCode.MessageTooLarge) {
        console.warn(`Message at offset ${offset} is too large for TCP fetch`);
        // Option 1: Skip it
        offset = offset + 1;
        // Option 2: Fetch via HTTP (4MB default limit)
        const batch = await httpClient.fetch("my-topic", offset);
        // Option 3: Surface to user
        throw new Error(`Cannot fetch offset ${offset}: ${err.message}`);
    }
}
```

**6. Polling patterns**

```
// Long-poll style with adaptive backoff
async function pollLoop(topic, startOffset) {
    let offset = startOffset;
    let backoff = 1000;  // start at 1s
    while (running) {
        const batch = await client.fetch(topic, offset, { max_count: 100 });
        if (batch.messages.length > 0) {
            for (const msg of batch.messages) {
                await process(msg);
            }
            offset = batch.next_offset;
            backoff = 1000;  // reset backoff on data received
        } else {
            await sleep(backoff);
            backoff = Math.min(backoff * 2, 30000);  // exponential backoff, cap 30s
        }
    }
}
```

**7. Retention race and gap detection**

When a client is mid-pagination and retention deletes segments between fetch calls, the client will experience a gap:

```
Fetch offset=1000 → returns messages 1000-1099, next_offset=1100
  (retention task runs, deletes segments up to offset 1500)
Fetch offset=1100 → receives OffsetNotAvailable warning, response starts at offset 1500
  gap: messages 1100-1499 were deleted
```

SDKs should detect this by comparing the first message's offset against the requested offset:

```
const batch = await client.fetch(topic, expectedOffset);
if (batch.messages.length > 0 && batch.messages[0].offset > expectedOffset) {
    const gap = batch.messages[0].offset - expectedOffset;
    console.warn(`Gap detected: ${gap} messages skipped (offsets ${expectedOffset}-${batch.messages[0].offset - 1} deleted by retention)`);
    // Emit a "gap" event so the application can decide how to handle it
    emit('gap', { from: expectedOffset, to: batch.messages[0].offset, count: gap });
}
```

This is expected behavior, not a bug. Retention runs independently of consumers. If your application cannot tolerate gaps, configure longer retention policies.

**8. Error recovery**

- **OffsetNotAvailable**: Your stored offset was deleted by retention. Update to `next_offset` from the response. Log the gap (see section 7). Consider whether skipped messages are acceptable for your use case.
- **MessageTooLarge**: Message at offset exceeds frame size. Skip with `offset + 1`, use HTTP endpoint (4MB budget), or contact server operator to increase `max_message_size`.
- **Connection lost mid-fetch**: Retry from the same offset. Fetch is idempotent — re-reading the same offset returns the same data (unless retention deleted it in the interim).
- **Empty responses**: Topic is caught up. Implement exponential backoff in poll interval (1s → 2s → 4s → max 30s), reset to 1s when messages are received.

**9. Push + pull hybrid pattern**

For SDK developers who want to offer both models:

```
// Subscribe for real-time, fall back to fetch for catch-up
if (behindByMoreThan(1000, currentOffset, headOffset)) {
    // Too far behind for push — bulk fetch to catch up
    while (currentOffset < headOffset - 100) {
        const batch = await client.fetch(topic, currentOffset, { max_count: 500 });
        if (batch.messages.length === 0) break;
        for (const msg of batch.messages) await process(msg);
        currentOffset = batch.next_offset;
    }
    // Close enough — switch to push for real-time tail
    await client.subscribeAt(topic, currentOffset);
}
```

**10. Performance characteristics**

| Read path | Latency | Throughput | When |
|-----------|---------|------------|------|
| Hot (in-memory) | < 1ms per fetch | Limited by CPU/serialization | Offset is within in-memory buffer |
| Cold (disk) | 1-10ms per fetch | Limited by disk I/O | Offset is below base_offset (evicted to segments) |
| Mixed | Variable | Transitions mid-batch | Batch spans memory/disk boundary |

> **SDK guidance:** For maximum throughput on cold reads, use `max_count` values of 100-500. Smaller batches waste round-trips. Larger batches may hit frame size limits. The server reads from disk in batches of `cold_batch_size` (default 256) internally, so requesting ~256 messages aligns well with the server's I/O pattern.

---

## Phase E: Relative (Negative) Offsets

### Problem

There is no way to fetch the last N messages from a topic without already knowing the head offset. Clients must either subscribe (push, starts at head, doesn't return existing messages) or make a throwaway Fetch to discover `head_offset`, then a second Fetch at `head_offset - N`. This two-step dance is awkward for common use cases: tailing a topic, inspecting the most recent message, sampling recent data for debugging, or building dashboards that show "last 10 messages".

### Solution

Support negative offsets as relative-from-head shorthand. `offset = -1` means "last message", `offset = -N` means "N messages from the head". Two resolution strategies:

1. **HTTP (server-side):** `FetchParams.offset` accepts `i64`. Negative values are resolved server-side against `head_offset` before calling `manager.fetch()`.
2. **TCP (client-side):** `FafbClient::fetch_tail(topic, count)` resolves client-side — a lightweight probe fetch discovers `head_offset`, then a second fetch reads `head - count`. No TCP protocol changes needed.

The manager, engine, and storage layers are completely unaffected. `TopicOffset` stays `u64` internally.

### Design Decisions

**No TCP protocol changes.** Adding a capability flag (`CAP_RELATIVE_OFFSETS`) to support negative offsets on the TCP wire protocol would be permanent protocol complexity for a convenience feature. Instead:
- HTTP gets native server-side resolution (trivially backward-compatible since positive `i64` values parse identically to `u64`)
- TCP gets client-side resolution via `FafbClient::fetch_tail()` (two round-trips, but this is for CLI/debugging, not hot-path consumption)
- If SDK authors later need single-round-trip tail fetches over TCP, a capability flag can be added as a follow-up

**Clamping behavior.** If `head_offset - N < base_offset` (the resolved position was deleted by retention), `saturating_sub` clamps to 0 and the existing clamping system in `manager.fetch()` handles the rest (clamp to `base_offset`, send `OffsetNotAvailable` warning).

**Edge case: empty topic.** If `head_offset == 0` (topic has no messages), negative offsets resolve to offset 0, returning an empty response. `FafbClient::fetch_tail()` short-circuits on empty probe result.

### E.1: HTTP negative offset support

**File: `src/server/http/routes.rs`**

- [x] Change `FetchParams.offset` type from `Option<u64>` to `Option<i64>`
- [x] Add resolution logic before calling `manager.fetch()`:
  - If `offset >= 0`: convert to `TopicOffset(offset as u64)` (existing behavior)
  - If `offset < 0`: call `manager.head_offset(ns, topic)`, compute `resolved = head.0.saturating_sub(offset.unsigned_abs())`, use `TopicOffset(resolved)` — existing clamping in `manager.fetch()` handles the `< base_offset` case
  - If `offset < 0` and topic doesn't exist (`head_offset == 0`): resolves to offset 0, returns empty response

### E.2: Client-side fetch_tail

**File: `src/client.rs`**

- [x] Add `FafbClient::fetch_tail(topic, count)` method:
  - Probe: `fetch(topic, TopicOffset::ZERO, 1)` to discover `head_offset`
  - If `head == 0`: return probe result directly (empty topic)
  - Compute `start = head.saturating_sub(count)`, then `fetch(topic, start, count)`

### E.3: CLI --from negative and --tail flag

**File: `src/main.rs`, `src/cli/mod.rs`, `src/cli/fetch.rs`**

- [x] Change `FetchArgs.from` from `Option<u64>` to `Option<i64>` with `allow_hyphen_values = true`
- [x] Add `--tail <N>` flag (`tail: Option<u32>`), mutually exclusive with `--from` and `--since` via `conflicts_with`
- [x] Validation: `--tail 0` returns error ("--tail count must be > 0")
- [x] `--tail N` resolves to `fetch_tail(topic, N)` call
- [x] Negative `--from` resolves to `fetch_tail(topic, |from|)` call
- [x] Update error message to include `--tail` option

### E.4: Integration tests

**File: `tests/http_integration.rs`**

- [x] `http_fetch_negative_offset_last` — publish 5 messages, `GET /fetch?offset=-1&max_count=1`, verify returns last message (offset 4)
- [x] `http_fetch_negative_offset_last_n` — publish 20 messages, `GET /fetch?offset=-5&max_count=10`, verify returns messages from offset 15
- [x] `http_fetch_negative_offset_exceeds_depth` — publish 3 messages, `GET /fetch?offset=-100`, verify returns all 3 from offset 0
- [x] `http_fetch_negative_offset_empty_topic` — `GET /fetch?offset=-1`, verify empty 200 response

**File: `tests/deploy_integration.rs`**

- [x] `client_fetch_tail_last_message` — publish 5, `fetch_tail(1)` returns last message (offset 4)
- [x] `client_fetch_tail_last_n` — publish 20, `fetch_tail(5)` returns offsets 15–19
- [x] `client_fetch_tail_exceeds_depth` — publish 3, `fetch_tail(100)` returns all 3 from offset 0
- [x] `client_fetch_tail_empty_topic` — `fetch_tail(10)` on nonexistent topic returns empty

### E.5: Documentation updates

- [x] Update `docs/api.md` — document negative offset on HTTP fetch endpoint with examples
- [x] Update `docs/fetch_consumer_guide.md` — add "Tail / last N messages" section (4.5) with HTTP, TCP, and CLI examples
- [x] Update `CLAUDE.md` — add `FafbClient::fetch_tail()`, HTTP negative offset behavior, CLI `--tail`/`--from` negative

---

## Future Work

These items are explicitly out of scope for Plan 15 but should be considered for follow-up:

### Long-poll / `wait_timeout_ms`

When a pull consumer is caught up, it must busy-poll: fetch → 0 messages → sleep → fetch → repeat. Adding an optional `wait_timeout_ms` field to the Fetch frame would let the server hold the response until either a message arrives or the timeout expires — turning busy-polling into efficient long-polling (similar to Kafka's `fetch.max.wait.ms`).

This would change the Fetch wire format, so it's worth considering early. However, it adds complexity to the server (the fetch handler becomes stateful for the duration of the wait, holding a `Notify` listener). Deferring to a separate plan allows Plan 15 to land the core pull mechanism cleanly.

### Per-connection fetch rate limiting

If fetch abuse becomes a problem (clients hammering cold storage with rapid requests), add a `tcp.max_fetch_requests_per_second` config with token-bucket enforcement in the session layer. The `fafb_fetch_requests_total` and `fafb_fetch_cold_reads_total` metrics from Phase B.5 provide the observability needed to justify and tune this.

---

## Files Changed Summary

### Phase A — completed

| File | Changes |
|------|---------|
| `src/codec/frame.rs` | Add `Fetch`, `FetchFrom`, `FetchResponse` variants, `FetchedMessage` struct, `FETCH`/`FETCH_FROM`/`FETCH_RESPONSE` message type constants, `ErrorCode::MessageTooLarge`, `FETCH_MESSAGE_OVERHEAD` shared constant |
| `src/codec/encoder.rs` | Encode `Fetch`, `FetchFrom`, and `FetchResponse` frames |
| `src/codec/decoder.rs` | Decode `Fetch`, `FetchFrom`, and `FetchResponse` frames, validation, 13 roundtrip tests |
| `src/namespace.rs` | Add `head_offset()` to `NamespaceManager` trait (prep for Phase B) |
| `src/namespace/manager.rs` | Implement `head_offset()` on `DefaultNamespaceManager` (prep for Phase B) |
| `src/server/http/error.rs` | Map `ErrorCode::MessageTooLarge` → `"MESSAGE_TOO_LARGE"` in `error_code_name()` |

### Phase B — completed

| File | Changes |
|------|---------|
| `src/types.rs` | Add `Permission::Fetch` variant, `FetchResult` struct |
| `src/error.rs` | Add `Error::MessageTooLarge { offset, message_bytes, budget_bytes }` variant, `is_message_too_large()` helper |
| `src/auth/permissions.rs` | Add `FETCH_BIT = 0b1000` to permission bitmask, update `bit_for()` and `Display` |
| `src/codec/frame.rs` | Map `Error::MessageTooLarge` → `ErrorCode::MessageTooLarge` in `from_error()` |
| `src/namespace.rs` | Add `fetch()` to `NamespaceManager` trait with `FetchResult` import |
| `src/namespace/manager.rs` | Implement `fetch()` on `DefaultNamespaceManager` — uses `consume_from()` loop (handles hot/cold internally), frame size budget, oversized detection, `clamped_from`, 4 metrics counters/histograms, `tracing::debug!` log. 10 unit tests. |
| `src/server/tcp/session.rs` | Add `max_frame_size` to `PreAuthState` and `AuthenticatedState`; replace placeholder `Fetch`/`FetchFrom` handlers with real permission check + budget computation + `manager.fetch()` dispatch; add `handle_fetch_at_offset()` shared helper; document Fetch-during-drain decision; update `StubNamespaceManager` with `fetch()` stub |
| `src/server/tcp/connection.rs` | Pass `max_frame_size` to `Session::new()` |
| `config/config.example.toml` | Add `Fetch` to default permissions, add `batch-consumer` (Fetch-only) and `streaming-only` example namespaces |
| `config/config.sdk-test.toml` | Add `Fetch` to test namespace permissions, add `subscribe-only` namespace |
| `tests/common/mod.rs` | Add `Permission::Fetch` to default test server permissions |
| `tests/tcp_integration.rs` | 10 fetch integration tests: basic, timestamp, no-match, permission-denied, zero-count, offset-clamping, pagination, interleaved-with-subscribe, message-too-large, batch-truncated |

### Phase C — completed

| File | Changes |
|------|---------|
| `Cargo.toml` | Add `base64 = "0.22"` dependency |
| `src/config.rs` | Add `http.max_fetch_response_bytes` (default 4MB) to `HttpConfig` |
| `src/server/http.rs` | Wire `fetch` route into `/v1` router, inject `MaxFetchResponseBytes` via axum `Extension` from `HttpConfig` |
| `src/server/http/routes.rs` | Add `MaxFetchResponseBytes` newtype, `FetchParams` query struct, `GET /v1/{ns}/{topic}/fetch` handler with `offset`/`timestamp` query params, `max_count` validation (default 100, max 10,000), base64 payload encoding, `clamped_from` warning field |
| `src/server/http/error.rs` | Map `Error::MessageTooLarge` → 413 HTTP status |
| `config/config.example.toml` | Add `max_fetch_response_bytes` to `[http]` section with comment |
| `tests/common/mod.rs` | Add `start_test_http_server_all_perms()` helper (includes `Permission::Fetch`), default HTTP test server unchanged |
| `tests/http_integration.rs` | 10 HTTP fetch integration tests: basic, by_timestamp (with timestamp resolution), pagination, permission_denied, missing_offset, both_offset_and_timestamp, base64_payload, message_too_large, max_count_exceeds_limit, empty_topic_returns_200 |
| `tests/lifecycle_integration.rs` | Use `..Default::default()` for `HttpConfig` construction |
| `docs/protocol.md` | Add Fetch (0x13), FetchFrom (0x15), FetchResponse (0x14) to message type table and wire format sections; add MessageTooLarge (0x000E) to error codes; update connection lifecycle and frame dispatch |
| `docs/api.md` | Add `GET /v1/{ns}/{topic}/fetch` endpoint documentation with query params, response format, pagination, error mapping |
| `docs/architecture.md` | Update delivery model to include fetch/pull as third consumption model |
| `CLAUDE.md` | Add HTTP fetch endpoint, `MaxFetchResponseBytes`, `HttpConfig.max_fetch_response_bytes` |

### Phase D — completed

| File | Changes |
|------|---------|
| `docs/client_sdk_spec.md` | Add Fetch/FetchFrom/FetchResponse to MessageType, Frame variants, wire layouts; Permission::Fetch; ErrorCode::MessageTooLarge; state machine updates; HTTP fetch endpoint; recommended API surface; error handling; 10 TCP + 6 HTTP compat test scenarios; implementation checklist; version history v1.2 |
| `docs/fetch_consumer_guide.md` | New file: practical guide with fetch vs subscribe matrix, offset management patterns, pagination, polling, gap detection, MessageTooLarge handling, error recovery, hybrid pattern, performance, TCP vs HTTP comparison |

### Phase E — completed

| File | Changes |
|------|---------|
| `src/server/http/routes.rs` | `FetchParams.offset` changed from `Option<u64>` to `Option<i64>`. Negative offset resolution via `head_offset.saturating_sub()` before calling `manager.fetch()` |
| `src/client.rs` | Add `FafbClient::fetch_tail(topic, count)` — client-side two-step resolution (probe + fetch) |
| `src/main.rs` | `--from` changed to `i64` with `allow_hyphen_values`, `--tail <N>` flag added with `conflicts_with_all` |
| `src/cli/mod.rs` | `FetchArgs.from` → `Option<i64>`, `FetchArgs.tail: Option<u32>` added |
| `src/cli/fetch.rs` | `fetch_tail()` helper, routing for `--tail` and negative `--from` via `FafbClient::fetch_tail()` |
| `tests/http_integration.rs` | 4 HTTP negative offset tests: last, last-N, exceeds-depth, empty-topic |
| `tests/deploy_integration.rs` | 4 `FafbClient::fetch_tail` tests: last-message, last-N, exceeds-depth, empty-topic. Existing `FetchArgs` constructors updated with `tail: None` |
| `docs/api.md` | `offset` type updated to `i64`, negative offset examples and documentation added |
| `docs/fetch_consumer_guide.md` | New section 4.5 "Tail / Last N Messages" with HTTP, TCP, and CLI examples |
| `CLAUDE.md` | HTTP negative offset, `FafbClient::fetch_tail()`, CLI `--tail`/`--from` negative |
| `plans/plan_15_fetch.md` | Phase E rewritten: simplified approach (no capability flag), all tasks completed |
| `plans/plan_23_cli_tool.md` | Phase 23.2a updated with implementation notes, fetch command examples |

### Deferred

| File | Changes |
|------|---------|
| `benches/queue_bench.rs` | Fetch throughput benchmarks (hot path, cold path, varying batch sizes) — deferred to follow-up |

## Verification

```bash
# Phase A: codec compiles and tests pass
cargo test codec -- --nocapture        # ✅ 13 fetch codec tests pass
cargo clippy                           # ✅ clean

# Phase B: server compiles and all tests pass
cargo test -- --nocapture              # ✅ 518 total tests pass (0 failures)
cargo clippy                           # ✅ clean (1 pre-existing warning in queue/topic.rs)
cargo fmt -- --check                   # ✅ clean

# Phase B unit tests
cargo test namespace::manager::tests::fetch  # ✅ 10 fetch unit tests
# Phase B integration tests
cargo test --test tcp_integration tcp_fetch  # ✅ 10 fetch integration tests

# Phase C: HTTP fetch tests
cargo test --test http_integration http_fetch  # ✅ 8 fetch integration tests

# Full regression
cargo test                             # ✅ 526 pass, 0 fail
cargo fmt -- --check                   # ✅ clean
cargo clippy -- -D warnings            # 1 pre-existing derivable_impls warning only

# Phase E: negative offset / fetch_tail tests
cargo test --test http_integration http_fetch_negative  # ✅ 4 tests pass
cargo test --test deploy_integration client_fetch_tail  # ✅ 4 tests pass

# Full regression (after Phase E)
cargo test                             # ✅ 1048 pass, 0 fail
cargo fmt -- --check                   # ✅ clean
cargo clippy                           # ✅ clean
```

## Done When

### Phase A (codec) — complete
- [x] `Fetch` (0x13), `FetchFrom` (0x15), and `FetchResponse` (0x14) frames encode/decode correctly with full roundtrip tests — 13 codec tests
- [x] `ErrorCode::MessageTooLarge` (0x000E) added, non-fatal, returned when first message exceeds frame budget
- [x] `FETCH_MESSAGE_OVERHEAD` constant (20 bytes) shared between codec and manager

### Phase B (server) — complete
- [x] `Permission::Fetch` variant added, wired through config and auth — `FETCH_BIT = 0b1000`, config TOML files updated
- [x] `NamespaceManager::fetch()` reads via `consume_from()` loop (handles hot/cold internally), respects frame size budget, returns `head_offset` — 10 unit tests
- [x] `fetch()` returns `MessageTooLarge` error (not empty result) when first message exceeds budget — unit + integration tested
- [x] TCP session dispatches Fetch and FetchFrom frames, returns FetchResponse (or OffsetNotAvailable + FetchResponse when clamped) — `handle_fetch_at_offset()` shared helper
- [x] FetchFrom resolves timestamp to offset via `offset_at_timestamp()`, requires only Fetch permission
- [x] Fetch does NOT create any subscription state, delivery tasks, or inflight tracking — verified by `fetch_does_not_create_subscription_state` test
- [x] Frame size truncation works correctly — server never exceeds `max_message_size` — verified by `tcp_fetch_large_batch_truncated` test
- [x] Fetch allowed during drain (stateless, no server state created) — documented in session.rs
- [x] `config.sdk-test.toml` includes Fetch permission for the `test` namespace + `subscribe-only` namespace for permission tests
- [x] `config.example.toml` shows full, pull-only (`batch-consumer`), and legacy (`streaming-only`) permission patterns
- [x] Mock implementations updated for `fetch()` trait method — `StubNamespaceManager` in session tests
- [x] `tracing::debug!` log in `fetch()` with namespace, topic, offsets, count, clamped status
- [x] 4 metrics: `fafb_fetch_requests_total`, `fafb_fetch_messages_returned`, `fafb_fetch_cold_reads_total`, `fafb_fetch_message_too_large_total`
- [x] 10 TCP integration tests: basic, timestamp, no-match, permission-denied, zero-count, offset-clamping, pagination, interleaved-with-subscribe, message-too-large, batch-truncated
- [x] All existing tests still pass — 518 total, 0 failures

### Phase C (HTTP endpoint) — complete
- [x] HTTP `GET /v1/{ns}/{topic}/fetch` endpoint works with `offset` and `timestamp` query params, returns JSON with base64 payloads and `head_offset`
- [x] `http.max_fetch_response_bytes` config (default 4MB) added to `HttpConfig`, injected via `MaxFetchResponseBytes` axum Extension (not on `AppState`)
- [x] `Error::MessageTooLarge` → 413 HTTP status mapping
- [x] 10 HTTP integration tests: basic, by_timestamp (timestamp resolution), pagination, permission_denied, missing_offset, both_offset_and_timestamp, base64_payload, message_too_large, max_count_exceeds_limit, empty_topic_returns_200
- [x] `config/config.example.toml` updated with `max_fetch_response_bytes` in `[http]` section
- [x] `docs/protocol.md` updated with Fetch (0x13), FetchFrom (0x15), FetchResponse (0x14) wire formats, MessageTooLarge (0x000E) error code, frame dispatch table, connection lifecycle
- [x] `docs/api.md` updated with HTTP fetch endpoint documentation
- [x] `docs/architecture.md` updated with fetch/pull as third consumption model
- [x] `CLAUDE.md` updated with HTTP fetch endpoint, `MaxFetchResponseBytes`, `HttpConfig.max_fetch_response_bytes`
- [x] All existing tests still pass — 528 total, 0 failures

### Phase D (documentation) — complete
- [x] `docs/client_sdk_spec.md` updated: MessageType enum (+3), ErrorCode table (+MessageTooLarge), Permission enum (+Fetch), Frame variants table (+3), FetchedMessage/FetchBatch composite types, valid messages per state (+Fetch/FetchFrom/FetchResponse in Authenticated/Subscribed/Draining), wire protocol field layouts (Fetch 0x13, FetchResponse 0x14, FetchFrom 0x15), HTTP fetch endpoint with base64/warning/max_count notes, recommended client API surface (fetch/fetchFrom), error handling (MessageTooLarge/OffsetNotAvailable special cases), 10 TCP + 6 HTTP compat test scenarios, implementation checklist (Fetch + HTTP fetch sections), version history (v1.2)
- [x] `docs/fetch_consumer_guide.md` created: fetch vs subscribe decision matrix, 3 offset management patterns (in-memory/persistent/per-message), pagination, head_offset/has_more usage, frame size truncation with concrete numbers, MessageTooLarge handling, adaptive backoff polling, retention race/gap detection, error recovery table, push+pull hybrid pattern, performance characteristics, TCP vs HTTP comparison table

### Phase E (relative offsets) — complete
- [x] HTTP `offset` query param accepts `i64` — negative values resolve server-side to `head_offset - |offset|`
- [x] `FafbClient::fetch_tail(topic, count)` resolves client-side (probe + fetch) — no TCP protocol changes
- [x] CLI `--from` accepts negative `i64` values, `--tail <N>` convenience flag added
- [x] Clamping to `base_offset` via existing `manager.fetch()` clamping when resolved offset is below retained range
- [x] Empty topic (`head_offset == 0`) returns empty response for any negative offset / fetch_tail
- [x] 4 HTTP integration tests: last, last-N, exceeds-depth, empty-topic
- [x] 4 FafbClient integration tests: last, last-N, exceeds-depth, empty-topic
- [x] Documentation updated: `api.md`, `fetch_consumer_guide.md`, `CLAUDE.md`

### Deferred
- [ ] Fetch throughput benchmarks added (hot and cold path) — deferred to follow-up
