# Fetch Consumer Guide

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

## 1. When to Use Fetch vs. Subscribe

| 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) |

**Key difference:** Subscribe is server-driven (the server pushes messages as fast as the client can handle). Fetch is client-driven (the client requests messages when ready). Subscribe creates server-side state (delivery tasks, inflight tracking). Fetch is stateless — the server forgets about each request immediately after responding.

## 2. Offset Management Patterns

Fetch consumers are responsible for their own offset tracking. The server does not remember which offsets a fetch consumer has read.

### Pattern A: In-memory offset (simplest, at-most-once)

```python
offset = 0  # or head_offset from a previous session
while running:
    batch = client.fetch("my-topic", offset, max_count=100)
    for msg in batch.messages:
        process(msg)
    offset = batch.next_offset
```

- On restart, start from offset 0 or head
- Messages may be reprocessed or skipped on restart
- Suitable for monitoring, sampling, or idempotent workloads

### Pattern B: Persistent offset (at-least-once)

```python
offset = read_offset_from_file("my-topic")  # 0 if first run
while running:
    batch = client.fetch("my-topic", offset, max_count=100)
    for msg in batch.messages:
        process(msg)
    offset = batch.next_offset
    write_offset_to_file("my-topic", offset)  # checkpoint after batch
```

- On restart, resumes from last checkpointed offset
- Messages may be reprocessed if crash occurs between processing and checkpoint
- Requires idempotent processing or deduplication

### Pattern C: Per-message checkpointing (exactly-once with external store)

```python
offset = db.query("SELECT MAX(offset) + 1 FROM processed WHERE topic = 'my-topic'") or 0
while running:
    batch = client.fetch("my-topic", offset, max_count=100)
    for msg in batch.messages:
        with db.transaction():
            result = process(msg)
            db.insert("processed", offset=msg.offset, result=result, topic="my-topic")
    offset = batch.next_offset
```

- On restart, queries the database for the highest processed offset
- Exactly-once semantics when processing and offset storage are in the same transaction
- Higher overhead, suitable for financial or compliance workloads

## 3. Pagination and Batch Draining

To drain an entire topic, chain fetches using `next_offset`:

```typescript
let offset = 0;
while (true) {
    const batch = await client.fetch("my-topic", offset, { max_count: 100 });
    for (const msg of batch.messages) {
        await process(msg);
    }
    if (!batch.has_more) break;  // caught up: next_offset >= head_offset
    offset = batch.next_offset;
    await persistOffset(offset);  // checkpoint
}
```

The server may return fewer messages than `max_count` due to frame size limits. This is normal — always use `has_more` (or `next_offset < head_offset`) to determine if more data exists, not `messages.length < max_count`.

## 4. Using `head_offset` and `has_more`

Every `FetchResponse` includes `head_offset` — the topic's current head (next offset to be assigned). This enables:

**Progress reporting:**

```typescript
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`);
```

**Efficient poll loops:**

```typescript
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.

## 4.5. Tail / Last N Messages

To fetch the most recent messages from a topic without knowing the head offset:

**HTTP (server-side resolution):**

The HTTP fetch endpoint accepts negative offsets: `-1` = last message, `-N` = N messages from head.

```bash
# Fetch the last message
curl ".../fetch?offset=-1&max_count=1"

# Fetch the last 10 messages
curl ".../fetch?offset=-10&max_count=10"
```

The server resolves `-N` to `head_offset - N` before fetching. If `N > depth`, you get all messages from offset 0 (no error — same clamping behavior as requesting an offset below the retained range).

**TCP (client-side resolution):**

The TCP Fetch frame uses unsigned offsets (`u64`), so negative offsets are not supported on the wire. Instead, SDK clients should resolve client-side:

```
// Two-step approach
const probe = await client.fetch(topic, 0, { max_count: 1 });
const head = probe.head_offset;
const start = Math.max(0, head - N);
const result = await client.fetch(topic, start, { max_count: N });
```

This adds one extra round-trip but is only needed for tail/debugging use cases, not hot-path consumption. The Rust `FafbClient` provides a `fetch_tail(topic, count)` helper that encapsulates this pattern.

**CLI:**

```bash
# Using --tail shorthand
fafb-q fetch myns/orders --tail 10

# Equivalent: negative --from
fafb-q fetch myns/orders --from -10 --count 10
```

## 5. Frame Size Truncation Behavior

The `FetchResponse` frame must fit within the server's `max_message_size` (default 1MB for TCP, 4MB for HTTP). The server fills messages into the response until the frame is full, with per-message wire overhead of 20 bytes (8 offset + 8 timestamp + 4 payload_len).

Concrete examples with 1MB TCP frame limit:

| Message size | Messages per fetch | Notes |
|--------------|-------------------|-------|
| 1 KB | ~900 | Accounting for 20-byte overhead per message |
| 10 KB | ~100 | |
| 100 KB | ~9 | |
| 500 KB | 1 | Second message would exceed 1MB |
| 1.5 MB | 0 (`MessageTooLarge`) | Single message exceeds frame limit |

### Handling MessageTooLarge

```typescript
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}`);
    }
}
```

SDKs should offer a `skip()` convenience and document the HTTP fallback prominently.

## 6. Polling Patterns

### Adaptive backoff poll loop

```typescript
async function pollLoop(topic: string, startOffset: number) {
    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
        }
    }
}
```

### Periodic batch job

```python
# Cron job that runs every 5 minutes
offset = read_checkpoint("my-topic")
batch = client.fetch("my-topic", offset, max_count=10000)
for msg in batch.messages:
    process(msg)
write_checkpoint("my-topic", batch.next_offset)
# Exit — next invocation resumes from checkpoint
```

## 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:

```typescript
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

| Error | Action |
|-------|--------|
| **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. |
| **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). |
| **PermissionDenied** | Namespace does not grant `Fetch` permission. This is a configuration error — do not retry. |
| **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:

```typescript
// 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);
}
```

This pattern works because Fetch and Subscribe are independent — they can coexist on the same connection and use different permissions. A client with both `Fetch` and `Subscribe` permissions can freely switch between pull and push.

## 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 |

**Tuning 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.
- For hot-path reads (tailing a topic), smaller `max_count` (10-50) with tight poll intervals gives better latency.
- HTTP fetch has higher per-request overhead than TCP but supports concurrent requests naturally (one fetch per HTTP request vs. sequential on TCP).
- For parallel fetches from multiple topics, use multiple TCP connections or HTTP.

## 11. TCP vs. HTTP Fetch Comparison

| Aspect | TCP Fetch | HTTP Fetch |
|--------|-----------|------------|
| Response budget | `max_message_size` (default 1MB) | `max_fetch_response_bytes` (default 4MB) |
| `max_count` limit | No upper bound (truncated by frame size) | 1-10,000 (returns 400 if exceeded) |
| Payload encoding | Raw bytes | Base64-encoded (`payload_base64` field) |
| Offset clamping | Separate `OffsetNotAvailable` error frame before `FetchResponse` | `"warning"` field in JSON response body |
| Connection model | Persistent TCP, sequential requests per connection | Stateless HTTP, naturally concurrent |
| Overhead | Lower (binary framing, no HTTP headers) | Higher (JSON, base64 ~33% expansion, HTTP headers) |
| Best for | High-throughput consumers, latency-sensitive | Serverless, batch jobs, simple scripts, oversized message fallback |
