# Phase 5: Storage

**Status: Complete**

## Goal
Filesystem-backed durability layer. Append-only log per topic, replay on startup.

## Files
- `src/storage.rs`
- `src/storage/fs.rs`
- `src/storage/replay.rs`
- `tests/storage_integration.rs`

## What Was Implemented

### storage.rs (root)
- [x] `trait StorageBackend` — async trait (`Send + Sync`, native async methods, no `async-trait` crate)
  - `async fn init(&self) -> Result<()>` — create directories, pre-populate writer caches from existing segments
  - `async fn append(&self, namespace: &NamespaceId, topic: &TopicName, msg: &PublishedMessage) -> Result<()>` — writes to OS page cache without fsync
  - `async fn flush(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<()>` — fsync a specific topic's active segment
  - `async fn replay(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<Vec<PublishedMessage>>` — replay all persisted messages in offset order
  - `async fn shutdown(&self) -> Result<()>` — flush and close all active writers
- [x] `pub use fs::FsStorage` re-export
- [x] `MockStorage` — in-memory `HashMap` behind `tokio::sync::Mutex`, gated by `#[cfg(any(test, feature = "test-support"))]` for downstream module tests
- [x] `test-support` feature flag added to `Cargo.toml`

### storage/fs.rs
- [x] `FsStorage` struct — `data_dir`, `segment_max_bytes`, per-topic writers
- [x] `impl StorageBackend for FsStorage`
- [x] Directory layout: `{data_dir}/{namespace}/{topic}/segments/`
- [x] Append-only segment files with size-based rotation (default 64 MiB, configurable via `with_segment_max_bytes`)
- [x] Segment naming: `{base_offset:020}.seg`
- [x] Record format on disk: `[4B u32 BE record_len][8B u64 BE offset][N bytes bincode MessageEnvelope]`
- [x] Single-writer guarantee via architecture (no file locking needed)
- [x] **Per-topic locking**: `RwLock<HashMap<TopicKey, Arc<Mutex<ActiveSegment>>>>` — concurrent appends to different topics proceed in parallel; read lock for lookups, write lock only when a new topic is first seen
- [x] **No per-write fsync**: `append` writes to the OS page cache only; callers control durability via `flush()` or `shutdown()`
- [x] **Init pre-population**: `init()` walks existing segment directories and opens the latest segment per topic, eliminating directory scans on first append

### storage/replay.rs
- [x] `replay_topic(seg_dir)` — scans segments dir, reads each in order, sorts by offset
- [x] `list_segments(seg_dir)` — lists `*.seg` files, returns empty Vec if dir doesn't exist
- [x] **Streaming replay**: `replay_segment` uses `BufReader` + `AsyncReadExt::read_exact` to read records incrementally instead of loading entire segment files into memory
- [x] Corruption handling: truncated or corrupt records log `tracing::warn!` and stop reading; all valid records returned

### Dependencies
- [x] `bincode = "1"` added to `[dependencies]`
- [x] `From<bincode::Error> for Error` added to `src/error.rs`

## Key Decisions Made
- **`PublishedMessage` as the storage unit** — bundles `MessageEnvelope` + `TopicOffset`, more type-safe than separate parameters
- **Append-only log, not random access** — simplest durable pattern
- **Segment rotation by size** (default 64 MiB) — prevents single huge files
- **No per-write fsync** — `append` writes to OS page cache; callers batch flushes for throughput. `flush(ns, topic)` for per-topic durability, `shutdown()` for flush-all before exit
- **Per-topic locking** — `RwLock` on the topic map + per-topic `Mutex` on the segment writer. Different topics never contend. Write lock on the map only on first-ever write to a topic
- **Streaming replay** — `BufReader` reads records incrementally; avoids loading 64 MiB segments into memory at startup
- **Feature-gated mock** — `MockStorage` behind `#[cfg(any(test, feature = "test-support"))]` so downstream modules (queue engine, integration tests) can enable it via `--features test-support`
- **Init pre-population** — `init()` walks existing data directories and opens the latest segment per topic into the writers cache, so first append to a known topic skips directory scanning
- **bincode for on-disk serialization** — compact binary format, already well-established in the Rust ecosystem

## Shared Abstraction: TopicLog

The storage layer is one of three subsystems that participate in the **TopicLog** abstraction:

- **Storage (this plan)**: Serializes the topic log to disk. The append-only segment files are the durable representation of the indexed log. Replay reconstructs the in-memory log on startup.
- **Queue Engine (plan 6)**: Manages the in-memory indexed log (`Vec<MessageEnvelope>` with `TopicOffset` indexing). The queue engine is the owner of the log's append path.
- **Namespace Manager (plan 7)**: Controls cursors into the log — subscriber offsets, consumer group offsets. Reports minimum live offset to the queue engine for GC/compaction.

### Trait Boundary Clarification

The `StorageBackend` trait aligns with the indexed log model:

- `append` writes a `PublishedMessage` (envelope + assigned offset) — the queue engine assigns the offset, storage persists it
- `replay` returns `Vec<PublishedMessage>` in offset order so the queue engine can reconstruct the log with correct offset state
- `list_topics` enumerates all persisted namespace/topic pairs — the queue engine calls this at startup to discover which topics need replay
- `flush` / `shutdown` give the queue engine control over fsync timing for batched durability

See also: [plan 6 (Queue Engine)](plan_6_queue.md#shared-abstraction-topiclog), [plan 7 (Namespace Manager)](plan_7_namespace.md#shared-abstraction-topiclog)

## Post-Completion Amendment: `list_topics` (Phase 6 dependency)

The queue engine (plan 6) needs to discover which namespace/topic pairs exist on disk at startup to drive replay. This required a new method on `StorageBackend`.

### Added
- [x] `async fn list_topics(&self) -> Result<Vec<(NamespaceId, TopicName)>>` on `StorageBackend` trait
- [x] `FsStorage` implementation — walks `data_dir/{ns}/{topic}/segments/`, returns pairs with at least one `.seg` file
- [x] `MockStorage` implementation — returns keys from internal `HashMap`
- [x] Unit tests: `list_topics_empty_data_dir`, `list_topics_returns_persisted_topics`

## Open Questions (deferred)
- Retention policy? TTL? Max segments? Out of scope for v1 (see plan 11 for retention design).
- Compaction (remove acked messages from disk)? In-memory compaction is plan 6; on-disk retention is plan 11.

## Tests

### Unit tests (13 tests)
**error.rs** (1): `bincode_error_converts`

**replay.rs** (6): `replay_empty_dir`, `replay_nonexistent_dir`, `replay_single_segment`, `replay_multiple_segments_in_order`, `replay_truncated_record_recovers`, `replay_ignores_non_seg_files`

**fs.rs** (8): `init_creates_data_dir`, `init_idempotent`, `init_pre_populates_writers`, `append_creates_segments_dir`, `flush_no_op_for_unknown_topic`, `shutdown_with_no_writers`, `segment_path_formatting`, `list_topics_empty_data_dir`, `list_topics_returns_persisted_topics`

### Integration tests (4 tests)
`append_and_replay_returns_messages_in_order` — 100 messages round-trip with payload verification

`replay_recovers_from_truncation` — 10 messages, truncate segment, recovers 8-9

`segment_rotation_creates_new_file` — small segment max (256B), 20 messages with 64-byte payloads, verifies multiple `.seg` files, replay returns all 20

`shutdown_persists_multiple_topics` — write to 2 topics, shutdown, fresh instance replays both correctly
