# Plan 12: Timestamp Exposure & Time-Based Retention

## Status: COMPLETE

All five phases implemented: u64 timestamp migration, wire protocol exposure, HTTP API timestamps, retention configuration, and background retention enforcement loop.

## Goal

Change the internal timestamp representation from `SystemTime` to `u64` millis, expose timestamps (and offsets) to clients via the wire protocol and HTTP API, and implement time-based retention so long-running deployments can bound disk usage by age, byte count, or segment count — per namespace.

## Prerequisites

- Plan 5 complete — `FsStorage` with append-only segments, per-topic locking, replay
- Plan 6 complete — `InMemoryQueueEngine` with background flush task
- Plan 10 complete — startup sequence with `storage.init()` + `engine.init()`
- Plan 11.4 designed — retention policies specified but not yet implemented

## Current State

**What exists:**
- `MessageEnvelope.timestamp: SystemTime` is set at publish time in both TCP (`session.rs:335`) and HTTP (`routes.rs:65`) paths
- Timestamp is serialized to segment files via bincode as part of the envelope
- Timestamp survives replay — deserialized correctly during `replay_topic()`
- Timestamp is used in exactly one place: delivery latency metric (`manager.rs:339`)
- Protocol version byte exists on `Connect` frame — server accepts v1 only (`session.rs:166`)
- This package/app has not been published — no deployed clients exist

**What's missing:**
- Wire protocol `Deliver` frame (0x09) does not include the timestamp or offset — clients can't see either
- HTTP publish response and topic info response don't include timestamps
- No retention enforcement loop
- No per-namespace retention config fields (`retention_max_age_secs`, `retention_max_bytes`, `retention_max_segments`)
- No `retention_check_interval_secs` in `StorageConfig`

## Scope Boundary

This plan does NOT include:
- LRU fd management (`max_open_fds`) — separate concern, tracked in plan 11.4
- Parallel startup replay — separate concern, tracked in plan 11.4
- Streaming replay — separate concern, tracked in plan 11.4
- Client-supplied timestamps — server is the sole clock authority
- Per-topic retention overrides — config struct is designed to accommodate this later (see Phase 4 notes) but not implemented in this plan

---

## Phase 1: Change Internal Timestamp to `u64` Millis

Change `MessageEnvelope.timestamp` from `SystemTime` to `u64` (milliseconds since Unix epoch). This eliminates per-use conversion throughout the codebase and makes the on-disk format portable and deterministic.

> **Why:** bincode serialization of `SystemTime` serializes the internal `Duration` struct, whose layout isn't guaranteed stable across Rust/platform versions. `u64` millis is compact, portable, trivially comparable, and needs zero conversion at the wire/HTTP/retention boundaries.

### 1.1 Change `MessageEnvelope.timestamp` type

**File: `src/types.rs`** (line 192)
- [x] Change `timestamp: SystemTime` to `timestamp: u64` with doc comment: `/// Publish timestamp — milliseconds since Unix epoch`
- [x] Add a helper function:

```rust
/// Returns the current time as milliseconds since Unix epoch.
pub fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}
```

### 1.2 Update all publish sites

**File: `src/server/tcp/session.rs`** (line 335)
- [x] Change `timestamp: SystemTime::now()` to `timestamp: now_millis()`

**File: `src/server/http/routes.rs`** (line 65)
- [x] Change `timestamp: SystemTime::now()` to `timestamp: now_millis()`

### 1.3 Update delivery latency metric

**File: `src/namespace/manager.rs`** (line 339)
- [x] Replace `msg.envelope.timestamp.elapsed()` with:

```rust
let now = now_millis();
let latency_ms = now.saturating_sub(msg.envelope.timestamp);
metrics::histogram!("fafb_delivery_latency_seconds", ...)
    .record(latency_ms as f64 / 1000.0);
```

### 1.4 Update all test code constructing `MessageEnvelope`

- [x] Grep for `timestamp: SystemTime::now()` across `src/` and `tests/` — replace with `timestamp: now_millis()` or a fixed test value
- [x] Update `src/storage/replay.rs` test helper `make_msg()` (line 110)

### 1.5 Storage format migration

Existing segment files contain bincode-serialized `SystemTime` in the envelope. After this change, they won't deserialize.

Since this app has not been published and there are no production deployments, **no migration path is needed.** Any existing local test data can be deleted and regenerated.

- [x] Add a note in the replay code (`src/storage/replay.rs`) as a comment: segments written before this change are incompatible and must be deleted

### 1.6 Tests

#### Unit tests — `src/types.rs` `#[cfg(test)]`

- [x] `now_millis_returns_nonzero` — `now_millis()` returns a value > 0 and within a reasonable range (> 1_700_000_000_000, i.e. after 2023)
- [x] `now_millis_is_monotonic` — two successive calls return non-decreasing values

#### Unit tests — `src/queue/topic.rs` `#[cfg(test)]`

Update `make_envelope` helper to use `now_millis()` instead of `SystemTime::now()`. All existing topic tests must continue to pass with the new `u64` timestamp.

- [x] Verify existing tests compile and pass — no new test cases needed here, just the fixture update

#### Unit tests — `src/storage/replay.rs` `#[cfg(test)]`

Update `make_msg` helper (line 110) to use `now_millis()`. All existing replay tests must continue to pass.

- [x] `replay_single_segment` — still passes with u64 timestamp in envelope
- [x] `replay_multiple_segments_in_order` — still passes
- [x] `replay_truncated_record_recovers` — still passes

#### Unit tests — `src/namespace/manager.rs` `#[cfg(test)]`

Update `make_envelope` helper to use `now_millis()`. Verify delivery latency metric path works with `u64` subtraction.

- [x] `delivery_latency_metric_uses_u64` — publish a message with a known timestamp (e.g., `now_millis() - 500`), deliver it, verify no panic in the latency calculation path

```bash
cargo test                     # all tests pass with u64 timestamp
cargo test storage             # storage roundtrip tests
cargo test replay              # replay tests
```

---

## Phase 2: Expose Timestamps and Offsets in Wire Protocol

Since no clients are deployed, we update the v1 protocol directly — no version bump needed.

### 2.1 Add `timestamp_millis` and `offset` to `Frame::Deliver`

The `Deliver` frame currently sends `message_id + topic + payload`. Add offset and timestamp.

**File: `src/codec/frame.rs`**
- [x] Add fields to `Frame::Deliver` variant:

```rust
Deliver {
    message_id: MessageId,
    offset: TopicOffset,        // NEW — position in topic log
    timestamp_millis: u64,      // NEW — millis since Unix epoch
    topic: TopicName,
    payload: Bytes,
},
```

> **Why add offset now?** Clients need it for offset tracking (resume after reconnect), debugging stale deliveries, and future consumer group rebalancing. Adding it later means another breaking wire format change. The cost is 8 bytes per delivery.

**File: `src/codec/encoder.rs`** (around line 89)
- [x] Encode in order: `message_id`, `offset`, `timestamp_millis`, then topic + payload

```text
// Updated Deliver wire format:
[8 bytes: message_id (u64 BE)]
[8 bytes: offset (u64 BE)]             ← NEW
[8 bytes: timestamp_millis (u64 BE)]   ← NEW
[2 bytes: topic_len (u16 BE)] [N bytes: topic UTF-8]
[4 bytes: payload_len (u32 BE)] [N bytes: payload]
```

**File: `src/codec/decoder.rs`** (around line 172)
- [x] Decode: check `remaining() >= 24` (was 8), read `message_id`, `offset`, `timestamp_millis` as three consecutive u64 BE values

### 2.2 Update delivery path to populate new fields

**File: `src/namespace/manager.rs`** (delivery task, around line 330)
- [x] The `PublishedMessage` already has `offset` and `envelope.timestamp` (now u64). Pass both through when constructing the `Deliver` frame:

```rust
Frame::Deliver {
    message_id: msg.envelope.id,
    offset: msg.offset,
    timestamp_millis: msg.envelope.timestamp,
    topic: msg.envelope.topic.clone(),
    payload: Bytes::from(msg.envelope.payload.clone()),
}
```

**File: `src/server/tcp/session.rs`** (anywhere `Frame::Deliver` is constructed)
- [x] Populate the new `offset` and `timestamp_millis` fields

### 2.3 Update all test code constructing `Frame::Deliver`

- [x] Grep for `Frame::Deliver {` across `src/` and `tests/` — add `offset` and `timestamp_millis` fields to every construction site
- [x] Update encoder/decoder roundtrip tests in `src/codec/` to include offset and timestamp assertions
- [x] Add a test that non-zero values for both fields survive encode → decode roundtrip

### 2.4 Tests

#### Unit tests — `src/codec/encoder.rs` `#[cfg(test)]`

- [x] `encode_deliver_includes_offset_and_timestamp` — encode a `Frame::Deliver` with known values (`message_id: 42`, `offset: TopicOffset(100)`, `timestamp_millis: 1_700_000_000_000`), verify the raw bytes contain all three u64 values at the correct positions (bytes 5-12: message_id, 13-20: offset, 21-28: timestamp)
- [x] `encode_deliver_frame_length_increased` — encode a Deliver frame, verify the 4-byte frame length prefix accounts for the 16 additional bytes (offset + timestamp)

#### Unit tests — `src/codec/decoder.rs` `#[cfg(test)]`

- [x] `decode_deliver_reads_offset_and_timestamp` — hand-craft a valid Deliver byte buffer with offset=7 and timestamp=1_700_000_000_000, decode it, assert both fields match
- [x] `decode_deliver_truncated_at_offset` — provide a Deliver frame with only 8 bytes after the type byte (message_id only, missing offset+timestamp), assert `Error::Codec` with "truncated" in the message
- [x] `decode_deliver_truncated_at_timestamp` — provide 16 bytes (message_id + offset, missing timestamp), assert `Error::Codec`

#### Roundtrip tests — `src/codec/encoder.rs` or `decoder.rs` `#[cfg(test)]`

- [x] `roundtrip_deliver_with_offset_and_timestamp` — construct a `Frame::Deliver` with non-zero offset and timestamp, encode then decode, assert all fields equal including `offset` and `timestamp_millis`
- [x] `roundtrip_deliver_zero_values` — roundtrip with `offset: TopicOffset(0)`, `timestamp_millis: 0` — edge case, both must survive
- [x] `roundtrip_deliver_max_values` — roundtrip with `offset: TopicOffset(u64::MAX)`, `timestamp_millis: u64::MAX`

#### Property tests — `src/codec/encoder.rs` `#[cfg(test)]`

- [x] Update `arb_frame()` proptest strategy to generate `offset` and `timestamp_millis` fields for the `Deliver` variant
- [x] Existing `roundtrip_never_panics` proptest continues to pass with the new fields

#### Unit tests — `src/codec/frame.rs` `#[cfg(test)]`

- [x] Update `frame_message_type` test — the `Frame::Deliver` construction must include the new fields
- [x] Update `frame_partial_eq_with_bytes` if it constructs Deliver frames

#### Integration-level — `src/server/tcp/session.rs` `#[cfg(test)]`

- [x] Update all test helpers that construct `Frame::Deliver` to include `offset` and `timestamp_millis`
- [x] `deliver_frame_contains_offset_and_timestamp` — process a Publish frame through the session, subscribe, receive the Deliver frame, assert `offset` is `TopicOffset(0)` (first message) and `timestamp_millis` is within 1000ms of `now_millis()`

```bash
cargo test                     # all tests pass with new frame layout
cargo test codec               # encoder/decoder roundtrip
cargo test session             # TCP session tests
```

---

## Phase 3: Expose Timestamps in HTTP API

### 3.1 Add `timestamp_millis` to HTTP publish response

**File: `src/server/http/routes.rs`** (`publish` handler, line 70)
- [x] Capture the timestamp from the envelope (now a u64) before publishing
- [x] Add `"timestamp_millis"` to the JSON response:

```rust
Ok(Json(serde_json::json!({
    "offset": offset.0,
    "message_id": msg_id,
    "timestamp_millis": envelope_timestamp,  // NEW
})))
```

### 3.2 Add oldest/newest timestamps to topic info

**File: `src/queue/topic.rs`** (`TopicStats` struct)
- [x] Add `oldest_timestamp: Option<u64>` and `newest_timestamp: Option<u64>` to `TopicStats`
- [x] Populate from `buffer.front().timestamp` and `buffer.back().timestamp` in `stats()` method

**File: `src/server/http/routes.rs`** (`topic_info` handler, line 94)
- [x] Add `"oldest_timestamp_millis"` and `"newest_timestamp_millis"` to the JSON response (null when topic is empty)

### 3.3 Tests

#### Unit tests — `src/queue/topic.rs` `#[cfg(test)]`

- [x] `stats_timestamps_empty_topic` — create an empty topic, call `stats()`, assert `oldest_timestamp` is `None` and `newest_timestamp` is `None`
- [x] `stats_timestamps_single_message` — publish one message with a known timestamp, call `stats()`, assert `oldest_timestamp == newest_timestamp == that timestamp`
- [x] `stats_timestamps_multiple_messages` — publish 3 messages with timestamps 1000, 2000, 3000, call `stats()`, assert `oldest_timestamp == Some(1000)` and `newest_timestamp == Some(3000)`
- [x] `stats_timestamps_after_compaction` — publish 5 messages (timestamps 1000-5000), compact to offset 3, assert `oldest_timestamp == Some(4000)` (the oldest remaining message)

#### Integration tests — HTTP routes (inline or `tests/`)

- [x] `publish_response_includes_timestamp` — POST to `/v1/{ns}/{topic}/publish`, parse JSON response, assert `timestamp_millis` field exists and is within 1000ms of `now_millis()`
- [x] `topic_info_includes_timestamps` — publish 2 messages, GET `/v1/{ns}/{topic}/info`, assert `oldest_timestamp_millis` and `newest_timestamp_millis` are both present and `oldest <= newest`
- [x] `topic_info_timestamps_null_when_empty` — GET info for a topic with no messages, assert `oldest_timestamp_millis` and `newest_timestamp_millis` are both `null`

```bash
cargo test http                # HTTP route tests
cargo test topic_info          # topic info endpoint tests
cargo test stats_timestamp     # topic stats timestamp tests
```

---

## Phase 4: Retention Configuration

### 4.1 Define `RetentionPolicy` as a standalone struct

**File: `src/config.rs`**
- [x] Create a reusable struct that can be embedded at namespace level now and at topic level later:

```rust
/// Retention policy for segment cleanup. All fields are optional —
/// None means no limit for that dimension.
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(default)]
pub struct RetentionPolicy {
    /// Delete segments whose newest message is older than this many seconds.
    pub max_age_secs: Option<u64>,
    /// Delete oldest segments when total topic size exceeds this byte count.
    pub max_bytes: Option<u64>,
    /// Keep at most this many segments per topic. Oldest deleted first.
    pub max_segments: Option<usize>,
}

impl RetentionPolicy {
    /// Returns true if any retention limit is configured.
    pub fn is_active(&self) -> bool {
        self.max_age_secs.is_some()
            || self.max_bytes.is_some()
            || self.max_segments.is_some()
    }
}
```

> **Design note:** Making `RetentionPolicy` a standalone struct (rather than inlining fields) means per-topic overrides can be added later as `HashMap<TopicName, RetentionPolicy>` without restructuring the config. Not implemented in this plan, but the shape accommodates it:
> ```toml
> # Future (not this plan):
> [namespaces.default.topic_retention_overrides.events]
> max_age_secs = 86400
> [namespaces.default.topic_retention_overrides."*__dlq"]
> max_age_secs = 2592000
> ```

### 4.2 Add retention to `NamespaceConfig`

**File: `src/config.rs`** (`NamespaceConfig` struct, around line 244)
- [x] Add:

```rust
/// Retention policy for this namespace's topics.
/// Defaults to no retention (keep forever).
#[serde(default)]
pub retention: RetentionPolicy,
```

### 4.3 Add `retention_check_interval_secs` to `StorageConfig`

**File: `src/config.rs`** (`StorageConfig` struct, around line 176)
- [x] Add:

```rust
/// Interval in seconds between retention enforcement sweeps. Default: 60.
#[serde(default = "default_retention_interval")]
pub retention_check_interval_secs: u64,
```

- [x] Add default function: `fn default_retention_interval() -> u64 { 60 }`
- [x] Update `Default` impl

### 4.4 Update config example

**File: `config/config.example.toml`**
- [x] Add retention fields with comments explaining each

### 4.5 Tests

#### Unit tests — `src/config.rs` `#[cfg(test)]`

- [x] `retention_policy_default_is_inactive` — `RetentionPolicy::default().is_active()` returns `false`
- [x] `retention_policy_active_with_max_age` — `RetentionPolicy { max_age_secs: Some(3600), ..Default::default() }.is_active()` returns `true`
- [x] `retention_policy_active_with_max_bytes` — same pattern with `max_bytes: Some(1_000_000)`
- [x] `retention_policy_active_with_max_segments` — same pattern with `max_segments: Some(10)`
- [x] `namespace_config_parses_retention` — parse TOML with `[namespaces.test.retention]` section containing `max_age_secs = 604800`, `max_bytes = 10737418240`, `max_segments = 64`, assert all three fields deserialize correctly
- [x] `namespace_config_retention_defaults_to_inactive` — parse TOML with a namespace that has no `[retention]` section, assert `retention.is_active() == false`
- [x] `storage_config_retention_interval_default` — parse TOML with no `retention_check_interval_secs`, assert it defaults to 60
- [x] `storage_config_retention_interval_custom` — parse TOML with `retention_check_interval_secs = 30`, assert it parses to 30

```bash
cargo test config              # config parsing tests
```

---

## Phase 5: Retention Enforcement Loop

### 5.1 Add `min_live_offset` to `NamespaceManager`

The retention task must not delete messages that any active subscriber or consumer group still needs. This is distinct from `base_offset` (which only reflects what has been compacted from the in-memory buffer).

**File: `src/namespace/manager.rs`** (or the `NamespaceManager` trait in `src/namespace.rs`)
- [x] Add method: `async fn min_live_offset(&self, ns: &NamespaceId, topic: &TopicName) -> Option<TopicOffset>`
  - Returns the **lowest offset across all active individual subscribers and consumer group cursors** for this topic
  - If no subscribers exist, returns `None` (retention is unconstrained)
  - The namespace manager already tracks subscriber offsets (`sub.next_offset`) and consumer group cursors — this method walks them to find the minimum

> **Why `min_live_offset` instead of `base_offset`?** `base_offset` reflects the in-memory compaction frontier. But a subscriber at offset 800 with `base_offset` at 0 means offset 800 is still needed. If retention compacts to offset 1000 based on age, that subscriber silently loses messages 800-999. `min_live_offset` prevents this — retention is bounded by the slowest consumer. If a consumer is too far behind, `SlowConsumerPolicy` handles it separately (disconnect, drop, DLQ).

### 5.2 Implement the retention sweep

**File: `src/retention.rs`** (new file, at crate root)

> **Why crate root?** Retention orchestrates across the storage layer, queue engine, AND namespace manager. It's a cross-cutting concern, not a storage detail. Placing it in `src/storage/` would create an awkward dependency on `NamespaceManager`.

- [x] Create `RetentionTask` struct:

```rust
pub struct RetentionTask<S: StorageBackend, M: NamespaceManager> {
    storage: Arc<S>,
    manager: Arc<M>,
    namespace_configs: HashMap<NamespaceId, RetentionPolicy>,
    interval: Duration,
}
```

- [x] Implement `run()` method — spawns a `tokio::spawn` loop:

```
loop {
    sleep(interval)
    for each (namespace, policy) where policy.is_active():
        topics = manager.list_topics(ns)
        for each topic:
            // Determine the retention floor — cannot delete below this
            let floor = manager.min_live_offset(ns, topic)

            // Scan segment files for this topic
            let seg_dir = storage.segments_dir(ns, topic)
            let segments = list_and_stat_segments(seg_dir)
            if segments.is_empty(): continue

            let mut delete_before_offset: Option<TopicOffset> = None

            // 1. Age-based: find segments past max_age
            if let Some(max_age) = policy.max_age_secs:
                cutoff = now_millis() - (max_age * 1000)
                // Use mtime as proxy for newest message timestamp
                for seg in segments (oldest first):
                    if seg.mtime_millis < cutoff:
                        delete_before_offset = Some(next segment's base_offset)

            // 2. Size-based: delete oldest until total <= max_bytes
            if let Some(max_bytes) = policy.max_bytes:
                total = sum(segment.size_bytes)
                for seg in segments (oldest first):
                    if total <= max_bytes: break
                    total -= seg.size_bytes
                    delete_before_offset = max(delete_before_offset, next segment's base_offset)

            // 3. Count-based: delete oldest until count <= max_segments
            if let Some(max_segs) = policy.max_segments:
                if segments.len() > max_segs:
                    excess = segments.len() - max_segs
                    delete_before_offset = max(delete_before_offset, segments[excess].base_offset)

            // Apply floor: never delete below min_live_offset
            if let (Some(delete), Some(floor)) = (delete_before_offset, floor):
                delete_before_offset = Some(min(delete, floor))

            // Execute: compact in-memory first, then delete files
            if let Some(offset) = delete_before_offset:
                manager.compact(ns, topic, offset)      // step 1: free memory
                storage.delete_segments(ns, topic, offset)  // step 2: remove files
}
```

### 5.3 Directory-scan approach (no in-memory segment index)

Rather than maintaining a parallel in-memory `HashMap<TopicKey, Vec<SegmentMeta>>` that must stay in sync with the filesystem, the retention task scans segment directories directly during each sweep.

For each topic:
1. `list_segments(seg_dir)` — already exists in `src/storage/replay.rs`, returns sorted `Vec<PathBuf>`
2. `fs::metadata(path)` on each segment — gives `len()` (size) and `modified()` (mtime)

Use `mtime` as the proxy for "newest message timestamp in this segment." This is correct because the last write to a segment IS the newest message, and `mtime` reflects the last write.

For a deployment with 1000 topics × 64 segments, that's 64K stat calls every 60 seconds — trivial I/O on any modern filesystem.

> **Why not an index?** An in-memory index that mirrors filesystem state is a synchronization bug waiting to happen. The index must be updated on every rotation, every deletion (by retention AND by `enforce_segment_limit`), and at startup. If any path forgets to update it, the retention task makes wrong decisions. The directory scan is stateless and always correct. Add caching only if profiling shows the sweep is a bottleneck (it won't be).

**File: `src/retention.rs`**
- [x] Add helper: `async fn scan_segments(seg_dir: &Path) -> Vec<SegmentInfo>`

```rust
struct SegmentInfo {
    path: PathBuf,
    base_offset: TopicOffset,
    size_bytes: u64,
    mtime_millis: u64,
}
```

- [x] Parse base offset from filename (strip `.seg`, parse u64)
- [x] Get size and mtime from `fs::metadata()`

**File: `src/storage.rs`** (`StorageBackend` trait)
- [x] Add method: `async fn delete_segments(&self, ns: &NamespaceId, topic: &TopicName, before_offset: TopicOffset) -> Result<usize>` — deletes all segment files with `base_offset < before_offset`, returns count deleted
- [x] Add method: `fn segments_dir(&self, ns: &NamespaceId, topic: &TopicName) -> PathBuf` — expose the directory path so the retention task can scan it
- [x] Provide no-op implementations for `MockStorage`

### 5.4 Coordination invariant

**The golden rule:** `engine.compact()` (via manager) must be called BEFORE segment deletion. If a crash occurs between segment deletion and compaction (wrong order), the engine would try to replay deleted data on restart.

The correct order:
1. Determine which segments to delete (by age/size/count)
2. Clamp against `min_live_offset` — never delete what active subscribers still need
3. Calculate `new_min_offset` (= `base_offset` of the first surviving segment)
4. Call `manager.compact(ns, topic, new_min_offset)` — frees in-memory messages
5. Call `storage.delete_segments(ns, topic, new_min_offset)` — removes files

**Crash safety:**
- Crash after step 4, before step 5: orphaned segment files on disk, harmless — cleaned up next sweep
- Crash after step 5, before step 4 (WRONG ORDER): replayed messages reference deleted data — **data loss**

### 5.5 Interaction with existing `enforce_segment_limit`

Today, `enforce_segment_limit()` in `fs.rs:160` deletes oldest segments during rotation based on `StorageConfig.max_segments_per_topic`. This is a global, count-based limit that runs synchronously during appends.

With per-namespace `retention.max_segments`, there are two segment-count mechanisms:
- **`StorageConfig.max_segments_per_topic`** — global hard cap, enforced inline during rotation. Acts as a safety net.
- **`NamespaceConfig.retention.max_segments`** — per-namespace policy, enforced by the background retention task.

These coexist: the inline limit is a ceiling that prevents runaway disk usage even if the retention task falls behind. The retention task handles the finer-grained per-namespace policy.

No changes needed to `enforce_segment_limit` — it continues to work as-is.

### 5.6 Wire into startup

**File: `src/main.rs`**
- [x] After engine init, spawn the retention task if any namespace has `retention.is_active()`
- [x] Pass `Arc<FsStorage>` and `Arc<DefaultNamespaceManager>` to the task
- [x] Build policy map from each `NamespaceConfig`'s retention field
- [x] Log: `tracing::info!("retention task started: interval={}s, namespaces={}", interval, count)`

### 5.7 Metrics

- [x] `fafb_retention_segments_deleted_total` — counter, labels: `namespace`, `topic`, `reason` (age/size/count)
- [x] `fafb_retention_bytes_freed_total` — counter, labels: `namespace`, `topic`
- [x] `fafb_retention_sweep_duration_seconds` — histogram

### 5.8 Tests

#### Unit tests — `src/namespace/manager.rs` `#[cfg(test)]` (or `src/namespace.rs`)

- [x] `min_live_offset_no_subscribers` — create a manager, publish messages to a topic, call `min_live_offset()` with no subscribers, assert `None`
- [x] `min_live_offset_single_subscriber` — subscribe one client at offset 5, assert `min_live_offset() == Some(TopicOffset(5))`
- [x] `min_live_offset_multiple_subscribers` — subscribe client A at offset 5 and client B at offset 10, assert `min_live_offset() == Some(TopicOffset(5))` (the minimum)
- [x] `min_live_offset_with_consumer_group` — create a consumer group with cursor at offset 3, add an individual subscriber at offset 7, assert `min_live_offset() == Some(TopicOffset(3))`
- [x] `min_live_offset_after_unsubscribe` — subscribe at offset 5, unsubscribe, assert `None`

#### Unit tests — `src/retention.rs` `#[cfg(test)]`

**`scan_segments` helper:**

- [x] `scan_segments_empty_dir` — scan an empty `TempDir`, assert returns empty `Vec`
- [x] `scan_segments_nonexistent_dir` — scan a path that doesn't exist, assert returns empty `Vec`
- [x] `scan_segments_parses_offsets` — create segment files `00000000000000000000.seg`, `00000000000000001000.seg`, `00000000000000002000.seg` in a TempDir, scan, assert 3 entries with `base_offset` 0, 1000, 2000 respectively
- [x] `scan_segments_returns_size_and_mtime` — create a segment file, write known bytes to it, scan, assert `size_bytes` matches and `mtime_millis` is recent (within 5000ms of `now_millis()`)
- [x] `scan_segments_ignores_non_seg_files` — create `notes.txt` and `data.log` alongside valid `.seg` files, assert only `.seg` files are returned
- [x] `scan_segments_sorted_by_offset` — create segments in reverse order (2000, 0, 1000), assert results are sorted ascending by `base_offset`

**Retention sweep logic (test with `MockStorage` and stub `NamespaceManager`):**

- [x] `retention_age_deletes_old_segments` — create 3 segment files, set mtime on the first two to 2 hours ago, configure `max_age_secs: 3600` (1 hour), run one sweep, assert the first two segments are deleted and the third remains
- [x] `retention_age_keeps_all_when_none_expired` — create 3 segments all with recent mtime, configure `max_age_secs: 3600`, run sweep, assert nothing deleted
- [x] `retention_bytes_deletes_until_under_limit` — create 3 segments of 100 bytes each (300 total), configure `max_bytes: 200`, run sweep, assert the oldest segment is deleted (bringing total to 200)
- [x] `retention_bytes_keeps_all_when_under_limit` — create 3 segments totaling 150 bytes, configure `max_bytes: 200`, run sweep, assert nothing deleted
- [x] `retention_segments_deletes_excess` — create 5 segments, configure `max_segments: 3`, run sweep, assert the 2 oldest are deleted
- [x] `retention_segments_keeps_all_when_under_limit` — create 3 segments, configure `max_segments: 5`, run sweep, assert nothing deleted
- [x] `retention_combined_policies_most_aggressive_wins` — create 5 segments, configure `max_age_secs` that would delete 1 and `max_segments` that would delete 2, assert 2 are deleted (the larger deletion wins)
- [x] `retention_respects_min_live_offset_floor` — create 3 old segments (offsets 0, 1000, 2000), all eligible for age-based deletion, but set `min_live_offset` to `TopicOffset(1500)`, assert only segment 0 is deleted (segment 1000 is protected because subscriber needs offset 1500 which is within it)
- [x] `retention_no_config_skips_namespace` — configure a namespace with `RetentionPolicy::default()` (inactive), run sweep, assert no segments scanned or deleted
- [x] `retention_compact_before_delete_order` — instrument or observe that `manager.compact()` is called before `storage.delete_segments()` for a given topic in a single sweep iteration (can verify via ordering of side effects in mocks)

#### Integration tests — `tests/retention_integration.rs`

- [x] `retention_task_deletes_old_segments_on_disk` — set up `FsStorage` with `TempDir`, publish enough messages to create 3+ segment files (use a small `segment_max_bytes`), manually backdate the mtime of older segments via `filetime` crate or `std::fs::set_permissions` workaround, configure `max_age_secs: 1`, sleep briefly, run a sweep, assert the old segment files are gone from disk and the newest segment remains
- [x] `retention_task_preserves_live_subscriber_data` — set up a full stack (engine + manager + storage), subscribe a client, publish messages across 3 segments, configure aggressive retention (`max_segments: 1`), run a sweep, assert the subscriber can still consume from its current offset (not silently skipped)
- [x] `retention_survives_empty_topic_directories` — create the directory structure for a topic but with no `.seg` files, run a sweep, assert no errors or panics
- [x] `retention_multiple_namespaces_independent` — configure two namespaces with different retention policies, create segments for both, run a sweep, assert each namespace's segments are managed according to its own policy

#### `StorageBackend` trait additions — `src/storage.rs` (MockStorage)

- [x] `delete_segments_removes_files_before_offset` — create segment files at offsets 0, 1000, 2000 in a `TempDir`, call `delete_segments(ns, topic, TopicOffset(1500))`, assert segments 0 and 1000 are deleted, segment 2000 remains, return value is 2
- [x] `delete_segments_returns_zero_when_none_match` — create segments at offsets 1000, 2000, call `delete_segments(ns, topic, TopicOffset(500))`, assert nothing deleted, return value is 0
- [x] `delete_segments_empty_directory` — call on a topic with no segments, assert returns 0 without error
- [x] `segments_dir_returns_correct_path` — verify `segments_dir(ns, topic)` returns `{data_dir}/{ns}/{topic}/segments/`

```bash
cargo test retention           # retention unit + integration tests
cargo test min_live_offset     # namespace manager offset floor tests
cargo test delete_segments     # storage deletion tests
cargo test scan_segments       # segment scanning tests
cargo test -- --nocapture retention  # with logs visible
```

---

## Files Changed

| File | Change |
|---|---|
| `src/types.rs` | Change `MessageEnvelope.timestamp` from `SystemTime` to `u64`, add `now_millis()` |
| `src/codec/frame.rs` | Add `offset: TopicOffset` and `timestamp_millis: u64` to `Frame::Deliver` |
| `src/codec/encoder.rs` | Encode offset + timestamp in Deliver frame |
| `src/codec/decoder.rs` | Decode offset + timestamp from Deliver frame |
| `src/server/tcp/session.rs` | Use `now_millis()`, populate new Deliver fields |
| `src/namespace/manager.rs` | Update latency metric to use u64 math, add `min_live_offset()`, populate Deliver fields |
| `src/namespace.rs` | Add `min_live_offset()` to `NamespaceManager` trait |
| `src/server/http/routes.rs` | Add timestamps to publish response and topic info |
| `src/queue/topic.rs` | Add `oldest_timestamp`/`newest_timestamp` (u64) to `TopicStats` |
| `src/config.rs` | `RetentionPolicy` struct, add to `NamespaceConfig`, interval on `StorageConfig` |
| `src/storage.rs` | Add `delete_segments()` and `segments_dir()` to trait |
| `src/retention.rs` | **New file** — `RetentionTask` with directory-scan sweep |
| `src/main.rs` | Spawn retention task at startup |
| `config/config.example.toml` | Document retention fields |
| `tests/retention_integration.rs` | **New file** — integration tests |

## Dependencies

No new crate dependencies. All needed functionality is available from `std::time`, `tokio`, and existing deps.

## Implementation Order

```
Phase 1 (u64 timestamp internally)  ← foundational, everything else builds on this
    ↓
Phase 2 (wire protocol: offset + timestamp in Deliver)  ← can ship with Phase 1
    ↓
Phase 3 (HTTP timestamps)           ← can ship with Phase 1
    ↓
Phase 4 (retention config)          ← required by Phase 5
    ↓
Phase 5 (retention task)            ← depends on 4
```

Phases 1, 2, and 3 should ship together as one PR (they all touch `MessageEnvelope.timestamp`). Phase 4 can be a separate PR. Phase 5 depends on Phase 4.

## Test Summary

| Phase | Unit Tests | Integration Tests | Total New |
|-------|-----------|-------------------|-----------|
| 1 — u64 timestamp | 3 new + existing pass | — | ~3 |
| 2 — wire protocol | 10 new + proptest update | 1 (session deliver) | ~11 |
| 3 — HTTP timestamps | 4 new (topic stats) | 3 (HTTP routes) | ~7 |
| 4 — retention config | 8 new | — | ~8 |
| 5 — retention task | 17 new (scan + sweep + min_live_offset) | 4 new + 4 storage trait | ~25 |
| **Total** | | | **~54** |

## Done When

- `MessageEnvelope.timestamp` is `u64` millis — no `SystemTime` anywhere in message handling
- `Frame::Deliver` includes `offset` and `timestamp_millis` on the wire
- HTTP publish response includes `timestamp_millis`
- HTTP topic info includes `oldest_timestamp_millis` and `newest_timestamp_millis`
- `RetentionPolicy` struct with `max_age_secs`, `max_bytes`, `max_segments` configurable per namespace
- Background retention task scans segment directories and deletes segments according to policy
- Retention bounded by `min_live_offset` — never deletes messages active subscribers still need
- Compact-then-delete ordering enforced — crash-safe
- Retention metrics emitted to Prometheus
- All existing tests pass, ~54 new tests cover timestamp and retention behavior
- Config example documents all new fields
