# Phase 6: Queue Engine — COMPLETED

## Goal
Per-topic in-memory indexed log with ordering guarantees, offset-based access, and GC/compaction.

## Files
- `src/queue.rs`
- `src/queue/engine.rs`
- `src/queue/topic.rs`

**Note:** `src/queue/delivery.rs` exists as a stub but is NOT part of this plan. Delivery semantics (fan-out, consumer groups, ack tracking) are entirely plan 7's responsibility. The `delivery.rs` file should be removed from the queue module — if shared delivery types are needed (e.g., `BackpressurePolicy`), they live in `src/types.rs` or `src/queue/topic.rs`.

## Work Items

### queue.rs (root)
- [x] `trait QueueEngine` — async interface for the indexed log
  - `async fn publish(&self, ns, topic, msg) -> Result<PublishResult>` — `PublishResult { offset, evicted_messages }` (changed from `Result<TopicOffset>` in retro — see DD-020)
  - `async fn consume_from(&self, ns, topic, offset) -> Result<Option<PublishedMessage>>`
  - `async fn head_offset(&self, ns, topic) -> Result<TopicOffset>`
  - `async fn compact(&self, ns, topic, min_offset) -> Result<CompactResult>` — `CompactResult { bytes_freed, messages_freed }` (changed from `Result<()>` in retro)
  - `async fn list_topics(&self, ns) -> Vec<TopicName>`
  - `async fn topic_stats(&self, ns, topic) -> Option<TopicStats>`
  - `async fn shutdown(&self) -> Result<()>`
- [x] Re-export Topic, InMemoryQueueEngine, BackpressurePolicy

### queue/topic.rs
- [x] `BackpressurePolicy` enum — `Reject`, `DropOldest`, `Block`
  - `Reject` (default): publish returns `Error::QueueFull` immediately
  - `DropOldest`: evict oldest undelivered messages to make room
  - `Block`: hold the publish until space is available (with timeout; deferred to plan 11)
- [x] `Topic` struct:
  - `buffer: VecDeque<MessageEnvelope>` — the in-memory indexed log
  - `next_offset: TopicOffset` — next offset to assign on publish (monotonically increasing)
  - `base_offset: TopicOffset` — lowest retained offset after compaction (initially 0)
  - `max_depth: usize` — capacity limit (number of messages)
  - `bytes_used: usize` — running sum of payload sizes for memory tracking
  - `backpressure: BackpressurePolicy` — policy when buffer is full
- [x] Messages are appended with monotonically increasing `TopicOffset` values (sequential within a topic, no gaps)
- [x] The log is append-only within a topic — no removal on consume
- [x] Messages stay in the buffer until a GC/compaction pass determines all cursors have advanced past them
- [x] `publish(msg) -> Result<(TopicOffset, usize, usize)>`:
  - If buffer is at `max_depth`, apply backpressure policy:
    - `Reject` → return `Error::QueueFull`
    - `DropOldest` → `pop_front()`, advance `base_offset`, subtract from `bytes_used`
  - Append message, assign `next_offset`, increment counter, add payload size to `bytes_used`
  - Returns `(assigned_offset, evicted_bytes, evicted_messages)` — engine uses evicted_bytes for global total_bytes; manager uses evicted_messages for namespace message_count (see DD-020)
- [x] `consume_from(offset) -> Option<&MessageEnvelope>`:
  - If `offset < base_offset` → return `None` (compacted away; namespace manager handles catchup)
  - If `offset >= next_offset` → return `None` (at/past head, caught up)
  - Otherwise: index = `offset.0 - base_offset.0`, return `&buffer[index]`
  - Pure read, no state mutation
- [x] `compact(min_offset) -> CompactResult`:
  - Drain all messages with offsets below `min_offset` from the front of the `VecDeque`
  - Update `base_offset` to `min_offset`
  - Subtract drained payload sizes from `bytes_used`
  - Returns `CompactResult { bytes_freed, messages_freed }` (changed from plain `usize` in retro)
- [x] `pub fn depth(&self) -> usize` — number of messages currently in buffer
- [x] `pub fn bytes_used(&self) -> usize` — accessor for memory tracking
- [x] `pub fn head_offset(&self) -> TopicOffset` — returns `next_offset` (one past the last written)
- [x] `pub fn base_offset(&self) -> TopicOffset` — returns lowest retained offset

### queue/engine.rs
- [x] `InMemoryQueueEngine<S: StorageBackend>` struct:
  - `topics: Arc<RwLock<TopicMap>>` (type alias for `HashMap<(NamespaceId, TopicName), Arc<Mutex<Topic>>>`)
  - `storage: Arc<S>`
  - `total_bytes: AtomicU64` — global byte tracking across all topics
  - `default_max_depth: usize` — default per-topic capacity (from config)
  - `default_backpressure: BackpressurePolicy` — default policy (from config)
  - `shutdown_tx: watch::Sender<bool>` — signals background flush task to stop
  - `flush_handle: Mutex<Option<JoinHandle<()>>>` — join handle for graceful shutdown
- [x] `impl QueueEngine for InMemoryQueueEngine<S>`
- [x] On publish:
  1. Read-lock the topic map to find existing topic
  2. If not found, write-lock to auto-create (double-check after acquiring write lock to avoid race)
  3. Lock the topic's Mutex
  4. Call `topic.publish(msg)` — assigns offset, enforces backpressure, returns `(offset, evicted_bytes, evicted_messages)`
  5. Capture topic `depth()` and `bytes_used()` while lock is held (for metrics)
  6. Call `storage.append()` (page cache write, NOT fsync — see flush task below)
  7. Update `total_bytes` atomic (net: added payload minus evicted bytes)
  8. Emit per-topic `fafb_queue_depth` and `fafb_queue_memory_bytes` gauges
  9. Return `PublishResult { offset, evicted_messages }`
- [x] On `consume_from`: read-lock map → lock topic Mutex → `topic.consume_from(offset)` → construct `PublishedMessage` if found
- [x] On `compact`: read-lock map → lock topic Mutex → `topic.compact(min_offset)` → update `total_bytes` atomic → return `CompactResult`
- [x] On `head_offset`: read-lock map → lock topic Mutex → return `topic.head_offset()`
- [x] On `shutdown`: signal flush task via `watch` channel → await flush handle → call `storage.shutdown()`
- [x] Startup replay:
  1. Call `storage.list_topics()` to discover all persisted namespace/topic pairs
  2. For each pair, call `storage.replay(ns, topic)` to get `Vec<PublishedMessage>`
  3. Populate `Topic` from replayed messages via `Topic::from_replay()`: restore `next_offset`, `base_offset`, `bytes_used`, respect `max_depth` by only retaining the tail
  4. Insert into the topic map
- [x] `pub fn topic_stats(&self, ns: &NamespaceId, topic: &TopicName) -> Option<TopicStats>` — returns depth, bytes, base/head offsets for observability
- [x] Background flush task: periodically (100ms) iterates all topics and calls `storage.flush()`. Runs as a `tokio::spawn` task, cancelled on shutdown via `watch` channel.

### TopicStats (observability helper)
- [x] `TopicStats` struct: `depth: usize`, `bytes_used: usize`, `base_offset: TopicOffset`, `head_offset: TopicOffset`
- [x] Used by HTTP info endpoints and future metrics exposition (plan 11)

## Implementation Notes (Deviations from Original Plan)

### `InMemoryQueueEngine<S>` is generic, not `Arc<dyn StorageBackend>`
The plan recommended `Arc<dyn StorageBackend>`, but `StorageBackend` uses async methods which are not dyn-compatible in Rust 2024. The engine is generic over `S: StorageBackend + 'static` instead. This means downstream consumers carry the type parameter (e.g., `InMemoryQueueEngine<FsStorage>`). Consider a type alias in `lib.rs` if the generic propagation becomes noisy.

### `StorageBackend` trait changed from `async fn` to `impl Future + Send`
To enable `tokio::spawn` in the background flush task, `StorageBackend` methods were changed from `async fn` to `fn -> impl Future<Output = ...> + Send`. This is backward-compatible — implementations still use `async fn`. The `#[allow(async_fn_in_trait)]` annotation was removed.

### `Topic::publish` returns `(TopicOffset, usize)` not just `TopicOffset`
The second value is `evicted_bytes` — non-zero only when `DropOldest` evicts a message. This lets the engine subtract evicted bytes from the `total_bytes` atomic on each publish, preventing drift that would otherwise accumulate over time with `DropOldest` backpressure.

### `Topic::from_replay` respects `max_depth`
If replayed messages exceed `max_depth`, only the most recent `max_depth` messages are retained. This prevents the buffer from exceeding capacity after a restart when messages were appended between compaction cycles.

### Publish ordering: in-memory first, then storage
The publish path writes to the in-memory topic first, then persists to the storage page cache. If `storage.append()` fails, the message is in-memory but not durable. The reverse ordering would risk offset gaps on storage failure. Page-cache writes virtually never fail, so the current ordering is acceptable. Documented in code.

## Test Summary

| Category | Count | Description |
|---|---|---|
| topic.rs unit tests | 17 | Offsets, consume, compact, backpressure, replay, bytes tracking |
| engine.rs unit tests | 14 | Auto-create, sequential offsets, consume, compact, stats, total_bytes, shutdown, concurrency, replay, DropOldest |
| queue_integration.rs | 4 | Concurrent publish, concurrent consume, publish+compact, replay from storage |
| **Total new tests** | **35** | |

## Key Decisions
- **Indexed log, not FIFO** — messages are assigned per-topic sequential `TopicOffset` values and retained until GC determines all consumers have advanced past them
- **`consume_from` is offset-aware** — callers (namespace manager) control their own cursor position; the queue engine has no concept of "current consumer position"
- **`consume_from` returns `Option<PublishedMessage>`** — uses the existing `PublishedMessage` type (envelope + offset pair) for consistency with the storage layer, rather than an ad-hoc tuple
- **GC/compaction is triggered by the namespace manager** reporting the minimum live offset across all subscribers and consumer groups for a topic
- **Compaction is in-memory only** — `compact()` removes messages from the `VecDeque` but does NOT delete on-disk segments. Storage-level cleanup (segment deletion) is handled by plan 11's retention policies, triggered separately
- **The queue engine does NOT track who has consumed what** — that responsibility belongs entirely to the namespace manager
- At-least-once delivery (simplest correct semantic)
- Queue engine is delivery-mode agnostic — it stores messages once; the namespace manager handles fan-out to individual subscribers and load-balanced delivery to consumer groups
- Topics are auto-created on first publish — no explicit `create_topic` call required from clients. Concurrency-safe via double-check locking (read-lock first, then write-lock with re-check)
- Redelivery timeout: 30 seconds default, configurable per-namespace via config
- StorageBackend is injected via dependency injection (constructor parameter)
- **`VecDeque` over `Vec` or ring buffer** — `VecDeque` supports O(1) front removal for compaction, grows dynamically (unlike fixed ring buffer), and index arithmetic is simple with `base_offset` tracking. Ring buffers have fixed capacity which conflicts with the dynamic capacity model. Benchmark against alternatives per plan 11 if needed.
- **`base_offset` tracks compaction state** — after `compact(min_offset=100)`, `base_offset` becomes 100 and `consume_from(50)` returns `None` (compacted away). The namespace manager is responsible for handling this case (e.g., resetting the subscriber to the current base offset).
- **`storage.append()` is a page cache write, not an fsync** — the publish hot path does not call `flush()`. A background flush task calls `storage.flush()` periodically to amortize fsync cost. This matches the storage layer's design (plan 5) and the batching recommendation from plan 11 section 4.
- **Locking strategy**: `tokio::sync::RwLock` on the topic map (read-lock for existing topic access, write-lock only on first-ever publish to a new topic). Each `Topic` wrapped in `Arc<tokio::sync::Mutex<Topic>>` so different topics never contend. Mirrors storage layer's proven two-level locking pattern. Plan 11 migrates the outer map to `DashMap` — the trait boundary insulates consumers from this change.
- **MessageId is assigned by the caller** — the namespace manager (plan 7) or TCP server (plan 8) assigns `MessageEnvelope.id` before calling `publish()`. The queue engine does not assign or validate `MessageId`.
- **Bytes conversion for fan-out happens at the boundary** — `MessageEnvelope.payload` remains `Vec<u8>` (compatible with bincode serialization for storage). When the namespace manager fans out a message to N subscribers, it converts `Vec<u8>` to `Bytes::from(payload)` once, then clones the `Bytes` handle for O(1) per-subscriber fan-out. This avoids touching the storage serialization path.

## Shared Abstraction: TopicLog

The queue engine is one of three subsystems that participate in the **TopicLog** abstraction:

- **Queue Engine (this plan)**: Owns the in-memory indexed log. Assigns `TopicOffset` values on publish. Serves reads via `consume_from(offset)`. Runs GC/compaction when notified of the minimum live offset.
- **Storage (plan 5)**: Persists the log to disk. The queue engine calls `StorageBackend::append()` on each publish (page cache write) and `StorageBackend::replay()` on startup to reconstruct the in-memory log. `StorageBackend::list_topics()` provides topic discovery for replay bootstrapping.
- **Namespace Manager (plan 7)**: The sole consumer of `consume_from()`. Manages all cursor state (subscriber offsets, group offsets). Periodically computes the minimum live offset across all cursors for a topic and calls `compact()`.

### Trait Boundary Clarification

The `QueueEngine` trait is the **write path and read path** for the log:

- `publish()` is the only way messages enter the log (assigns offset, persists to page cache, returns offset)
- `consume_from()` is a pure read — does not mutate state, does not advance any cursor. Returns `Option<PublishedMessage>` (not a tuple).
- `compact()` is the GC entry point — called by the namespace manager, not by consumers. In-memory only; does not affect on-disk segments.
- `head_offset()` returns the next offset to be assigned (one past the last written message)
- `shutdown()` flushes the storage layer and cancels the background flush task
- The queue engine does NOT track who has consumed what — that responsibility belongs entirely to the namespace manager

See also: [plan 5 (Storage)](plan_5_storage.md#shared-abstraction-topiclog), [plan 7 (Namespace Manager)](plan_7_namespace.md#shared-abstraction-topiclog)

## Recommendations

- **Batch appends and flush periodically** rather than per-message when wiring up the queue engine. The publish hot path calls `storage.append()` (page cache only). A background task calls `storage.flush()` on an interval (default 100ms) to amortize fsync cost. This matches the storage layer's design — `append` writes to the OS page cache, `flush` does the fsync.
- ~~**Use `Arc<dyn StorageBackend>` for the storage dependency**~~ — **Not viable.** `StorageBackend` async methods are not dyn-compatible. Use a generic `S: StorageBackend` parameter instead. See "Implementation Notes" above.
- **`MockStorage` visibility will need fixing.** When you first try to use `rust_fafb_q::storage::mock::MockStorage` from a test, it won't be publicly accessible. Flag it and fix it then (add `pub` to the mock module and struct).

## Resolved Questions

| Question | Resolution |
|---|---|
| Max inflight per consumer? | **No.** The queue engine is a log. Read-ahead limits are a delivery concern (plan 7). Keep the engine simple. |
| Ring buffer vs. `Vec` with index offset? | **`VecDeque` with `base_offset` tracking.** Ring buffers have fixed capacity which conflicts with the dynamic capacity model (plan 11 per-topic max depth). `VecDeque` grows/shrinks naturally and has O(1) front removal for compaction. Benchmark against alternatives per plan 11 if needed. |
| What happens on `consume_from` after compaction? | Returns `None` if `offset < base_offset`. The namespace manager handles catchup (e.g., resetting subscriber to current base offset). |
| Who assigns `MessageId`? | The caller (namespace manager or TCP server) assigns `MessageEnvelope.id` before calling `publish()`. The queue engine does not assign or validate `MessageId`. |
| How does startup replay discover topics? | Via `StorageBackend::list_topics()` (added as a plan 5 amendment). |
| Does `compact()` affect on-disk segments? | No. `compact()` is in-memory only. Storage-level retention is handled by plan 11's retention policies. |
| `Arc<dyn StorageBackend>` or generic? | **Generic `S: StorageBackend`.** Async fns in traits are not dyn-compatible. The generic approach is zero-cost. |
| Does `from_replay` respect `max_depth`? | **Yes.** Only the most recent `max_depth` messages are retained, preventing buffer overflow on restart. |
| How does `total_bytes` stay accurate with `DropOldest`? | `Topic::publish` returns evicted bytes. Engine subtracts them from the atomic on each publish. |

## Required Amendments to Other Plans

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

### Plan 5 (Storage) — Amendment: `StorageBackend` trait Send bounds (COMPLETED)
- [x] Changed all `StorageBackend` methods from `async fn` to `fn -> impl Future<Output = ...> + Send`
- [x] Removed `#[allow(async_fn_in_trait)]` from trait definition
- [x] Backward-compatible: all implementations still use `async fn`
- [x] Required so that `tokio::spawn` in background flush task can hold an `Arc<S>` reference

### Error Enum — Amendment: `QueueFull` (COMPLETED)
- [x] Added `Error::QueueFull { namespace: NamespaceId, topic: TopicName }` variant to `src/error.rs`
- [x] Mapped to `ErrorCode::InternalError` in codec's `ErrorCode::from_error()` (wire code `0x0009`)

## Done When
- [x] Unit tests: publish returns sequential offsets with no gaps
- [x] Unit tests: `consume_from(offset)` returns correct `PublishedMessage`
- [x] Unit tests: `consume_from` past head returns `None`
- [x] Unit tests: `consume_from` after compaction with offset below `base_offset` returns `None`
- [x] Unit tests: `consume_from` after compaction with valid offset returns correct message
- [x] Unit tests: compaction removes messages below `min_offset` and updates `base_offset`
- [x] Unit tests: compaction updates `bytes_used` correctly
- [x] Unit tests: buffer respects capacity limit (max_depth)
- [x] Unit tests: backpressure policy `Reject` returns `Error::QueueFull` when topic is at capacity
- [x] Unit tests: backpressure policy `DropOldest` evicts oldest message on full topic
- [x] Unit tests: `bytes_used` tracking is accurate after publish and compact
- [x] Unit tests: `depth()` returns correct count after publish and compact
- [x] Unit tests: topic auto-created on first publish
- [x] Unit tests: concurrent topic auto-creation does not duplicate topics
- [x] Unit tests: shutdown calls storage shutdown
- [x] Test with mock StorageBackend (no filesystem)
- [x] Integration test: concurrent publish from multiple tasks to the same topic — offsets are sequential with no gaps or duplicates
- [x] Integration test: concurrent `consume_from` across multiple tasks — all readers see consistent log state
- [x] Integration test: publish + compact under concurrent load — no messages lost that are still referenced by a live offset
- [x] Integration test: replay from storage correctly restores `base_offset` and `next_offset`
- [x] Additional: `DropOldest` eviction bytes tracking at engine level
- [x] Additional: `from_replay` truncation to `max_depth`
- [x] Additional: `Block` falls back to `Reject` in v1
