# Plan 12: Tiered Storage — Overflow to Disk

## Goal

Allow topics to accept publishes as long as disk has space, using the in-memory buffer as a hot cache over a larger on-disk log. Consumers that fall behind the in-memory window read transparently from disk segments. The operator configures a disk capacity threshold (default 90%); publishes are only rejected when disk is full.

## Status: COMPLETE

All phases (A through H) have been implemented. See git history for implementation details.

## Problem Statement

Today the system has two independent failure modes that both reject publishes:

1. **Per-topic limits** (`max_depth`, `max_bytes`) — `DropOldest` evicts from memory but the evicted messages become invisible to slow consumers because `consume_from` only reads the VecDeque.
2. **Global memory budget** (`global_max_memory_bytes`) — hard reject via `ResourceExhausted`, even though the message could safely land on disk.

The on-disk segment files already contain every published message, but there is no read path back to them after in-memory eviction. This plan bridges that gap.

There is also a **pre-existing bug** in delivery tasks: when `consume_from` returns `None` for an offset below `base_offset` (evicted by DropOldest), the delivery task treats it as "caught up" and waits for a `Notify`. It wakes on the next publish, retries the same stale offset, gets `None` again — a spin loop where the consumer is permanently stuck. This plan fixes that as Step 1.

## Design Principles

- **The VecDeque is a cache, not the log.** The segment files on disk are the authoritative log. The in-memory buffer is a read-ahead window for fast consumers.
- **Cold reads are explicit and metered.** Disk fallback is a separate code path with its own metrics so operators can see when consumers fall behind the hot tier.
- **Sequential cold reads use cursors, not point lookups.** A consumer catching up reads messages sequentially. The cold path keeps a file handle open across reads instead of reopening and rescanning per message.
- **Disk-backed mode is an engine-level config, not a backpressure variant.** `BackpressurePolicy` controls topic-level buffer eviction. Disk-backed mode is an engine-level decision about whether to reject on memory budget or fall through to disk. Keeping them separate avoids a leaky abstraction where a topic-level enum controls engine-level behavior.
- **No new crate dependencies for disk monitoring.** Use `libc::statvfs` (already a transitive dependency) behind a `#[cfg(unix)]` gate.
- **Segment deletion respects consumers.** Segments cannot be deleted while any consumer's cursor points into them.
- **Existing behavior is the default.** Disk-backed mode is opt-in. `Reject` and `DropOldest` behave exactly as before.

## Architecture

```
                    publish
                      │
                      ▼
         ┌────────────────────────┐
         │ disk_backed = true?    │
         │  yes → check disk pct │── over threshold? → reject
         │  no  → check RAM      │── over budget? → reject
         └────────────┬──────────┘
                      │
          ┌───────────▼───────────┐
          │  Topic::publish()     │
          │  DropOldest for memory│
          └───────────┬───────────┘
                      │
          ┌───────────▼───────────┐
          │  storage.append()     │── always succeeds (page cache)
          └───────────────────────┘

                   consume
                      │
                      ▼
          ┌───────────────────────┐
          │  Topic::consume_from  │── hit? → return (hot path)
          └───────────┬───────────┘
                      │ miss (offset < base_offset)
                      ▼
          ┌───────────────────────┐
          │  SegmentCursor::next  │── sequential disk read (cold path)
          │  (keeps fd open)      │   cursor advances, no rescan
          └───────────┬───────────┘
                      │ miss (segment deleted)
                      ▼
          ┌───────────────────────┐
          │  Skip to base_offset  │── log warning with gap size
          │  (data loss recovery) │
          └───────────────────────┘
```

---

## Execution Plan

### Phase A — Delivery Task Skip-Forward Fix (step 1) — **DONE**

Fix the pre-existing bug before introducing any new storage behavior. This is independent of tiered storage and makes the delivery task robust for all backpressure policies.

**Implemented**: Added `base_offset()` to `QueueEngine` trait, both individual and group delivery tasks now check `offset < base_offset` when `consume_from` returns `None` and skip forward with a warning + metric. Test: `delivery_skips_evicted_offsets`.

#### Step 1: Delivery task handles evicted offsets

Currently in `src/namespace/manager.rs`, both individual and group delivery tasks treat `Ok(None)` as "caught up — wait for notify." This is only correct when `offset >= head_offset`. When `offset < base_offset`, the message was evicted and the consumer is stuck.

Modify both delivery task loops to distinguish the two cases. Change `consume_from` to return enough context, or read `base_offset` from the topic state (already available in the delivery task's `topic_state` lock):

```rust
Ok(None) => {
    // Check if we're behind the buffer (evicted) or caught up (at head)
    let should_skip = {
        let state = topic_state.lock().await;
        if let Some(sub) = state.individual.get(&client) {
            // Will be refined in Phase B to try disk first
            sub.next_offset < engine_base_offset
        } else {
            break;
        }
    };
    if should_skip {
        // Data was evicted — skip forward to what's available
        let mut state = topic_state.lock().await;
        if let Some(sub) = state.individual.get_mut(&client) {
            let gap = engine_base_offset.0 - sub.next_offset.0;
            tracing::warn!(
                namespace = %ns, topic = %topic, client = %client,
                gap = gap, new_offset = %engine_base_offset,
                "consumer skipping evicted offsets"
            );
            sub.next_offset = engine_base_offset;
        }
        continue;
    }
    // Caught up — wait for notification or shutdown
    tokio::select! { ... }
}
```

Apply the same pattern to the group delivery task.

To get `engine_base_offset` without adding a new engine method, the delivery task can call `engine.consume_from(ns, topic, TopicOffset(0))` — but that's wasteful. Cleaner: add `QueueEngine::base_offset(ns, topic) -> Result<TopicOffset>` alongside the existing `head_offset`. Or: store the base_offset in the `DeliveryTaskContext` and update it when eviction is detected. The simplest option: add `base_offset` to `QueueEngine` trait — it mirrors `head_offset` and the Topic already exposes it.

**Files**: `src/queue.rs`, `src/queue/engine.rs`, `src/namespace/manager.rs`

**Tests**: Unit test in `manager.rs` — publish 100 messages to a topic with `max_depth=10` and `DropOldest`, subscribe a client, verify it receives the 10 most recent messages (not stuck at offset 0).

---

### Phase B — Storage Point-Read & Cursor (steps 2–5)

The foundation: give the engine a way to read messages from disk, optimized for sequential access.

#### Step 2: `RecordStream` — shared record parser as iterator

Extract the record-parsing logic from `storage/replay.rs` into a reusable async iterator. Both `replay_segment` and the new cold-read path use the same on-disk format.

Add to `src/storage/replay.rs`:

```rust
/// Streaming iterator over records in a segment file.
///
/// Reads one record at a time via `next()`. Stops at EOF or first
/// corruption (same recovery semantics as `replay_segment`).
pub struct RecordStream {
    reader: BufReader<File>,
}

impl RecordStream {
    pub async fn open(path: &Path) -> Result<Self>;

    /// Read the next record. Returns `None` at EOF or corruption.
    pub async fn next(&mut self) -> Result<Option<(TopicOffset, MessageEnvelope)>>;
}
```

Refactor `replay_segment` to use `RecordStream` internally — no behavior change, just deduplication.

#### Step 3: `StorageBackend::read_at` trait method

Add to `src/storage.rs`:

```rust
/// Read a single message by offset from persisted segments.
///
/// Returns `Ok(None)` if the offset is not found on disk (deleted by
/// segment rotation or never persisted). This is the cold-path
/// counterpart to the in-memory `Topic::consume_from`.
fn read_at(
    &self,
    namespace: &NamespaceId,
    topic: &TopicName,
    offset: TopicOffset,
) -> impl Future<Output = Result<Option<PublishedMessage>>> + Send;
```

**MockStorage**: linear scan of the in-memory `Vec<PublishedMessage>` — trivial.

#### Step 4: `FsStorage::read_at` implementation + segment cache

Add to `src/storage/fs.rs`:

1. **Segment index cache**: `segment_index: RwLock<HashMap<TopicKey, Vec<(TopicOffset, PathBuf)>>>` — sorted list of `(base_offset, path)` per topic. Populated on `init()` from the directory scan that already runs. Updated on segment rotation in `append()` (add new segment) and in `cleanup_segments()` (remove old entries).

2. **`find_segment_for_offset`**: binary search the cached index for the entry with the highest base_offset ≤ target offset. No directory listing syscall — pure in-memory lookup.

3. **`read_at` implementation**: look up the segment via `find_segment_for_offset`, open with `RecordStream`, scan until offset found or EOF. Return `Ok(None)` if not found.

This replaces the standalone `find_segment_for_offset` function from the original plan with a cached approach that avoids per-read directory listings.

#### Step 5: `SegmentCursor` for sequential cold reads

Add to `src/storage/fs.rs`:

```rust
/// Holds an open file handle for sequential cold reads within a segment.
///
/// The delivery task creates a cursor when it first hits the cold path
/// and advances it with `read_next()`. When the cursor reaches the end
/// of a segment, it automatically opens the next one. Avoids the cost
/// of segment lookup + file open + linear scan per message.
pub struct SegmentCursor {
    stream: RecordStream,
    current_segment_base: TopicOffset,
    topic_key: TopicKey,
    seg_dir: PathBuf,
}

impl SegmentCursor {
    /// Create a cursor positioned at (or just before) `start_offset`.
    pub async fn open(
        seg_dir: &Path,
        topic_key: TopicKey,
        start_offset: TopicOffset,
        segment_index: &[(TopicOffset, PathBuf)],
    ) -> Result<Option<Self>>;

    /// Read the next record. Automatically crosses segment boundaries.
    /// Returns `None` when no more segments or data exist on disk.
    pub async fn read_next(&mut self) -> Result<Option<PublishedMessage>>;
}
```

The engine can optionally cache one `SegmentCursor` per topic for the cold path. On first cold read, create the cursor. On subsequent sequential reads, call `read_next()` — O(1) per message with no reopening or rescanning.

If a publish happens between cold reads (advancing `head_offset`), the cursor remains valid — it reads from older segments and the new publish goes to the active segment. If compaction deletes the cursor's segment, the next `read_next()` returns an I/O error; the delivery task skips forward (Step 1's recovery).

**Files**: `src/storage.rs`, `src/storage/fs.rs`, `src/storage/replay.rs`

---

### Phase C — Engine Consume Fallback (steps 6–7)

Wire the point-read into the engine so `consume_from` transparently falls back to disk.

#### Step 6: `InMemoryQueueEngine::consume_from` disk fallback

Modify `src/queue/engine.rs`:

```rust
async fn consume_from(&self, ns, topic, offset) -> Result<Option<PublishedMessage>> {
    let map = self.topics.read().await;
    let Some(topic_lock) = map.get(&(ns.clone(), topic.clone())) else {
        return Ok(None);
    };

    // Hot path: read from in-memory buffer
    let (result, base_offset, next_offset) = {
        let t = topic_lock.lock().await;
        (
            t.consume_from(offset).map(|env| PublishedMessage {
                envelope: env.clone(),
                offset,
            }),
            t.base_offset(),
            t.head_offset(),
        )
    };
    drop(map); // release read lock before disk I/O

    if result.is_some() {
        return Ok(result);
    }

    // Cold path: offset was evicted from memory but may exist on disk
    if offset < base_offset && offset < next_offset {
        metrics::counter!("fafb_cold_reads_total",
            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
        ).increment(1);

        let cold_start = std::time::Instant::now();
        let disk_result = self.storage.read_at(ns, topic, offset).await?;
        metrics::histogram!("fafb_cold_read_latency_seconds",
            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
        ).record(cold_start.elapsed().as_secs_f64());

        return Ok(disk_result);
    }

    Ok(None)
}
```

**Key detail**: the topic lock is released before the disk I/O call. This ensures disk reads for slow consumers don't block publishes or fast consumers on the same topic.

#### Step 7: Update delivery task to use skip-forward with disk awareness

Refine Step 1's skip-forward logic. Now that `consume_from` can return `Some` from disk, the `Ok(None)` case in the delivery task means either "caught up" or "segment deleted." Update the delivery task:

```rust
Ok(None) => {
    if offset >= next_offset {
        // Caught up — wait for notification
        tokio::select! { ... }
    } else {
        // offset < base_offset AND not on disk (segment deleted)
        // Skip forward to base_offset — these messages are lost
        let mut state = topic_state.lock().await;
        if let Some(sub) = state.individual.get_mut(&client) {
            let gap = base_offset.0 - sub.next_offset.0;
            tracing::warn!(
                namespace = %ns, topic = %topic, client = %client,
                gap = gap, "skipping deleted offsets"
            );
            metrics::counter!("fafb_messages_skipped_total",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone()
            ).increment(gap);
            sub.next_offset = base_offset;
        }
        continue;
    }
}
```

The delivery task needs `base_offset` and `next_offset` from the engine. Add `QueueEngine::base_offset(ns, topic) -> Result<TopicOffset>` (mirrors existing `head_offset`). Read both in a single lock scope at the top of the delivery loop (where `offset` and `notified_future` are already read), so no extra locking.

**Files**: `src/queue/engine.rs`, `src/queue.rs`, `src/namespace/manager.rs`

---

### Phase D — Disk-Backed Mode (steps 8–10)

The engine-level config that makes memory a cache and disk the acceptance boundary.

#### Step 8: `disk_backed` engine config

Add to `EngineConfig` in `src/queue/engine.rs`:

```rust
pub struct EngineConfig {
    // ... existing ...
    /// When true, publishes are accepted as long as disk has space.
    /// The global memory budget becomes advisory (DropOldest keeps memory
    /// bounded) and the disk capacity threshold replaces it as the
    /// acceptance gate.
    pub disk_backed: bool,                    // default: false
    /// Reject publishes when disk usage exceeds this percentage.
    /// Only applies when `disk_backed = true`.
    pub disk_capacity_threshold_pct: f64,     // default: 90.0
}
```

This is NOT a `BackpressurePolicy` variant. When `disk_backed = true`:
- Per-topic backpressure is forced to `DropOldest` (memory eviction, never reject)
- The global `global_max_memory_bytes` check is skipped (memory is a cache)
- A disk capacity check gates publish acceptance instead
- `consume_from` uses the cold path from Phase C

When `disk_backed = false` (default): all existing behavior is unchanged.

Config file:
```toml
[queue]
disk_backed = false                    # NEW — opt-in
disk_capacity_threshold_pct = 90       # NEW
```

Env var: `FAFB_DISK_BACKED=true`, `FAFB_DISK_CAPACITY_THRESHOLD_PCT=90`.

Validation: `disk_capacity_threshold_pct` must be in `(0.0, 100.0)`. If `disk_backed = true` and storage backend is `MockStorage` (tests), skip disk checks.

#### Step 9: Disk capacity monitoring

Add `src/storage/disk.rs` (new file):

```rust
/// Query available disk space using statvfs (Unix).
#[cfg(unix)]
pub fn disk_usage(path: &std::path::Path) -> std::io::Result<DiskUsage> {
    let c_path = std::ffi::CString::new(
        path.to_str().ok_or(std::io::ErrorKind::InvalidInput)?
    )?;
    let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
    if unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) } != 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(DiskUsage {
        available_bytes: stat.f_bavail as u64 * stat.f_frsize as u64,
        total_bytes: stat.f_blocks as u64 * stat.f_frsize as u64,
    })
}

pub struct DiskUsage {
    pub available_bytes: u64,
    pub total_bytes: u64,
}
```

Cache the result in the engine as `AtomicU32` storing integer percentage (0–100). Updated by the background flush task on each cycle (every `flush_interval_ms`, default 100ms):

```rust
// In InMemoryQueueEngine:
disk_used_pct: AtomicU32,  // 0–100 integer percentage

fn disk_has_capacity(&self) -> bool {
    let pct = self.disk_used_pct.load(Relaxed);
    (pct as f64) < self.disk_capacity_threshold_pct
}
```

The flush task already runs periodically; add the `disk_usage` call after flushing all topics. No new background task needed.

#### Step 10: Engine publish changes for disk-backed mode

In `InMemoryQueueEngine::publish`:

```rust
if self.disk_backed {
    // Disk-backed mode: reject only when disk is full
    if !self.disk_has_capacity() {
        return Err(Error::ResourceExhausted(
            "disk capacity threshold exceeded".to_string(),
        ));
    }
    // Memory budget is advisory — DropOldest in Topic::publish
    // keeps in-memory usage bounded. No reject here.
} else if self.global_max_memory_bytes > 0 {
    // Existing behavior: hard reject on memory budget
    let current = self.total_bytes.load(Relaxed);
    if current + payload_len > self.global_max_memory_bytes {
        return Err(Error::ResourceExhausted(
            "global memory budget exceeded".to_string(),
        ));
    }
    // ... 80% warning ...
}
```

When `disk_backed = true`, the `Topic::publish` call inside the engine uses `DropOldest` regardless of the configured policy (forced at engine construction). The `total_bytes` counter and metrics continue to work correctly — `DropOldest` eviction adjusts bytes via `evicted_bytes` in the publish result, and the `net_add = payload_len - evicted_bytes` calculation is unchanged.

**Note on namespace quotas**: `max_total_messages` is tracked by the namespace manager and decremented on compaction. In disk-backed mode, compaction frees memory but doesn't delete disk data. The quota counter still tracks the in-memory publish/compact lifecycle. If namespace message quotas conflict with disk-backed mode, operators should set `max_total_messages = None` for disk-backed namespaces. Document this in Phase H.

**Files**: `src/queue/engine.rs`, `src/storage/disk.rs` (new), `src/config.rs`, `src/main.rs`

---

### Phase E — Segment Deletion Safety (steps 11–12)

Prevent deleting segments that slow consumers still need.

#### Step 11: `StorageBackend::cleanup_segments` with safe offset

Add to `src/storage.rs`:

```rust
/// Delete segment files whose highest offset is below `safe_below`.
///
/// Guarantees no segment containing offsets ≥ `safe_below` is deleted.
/// Also respects `max_segments` as a count limit. Falls back to
/// count-only cleanup if `safe_below` is `None` (no consumers).
fn cleanup_segments(
    &self,
    namespace: &NamespaceId,
    topic: &TopicName,
    safe_below: Option<TopicOffset>,
    max_segments: usize,
) -> impl Future<Output = Result<u64>> + Send;  // returns segments deleted
```

**FsStorage implementation**: list segments (from the cached index, Step 4), delete only those whose next segment's base offset ≤ `safe_below`, subject to the `max_segments` count. Update the segment index cache after deletion.

**MockStorage**: no-op, return 0.

#### Step 12: Wire cleanup into a dedicated cleanup cycle

Rather than coupling cleanup to `FsStorage::append` (current `enforce_segment_limit`) or adding a `set_min_consumer_offset` method to the engine trait, use a separate cleanup task spawned in `main.rs` where both the engine and namespace manager are available:

```rust
// In main.rs, after spawning servers:
let cleanup_handle = tokio::spawn({
    let engine = Arc::clone(&engine);
    let manager = Arc::clone(&manager);
    let storage = Arc::clone(&storage);
    let mut shutdown_rx = shutdown_rx.clone();
    async move {
        let mut interval = tokio::time::interval(Duration::from_secs(60));
        loop {
            tokio::select! {
                _ = shutdown_rx.changed() => break,
                _ = interval.tick() => {
                    for (ns, topic) in engine.all_topic_keys().await {
                        let min_offset = manager
                            .min_consumer_offset(&ns, &topic).await;
                        let _ = storage.cleanup_segments(
                            &ns, &topic, min_offset, max_segments
                        ).await;
                    }
                }
            }
        }
    }
});
```

Add `NamespaceManager::min_consumer_offset(ns, topic) -> Option<TopicOffset>` — reads the existing `minimum_live_offset()` from `TopicDeliveryState`. No new state; queries what's already tracked.

Add `QueueEngine::all_topic_keys() -> Vec<(NamespaceId, TopicName)>` — returns the keys from the topic map. The engine already has `list_topics(ns)` but this returns all namespaces at once for the cleanup sweep.

Remove the `enforce_segment_limit` call from `FsStorage::append` (segment rotation). Segment deletion is now exclusively handled by the cleanup task.

**Files**: `src/storage.rs`, `src/storage/fs.rs`, `src/queue.rs`, `src/queue/engine.rs`, `src/namespace.rs`, `src/namespace/manager.rs`, `src/main.rs`

---

### Phase F — Observability (step 13)

#### Step 13: New metrics and logging

| Metric | Type | Labels | Where |
|---|---|---|---|
| `fafb_cold_reads_total` | counter | `namespace`, `topic` | `engine.consume_from` cold path |
| `fafb_cold_read_latency_seconds` | histogram | `namespace`, `topic` | `engine.consume_from` cold path |
| `fafb_messages_skipped_total` | counter | `namespace`, `topic` | delivery task skip-forward |
| `fafb_disk_used_pct` | gauge | — | flush task |
| `fafb_disk_available_bytes` | gauge | — | flush task |
| `fafb_segments_deleted_total` | counter | `namespace`, `topic` | `cleanup_segments` |

Structured logging:

- `warn`: disk capacity above 80% (one-shot flag, same pattern as memory 80% warning)
- `warn`: disk capacity threshold reached, publishes rejected
- `warn`: consumer skipping evicted/deleted offsets (gap size)
- `info`: cold read served from disk (`namespace`, `topic`, `offset`, `latency_ms`)
- `info`: segments cleaned up (`namespace`, `topic`, `count`)

**Files**: `src/queue/engine.rs`, `src/storage/fs.rs`, `src/namespace/manager.rs`

---

### Phase G — Tests (steps 14–16)

#### Step 14: Unit tests

| Test | Location | What |
|---|---|---|
| `delivery_skips_evicted_offsets` | `namespace/manager.rs` | max_depth=10, publish 100, subscribe, client receives most recent 10 (not stuck at 0) |
| `record_stream_reads_all_records` | `storage/replay.rs` | RecordStream matches replay_segment output |
| `record_stream_stops_on_corruption` | `storage/replay.rs` | Truncated file, stream stops gracefully |
| `read_at_finds_message` | `storage/fs.rs` | Write 10 messages, `read_at` each offset |
| `read_at_missing_offset` | `storage/fs.rs` | Offset not in any segment → `None` |
| `read_at_across_segments` | `storage/fs.rs` | Messages spanning 3 segments, read from each |
| `read_at_after_segment_deletion` | `storage/fs.rs` | Delete oldest segment, `None` for old offsets, `Some` for retained |
| `segment_cursor_sequential_read` | `storage/fs.rs` | Cursor reads 1000 messages sequentially across 3 segments |
| `segment_cursor_crosses_boundary` | `storage/fs.rs` | Cursor auto-opens next segment at boundary |
| `find_segment_for_offset_cached` | `storage/fs.rs` | Binary search on cached index, edge cases |
| `mock_storage_read_at` | `storage.rs` | MockStorage implementation |
| `cleanup_segments_respects_safe_offset` | `storage/fs.rs` | Segments below safe offset deleted, others kept |
| `cleanup_segments_respects_max_count` | `storage/fs.rs` | Count limit still applies |
| `cleanup_segments_updates_index_cache` | `storage/fs.rs` | Cached index reflects deletions |
| `disk_backed_skips_memory_budget` | `queue/engine.rs` | disk_backed=true, memory budget exceeded, publish succeeds |
| `disk_backed_rejects_on_disk_full` | `queue/engine.rs` | Cached disk pct > threshold, publish rejected |
| `disk_usage_returns_valid_data` | `storage/disk.rs` | Sanity check on real filesystem |

#### Step 15: Integration tests

| Test | Location | What |
|---|---|---|
| `cold_read_after_memory_eviction` | `tests/queue_integration.rs` | Publish 1000, max_depth=50, consume_from offset 0 returns message from disk |
| `disk_backed_publish_past_memory_budget` | `tests/queue_integration.rs` | 1KB budget, disk_backed=true, publish 100 messages, all succeed |
| `segment_cleanup_preserves_consumer_offsets` | `tests/queue_integration.rs` | Consumer at offset 50, segments covering 0–100, cleanup only removes segments fully below 50 |
| `skip_forward_after_segment_deletion` | `tests/queue_integration.rs` | Delete segment, delivery task skips to base_offset, continues delivering |

#### Step 16: Load test

Add to `tests/load_test.rs`:

| Test | What |
|---|---|
| `disk_backed_sustained_load` | disk_backed=true with 1MB memory budget, publish 100K messages, verify all consumable from disk, measure cold read latency distribution |

**Files**: `src/namespace/manager.rs` (tests), `src/storage/fs.rs` (tests), `src/storage/replay.rs` (tests), `src/storage/disk.rs` (tests), `src/queue/engine.rs` (tests), `tests/queue_integration.rs`, `tests/load_test.rs`

---

### Phase H — Documentation (step 17)

#### Step 17: Update docs

- `docs/architecture.md` — add "Tiered Storage" section explaining hot/cold tiers
- `docs/design_decisions.md`:
  - DD: `disk_backed` as engine config vs. BackpressurePolicy variant (separation of concerns)
  - DD: `SegmentCursor` for sequential cold reads vs. per-message `read_at`
  - DD: `AtomicU32` integer percentage for disk usage cache (not f64 bit tricks)
  - DD: Cleanup task in `main.rs` with manager query vs. engine-owned consumer offset map
  - DD: Namespace message quotas are orthogonal to disk-backed mode (document incompatibility)
- `CLAUDE.md` — note `disk_backed` config and `read_at` / `SegmentCursor` in subsystem table
- `README.md` — document `disk_backed`, `disk_capacity_threshold_pct` config
- `config/config.example.toml` — add commented example

---

## Files Modified

| File | Steps | Change |
|---|---|---|
| `src/namespace/manager.rs` | 1, 7, 12 | ~~Delivery task skip-forward~~ ✅, `min_consumer_offset`, base_offset read |
| `src/namespace.rs` | 12 | Add `min_consumer_offset` to trait |
| `src/storage/replay.rs` | 2 | Extract `RecordStream` iterator |
| `src/storage.rs` | 3, 11 | Add `read_at` and `cleanup_segments` to trait + MockStorage impls |
| `src/storage/fs.rs` | 4, 5, 11 | `read_at`, segment index cache, `SegmentCursor`, `cleanup_segments`, remove `enforce_segment_limit` from append |
| `src/storage/disk.rs` | 9 | **New** — `disk_usage()` via statvfs |
| `src/queue.rs` | 1, 7, 12 | ~~Add `base_offset`~~ ✅, `all_topic_keys` to `QueueEngine` trait |
| `src/queue/engine.rs` | 1, 6, 8, 9, 10 | ~~`base_offset` impl~~ ✅, `consume_from` fallback, `disk_backed` config, disk cache, publish changes |
| `src/config.rs` | 8 | `disk_backed`, `disk_capacity_threshold_pct` config |
| `src/main.rs` | 8, 12 | Wire config, spawn cleanup task |
| `tests/queue_integration.rs` | 15 | Cold read + disk-backed integration tests |
| `tests/load_test.rs` | 16 | `disk_backed_sustained_load` test |
| `docs/*` | 17 | Architecture, design decisions, README |

---

## Execution Order

```
Phase A: Delivery task fix          (step 1)       — ✅ DONE
Phase B: Storage point-read + cursor (steps 2–5)   — ✅ DONE
Phase C: Engine consume fallback    (steps 6–7)    — ✅ DONE
Phase D: Disk-backed mode           (steps 8–10)   — ✅ DONE
Phase E: Segment deletion safety    (steps 11–12)  — ✅ DONE
Phase F: Observability              (step 13)      — ✅ DONE
Phase G: Tests                      (steps 14–16)  — ✅ DONE
Phase H: Documentation              (step 17)      — ✅ DONE
```

Phases A–C can ship independently as non-breaking enhancements — cold reads and skip-forward work with any backpressure policy. Phase D introduces the disk-backed mode. Phase E hardens segment deletion. Phases F–H are polish.

---

## Verification

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

After Phase G:
```bash
cargo test --test load_test -- --ignored disk_backed_sustained_load --nocapture
```

---

## Design Decisions

### DD-021: `disk_backed` as engine config, not `BackpressurePolicy` variant

**Problem**: The original plan added `Overflow` to `BackpressurePolicy`. But in `Topic::publish`, `Overflow` behaves identically to `DropOldest` — same eviction logic, same code paths. The only difference is at the engine level (skip global memory check, check disk instead).

**Decision**: `disk_backed` is a `bool` on `EngineConfig`. When true, the engine forces `DropOldest` for all topics and replaces the memory budget gate with a disk capacity gate. `BackpressurePolicy` remains a topic-level concept with clean semantics.

**Why**: A topic-level enum that controls engine-level behavior is a leaky abstraction. Keeping them separate means `BackpressurePolicy` docs and behavior are self-contained, and `disk_backed` docs and behavior are self-contained.

### DD-022: `SegmentCursor` for sequential cold reads

**Problem**: A naive `read_at` implementation does `find_segment → open file → BufReader scan → find offset → close` per message. A consumer catching up 10K messages behind would do this 10K times, each time reopening the same file and rescanning from the start.

**Decision**: `SegmentCursor` keeps the file handle and `BufReader` position open. Sequential calls to `read_next()` advance through the segment with no rescanning. The cursor auto-opens the next segment at boundaries.

**Why**: Turns O(n * records_in_segment) catchup into O(n). The delivery task already loops sequentially — the cursor matches its access pattern naturally.

### DD-023: Segment index cache, not per-read directory listing

**Problem**: `find_segment_for_offset` in the original plan called `fs::read_dir` on every cold read — a syscall per message for catching-up consumers.

**Decision**: Cache a sorted `Vec<(TopicOffset, PathBuf)>` per topic, populated on `init()`, updated on rotation and cleanup. Binary search is O(log segments) with no I/O.

### DD-024: Cleanup task in `main.rs`, not engine-owned consumer offsets

**Problem**: The original plan added `set_min_consumer_offset` to `QueueEngine`, called on every ack. This maintained a persistent map on the hot path for something that only matters once per minute (segment cleanup).

**Decision**: Spawn a cleanup task in `main.rs` that queries `NamespaceManager::min_consumer_offset()` when it runs. The manager already tracks this state via `minimum_live_offset()`. No new trait method on the engine, no hot-path overhead.

### DD-025: Integer percentage for disk usage cache

**Problem**: The original plan stored `f64` bits in `AtomicU64` via `to_bits()`/`from_bits()`. This works but is non-obvious — a reviewer has to reason about IEEE 754 bit representation.

**Decision**: Store integer percentage (0–100) in `AtomicU32`. The threshold comparison is `cached_pct < threshold_pct` — both integers when compared, trivially correct.

---

## Future Work (out of scope)

| Item | Why deferred |
|---|---|
| **Offset index files** (`.idx` alongside `.seg`) | O(1) seek within segments. Only needed if `SegmentCursor` sequential reads aren't fast enough. |
| **Per-namespace disk quotas** | Global threshold means one namespace can starve others. Per-namespace quotas need per-namespace byte tracking on disk. |
| **Cross-platform disk monitoring** | `statvfs` is Unix-only. Windows needs `GetDiskFreeSpaceExW`. Deferred until there's a Windows deployment target. |
| **Segment compaction on disk** | Merge small segments or remove acked messages. Append-only + delete-whole is simpler and sufficient. |
| **mmap for cold reads** | Memory-map segments instead of read+deserialize. Reduces copies for large payloads but adds `unsafe`. Benchmark first. |
| **Consumer seek/rewind protocol** | `SubscribeAt { topic, offset }` or `SubscribeFrom { topic, timestamp }`. Requires timestamp-to-offset index. Natural follow-on once cold reads work. |
| **Time-based retention** | `retention_max_age_secs` for segment files. Designed in `plan_11.4` but not yet implemented. Should coordinate with this plan's `cleanup_segments` — retention deletes by age, cleanup deletes by consumer offset, the more conservative constraint wins. |

---

## Post-Implementation Refinements

- Consolidated `delete_segments` + `cleanup_segments` into unified method
- Added `disk_capacity_threshold_pct` validation (clamping)
- Added `SegmentCursor::skip_to()` for efficient cold reads
- Made `COLD_BATCH_SIZE` configurable via `ManagerConfig.cold_batch_size`
- Added `cold_deliver_individual`/`cold_deliver_group` helpers to deduplicate delivery logic
- Cleanup task now compacts before deleting (crash-safe ordering)

---

## Risk Assessment

| Risk | Mitigation |
|---|---|
| Cold reads slow down delivery tasks | Each subscriber has its own tokio task — one slow consumer doesn't block others. `SegmentCursor` eliminates per-read overhead. Metrics (`fafb_cold_reads_total`) alert operators. |
| Segment scan within cursor | `SegmentCursor` reads sequentially — O(1) per `read_next()` call. Only the initial seek to the starting offset scans. |
| statvfs overhead | Called once per flush interval (default 100ms), not per publish. Cached as `AtomicU32`, checked with a single atomic load. |
| Disk-backed mode hides memory pressure | The 80% memory warning (already implemented) still fires. Cold read metrics make the problem visible. Document that `max_total_messages` quotas should be disabled for disk-backed namespaces. |
| Segment deletion race with concurrent cold read | On Unix, an open fd remains valid after unlink — the read completes. On non-Unix, `cleanup_segments` is a no-op (disk monitoring is `#[cfg(unix)]`). |
| Consumer cursor points to deleted segment | Delivery task skip-forward (Step 1) handles this: skip to `base_offset`, log warning with gap size, emit `fafb_messages_skipped_total` metric. |
| Namespace quotas conflict with disk-backed mode | Document incompatibility. `max_total_messages` counts in-memory publishes; with disk-backed mode, operators should set it to `None`. |
