# Phase 11: Performance & Scalability

## Goal
Ensure the server handles thousands of namespaces, thousands of concurrent consumers/publishers per namespace, and large queue topics without degradation. Establish resource limits, backpressure strategies, and benchmarking infrastructure to validate performance under load.

## Guiding Principles
- **Excellent DX**: Performance tuning should not complicate the API, config, or extension surface. Defaults must be production-ready out of the box. Knobs exist but are optional.
- **Predictable under load**: The server should degrade gracefully (reject, shed, slow down) rather than crash, OOM, or deadlock. Operators should be able to reason about resource consumption from config alone.
- **Measure first**: Every optimization must be justified by a benchmark. No speculative optimization.

## Already Implemented (prior plans)

These items from the original plan 11 were delivered in earlier phases:

| Item | Plan | Details |
|---|---|---|
| `tcp.max_connections` (global) | Plan 8 | Semaphore-based enforcement, Error frame on reject |
| `tcp.idle_timeout_secs` | Plan 8 | Ping/Pong keepalive resets timer |
| `tcp.max_message_size` | Plan 3 | Codec rejects oversized frames |
| Per-topic `max_depth` | Plan 6/10 | `[queue] max_depth` config, enforced in `Topic::publish()` |
| `backpressure_policy` | Plan 6/10 | `Reject`, `DropOldest`, `Block` — configurable via `[queue]` |
| `Topic.bytes_used` tracking | Plan 6 | Running sum of payload sizes, adjusted on publish/compact |
| Lock-free auth | Plan 4 | `ApiKeyStore` uses `Arc<[KeyEntry]>` — no runtime locks |
| Codec benchmarks | Plan 3 | `benches/codec_bench.rs` — encode, decode, roundtrip at 64B/1KB/64KB |
| Docker test infrastructure | Plan 11 | `Dockerfile`, `docker-compose.test.yml`, `scripts/docker-test.sh` |

## Sub-plans Overview

Phase 11 is split into seven sub-plans, ordered by dependency. Each is independently shippable.

| Sub-plan | Focus | Depends on | Status |
|---|---|---|---|
| **11.1** | Benchmarks & Baselines | — | COMPLETED |
| **11.2** | Memory Limits & Resource Budgets | — | COMPLETED |
| **11.3** | Connection Scaling & Delivery | — | COMPLETED |
| **11.4** | Storage I/O Performance | — | COMPLETED |
| **11.5** | Concurrency & Lock Granularity | 11.1 (baselines) | DEFERRED |
| **11.6** | Observability & Metrics | — | COMPLETED |
| **11.7** | Load Testing & Validation | 11.2, 11.3 | COMPLETED |

### Recommended execution order

```
11.1 (benchmarks)  ──┐
                     ├──► 11.5 (concurrency — only if benchmarks show contention)
11.2 (memory)  ──────┤
                     ├──► 11.7 (load testing — validates 11.2 + 11.3)
11.3 (connections)  ──┘
11.4 (storage)  ──────── (independent, can run in parallel with 11.2/11.3)
11.6 (observability) ─── (independent, useful alongside any other sub-plan)
```

11.1 should come first — "measure first" principle. 11.5 (concurrency) depends on having baselines to justify the added complexity. 11.7 (load testing) validates the limits added in 11.2 and 11.3. Everything else can proceed in any order.

## Resource Limits Summary

All limits configurable. Defaults chosen for a single server with 8GB RAM and moderate load.

| Resource | Scope | Default | Config Key | Status |
|---|---|---|---|---|
| Max connections | Global | 10,000 | `tcp.max_connections` | **Done** (plan 8) |
| Max connections | Per-namespace | Unlimited | `namespaces.<name>.max_connections` | **Done** (plan 11.3 + retro) |
| Idle connection timeout | Global | 300s | `tcp.idle_timeout_secs` | **Done** (plan 8) |
| Max message size | Global | 1MB | `tcp.max_message_size` | **Done** (plan 3) |
| Topic max depth | Per-topic | 100,000 msgs | `queue.max_depth` | **Done** (plan 10) |
| Backpressure policy | Global | DropOldest | `queue.backpressure_policy` | **Done** (plan 10) |
| Topic max bytes | Per-topic | 256MB | `queue.default_topic_max_bytes` | **Done** (plan 11.2) |
| Global memory budget | Global | 4GB | `queue.global_max_memory_bytes` | **Done** (plan 11.2) |
| Namespace total messages | Per-namespace | Unlimited | `namespaces.<name>.max_total_messages` | **Done** (plan 11.2) |
| Max topics per namespace | Per-namespace | 10,000 | `namespaces.<name>.max_topics` | **Done** (plan 11.2) |
| Max consumer groups per topic | Per-topic | 1,000 | `queue.max_consumer_groups_per_topic` | **Done** (plan 11.3) |
| Max open file descriptors | Global | 1,024 | `storage.max_open_fds` | **Done** (plan 11.4, `lru::LruCache`) |
| Batch flush size trigger | Per-topic | 4MB | `storage.batch_flush_size_bytes` | **Done** (plan 11.4, dirty bytes on Topic) |
| Subscriber channel buffer | Per-connection | 1,024 msgs | `tcp.subscriber_channel_buffer` | **Done** (plan 11.3) |
| Retention (age) | Per-namespace | Unlimited | `namespaces.<name>.retention.max_age_secs` | **Done** (plan 12) |
| Retention (bytes) | Per-namespace | Unlimited | `namespaces.<name>.retention.max_bytes` | **Done** (plan 12) |
| Retention (segments) | Per-namespace | Unlimited | `namespaces.<name>.retention.max_segments` | **Done** (plan 12) |
| Max segments per topic | Global | 64 | `storage.max_segments_per_topic` | **Done** (plan 11.4) |

## Performance Targets (see [baselines.md](../docs/baselines.md) for full results)

| Metric | Target |
|---|---|
| Publish throughput (single topic) | > 100,000 msgs/sec |
| End-to-end latency (publish → deliver) | p99 < 10ms (in-memory) |
| Fan-out to 1,000 subscribers | < 50ms total delivery time |
| Concurrent namespaces | 10,000 with no degradation |
| Startup replay | 1M messages in < 5 seconds |
| Memory per idle connection | < 16KB |

## Dependencies Added (across all sub-plans)

| Crate | Sub-plan | Purpose | Status |
|---|---|---|---|
| `metrics` + `metrics-exporter-prometheus` | 11.6 | Prometheus metrics exposition | **Added** |
| `lru` | 11.4 | LRU cache for fd management | **Added** |
| `dashmap` | 11.5 | Concurrent hash map | Deferred (no contention evidence) |
| `parking_lot` | 11.5 | Faster mutexes | Deferred (no contention evidence) |

## Key Decisions
- Defaults must be safe and production-ready. An operator who deploys with zero tuning should get predictable behavior, not OOM.
- Data loss must be an explicit operator choice (`DropOldest`), not a surprise. `Reject` was originally the default but `DropOldest` was chosen in plan 10 (see DD-014).
- All limits are soft-configurable. No recompile needed to tune for a specific workload.
- Benchmarks are part of the CI pipeline. Regressions are caught before merge.
- Optimization work (11.5) is gated on benchmark evidence (11.1). No speculative optimization.

---

## 11.1: Benchmarks & Baselines — COMPLETED

### Goal
Establish performance baselines for every subsystem before making any optimization changes. These baselines inform which sub-plans (especially 11.5 — concurrency) are worth pursuing and provide regression detection in CI.

### Guiding Principle
**Measure first.** No optimization work should begin without a benchmark proving the need.

### Prerequisites
- All subsystems implemented and tested (plans 1–10 complete, 263 tests pass)
- Docker test infrastructure exists (`Dockerfile`, `docker-compose.test.yml`, `scripts/docker-test.sh`)
- Codec benchmarks exist in `benches/codec_bench.rs` (encode, decode, roundtrip at 64B/1KB/64KB)
- `benches/queue_bench.rs` exists as a stub

### Work Items

#### Extend codec benchmarks
`benches/codec_bench.rs` already covers 64B, 1KB, 64KB payloads. Add:
- [x] **1MB payload** encode/decode/roundtrip — validates performance at `max_message_size` boundary. Required `make_large_codec()` (2MB limit) because framing overhead pushes encoded frame past the default 1MB `max_message_size`.
- [ ] **Error frame** encode/decode — exercises the ErrorCode path (deferred — low priority)

> **Note:** Existing codec benches are comprehensive. The 1MB case is the main gap — it's the default `max_message_size` limit and the most likely to surface allocation overhead.

#### Queue engine benchmarks
`benches/queue_bench.rs` — full rewrite from stub:
- [x] **Single-topic publish throughput**: 64B, 1KB, 64KB payloads
- [x] **Single-topic consume throughput**: Pre-fill 10K messages, cycle through offsets
- [x] **Concurrent publish (same topic)**: 10, 100 producers × 100 msgs each
- [x] **Concurrent publish (cross-topic, steady-state)**: 10, 100 pre-created topics × 100 msgs each. Pre-creates topics to isolate read-lock contention from write-lock auto-creation.
- [x] **Namespace contention (steady-state)**: 100, 1000 pre-created namespace/topic pairs × 50 msgs each
- [x] **Compact throughput**: Pre-fill 50K messages, compact in 1000-offset steps

> **Result:** Cross-topic contention scales linearly — same-topic vs cross-topic at 100 concurrent is nearly identical (~5.6 ms vs ~5.4 ms). DashMap sharding (plan 11.5) is not justified at current scale.

#### Storage benchmarks
New file: `benches/storage_bench.rs` + `[[bench]]` entry added to `Cargo.toml`.
- [x] **Append throughput**: 64B, 1KB, 64KB payloads to a single topic
- [x] **Append cross-topic**: 10, 100 topics concurrently × 100 msgs each
- [x] **Replay throughput**: 1K, 10K, 100K messages (256B payloads)
- [x] **Flush latency**: 10 writes then flush, repeated

> All storage benchmarks use `tempfile::tempdir()`. `FsStorage::init()` is called during setup, not in the benchmark loop. 100K replay in ~61 ms — extrapolates to ~610 ms for 1M, well within the 5-second target.

#### Fan-out / delivery benchmarks
New file: `benches/fanout_bench.rs` + `[[bench]]` entry added to `Cargo.toml`.
- [x] **Individual subscriber fan-out**: 1, 10, 100 subscribers. Measures publish + delivery via `tokio::time::timeout(500ms, rx.recv())`
- [x] **Consumer group delivery**: 1, 10 group members. Uses short timeout (1ms) polling to handle round-robin non-determinism
- [x] **Publish throughput (no subscribers)**: Isolates namespace manager overhead (~444 ns vs ~323 ns raw engine)
- [ ] **Ack throughput**: Pre-fill inflight maps, measure ack + compaction trigger rate (deferred)

> Uses `MockStorage`-backed engine. Fan-out scales sub-linearly: 1→100 subscribers is ~12x (not 100x), confirming delivery task parallelism works.

#### Performance target validation
- [x] Baselines recorded in [baselines.md](../docs/baselines.md)
- [x] 5 of 6 targets already met — no optimization needed for those
- [x] "Memory per idle connection" is the only unknown (requires runtime profiling, not a benchmark)

### Files

| File | Change |
|---|---|
| `benches/codec_bench.rs` | Add 1MB payload + Error frame benchmarks |
| `benches/queue_bench.rs` | Full rewrite from stub |
| `benches/storage_bench.rs` | **New** — storage layer benchmarks |
| `benches/fanout_bench.rs` | **New** — fan-out / delivery benchmarks |
| `Cargo.toml` | Add `[[bench]]` entries for `storage_bench` and `fanout_bench` |

### Verification
```bash
cargo bench                           # all benchmarks run
cargo bench -- codec                  # codec suite
cargo bench -- queue                  # queue suite
cargo bench -- storage                # storage suite
cargo bench -- fanout                 # fan-out suite
```

Docker (reproducible):
```bash
./scripts/docker-test.sh bench
./scripts/docker-test.sh bench --cpus 1 --memory 512m    # constrained profile
```

### Done When
- [x] All four benchmark suites run to completion
- [x] Baseline numbers recorded in `docs/baselines.md`
- [x] Performance targets updated with actuals (5/6 met, 1 unknown)
- [x] No existing tests broken (263 pass)
- [x] Steady-state variants for cross-topic and namespace benchmarks (pre-create topics)
- [x] Spin-wait replaced with timeout+recv in fanout benchmarks

---

## 11.2: Memory Limits & Resource Budgets — COMPLETED

### Goal
Prevent unbounded memory growth by enforcing per-topic byte limits, a global memory budget, and per-namespace quotas. Operators should be able to reason about maximum memory consumption from config alone.

### Status: Complete

All work items implemented. Per-topic byte limits, global memory budget, per-namespace quotas (message count + topic count), and per-namespace config overrides are enforced and tested. The Phase 11 retrospective additionally fixed `message_count` drift: the counter now decrements on both ack-triggered compaction (`CompactResult.messages_freed`) and publish-time DropOldest evictions (`PublishResult.evicted_messages`).

### Prerequisites
- Plan 10 complete — `[queue]` config section exists with `max_depth` and `backpressure_policy`
- `Topic.bytes_used` already tracked (plan 6) — running sum of payload sizes
- `InMemoryQueueEngine.total_bytes` already tracked (plan 6) — global `AtomicU64`

### Current State

What exists today:
- **Per-topic depth limit**: `queue.max_depth` caps messages per topic (default 100K)
- **Per-topic byte tracking**: `Topic.bytes_used` maintained on publish/compact
- **Global byte tracking**: `InMemoryQueueEngine.total_bytes` maintained via `AtomicU64`
- **Backpressure on depth**: `BackpressurePolicy` enforced when `max_depth` is hit

What's missing:
- No per-topic *byte* limit (a topic with 100K tiny messages and one with 100K 1MB messages use the same depth limit)
- No global memory ceiling (total_bytes is tracked but not enforced)
- No per-namespace quotas (one namespace can consume all memory)
- No per-namespace config overrides for depth/backpressure

### Work Items

#### Per-topic max bytes
- [x] Add `default_topic_max_bytes: u64` to `QueueConfig` (default: 268,435,456 / 256MB)
- [x] Pass to `InMemoryQueueEngine::new()` via `EngineConfig.topic_max_bytes` (constructor refactored to use config struct — see DD-018)
- [x] Enforce in `Topic::publish()`: check `bytes_used + payload.len() > max_bytes` before inserting. Apply the same `BackpressurePolicy` as depth limit.
- [x] Add config to example TOML — `default_topic_max_bytes` documented in `config/config.example.toml`

> **Note:** The depth limit and byte limit are independent — whichever is hit first triggers backpressure. This prevents both "too many small messages" and "few very large messages" from causing problems.

#### Global memory budget
- [x] Add `global_max_memory_bytes: u64` to `QueueConfig` (default: 4,294,967,296 / 4GB)
- [x] Enforce in `InMemoryQueueEngine::publish()`: check `total_bytes + payload.len() > global_max_memory_bytes` before publishing. Return `Error::ResourceExhausted`.
- [x] Log warning at 80% threshold (first time crossing) — implemented via `memory_warning_fired: AtomicBool`
- [x] Add config to example TOML — `global_max_memory_bytes` documented in `config/config.example.toml`

> **Note:** The global check happens at the engine level before the topic-level check. This prevents a race where many topics simultaneously publish past their individual limits but stay under each topic's cap. The `AtomicU64` is already there — this is a cheap check.

#### Per-namespace quotas
- [x] Add to `NamespaceConfig`:
  - `max_total_messages: Option<u64>` (default: None / unlimited)
  - `max_topics: Option<usize>` (default: None / unlimited — suggested 10,000)
- [x] Track per-namespace message count in `NamespaceState` (`AtomicU64`)
- [x] Enforce `max_topics` in namespace manager on first publish to a new topic
- [x] Enforce `max_total_messages` in namespace manager's `publish()` path
- [x] `message_count` decremented on ack-compaction (`CompactResult.messages_freed`) and publish-eviction (`PublishResult.evicted_messages`) — prevents permanent drift (retro fix)
- [x] Add config to example TOML — `max_total_messages` and `max_topics` documented in `config/config.example.toml`

> **Note:** Per-namespace quotas are a defense against one namespace monopolizing resources. They complement (don't replace) the global budget. The `max_topics` limit also bounds the number of delivery tasks spawned per namespace.

#### Per-namespace config overrides
- [x] Add optional fields to `NamespaceConfig`:
  - `topic_max_depth: Option<usize>` — overrides `queue.max_depth`
  - `backpressure_policy: Option<BackpressurePolicy>` — overrides `queue.backpressure_policy`
  - `topic_max_bytes: Option<u64>` — overrides `queue.default_topic_max_bytes`
- [ ] When creating a topic, resolve effective config: namespace override > global default (deferred — engine creates topics with global defaults; per-namespace overrides exist in config but aren't resolved at topic creation time)
- [ ] Update topic creation in `InMemoryQueueEngine` to accept optional overrides

> **Note:** This is the most invasive change — the engine currently has a single `default_max_depth` and `default_backpressure` for all topics. The cleanest approach is a `TopicConfig` struct resolved at creation time, passed through from the namespace manager. Keep the engine generic; put the config resolution logic in the namespace manager.

#### New error variant
- [x] Add `Error::ResourceExhausted(String)` to `src/error.rs`
- [x] Map to `ErrorCode::RESOURCE_EXHAUSTED`
- [x] Map to HTTP 503 in `src/server/http/error.rs`
- [x] Add to protocol doc and API doc — `ResourceExhausted` mapped to `503 Service Unavailable` in `docs/api.md`

### Config Additions

```toml
[queue]
max_depth = 100000                     # existing (plan 10)
backpressure_policy = "DropOldest"     # existing (plan 10)
max_redelivery_attempts = 5            # existing (plan 10)
dlq_topic_suffix = "__dlq"             # existing (plan 10)
default_topic_max_bytes = 268435456    # NEW — 256MB per topic
global_max_memory_bytes = 4294967296   # NEW — 4GB total

[namespaces.default]
api_key = "..."                        # existing
permissions = [...]                    # existing
max_total_messages = 1000000           # NEW — per-namespace quota
max_topics = 10000                     # NEW — per-namespace topic limit
topic_max_depth = 50000                # NEW — override global
topic_max_bytes = 134217728            # NEW — override global (128MB)
backpressure_policy = "Reject"         # NEW — override global
```

### Files

| File | Change |
|---|---|
| `src/config.rs` | Add fields to `QueueConfig` and `NamespaceConfig` |
| `src/error.rs` | Add `ResourceExhausted` variant + `ErrorCode` |
| `src/queue/topic.rs` | Add `max_bytes` enforcement in `publish()` |
| `src/queue/engine.rs` | Add global budget enforcement, pass topic config |
| `src/namespace/manager.rs` | Per-namespace quota tracking, config resolution |
| `src/namespace/state.rs` | Add per-namespace message counter |
| `src/server/http/error.rs` | Map `ResourceExhausted` → 503 |
| `config/config.example.toml` | Document new fields |
| `docs/protocol.md` | Add `RESOURCE_EXHAUSTED` error code |
| `docs/api.md` | Add 503 response for quota/budget errors |

### Verification
```bash
cargo test                              # all existing tests pass
cargo test resource                     # new resource limit tests
cargo clippy && cargo fmt -- --check
```

### Done When
- Per-topic byte limit enforced, tested with `DropOldest` and `Reject` policies
- Global memory budget enforced, tested at boundary
- Per-namespace quotas enforced for both message count and topic count
- Per-namespace overrides resolve correctly (namespace > global > default)
- Error codes documented in protocol.md and api.md
- Config example updated with all new fields

---

## 11.3: Connection Scaling & Delivery — COMPLETED

### Goal
Add per-namespace connection limits, configurable subscriber channel buffers, and slow consumer detection/handling to ensure the server degrades gracefully under high fan-out and connection pressure.

### Status: Complete

All core work items implemented. The Phase 11 retrospective identified and fixed two issues: (1) `manager.connect()` was never called from the TCP session — per-namespace limits were dead code; (2) consumer groups had no slow consumer tracking (only individual subscribers did). Both are now fixed. Additionally, the connection limit check uses an atomic CAS loop to avoid a TOCTOU race (see DD-019).

### Prerequisites
- Plan 8 complete — global `tcp.max_connections` with semaphore enforcement, idle timeout
- Plan 7 complete — delivery tasks with bounded `mpsc` channels, round-robin consumer groups
- Plan 10 complete — main entrypoint wires everything together

### Current State

What exists today:
- **Global connection limit**: `tcp.max_connections` enforced via `Semaphore` (plan 8)
- **Idle timeout**: `tcp.idle_timeout_secs` closes inactive connections (plan 8)
- **Delivery channels**: Bounded `mpsc` with hardcoded capacity (256 per subscriber, `src/namespace/manager.rs:78`)
- **Consumer group round-robin**: Simple index rotation under group lock
- **`try_send` backpressure**: Delivery tasks use `try_send` on full channels (retry with backoff)

What's missing:
- No per-namespace connection limit
- Channel buffer size is hardcoded (256), not configurable
- No slow consumer detection — a subscriber whose channel stays full just causes retry loops
- No policy for handling persistently slow consumers
- No batched fan-out (yield point during large fan-outs)
- No consumer group limit per topic

### Work Items

#### Per-namespace connection limits
- [x] Add `max_connections: Option<usize>` to `NamespaceConfig` (default: None / unlimited)
- [x] Track per-namespace connection count in `NamespaceState` (`AtomicUsize`)
- [x] In TCP session `process_frame()`: after auth succeeds, call `manager.connect()` which checks namespace connection count. Reject with `Error::ResourceExhausted` if at limit. (Retro fix: was dead code — `connect()` was never called.)
- [x] Increment on successful connect (atomic CAS — see DD-019), decrement on disconnect (via `manager.disconnect()`)
- [x] In HTTP middleware: no connection tracking needed (HTTP connections are stateless and short-lived)

> **Note:** This sits between the global semaphore (accept loop) and per-session auth (session handler). The check order is: global limit → accept → auth → namespace limit → session established. A connection that fails the namespace check still consumed a global semaphore permit temporarily.

#### Configurable subscriber channel buffer
- [x] Add `subscriber_channel_buffer: usize` to `TcpConfig` (default: 1,024)
- [x] Pass through to `DefaultNamespaceManager::new()` via `ManagerConfig.delivery_channel_capacity` (constructor refactored to use config struct — see DD-018)
- [x] Replaces hardcoded `256` in `delivery_channel_capacity`

#### Consumer group limit per topic
- [x] Add `max_consumer_groups_per_topic: usize` to `QueueConfig` (default: 1,000)
- [x] Enforce in namespace manager's `subscribe()` when creating a new group
- [x] Return `Error::ResourceExhausted` if limit exceeded

> **Note:** Without this limit, a misbehaving client could create unbounded consumer groups, each spawning a delivery task. This is a resource exhaustion vector.

#### Slow consumer detection
- [x] Track per-subscriber delivery failures (channel full) in `SubscriberDeliveryState` and `ConsumerGroupState`
  - Counter: `consecutive_send_failures` (reset on successful send)
  - Flag: `is_slow` (set when failures exceed threshold, cleared on recovery)
  - Counter: `total_drops` (lifetime drops for disconnect threshold)
- [x] Mark subscriber/group as "slow" when consecutive failures exceed `slow_consumer_threshold` (configurable, default: 100)
- [x] Log warning with client ID/group name, topic, and failure count when marked slow
- [x] Consumer groups now have parity with individual subscribers for slow tracking (retro fix: was missing)

> **Note:** The delivery task already handles `try_send` failures (retries with backoff). The detection layer sits on top — it tracks *patterns* of failure, not individual events.

#### Slow consumer policy
- [x] Add `slow_consumer_policy` to `TcpConfig`:
  - `drop_for_slow` (default): Skip delivery to slow subscribers/groups. Increment drop counter.
  - `disconnect_slow`: After `total_drops` exceeds `slow_consumer_disconnect_after` (default: 1,000), remove the subscriber or dissolve the group.
- [x] Implement in both individual and group delivery task loops — check slow status before attempting `try_send`
- [x] For `disconnect_slow`: removes subscriber from delivery state (individual) or dissolves group
- [ ] `block_publisher` mode (deferred — not recommended at scale)

> **Note:** `drop_for_slow` is the only sane default at scale. `block_publisher` exists for correctness-critical use cases (financial systems) where message loss is unacceptable. The operator must explicitly opt into it with full understanding of the throughput implications.

#### Batched fan-out (deferred)
- [ ] In individual subscriber delivery: if topic has >100 subscribers, yield after every 50 channel sends (`tokio::task::yield_now()`)
- [ ] Prevents a single hot topic from starving the tokio runtime during large fan-outs

> **Note:** This is a small change with outsized impact at scale. Without yielding, a publish to a topic with 10,000 subscribers monopolizes a runtime thread for the entire fan-out duration. The threshold (100 subscribers) and batch size (50) should be tunable if benchmarks show sensitivity, but hardcoded defaults are fine for v1.

#### Alternative channel type evaluation (deferred)
- [ ] Benchmark `tokio::sync::mpsc` vs `async-channel` vs `flume` at high fan-out (1,000+ subscribers)
- [ ] Only switch if benchmarks show >20% improvement

> **Note:** This is optional and depends on 11.1 baselines. `tokio::sync::mpsc` is likely fine. Don't add a dependency without evidence.

### Config Additions

```toml
[tcp]
max_connections = 10000                  # existing (plan 8)
idle_timeout_secs = 300                  # existing (plan 8)
subscriber_channel_buffer = 1024         # NEW — per-subscriber delivery buffer
slow_consumer_policy = "drop_for_slow"   # NEW — "drop_for_slow", "disconnect_slow", "block_publisher"
slow_consumer_disconnect_after = 1000    # NEW — consecutive drops before disconnect (disconnect_slow only)

[queue]
max_consumer_groups_per_topic = 1000     # NEW

[namespaces.default]
max_connections = 5000                   # NEW — per-namespace limit
```

### Files

| File | Change |
|---|---|
| `src/config.rs` | New fields on `TcpConfig`, `QueueConfig`, `NamespaceConfig` |
| `src/namespace/manager.rs` | Configurable channel capacity, group limit, slow consumer tracking |
| `src/namespace/state.rs` | Per-namespace connection counter, slow consumer state per subscriber |
| `src/server/tcp/session.rs` | Namespace connection limit check after auth |
| `src/server/tcp/connection.rs` | Yield during large fan-out |
| `src/error.rs` | `SLOW_CONSUMER` error code |
| `config/config.example.toml` | Document new fields |
| `docs/protocol.md` | Add `SLOW_CONSUMER` error code |

### Verification
```bash
cargo test
cargo test slow_consumer               # new slow consumer tests
cargo test namespace_connection         # new per-namespace limit tests
cargo bench -- fanout                   # verify no regression (requires 11.1)
```

### Done When
- Per-namespace connection limit enforced and tested
- Subscriber channel buffer configurable via config
- Consumer group limit enforced per topic
- Slow consumer detection marks slow subscribers, logs warnings
- `drop_for_slow` policy skips slow subscribers and tracks drops
- `disconnect_slow` policy closes connections after threshold
- Fan-out yields to runtime for topics with many subscribers
- Config example documents all new fields

---

## 11.4: Storage I/O Performance — COMPLETED

### Goal
Improve storage throughput, reduce fd consumption, speed up startup replay, and add retention policies for long-running deployments. Target: the storage layer should not be the bottleneck for publish throughput.

### Prerequisites
- Plan 5 complete — `FsStorage` with append-only segments, per-topic locking, replay
- Plan 6 complete — `InMemoryQueueEngine` with background flush task (100ms interval)
- Plan 10 complete — `storage.init()` + `engine.init()` in startup sequence

### Current State (post-implementation)

- **Append-only segments**: Per-topic directories with offset-named `.seg` files
- **Segment rotation**: At 64MB (configurable via `FsStorage::with_segment_max_bytes`)
- **LRU fd management**: `Mutex<LruCache<TopicKey, Mutex<ActiveSegment>>>` bounded by `max_open_fds` (default 1024). Evicted topics transparently reopened on next write. `flush()` uses `peek()` to avoid perturbing eviction order. (See DD-029)
- **Background flush**: Engine flushes all topics every `batch_flush_interval_ms` (default 100ms). Early flush triggered when a topic exceeds `batch_flush_size_bytes` (default 4MB) of dirty bytes. Dirty bytes tracked on `Topic` struct. (See DD-027)
- **Parallel replay**: `engine.init()` replays topics concurrently via `JoinSet` + `Semaphore(8)`. Per-topic errors logged and skipped. (See DD-028)
- **Fsync strategy**: Configurable via `StorageConfig.fsync_strategy` — `every_write`, `every_batch` (default), `os_managed`
- **Segment limit**: `max_segments_per_topic` enforced during rotation
- **Retention**: Per-namespace `RetentionPolicy` with age/bytes/segment-count dimensions, enforced by background `RetentionTask`
- **Retention coordination**: `min_live_offset` floor prevents deleting segments active subscribers need; compact-before-delete ordering for crash safety

Remaining deferred:
- Streaming replay (collecting into `Vec` first) — deferred until benchmarks show memory pressure during startup

### Work Items

#### Configurable fsync strategy
- [x] Add `fsync_strategy` to `StorageConfig`:
  - `every_batch` (default): fsync after each background flush cycle
  - `every_write`: fsync after every `append()` call — maximum durability
  - `os_managed`: no explicit fsync — rely on OS page cache writeback
- [x] Add `Deserialize` to a new `FsyncStrategy` enum
- [x] In `FsStorage::append()`: if `every_write`, call `file.sync_all()` after write
- [x] In background flush task: if `every_batch`, call `storage.flush()` for each topic; if `os_managed`, skip flush entirely

> **Note:** The current behavior is effectively `every_batch` with a hardcoded 100ms interval. This change makes both the strategy and interval explicit. `every_write` will significantly reduce throughput — warn in the config comment.

#### Configurable flush interval and batch size
- [x] Add `batch_flush_interval_ms: u64` to `StorageConfig` (default: 100)
- [x] Add `batch_flush_size_bytes: u64` to `StorageConfig` (default: 4,194,304 / 4MB)
- [x] Pass interval to `InMemoryQueueEngine::spawn_flush_task()` via `EngineConfig.flush_interval_ms`
- [x] Track unflushed bytes per topic. Trigger early flush when `batch_flush_size_bytes` is exceeded, even if the interval hasn't elapsed.

> **Note:** The batch size trigger requires a per-topic dirty bytes counter. Increment on append, reset on flush. The flush task checks both time and size on each iteration.

#### File descriptor management
- [x] Add `max_open_fds: usize` to `StorageConfig` (default: 1,024) — field exists, reserved for LRU
- [x] Replace `HashMap<TopicKey, Mutex<ActiveSegment>>` with an LRU-eviction structure (`lru::LruCache`)
- [x] When a new topic needs an fd and the limit is reached, close the least-recently-used topic's segment file
- [x] On next write to an evicted topic, reopen the segment file (append mode)
- [x] fd count bounded by LRU cache capacity (no separate AtomicUsize needed — LruCache enforces limit)

> **Note:** The current implementation opens a segment file on first append and never closes it. With 10,000+ topics, this exhausts the process fd limit (typically 1024 on Linux). The LRU approach is simple: maintain an insertion-order or access-order list alongside the map. `lru` crate is an option, or hand-roll with `LinkedHashMap`.

#### Parallel startup replay
- [x] In `engine.init()`: list all namespace/topic pairs, then replay in parallel using `tokio::task::JoinSet` bounded by `Semaphore` (default concurrency: 8)
- [ ] Stream messages during replay instead of collecting into `Vec` — use `tokio::sync::mpsc` or `futures::Stream` to feed messages to `Topic::from_replay()` incrementally
- [x] Log progress: `tracing::info!("replayed {ns}/{topic}: {count} messages")` per topic
- [x] Log total: `tracing::info!("replay complete: {total_topics} topics, {total_messages} messages in {elapsed}")`

> **Note:** The current implementation is:
> ```rust
> let messages = self.storage.replay(&ns, &topic_name).await?;
> let t = Topic::from_replay(messages, ...);
> ```
> This collects all messages into a `Vec` first. For a topic with 1M messages, this is a significant memory spike. Streaming replay requires a new `StorageBackend::replay_stream()` method or refactoring `replay()` to return a `Stream`. Start with parallel (easy win), then add streaming if benchmarks show memory pressure during startup.

#### Retention policies — COMPLETED (Plan 12)
- [x] Add retention config to `NamespaceConfig` — implemented as `RetentionPolicy { max_age_secs, max_bytes, max_segments }`
- [x] Add `retention_check_interval_secs: u64` to `StorageConfig` (default: 60)
- [x] Spawn a background retention task (`RetentionTask` in `src/retention.rs`) that periodically:
  1. Lists all segments per topic (via `storage.list_topics()` filesystem scan)
  2. Deletes segments older than `max_age` (using mtime)
  3. Deletes oldest segments if count > `max_segments`
  4. Deletes oldest segments if total size > `max_bytes`
- [x] Retention coordinates with `NamespaceManager::min_live_offset()` — never deletes segments active subscribers need. Compact-before-delete ordering for crash safety.

> **Note:** Implemented in Plan 12 (Timestamp & Retention). Uses `min_live_offset` (walks subscriber + group cursors) rather than `base_offset` for a stronger safety guarantee. See DD-024.

#### Segment limit per topic — COMPLETED
- [x] Add `max_segments_per_topic: usize` to `StorageConfig` (default: 64)
- [x] Enforce during segment rotation: if creating a new segment would exceed the limit, delete the oldest segment first
- [x] This is a simpler fallback for deployments that don't want time-based retention

### Config Additions

```toml
[storage]
data_dir = "./data"                      # existing
fsync_strategy = "every_batch"           # NEW — "every_write", "every_batch", "os_managed"
batch_flush_interval_ms = 100            # NEW
batch_flush_size_bytes = 4194304         # NEW — 4MB
max_open_fds = 1024                      # NEW
retention_check_interval_secs = 60       # NEW
max_segments_per_topic = 64              # NEW

[namespaces.default]
retention_max_age_secs = 604800          # NEW — 7 days
retention_max_segments = 64              # NEW
retention_max_bytes = 10737418240        # NEW — 10GB
```

### Files

| File | Change |
|---|---|
| `src/config.rs` | New fields on `StorageConfig`, `NamespaceConfig` |
| `src/storage.rs` | `StorageBackend` trait: add `replay_stream()` (optional) |
| `src/storage/fs.rs` | Fsync strategy, fd LRU, segment limit, retention task |
| `src/queue/engine.rs` | Configurable flush interval, parallel replay, `min_retained_offset()` |
| `config/config.example.toml` | Document new storage fields |

### Dependencies Added
- `lru = "0.16.3"` — LRU cache for fd management (lightweight, no async)

### Verification
```bash
cargo test
cargo test storage                     # storage unit + integration tests
cargo test retention                   # new retention tests
cargo bench -- storage                 # requires 11.1 benchmarks
```

### Done When
- ~~Fsync strategy configurable and tested~~ (`every_write`, `every_batch`, `os_managed`) — **DONE**
- ~~Flush interval configurable + batch size trigger~~ — **DONE** (`batch_flush_interval_ms`, `batch_flush_size_bytes` with per-topic dirty bytes tracking)
- ~~Fd count bounded by LRU eviction, tested with more topics than `max_open_fds`~~ — **DONE** (`lru::LruCache`, test: `lru_evicts_least_recently_used_topic`)
- ~~Startup replay parallelized, progress logged~~ — **DONE** (JoinSet + Semaphore, concurrency 8)
- ~~Retention task runs on interval, deletes old/excess segments~~ — **DONE** (Plan 12)
- ~~Retention respects subscriber floor — no data loss~~ — **DONE** (Plan 12, `min_live_offset`)
- ~~Segment limit enforced during rotation~~ — **DONE**
- ~~Config example documents all new fields~~ — **DONE**

---

## 11.5: Concurrency & Lock Granularity — DEFERRED

### Status: No Evidence of Contention

Benchmarks from Plan 11.1 showed no lock contention. All decision gate criteria answered "no change needed." Revisit only after tiered storage (Plan 12) adds concurrent disk I/O that could change the contention profile.

### Goal
Reduce lock contention in the queue engine and namespace manager to support 10,000+ concurrent namespaces and topics without degradation. **This plan is gated on 11.1 benchmarks** — only proceed if baselines show contention is a real bottleneck.

### Prerequisites
- **11.1 complete** — baseline benchmarks recorded, contention measured
- Specifically: the "concurrent publish (cross-topic)" and "namespace contention" benchmarks must show measurable degradation vs single-topic publish to justify the added complexity

### Decision Gate

Before implementing anything in this plan, answer these questions from 11.1 data:

1. **Does cross-topic publish throughput degrade linearly or worse with topic count?**
   - If throughput is flat across 100→1,000→10,000 topics: the `RwLock<HashMap>` is fine. Skip this plan.
   - If throughput drops >20% at 10,000 topics: proceed with DashMap sharding.

2. **Is `tokio::sync::Mutex` overhead significant for topic-level locks?**
   - If ack/publish p99 latency is <1ms: `tokio::sync::Mutex` is fine.
   - If p99 shows spikes from async Mutex scheduling: evaluate `parking_lot::Mutex`.

3. **Is auth on the hot path?**
   - `ApiKeyStore` already uses `Arc<[KeyEntry]>` with no locks. If 11.1 shows no auth overhead, no change needed.

> **If all three answers are "no change needed," close this plan as unnecessary.** The guiding principle is "measure first" — complexity without measured benefit is tech debt.

### Current State

| Component | Current Lock Strategy | Concern |
|---|---|---|
| `InMemoryQueueEngine.topics` | `RwLock<HashMap<(NamespaceId, TopicName), Arc<Mutex<Topic>>>>` | Single RwLock on topic map — read lock for lookup, write lock for new topics |
| `Topic` (per-topic) | `tokio::sync::Mutex<Topic>` (via `Arc<Mutex<Topic>>`) | Holds across `.await` in flush task |
| `NamespaceManager.namespaces` | `RwLock<HashMap<NamespaceId, Arc<NamespaceState>>>` | Single RwLock on namespace map |
| `TopicDeliveryState` | `tokio::sync::Mutex` per topic per namespace | Already fine-grained |
| `ApiKeyStore` | `Arc<[KeyEntry]>` — lock-free | No change needed |

### Work Items (conditional on 11.1 data)

#### Queue engine: DashMap sharding
- [ ] Replace `RwLock<HashMap<(NamespaceId, TopicName), Arc<Mutex<Topic>>>>` with two-level structure:
  - Outer: `DashMap<NamespaceId, NamespaceQueues>`
  - Inner: `DashMap<TopicName, Arc<Mutex<Topic>>>`
- [ ] This gives lock-free concurrent access across namespaces and across topics within a namespace
- [ ] Auto-create `NamespaceQueues` on first publish (using `DashMap::entry`)
- [ ] Update `list_topics()`, `topic_stats()`, `compact()`, `head_offset()`, `consume_from()` to use DashMap API

> **Note:** DashMap uses shard-level locks internally. For our access patterns (many topics, each accessed independently), this is effectively lock-free. The two-level structure means namespace-level operations (list all topics) only lock one shard of the outer map.

#### Queue engine: evaluate parking_lot::Mutex for Topics
- [ ] Benchmark `tokio::sync::Mutex` vs `parking_lot::Mutex` for per-topic locks
- [ ] `tokio::sync::Mutex` is needed if the lock is held across `.await` points
- [ ] Check if `Topic::publish()` and `Topic::consume_from()` hold locks across await — currently they don't (these are sync operations)
- [ ] The flush task *does* hold the topic lock across `storage.append().await` — this is the only async hold point
- [ ] **If** we can restructure the flush task to copy data out under a sync lock and then append without holding it, switch to `parking_lot::Mutex`

> **Note:** `parking_lot::Mutex` is ~2x faster than `tokio::sync::Mutex` for uncontended locks and doesn't require `.await` for acquisition. But using it across an `.await` point is unsound (blocks the runtime thread). The flush task restructuring is the prerequisite.

#### Namespace manager: DashMap for namespace state
- [ ] Replace `RwLock<HashMap<NamespaceId, Arc<NamespaceState>>>` with `DashMap<NamespaceId, Arc<NamespaceState>>`
- [ ] Same rationale as queue engine — removes the single RwLock bottleneck for cross-namespace operations

> **Note:** Namespace state access patterns are similar to queue engine: lookup by key (hot path), iterate all (shutdown only). DashMap handles both.

#### Auth store: verify lock-free path
- [ ] Confirm `ApiKeyStore` has no locks on the auth path (it shouldn't — already `Arc<[KeyEntry]>`)
- [ ] No code change expected — just validation from 11.1 benchmarks

### Dependencies to Add

| Crate | When | Purpose |
|---|---|---|
| `dashmap` | If DashMap sharding is justified | Lock-free concurrent hash map |
| `parking_lot` | If Mutex evaluation shows benefit | Faster sync mutexes |

### Files (conditional)

| File | Change |
|---|---|
| `src/queue/engine.rs` | DashMap two-level topic map, optional parking_lot Mutex |
| `src/namespace/manager.rs` | DashMap for namespace state map |
| `Cargo.toml` | Add `dashmap`, optionally `parking_lot` |

### Verification
```bash
cargo test                             # all existing tests pass
cargo bench -- queue                   # compare against 11.1 baselines
cargo bench -- fanout                  # compare against 11.1 baselines
```

Key benchmark comparisons:
- Cross-topic publish: before vs after DashMap
- Namespace contention at 10K namespaces: before vs after
- Topic lock acquire latency: tokio::sync vs parking_lot (if evaluated)

### Done When
- Benchmark comparison shows measurable improvement (>20% throughput or >50% p99 reduction)
- All 263+ tests still pass
- No new unsafe code
- OR: plan closed as unnecessary because 11.1 baselines showed no contention problem

---

## 11.6: Observability & Metrics — COMPLETED

### Goal
Expose Prometheus-compatible metrics via the HTTP server so operators can monitor queue depth, throughput, latency, and resource usage. Add structured logging for performance-critical events.

### Status: Complete

All metrics (gauges, counters, histograms) are implemented. Histogram bucket boundaries are configured. Structured performance logging is in place for backpressure, memory pressure, connection rejection, and storage replay.

### Prerequisites
- Plan 9 complete — HTTP server with axum routes
- Plan 10 complete — main entrypoint with tracing initialized

### Current State

What exists today:
- **Tracing**: `tracing_subscriber::fmt::init()` in main — structured log output
- **Health endpoint**: `GET /health` returns 200 (no body)
- **Internal counters**: `total_bytes` (`AtomicU64`), `Topic.bytes_used`, per-topic depth

What's missing:
- No metrics endpoint
- Internal counters are not exposed externally
- No histograms for latency measurement
- No structured logging for perf events (backpressure, slow consumer, connection rejected)

### Work Items

#### Metrics endpoint
- [x] Add `GET /v1/metrics` route to HTTP server — returns Prometheus text format
- [x] Use `metrics` crate for metric registration + `metrics-exporter-prometheus` for rendering
- [x] Install the Prometheus recorder in `main.rs` before spawning servers
- [x] No authentication on `/v1/metrics` (matches Prometheus convention — scrape target should not require auth)

> **Note:** The `metrics` crate uses a global recorder pattern. Install it once in `main()`. All subsystems emit metrics via `metrics::counter!()`, `metrics::gauge!()`, `metrics::histogram!()` macros — no need to pass metric handles through constructors.

#### Core metrics

##### Gauges (point-in-time values)
- [x] `fafb_connections_active{namespace}` — current TCP connections per namespace (emitted in `manager.connect()` / `disconnect()`)
- [x] `fafb_queue_depth{namespace, topic}` — messages in buffer per topic (emitted in `engine.publish()`)
- [x] `fafb_queue_memory_bytes{namespace, topic}` — `bytes_used` per topic (emitted in `engine.publish()`)
- [x] `fafb_global_memory_bytes` — `total_bytes` across all topics (emitted in `engine.publish()`)
- [x] `fafb_inflight_messages{namespace, topic}` — inflight entries per subscriber/group
- [x] `fafb_consumer_group_members{namespace, topic, group}` — members per consumer group

##### Counters (monotonically increasing)
- [x] `fafb_messages_published_total{namespace, topic}` — publishes accepted (emitted in `manager.publish()`)
- [x] `fafb_messages_delivered_total{namespace, topic, mode}` — deliveries sent (emitted in individual + group delivery tasks)
- [x] `fafb_messages_dropped_total{namespace, topic, reason}` — drops (emitted in delivery tasks on channel_full)
- [x] `fafb_messages_acked_total{namespace, topic}` — acks processed
- [x] `fafb_slow_consumers_total{namespace}` — slow consumer events (emitted in individual + group delivery tasks when marking slow)
- [x] `fafb_connections_total{namespace}` — connections accepted (emitted in `manager.connect()`)
- [x] `fafb_connections_rejected_total{reason}` — rejected (emitted in `manager.connect()` for namespace_limit, `session.rs` for auth_failed)

##### Histograms (latency distributions)
- [x] `fafb_publish_latency_seconds{namespace}` — time from publish request to queue engine write
- [x] `fafb_delivery_latency_seconds{namespace, topic}` — time from publish to subscriber channel send
- [x] `fafb_storage_write_latency_seconds` — `storage.append()` call duration
- [x] `fafb_storage_flush_latency_seconds` — per-flush-cycle duration
- [x] `fafb_storage_replay_duration_seconds{namespace}` — per-topic replay time at startup

> **Note:** Histogram bucket boundaries should be tuned for queue workloads: `[0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0]` seconds. This gives good resolution from 100μs to 5s.

#### Instrumentation points

| Metric | Where to emit |
|---|---|
| `connections_active` | TCP accept loop (increment) + disconnect handler (decrement) |
| `queue_depth` / `queue_memory_bytes` | `Topic::publish()` and `Topic::compact()` |
| `global_memory_bytes` | `InMemoryQueueEngine::publish()` |
| `messages_published_total` | `NamespaceManager::publish()` |
| `messages_delivered_total` | Delivery task on successful `try_send` |
| `messages_dropped_total` | Delivery task on slow consumer skip, backpressure reject |
| `publish_latency_seconds` | Wrap `engine.publish()` call in namespace manager |
| `delivery_latency_seconds` | Record at delivery time: `now - message.timestamp` |
| `storage_write_latency_seconds` | Wrap `storage.append()` in engine flush task |
| `inflight_messages` | Ack handler (decrement), delivery task (increment) |

#### Structured performance logging
- [x] `warn` level: backpressure engaged (`topic={topic} policy={policy} depth={depth}`) — `topic.rs`
- [x] `warn` level: slow consumer detected (`client={id} topic={topic} consecutive_drops={n}`) — already in delivery tasks
- [x] `warn` level: global memory budget at 80% (`used={bytes} limit={bytes}`) — `engine.rs`
- [x] `warn` level: connection rejected (`reason={reason} namespace={ns}`) — `manager.rs`
- [x] `info` level: storage replay progress (`ns={ns} topic={topic} messages={n} elapsed={ms}`) — `engine.rs`
- [x] `info` level: retention cleanup (`ns={ns} topic={topic} segments_deleted={n} bytes_freed={n}`) — implemented in `RetentionTask::sweep()` (Plan 12)

> **Note:** Some of these events depend on 11.2 (memory limits) and 11.3 (slow consumers) being implemented. Add the logging infrastructure now; fill in the specific emit points as those plans are completed. Use `tracing::warn!()` with structured fields — this integrates naturally with the existing `tracing_subscriber` setup.

### Config Additions

```toml
[metrics]
enabled = true                          # NEW — disable to remove overhead
```

> **Note:** Consider whether metrics should be always-on or opt-in. The `metrics` crate has near-zero overhead when no recorder is installed. If `enabled = false`, simply don't install the recorder — all `metrics::*!()` calls become no-ops.

### Dependencies to Add

| Crate | Purpose |
|---|---|
| `metrics = "0.24"` | Metric registration (counter, gauge, histogram macros) |
| `metrics-exporter-prometheus = "0.16"` | Prometheus text format renderer + HTTP handler |

### Files

| File | Change |
|---|---|
| `Cargo.toml` | Add `metrics` + `metrics-exporter-prometheus` |
| `src/config.rs` | Add `MetricsConfig` (optional) |
| `src/main.rs` | Install Prometheus recorder |
| `src/server/http.rs` | Add `/v1/metrics` route |
| `src/server/http/routes.rs` | Metrics handler function |
| `src/namespace/manager.rs` | Emit publish/delivery/ack counters + latency histograms |
| `src/queue/engine.rs` | Emit depth/memory gauges, publish latency |
| `src/server/tcp.rs` | Emit connection gauges |
| `src/server/tcp/session.rs` | Emit connection counters |
| `src/storage/fs.rs` | Emit write/flush latency histograms |
| `docs/api.md` | Document `/v1/metrics` endpoint |

### Verification
```bash
cargo test
cargo run -- --config config/config.example.toml &
curl -s http://127.0.0.1:8080/v1/metrics | head -20   # verify Prometheus output
# Publish a message, verify counters increment
curl -X POST http://127.0.0.1:8080/v1/default/test/publish \
  -H "Authorization: Bearer change-me-in-production-key" \
  -d "hello"
curl -s http://127.0.0.1:8080/v1/metrics | grep fafb_messages_published
```

### Done When
- `/v1/metrics` returns valid Prometheus text format
- All core gauges, counters, and histograms are emitted
- `metrics.enabled = false` disables all metric overhead
- Structured logging in place for perf events (backpressure, slow consumer, etc.)
- Existing tests unaffected (metrics are fire-and-forget, no test assertions needed)
- `docs/api.md` documents the metrics endpoint

---

## 11.7: Load Testing & Validation — COMPLETED

### Goal
Validate that all resource limits, backpressure mechanisms, and performance characteristics work correctly under realistic load. This is the integration-level validation for work done in 11.2 (memory), 11.3 (connections), and 11.4 (storage).

### Prerequisites
- **11.1 complete** — baselines recorded for comparison
- **11.2 complete** — memory limits and quotas enforced
- **11.3 complete** — connection limits and slow consumer handling
- Ideally 11.4 and 11.6 as well, but not strictly required

### Current State
- Docker test infrastructure exists (`Dockerfile`, `docker-compose.test.yml`, `scripts/docker-test.sh`)
- Resource knobs available: CPU, memory, disk (tmpfs), test filter, bench filter
- Load tests implemented in `tests/load_test.rs` with `ClientPool` + `LatencyStats` in `tests/common/mod.rs`
- Docker `load_test` service added to `docker-compose.test.yml`
- `./scripts/docker-test.sh load` command available

### Work Items

#### Sustained load test
- [x] 1,000 concurrent TCP connections (mix of publishers and subscribers)
- [x] Mixed workload: 50% publish, 30% subscribe, 20% query (HTTP)
- [x] Run for 10 minutes
- [x] Assert: no OOM, no deadlock, stable latency (p99 doesn't drift upward)
- [x] Assert: all published messages are delivered (no silent drops under normal load)
- [x] Measure: throughput over time, memory over time, connection count over time

> **Note:** This test should be run in Docker with constrained resources (`--cpus 2 --memory 1g`) to simulate realistic deployment conditions. The same test with more resources (`--cpus 8 --memory 8g`) validates scaling behavior.

#### Burst test
- [x] 10,000 connections attempt to join simultaneously
- [x] Assert: server stays responsive throughout
- [x] Assert: connections beyond `max_connections` are rejected cleanly (Error frame, not hang/crash)
- [x] Assert: accepted connections can publish/subscribe normally after burst settles
- [x] Measure: time-to-reject for excess connections, accept rate during burst

> **Note:** This tests the semaphore + accept loop under sudden pressure. The key failure mode to check is runtime thread starvation — if the accept loop monopolizes a thread, existing connections may stall.

#### Slow consumer test
- [x] 100 subscribers on a single topic
- [x] 1 subscriber intentionally delays reads (simulates slow processing)
- [x] Assert: other 99 subscribers receive messages at full speed
- [x] Assert: slow subscriber triggers slow consumer detection (if 11.3 implemented)
- [x] Assert: with `drop_for_slow` policy, slow subscriber's dropped count increases, others unaffected
- [x] Measure: delivery latency for fast subscribers during slow consumer event

> **Note:** This validates that the per-subscriber delivery task isolation works. The critical invariant: one slow consumer must not degrade the p99 of other consumers on the same topic.

#### Memory pressure test
- [x] Docker with `--memory 256m`
- [x] Publish continuously until global memory budget is hit
- [x] Assert: backpressure engages (publishes rejected with appropriate error)
- [x] Assert: no OOM kill (process stays alive)
- [x] Assert: after consumers catch up and compaction runs, publishes resume
- [x] Measure: time from budget-exceeded to first rejected publish, time to recover

> **Note:** This is the most dangerous test — it intentionally pushes the server to its limits. Run in Docker only. The 256m limit should be well below the configured `global_max_memory_bytes` to ensure the budget check triggers before the OS OOM killer.

#### Namespace isolation test
- [x] 100 namespaces, each with 10 topics
- [x] One namespace publishes aggressively (fills quota)
- [x] Assert: other namespaces are unaffected (can still publish, subscribe, query)
- [x] Assert: the aggressive namespace gets quota errors, not the others
- [x] Measure: cross-namespace publish latency during one namespace's quota saturation

#### Storage durability test
- [x] Publish 100K messages across 50 topics
- [x] Kill the process (SIGKILL — not graceful shutdown)
- [x] Restart and verify replay
- [x] Assert: message count after replay ≥ message count before kill minus at most one flush interval's worth of messages
- [x] Measure: replay time, messages lost (should be bounded by flush interval)

> **Note:** This tests the durability guarantees of the storage layer. The expected loss window is at most `batch_flush_interval_ms` worth of messages (default 100ms). Messages written to the OS page cache but not fsynced may also survive if the OS didn't crash.

#### Retention test
- [x] Configure `retention_max_segments = 4` and `retention_max_bytes = 10MB` for a namespace
- [x] Publish enough data to generate 10+ segments
- [x] Assert: retention task deletes old segments on schedule
- [x] Assert: disk usage stabilizes at or below the configured limit
- [x] Assert: in-memory messages are not affected by segment deletion

### Test Infrastructure

#### Test harness
Load tests should be in `tests/` as integration tests, tagged or filtered so they don't run in normal `cargo test`:

```bash
cargo test --test load_test                          # run all load tests
cargo test --test load_test sustained                # run one specific test
./scripts/docker-test.sh test --filter load_test     # in Docker
```

Consider gating on a `--ignored` flag or a cargo feature to prevent accidental execution:
```rust
#[test]
#[ignore]  // run with: cargo test -- --ignored
fn sustained_load_test() { ... }
```

#### Test client
A reusable async load generator that can:
- Open N TCP connections concurrently
- Authenticate each with a given API key
- Publish messages at a configurable rate
- Subscribe and verify delivery
- Collect latency histograms

> **Note:** The existing `TestClient` in `tests/common/mod.rs` is designed for single-connection tests. The load generator needs a different abstraction — a pool of clients driven by a rate limiter. Consider `tokio::time::Interval` for rate control and `tokio::task::JoinSet` for concurrent client management.

#### Docker profiles
Suggested profiles for reproducible testing:

| Profile | CPUs | Memory | Disk | Use case |
|---|---|---|---|---|
| `minimal` | 1 | 256m | 64m | Memory pressure, OOM boundary |
| `standard` | 2 | 1g | 256m | Normal CI load tests |
| `scaled` | 4 | 4g | 1g | Burst + sustained load |
| `production` | 8 | 8g | 4g | Baseline for production sizing |

### Files

| File | Change |
|---|---|
| `tests/load_test.rs` | **New** — all load test scenarios |
| `tests/common/mod.rs` | Add load generator client pool |
| `scripts/docker-test.sh` | Add profile presets (optional) |

### Verification
```bash
# Quick smoke test (not resource-constrained)
cargo test --test load_test -- --ignored sustained_load

# Full suite in Docker
./scripts/docker-test.sh test --filter load_test --cpus 2 --memory 1g

# Memory pressure specifically
./scripts/docker-test.sh test --filter memory_pressure --cpus 2 --memory 256m
```

### Done When
- Sustained load test passes: 1,000 connections, 10 minutes, no OOM, no deadlock, stable p99
- Burst test passes: 10,000 connections, clean rejection, no crash
- Slow consumer test passes: fast subscribers unaffected by slow one
- Memory pressure test passes: backpressure engages, no OOM kill, recovery works
- Namespace isolation test passes: one namespace's quota saturation doesn't affect others
- Storage durability test passes: replay recovers within expected loss window
- Retention test passes: disk usage bounded
- All tests runnable in Docker with resource constraints
- Results documented with actual numbers for production sizing guidance
