# Phase 7: Namespace Manager — COMPLETED

## Goal
Per-namespace orchestration: manage subscribers, consumers, and route messages between queue engine and connected clients.

## Status: Complete

Implementation completed with all unit tests (27 namespace-specific tests) and integration tests (4) passing. Post-implementation review identified and resolved 7 additional correctness and design improvements.

## Files
- `src/namespace.rs` — trait definitions
- `src/namespace/manager.rs` — `DefaultNamespaceManager<Q>` implementation + 22 unit tests
- `src/namespace/state.rs` — delivery state structs + 7 unit tests
- `src/error.rs` — 3 new error variants added
- `src/config.rs` — `max_redelivery_attempts`, `dlq_topic_suffix` added to `NamespaceConfig`
- `src/codec/frame.rs` — new error variants mapped in `ErrorCode::from_error()`
- `src/queue.rs` — `QueueEngine` trait converted to `fn -> impl Future + Send`
- `src/auth.rs` — `MockAuthenticator` visibility widened for integration tests
- `tests/common/mod.rs` — shared test helpers
- `tests/namespace_integration.rs` — 4 integration tests
- `Cargo.toml` — self dev-dependency for `test-support` feature

## Implemented API Surface

### namespace.rs (traits)

```rust
pub trait NamespaceManager: Send + Sync {
    fn connect(&self, ns, client: ClientId)
        -> impl Future<Output = Result<()>> + Send;
    fn subscribe(&self, ns, topic, client, group: Option<GroupName>)
        -> impl Future<Output = Result<mpsc::Receiver<PublishedMessage>>> + Send;
    fn unsubscribe(&self, ns, topic, client, group: Option<GroupName>)
        -> impl Future<Output = Result<()>> + Send;
    fn publish(&self, ns, topic, msg: MessageEnvelope)
        -> impl Future<Output = Result<TopicOffset>> + Send;
    fn ack(&self, ns, topic, client, message_id: MessageId)   // NOTE: MessageId, not TopicOffset
        -> impl Future<Output = Result<()>> + Send;
    fn disconnect(&self, client: &ClientId)
        -> impl Future<Output = Result<()>> + Send;
    fn list_topics(&self, ns) -> impl Future<Output = Result<Vec<TopicName>>> + Send;
    fn topic_info(&self, ns, topic) -> impl Future<Output = Result<TopicInfo>> + Send;
    fn shutdown(&self)
        -> impl Future<Output = Result<()>> + Send;
}

pub trait PublishInterceptor: Send + Sync {
    fn intercept_publish(&self, ns, topic, msg: MessageEnvelope)
        -> Pin<Box<dyn Future<Output = Result<MessageEnvelope>> + Send + '_>>;
}

pub trait DeliveryInterceptor: Send + Sync {
    fn intercept_delivery(&self, ns, topic, client, msg: MessageEnvelope)
        -> Pin<Box<dyn Future<Output = Result<Option<MessageEnvelope>>> + Send + '_>>;
}
```

Key differences from plan:
- `connect` added in plan 11.3 — registers client connection, checks per-namespace limit (atomic CAS), records client→namespace mapping for disconnect.
- `subscribe` returns `mpsc::Receiver<PublishedMessage>` (not `Result<()>`). The manager holds the `Sender` side and pushes via delivery tasks.
- `ack` takes `MessageId` (not `TopicOffset`). Matches the wire protocol (`Frame::Ack { message_id }`). Internal lookup resolves MessageId → TopicOffset via inflight map scan.
- `list_topics` and `topic_info` added in plan 9 for HTTP query endpoints.
- `shutdown` added for clean delivery task lifecycle management.
- Interceptor traits use `Pin<Box<dyn Future>>` (not `#[async_trait]`) for dyn-compatibility.
- All `NamespaceManager` methods use `fn -> impl Future + Send` pattern (not `async fn`) matching `StorageBackend`.

### manager.rs (constructor)

```rust
// Constructor uses config structs (DD-018) instead of positional params:
DefaultNamespaceManager::new(
    engine: Arc<Q>,
    namespace_ids: impl IntoIterator<Item = NamespaceId>,
    quotas: impl IntoIterator<Item = (NamespaceId, NamespaceQuota)>,
    config: ManagerConfig,
) -> Self

// ManagerConfig with defaults:
ManagerConfig {
    max_redelivery_attempts: 5,
    dlq_topic_suffix: "__dlq",
    delivery_channel_capacity: 256,
    max_consumer_groups_per_topic: 1_000,
    slow_consumer_policy: SlowConsumerPolicy::DropForSlow,
    slow_consumer_threshold: 100,
    slow_consumer_disconnect_after: 1_000,
}
```

### namespace/state.rs (data structures)

```rust
InflightEntry { offset, message_id, delivered_to, delivered_at, delivery_count }
SubscriberDeliveryState { next_offset, inflight, tx: mpsc::Sender, last_activity }
ConsumerGroupState { name, members: HashMap<ClientId, Sender>, next_offset, inflight,
                     round_robin_counter: AtomicU64, last_activity, emptied_at: Option<Instant> }
TopicDeliveryState { individual, groups, notify: Arc<Notify>,
                     individual_handles: HashMap<ClientId, JoinHandle<()>>,
                     group_handles: HashMap<GroupName, JoinHandle<()>> }
NamespaceState { topics: RwLock<HashMap<TopicName, Arc<Mutex<TopicDeliveryState>>>> }
```

Key additions vs plan:
- `ConsumerGroupState.members` is `HashMap<ClientId, mpsc::Sender>` (not `HashSet<ClientId>`) — each member has its own delivery channel.
- `TopicDeliveryState` stores `JoinHandle`s for delivery task lifecycle management.
- `notify: Arc<Notify>` per topic for waking delivery tasks on publish.
- `emptied_at: Option<Instant>` on groups for grace period tracking.
- `evict_expired_inflight(timeout)` method for clearing stale inflight entries that block compaction.
- `abort_all_handles()` method for shutdown.

### namespace/manager.rs (DefaultNamespaceManager)

```rust
pub struct DefaultNamespaceManager<Q: QueueEngine> {
    engine: Arc<Q>,
    namespaces: Arc<RwLock<HashMap<NamespaceId, Arc<NamespaceState>>>>,
    publish_interceptors: Vec<Box<dyn PublishInterceptor>>,
    delivery_interceptors: Arc<Vec<Box<dyn DeliveryInterceptor>>>,
    client_subscriptions: ClientSubscriptionMap,  // reverse index
    max_redelivery_attempts: u32,
    dlq_topic_suffix: String,
    delivery_channel_capacity: usize,
    inflight_timeout: Duration,
    shutdown_tx: watch::Sender<bool>,
}
```

Generic over `Q: QueueEngine` (not `Box<dyn>` — trait uses `impl Future + Send`, not dyn-compatible).

## Prerequisites Completed (Step 0)

- [x] `QueueEngine` trait converted from `async fn` to `fn -> impl Future + Send` (`src/queue.rs`)
- [x] Error variants added: `NotSubscribed`, `DuplicateSubscription`, `GroupNotFound` (`src/error.rs`)
- [x] Error variants mapped in `ErrorCode::from_error()` (`src/codec/frame.rs`)
- [x] `NamespaceConfig` extended with `max_redelivery_attempts`, `dlq_topic_suffix` (`src/config.rs`)
- [x] `MockAuthenticator` visibility widened to `#[cfg(any(test, feature = "test-support"))]` (`src/auth.rs`)

## Decisions Made

### Resolved from Open Questions

- **Namespace lifecycle**: Namespaces are pre-populated from config at construction time. Not dynamic.
- **Consumer group load-balancing**: Round-robin only (via `AtomicU64` counter modulo member count).
- **Grace period for empty groups**: 60 seconds (`EMPTY_GROUP_GRACE_PERIOD`). Lazy cleanup on next subscribe/publish.
- **Max members per consumer group**: No limit imposed.

### Key Implementation Decisions

- **Delivery via Notify-based per-subscriber tasks**: Each subscriber/consumer group gets a spawned `tokio::task` that loops: read offset → consume from engine → send to channel. Tasks wait on `Arc<Notify>` when caught up, woken by publish.
- **Notify race fix**: Delivery tasks register `notify.notified_owned()` *before* calling `engine.consume_from()`. If a publish happens between consume returning `None` and awaiting, the notification is not lost.
- **Three-level locking**: `RwLock<HashMap<NamespaceId, Arc<NamespaceState>>>` → `RwLock<HashMap<TopicName, Arc<Mutex<TopicDeliveryState>>>>` → individual/group state under Mutex. Avoids cross-topic contention within a namespace.
- **Lock discipline**: Never hold `TopicDeliveryState` mutex while calling `engine.consume_from()`. Acquire briefly to read offset, release, do engine call, re-acquire to update.
- **Client reverse index**: `HashMap<ClientId, Vec<(NamespaceId, TopicName, Option<GroupName>)>>` enables O(1) lookup for disconnect cleanup and ack group resolution.
- **Ack by MessageId**: The `ack` method accepts `MessageId` (matching the wire protocol) and scans the inflight map to find the corresponding `TopicOffset`. Uses reverse index for O(1) group lookup.
- **Task lifecycle management**: `JoinHandle`s stored in `TopicDeliveryState`. `shutdown()` sends signal via `watch::channel` and aborts all handles.
- **Group task respawn**: On rejoin after all members left, checks `JoinHandle::is_finished()` and respawns if dead.
- **Inflight timeout**: Expired inflight entries (default 300s) are evicted during `ack` before computing `minimum_live_offset`, preventing stale entries from blocking compaction.
- **DLQ guard**: Messages published to DLQ topics (suffix match) skip publish interceptors to avoid infinite loops.
- **Channel capacity**: Default 256 messages per subscriber channel.

## Test Coverage

### Unit Tests (27 namespace tests)

State tests:
- [x] `minimum_live_offset` returns correct minimum across mixed individual/group cursors
- [x] `minimum_live_offset` includes inflight entries in minimum
- [x] `minimum_live_offset` returns `None` when no subscribers
- [x] `pick_member` round-robins correctly across N members
- [x] `pick_member` returns `None` for empty group
- [x] `evict_expired_inflight` removes old entries
- [x] `evict_expired_inflight` keeps fresh entries

Manager tests:
- [x] Construction with configured namespaces initializes correct entries
- [x] `DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>>` is `Send + Sync`
- [x] Publish to configured namespace succeeds
- [x] Publish to unconfigured namespace → `NamespaceNotFound`
- [x] Publish interceptor modifies message
- [x] Publish interceptor rejects message
- [x] Individual subscribe → publish → subscriber receives message
- [x] Multiple individual subscribers fan-out
- [x] Consumer group delivers to exactly one member
- [x] Consumer group round-robins across members
- [x] Mixed individual + group delivery
- [x] Ack by MessageId advances cursor and triggers compaction
- [x] DLQ routing on max redelivery
- [x] DLQ disabled drops message
- [x] Disconnect cleans up individual subscriptions
- [x] Disconnect group member keeps group with remaining members
- [x] Empty group retains state during grace period
- [x] Shutdown stops delivery tasks
- [x] Group task respawns after all members leave and rejoin
- [x] Ack group member uses reverse index

### Integration Tests (4)

- [x] Concurrent subscribe/unsubscribe during publish (10 tasks, 50 messages) — no deadlocks
- [x] Concurrent consumer group join/leave (5 tasks, 50 messages) — no message loss
- [x] Fan-out to 20 individual subscribers + 3 consumer groups × 3 members, 10 messages — correct delivery
- [x] Slow subscriber doesn't block others

## Post-Implementation Improvements Applied

After initial implementation, a review identified and resolved 7 issues:

1. **Task lifecycle management**: Delivery task `JoinHandle`s stored in state. `shutdown()` method added to trait and implementation.
2. **Group task respawn on rejoin**: Subscribe checks `is_finished()` on existing group handles and respawns dead tasks.
3. **Ack by MessageId**: Changed from `TopicOffset` to `MessageId` to match wire protocol.
4. **Notify race fix**: `notified_owned()` registered before `consume_from()` call.
5. **Inflight timeout**: Expired entries evicted during ack to prevent compaction blocking.
6. **Integration test timing**: Fan-out test reduced from 100×50 to 20×10 for CI reliability.
7. **Ack O(groups) scan**: Uses reverse index for O(1) group lookup instead of linear scan.

---

## Implementation Guidance from Plan 6

> **These notes capture lessons learned and concrete technical decisions from implementing the queue engine (plan 6) that directly affect how plan 7 should be built.**

### Step 0: Resolve these before writing code — ALL RESOLVED

**1. Generic type propagation from QueueEngine** — Resolved: `DefaultNamespaceManager<Q: QueueEngine>` mirrors how `InMemoryQueueEngine<S>` is generic over `StorageBackend`.

**2. Trait async method Send bounds** — Resolved: `QueueEngine` converted to `fn -> impl Future + Send` pattern. `NamespaceManager` trait uses the same pattern. Delivery tasks call engine methods from `tokio::spawn` successfully.

**3. Error enum additions** — Resolved: `NotSubscribed`, `DuplicateSubscription`, `GroupNotFound` added.

**4. Don't use `#[async_trait]`** — Resolved: Interceptor traits use `Pin<Box<dyn Future + Send>>` for dyn-compatibility. Main traits use `fn -> impl Future + Send`.

### Concrete API surface from plan 6 — All utilized correctly

```rust
// All 5 QueueEngine methods called by namespace manager:
engine.publish(ns, topic, msg)        // in publish path
engine.consume_from(ns, topic, offset) // in delivery tasks
engine.head_offset(ns, topic)          // in subscribe (initial offset)
engine.compact(ns, topic, min_offset)  // in ack (after minimum_live_offset)
engine.shutdown()                      // not called by namespace manager (separate concern)
```

### Testing pattern — Applied

Used `InMemoryQueueEngine<MockStorage>` for all tests (real engine, mock storage). `MockStorage` gated behind `#[cfg(any(test, feature = "test-support"))]`. Integration tests use `test-support` feature via self dev-dependency in `Cargo.toml`.

### Concurrency pattern — Applied with enhancement

Used the two-level locking pattern from queue engine but with three levels for finer granularity:
- `RwLock<HashMap<NamespaceId, Arc<NamespaceState>>>`
- `RwLock<HashMap<TopicName, Arc<Mutex<TopicDeliveryState>>>>`
- Individual/group state under the topic Mutex

This avoids cross-topic contention within a namespace.
