# Plan 13: Advanced Storage

## Goal

Build on Plan 12's tiered storage foundation with four independent enhancements: sparse offset index files for O(1) seek within segments, per-namespace disk quotas, consumer seek/rewind protocol commands, and segment merge compaction. Each phase can ship independently.

## Prerequisites

- Plan 12 complete — tiered storage with `read_at`, `read_cold_batch`, `SegmentCursor`, `disk_backed` mode, unified `delete_segments`
- Plan 12 timestamp retention complete — `u64` timestamps on `MessageEnvelope`, `RetentionPolicy`

## Dependency Graph

```
Phase A (index files) ───────────> Phase C (seek/rewind)
                                       ↑ uses index for timestamp lookup
Phase B (disk quotas) ── independent
Phase D (compaction)  ── independent, but writes .idx files if Phase A is done
```

**Recommended order**: A → B → C → D (but B and D can land in any order)

---

## Phase A: Offset Index Files (`.idx`) — COMPLETE

### Problem

`read_at` and `SegmentCursor::skip_to` scan linearly from segment start to find a target offset. For a 64MB segment with ~100-byte messages (~640K records), seeking to the middle means scanning ~320K records. This is O(n) where n is position within the segment.

### Solution

Sparse index files alongside each `.seg` file. Every N records (default 256), write a fixed-size entry mapping `(offset, byte_position, timestamp)`. Binary search the index to find the closest entry, then seek the segment file to that byte position and scan forward at most N records.

### Index file format

File naming: `{base_offset:020}.idx` alongside `{base_offset:020}.seg`

```text
// No header — consecutive fixed-size entries, 24 bytes each
[8 bytes: u64 BE offset]
[8 bytes: u64 BE byte_position]       // position within .seg file
[8 bytes: u64 BE timestamp_millis]    // from MessageEnvelope.timestamp
... repeated ...
```

For a 64MB segment with 100-byte records at interval 256: ~2500 entries × 24 bytes = 60KB. Small enough to load entirely into memory.

### A.1: `SegmentIndex` struct

**File: `src/storage/fs.rs`**

```rust
/// Sparse offset/timestamp index for a single segment file.
///
/// Entries are sorted by offset. Loaded entirely into memory for
/// binary search. Typically < 100KB per segment.
struct SegmentIndex {
    entries: Vec<IndexEntry>,
}

struct IndexEntry {
    offset: TopicOffset,
    byte_position: u64,
    timestamp_millis: u64,
}
```

Methods:
- `load(path: &Path) -> Result<Self>` — read file, parse 24-byte entries
- `find_position(offset: TopicOffset) -> Option<u64>` — binary search by offset, return byte_position of closest entry ≤ target
- `find_offset_at_timestamp(ts: u64) -> Option<(TopicOffset, u64)>` — binary search on timestamp, return (offset, byte_position) of closest entry ≤ target timestamp (used by Phase C)

### A.2: Add seek to `RecordStream`

**File: `src/storage/replay.rs`**

```rust
impl RecordStream {
    /// Seek the underlying reader to a byte position within the segment.
    /// Subsequent `next()` calls read from this position.
    pub async fn seek_to_position(&mut self, pos: u64) -> Result<()> {
        use tokio::io::AsyncSeekExt;
        self.reader.seek(std::io::SeekFrom::Start(pos)).await?;
        Ok(())
    }
}
```

`BufReader<tokio::fs::File>` supports `AsyncSeek` since `tokio::fs::File` implements it. The seek invalidates the BufReader's internal buffer, which is correct — we want to read from the new position.

### A.3: Add index tracking to `ActiveSegment`

**File: `src/storage/fs.rs`**

```rust
struct ActiveSegment {
    file: File,
    size: u64,
    base_offset: TopicOffset,
    index_file: Option<File>,          // .idx file handle
    records_since_last_index: u32,     // counter for sparse indexing
}
```

### A.4: Write index entries during `append()`

**File: `src/storage/fs.rs`** (in `append()`)

After writing a record to the segment:

```rust
let record_position = segment.size; // byte position BEFORE writing
segment.file.write_all(&record).await?;
segment.size += record.len() as u64;

// Write sparse index entry every INDEX_INTERVAL records
segment.records_since_last_index += 1;
if segment.records_since_last_index >= INDEX_INTERVAL {
    if let Some(ref mut idx_file) = segment.index_file {
        let mut entry = [0u8; 24];
        entry[..8].copy_from_slice(&msg.offset.0.to_be_bytes());
        entry[8..16].copy_from_slice(&record_position.to_be_bytes());
        entry[16..24].copy_from_slice(&msg.envelope.timestamp.to_be_bytes());
        idx_file.write_all(&entry).await?;
    }
    segment.records_since_last_index = 0;
}
```

**On segment rotation:** flush and close old `.idx`, open new `.idx` alongside the new `.seg`, reset counter.

**On init():** when opening existing segments, open the corresponding `.idx` file (create if missing).

**Constant:** `const INDEX_INTERVAL: u32 = 256;`

### A.5: Use index in `SegmentCursor::skip_to`

**File: `src/storage/fs.rs`**

Before scanning forward, load the segment's `.idx` file (if it exists), binary search for the closest entry ≤ target offset, and seek the RecordStream to that byte position. Then scan forward at most INDEX_INTERVAL records instead of the entire segment.

```rust
pub async fn skip_to(&mut self, target: TopicOffset) -> Result<()> {
    // Try to use index for fast seek
    let idx_path = self.segments[self.current_idx].1
        .with_extension("idx");
    if let Ok(index) = SegmentIndex::load(&idx_path).await {
        if let Some(byte_pos) = index.find_position(target) {
            self.stream.seek_to_position(byte_pos).await?;
        }
    }

    // Scan forward from indexed position (at most INDEX_INTERVAL records)
    // ... existing skip_to scan logic ...
}
```

### A.6: Use index in `read_at`

**File: `src/storage/fs.rs`**

Same pattern: before opening `RecordStream`, check for `.idx`, seek to closest entry, then scan forward.

### A.7: Delete `.idx` with `.seg`

**File: `src/storage/fs.rs`** (in `delete_segments()`)

When deleting a `.seg` file, also delete the corresponding `.idx` file:
```rust
let idx_path = path.with_extension("idx");
let _ = fs::remove_file(&idx_path).await; // ignore if missing
```

### A.8: Rebuild missing indexes

**File: `src/storage/fs.rs`**

Add `rebuild_index(seg_path: &Path) -> Result<()>` — scans a segment with `RecordStream`, writes `.idx` entries every INDEX_INTERVAL records. Called during `init()` for segments that lack an `.idx` file, or lazily on first read.

### A.9: Config

**File: `src/config.rs`** (StorageConfig)

```rust
/// Records between sparse index entries. Default: 256.
/// Lower values = faster seek but larger index files.
pub index_interval: u32,
```

### A.10: Tests

- `index_entry_roundtrip` — write 10 entries, load, verify all fields match
- `index_find_position_exact` — binary search finds exact offset
- `index_find_position_between` — returns closest entry ≤ target
- `index_find_position_before_all` — returns None when target < first entry
- `skip_to_uses_index` — create segment with 1000 records + index, skip_to(500), verify only ~256 records scanned (not 500)
- `read_at_uses_index` — read_at with index is correct
- `append_writes_index_entries` — append 1000 records, verify `.idx` file has ~4 entries (1000/256)
- `delete_segments_removes_idx` — delete segment, verify `.idx` also deleted
- `rebuild_index_matches_append` — rebuild index from segment, compare to index written during append

---

## Phase B: Per-Namespace Disk Quotas — COMPLETE

### Problem

Global `disk_capacity_threshold_pct` means one namespace can consume all disk space, starving others. Operators need per-namespace byte limits.

### Solution

Add `max_disk_bytes: Option<u64>` to `NamespaceConfig`. Track disk bytes per namespace incrementally in `FsStorage`. Reject publishes when a namespace exceeds its quota.

### B.1: Config

**File: `src/config.rs`** (NamespaceConfig)

```rust
/// Maximum total bytes on disk for this namespace's segments.
/// None means unlimited (subject to global disk threshold only).
#[serde(default)]
pub max_disk_bytes: Option<u64>,
```

### B.2: Track disk bytes in `FsStorage`

**File: `src/storage/fs.rs`**

Add field:
```rust
pub struct FsStorage {
    // ... existing ...
    /// Tracked disk bytes per namespace, updated incrementally.
    namespace_disk_bytes: RwLock<HashMap<String, AtomicU64>>,
}
```

Update in three places:
1. **`init()`** — scan all `.seg` and `.idx` files, sum sizes per namespace
2. **`append()`** — add record byte count + index entry bytes (when written) to namespace counter
3. **`delete_segments()`** — stat `.seg` and `.idx` files before deleting, subtract sizes from counter

### B.3: Expose via `StorageBackend`

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

```rust
/// Return the total bytes on disk for a namespace's segments.
fn namespace_disk_bytes(&self, namespace: &NamespaceId) -> u64;
```

`MockStorage`: return 0. `FsStorage`: read from cached counter.

### B.4: Expose via `QueueEngine`

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

```rust
fn namespace_disk_bytes(&self, ns: &NamespaceId) -> u64;
```

`InMemoryQueueEngine`: delegate to `self.storage.namespace_disk_bytes(ns)`.

### B.5: Enforce in publish path

**File: `src/queue/engine.rs`** (publish method)

The engine doesn't hold namespace config, so pass the quota limit through. Two approaches:

**Option A**: Add `namespace_quotas: HashMap<NamespaceId, u64>` to `EngineConfig` — built from config at startup.

**Option B**: Check at the namespace manager level (manager holds config).

Option B is cleaner — the engine remains config-agnostic:

**File: `src/namespace/manager.rs`** (publish method)

Before calling `engine.publish()`:
```rust
if let Some(max_bytes) = ns_config.max_disk_bytes {
    let current = self.engine.namespace_disk_bytes(ns);
    if current + payload_len > max_bytes {
        return Err(Error::ResourceExhausted(
            format!("namespace disk quota exceeded ({current} bytes, limit {max_bytes})")
        ));
    }
}
```

### B.6: Metrics

- `fafb_namespace_disk_bytes` gauge (labels: `namespace`) — set on each append, updated on delete
- `fafb_namespace_disk_quota_rejected_total` counter (labels: `namespace`) — on quota rejection

### B.7: Config example

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

```toml
# max_disk_bytes = 10737418240        # 10 GB per namespace
```

### B.8: Tests

- `namespace_disk_bytes_tracked_on_append` — append 10 messages, verify counter matches total record bytes
- `namespace_disk_bytes_decremented_on_delete` — append, delete segments, verify counter decreased
- `namespace_disk_bytes_populated_on_init` — create segments manually, init(), verify counter matches file sizes
- `publish_rejected_when_quota_exceeded` — set `max_disk_bytes = 1000`, publish until rejected
- `publish_succeeds_when_under_quota` — set `max_disk_bytes = 1_000_000`, publish small messages, all succeed
- `quota_none_means_unlimited` — no `max_disk_bytes`, publish large amount, never rejected

---

## Phase C: Consumer Seek/Rewind Protocol

### Problem

Consumers can only subscribe at the current head offset. There's no way to replay from a specific offset or timestamp (e.g., "replay from 1 hour ago"). This prevents debugging, reprocessing, and consumer recovery after extended downtime.

### Solution

Add two new frame types: `SubscribeAt` (seek to offset) and `SubscribeFrom` (seek to timestamp). Add `start_offset` to `SubscribeAck` so the client knows where it's starting.

### C.1: New frame types

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

```rust
/// Subscribe starting from a specific offset.
SubscribeAt {
    topic: TopicName,
    offset: TopicOffset,
    group: Option<GroupName>,
},                                     // 0x0E

/// Subscribe starting from the first message at or after a timestamp.
SubscribeFrom {
    topic: TopicName,
    timestamp_millis: u64,
    group: Option<GroupName>,
},                                     // 0x0F
```

Update `SubscribeAck` to include starting offset:
```rust
SubscribeAck {
    topic: TopicName,
    group: Option<GroupName>,
    start_offset: TopicOffset,         // NEW — actual starting offset
},                                     // 0x04 (same type byte, new field)
```

### C.2: Encoder/decoder updates

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

`SubscribeAt` wire format:
```text
[1 byte: 0x0E]
[2 bytes: u16 BE topic_len] [N bytes: topic]
[8 bytes: u64 BE offset]
[2 bytes: u16 BE group_len] [M bytes: group]
```

`SubscribeFrom` wire format:
```text
[1 byte: 0x0F]
[2 bytes: u16 BE topic_len] [N bytes: topic]
[8 bytes: u64 BE timestamp_millis]
[2 bytes: u16 BE group_len] [M bytes: group]
```

`SubscribeAck` updated wire format:
```text
[1 byte: 0x04]
[2 bytes: u16 BE topic_len] [N bytes: topic]
[8 bytes: u64 BE start_offset]        ← NEW
[2 bytes: u16 BE group_len] [M bytes: group]
```

**File: `src/codec/decoder.rs`** — decode the new frame types and updated SubscribeAck.

### C.3: Timestamp-to-offset lookup

**File: `src/storage.rs`** (StorageBackend trait)

```rust
/// Find the offset of the first message at or after `timestamp_millis`.
/// Returns None if no message exists at or after the timestamp.
fn offset_at_timestamp(
    &self,
    namespace: &NamespaceId,
    topic: &TopicName,
    timestamp_millis: u64,
) -> impl Future<Output = Result<Option<TopicOffset>>> + Send;
```

**FsStorage implementation** (depends on Phase A):
1. List segments sorted by base offset
2. Use segment mtime to narrow down candidate segments (skip segments whose mtime < timestamp)
3. For the candidate segment, load its `.idx` file
4. Binary search on `timestamp_millis` field to find closest entry
5. Open `RecordStream` at that byte position, scan forward to find exact first record with `timestamp >= target`
6. Return that record's offset

**Fallback (no `.idx`):** scan segment from start, checking each record's timestamp. Slow but correct.

**MockStorage:** linear scan of in-memory messages by timestamp.

### C.4: Manager `subscribe_at` and `subscribe_from`

**File: `src/namespace.rs`** (NamespaceManager trait)

```rust
fn subscribe_at(
    &self, ns, topic, client, group,
    start_offset: TopicOffset,
) -> impl Future<Output = Result<(mpsc::Receiver<PublishedMessage>, TopicOffset)>> + Send;

fn subscribe_from(
    &self, ns, topic, client, group,
    timestamp_millis: u64,
) -> impl Future<Output = Result<(mpsc::Receiver<PublishedMessage>, TopicOffset)>> + Send;
```

Both return the actual starting offset alongside the receiver (for the SubscribeAck response).

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

`subscribe_at`: same as `subscribe` but sets `next_offset = start_offset` instead of `head_offset`. If `start_offset > head_offset`, clamp to head. If `start_offset` is below the oldest available offset (not on disk), start at oldest available and log a warning.

`subscribe_from`: calls `storage.offset_at_timestamp()` (via engine), then calls `subscribe_at` with the result. If no offset found, falls back to `head_offset`.

### C.5: TCP session handling

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

Add match arms for `Frame::SubscribeAt` and `Frame::SubscribeFrom`:

```rust
Frame::SubscribeAt { topic, offset, group } => {
    // Permission check (same as Subscribe)
    // Call manager.subscribe_at(ns, &topic, client_id, group, offset)
    // Return SubscribeAck { topic, group, start_offset }
}
Frame::SubscribeFrom { topic, timestamp_millis, group } => {
    // Permission check
    // Call manager.subscribe_from(ns, &topic, client_id, group, timestamp_millis)
    // Return SubscribeAck { topic, group, start_offset }
}
```

Also update the existing `Frame::Subscribe` handler to return `start_offset` in the new SubscribeAck format.

### C.6: Update existing Subscribe handler

The existing `subscribe()` method should also return the starting offset so the SubscribeAck can include it. Update its return type to `Result<(mpsc::Receiver<PublishedMessage>, TopicOffset)>`.

### C.7: Update protocol docs

**File: `docs/protocol.md`** — add SubscribeAt (0x0E) and SubscribeFrom (0x0F) frame specs.

**File: `docs/client_sdk_spec.md`** — add new frame types, update SubscribeAck format.

### C.8: Tests

**Codec tests:**
- `encode_decode_subscribe_at` — roundtrip with offset=42
- `encode_decode_subscribe_from` — roundtrip with timestamp
- `encode_decode_subscribe_ack_with_offset` — roundtrip with start_offset
- Update proptest `arb_frame()` to generate new frame variants

**Manager tests:**
- `subscribe_at_starts_from_offset` — subscribe at offset 50, verify first delivered message is at offset 50
- `subscribe_at_clamps_to_head` — subscribe at offset past head, starts at head
- `subscribe_at_below_available` — subscribe at offset 0 when oldest on disk is 100, starts at 100
- `subscribe_from_finds_timestamp` — publish messages with known timestamps, subscribe_from a timestamp in the middle, verify starting offset matches

**Integration tests:**
- `tcp_subscribe_at_receives_from_offset` — full TCP roundtrip with SubscribeAt frame
- `tcp_subscribe_from_receives_from_timestamp` — full TCP roundtrip with SubscribeFrom frame

---

## Phase D: Segment Compaction (Merge)

### Problem

Segment rotation can produce many small segments (e.g., during low-traffic periods, topic with bursty writes, or after retention deletes). Many small files increase fd pressure, slow down segment listing, and make cold reads less efficient (more segment boundary crossings).

### Solution

Background merge compaction: find runs of consecutive small segments, read their records, write them into a single merged segment, delete the originals. No data loss — records are preserved, just reorganized into fewer files.

> **Scope note:** This plan covers merge compaction only (combining small segments). Ack-based compaction (removing fully-acknowledged messages) is a separate concern with significantly more complexity (tracking per-consumer ack state) and is deferred.

### D.1: Config

**File: `src/config.rs`** (StorageConfig)

```rust
/// Merge consecutive segments smaller than this byte count.
/// Default: segment_max_bytes / 4 (16MB for 64MB segments).
/// Set to 0 to disable merge compaction.
pub segment_merge_threshold_bytes: u64,

/// Interval in seconds between compaction sweeps. Default: 300 (5 min).
pub compaction_interval_secs: u64,
```

### D.2: `compact_segments` on `StorageBackend`

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

```rust
/// Merge runs of consecutive small segments for a topic.
/// Returns the number of merge operations performed (each merging 2+ segments).
fn compact_segments(
    &self,
    namespace: &NamespaceId,
    topic: &TopicName,
    merge_threshold_bytes: u64,
) -> impl Future<Output = Result<usize>> + Send;
```

### D.3: `FsStorage::compact_segments`

**File: `src/storage/fs.rs`**

Algorithm:
1. List segments sorted by base offset
2. Find runs of consecutive segments where each is `< merge_threshold_bytes`
3. For each run of 2+ segments:
   a. Create a new merged segment file at a temp path (base_offset = first segment's base_offset)
   b. Open `RecordStream` on each segment in the run, write all records to the merged file
   c. If Phase A is done, write a new `.idx` file for the merged segment
   d. Atomically rename temp file → final path (replaces first segment)
   e. Delete the remaining original segment files (and their `.idx` files)
   f. Update `segment_index` cache
4. Update `namespace_disk_bytes` if Phase B tracking is active (merged size ≤ sum of originals due to no overhead)

**Crash safety:**
- Write merged segment first, then delete originals
- If crash after write, before delete: duplicate records on disk — `replay_topic` handles this by deduplicating on offset (records are sorted, duplicates are adjacent)
- If crash during write: partial merged segment — `replay_segment` truncates at corruption boundary, originals still intact

**Concurrency:**
- Don't merge segments that might have active `SegmentCursor` readers. Since cursors hold a snapshot of the segment index, they'll reference old paths. On I/O error (file deleted), the delivery task skips forward — acceptable but noisy.
- Alternative: merge only segments older than `min_live_offset` (subscribers have moved past them). This is safe and simple.

### D.4: Spawn compaction task

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

After the cleanup task, spawn a compaction task:
```rust
let compaction_handle = tokio::spawn({
    let storage = Arc::clone(&storage);
    let manager = Arc::clone(&manager);
    let mut shutdown_rx = shutdown_tx.subscribe();
    let merge_threshold = config.storage.segment_merge_threshold_bytes;
    let interval_secs = config.storage.compaction_interval_secs;
    async move {
        let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
        interval.tick().await;
        loop {
            tokio::select! {
                _ = shutdown_rx.changed() => break,
                _ = interval.tick() => {
                    if merge_threshold == 0 { continue; }
                    let topics = storage.list_topics().await.unwrap_or_default();
                    for (ns, topic) in &topics {
                        match storage.compact_segments(ns, topic, merge_threshold).await {
                            Ok(n) if n > 0 => {
                                metrics::counter!("fafb_segments_merged_total",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                ).increment(n as u64);
                            }
                            Err(e) => tracing::warn!(namespace = %ns, topic = %topic,
                                error = %e, "compaction failed"),
                            _ => {}
                        }
                    }
                }
            }
        }
    }
});
```

### D.5: Metrics

- `fafb_segments_merged_total` counter (labels: `namespace`, `topic`)
- `fafb_compaction_sweep_duration_seconds` histogram

### D.6: Tests

- `compact_segments_merges_small_segments` — create 5 segments of 100 bytes each, threshold=1000, verify merged into 1
- `compact_segments_skips_large_segments` — create 3 large + 2 small segments, verify only small ones merged
- `compact_segments_preserves_records` — merge segments, replay topic, verify all records present and in order
- `compact_segments_noop_when_none_small` — all segments above threshold, returns 0
- `compact_segments_writes_idx` — after merge, `.idx` file exists for merged segment (if Phase A is done)
- `compact_segments_updates_index_cache` — segment index cache reflects merged state

---

## Files Changed Summary

| File | Phases | Changes |
|------|--------|---------|
| `src/storage/fs.rs` | A, B, D | SegmentIndex, index writes in append, seek in skip_to/read_at, namespace_disk_bytes tracking, compact_segments |
| `src/storage/replay.rs` | A | `RecordStream::seek_to_position` |
| `src/storage.rs` | B, C, D | `namespace_disk_bytes`, `offset_at_timestamp`, `compact_segments` on trait + MockStorage |
| `src/config.rs` | A, B, D | `index_interval`, `max_disk_bytes`, `segment_merge_threshold_bytes`, `compaction_interval_secs` |
| `src/codec/frame.rs` | C | `SubscribeAt`, `SubscribeFrom` variants, `start_offset` on `SubscribeAck` |
| `src/codec/encoder.rs` | C | Encode new frame types + updated SubscribeAck |
| `src/codec/decoder.rs` | C | Decode new frame types + updated SubscribeAck |
| `src/namespace.rs` | C | `subscribe_at`, `subscribe_from` on trait |
| `src/namespace/manager.rs` | B, C | Quota check in publish, subscribe_at/subscribe_from impl |
| `src/queue.rs` | B | `namespace_disk_bytes` on QueueEngine trait |
| `src/queue/engine.rs` | B | Delegate `namespace_disk_bytes` to storage |
| `src/server/tcp/session.rs` | C | Handle SubscribeAt/SubscribeFrom frames |
| `src/main.rs` | D | Spawn compaction task |
| `config/config.example.toml` | A, B, D | New config fields |
| `docs/protocol.md` | C | New frame type documentation |
| `docs/client_sdk_spec.md` | C | SDK type updates |

## Implementation Order

```
Phase A: Index files                — COMPLETE
    A.1  SegmentIndex struct ✓
    A.2  RecordStream::seek_to_position + position() tracking ✓
    A.3  ActiveSegment index tracking ✓
    A.4  Write index entries in append() (first record + every interval) ✓
    A.5  Use index in skip_to ✓
    A.6  Use index in read_at (with SegmentIndex cache) ✓
    A.7  Delete .idx with .seg (+ cache eviction) ✓
    A.8  Rebuild missing indexes (using RecordStream.position()) ✓
    A.9  Config (index_interval) ✓
    A.10 Tests ✓

Phase B: Per-namespace disk quotas  — COMPLETE
    B.1  Config (max_disk_bytes on NamespaceConfig + NamespaceQuota) ✓
    B.2  Track bytes in FsStorage (init/append/delete_segments) ✓
    B.3  Expose via StorageBackend trait ✓
    B.4  Expose via QueueEngine trait ✓
    B.5  Enforce in publish (manager-level check) ✓
    B.6  Metrics (gauge + rejection counter) ✓
    B.7  Config example ✓
    B.8  Tests ✓

Phase C: Consumer seek/rewind      — COMPLETE
    C.1  SubscribeStart enum + Frame types (SubscribeAt 0x0E, SubscribeFrom 0x0F) ✓
    C.2  Encoder/decoder + SubscribeAck start_offset field ✓
    C.3  offset_at_timestamp on StorageBackend + QueueEngine ✓
    C.4  Single subscribe() with SubscribeStart param (not 3 methods) ✓
    C.5  TCP session handle_subscribe shared handler ✓
    C.6  ErrorCode::OffsetNotAvailable (0x000C) ✓
    C.7  Protocol docs + CLAUDE.md updated ✓
    C.8  Tests (codec roundtrips, storage timestamp lookup, manager seek, proptest) ✓

Phase D: Segment compaction         — COMPLETE
    D.1  Config (segment_merge_threshold_bytes, compaction_interval_secs) ✓
    D.2  compact_segments on StorageBackend trait + MockStorage ✓
    D.3  FsStorage implementation (scan, filter, merge, inline idx, atomic rename, cache update) ✓
    D.4  Spawn compaction task in main.rs ✓
    D.5  Metrics (fafb_segments_merged_total, fafb_compaction_sweep_duration_seconds) ✓
    D.6  Tests (12 tests: merge, skip-large, preserve, noop, idx, safe_below, nonexistent, tmp cleanup, LRU eviction, error handling, disk bytes, candidate exclusion) ✓
```

## Verification

After each phase:
```bash
cargo fmt -- --check && cargo clippy && cargo test
```

## Done When

- ~~`.idx` files written alongside `.seg` files during append, used by `skip_to`/`read_at` for O(log n) seek~~ ✓ (Phase A)
- ~~Per-namespace `max_disk_bytes` enforced on publish, tracked incrementally~~ ✓ (Phase B)
- ~~`SubscribeAt` and `SubscribeFrom` protocol commands work end-to-end~~ ✓ (Phase C)
- ~~Small segments automatically merged by background task~~ ✓ (Phase D)
- ~~All existing tests pass, new tests cover each phase~~ ✓
- ~~Protocol and SDK docs updated for new frame types~~ ✓ (Phase C)
