# Architecture

## Subsystem Interaction

```
                    ┌─────────────┐
                    │   Clients   │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              ▼                         ▼
       ┌────────────┐           ┌────────────┐
       │ TCP Server │           │ HTTP Server │
       │  (plan 8)  │           │  (plan 9)   │
       └─────┬──────┘           └──────┬──────┘
             │                         │
             └────────┬────────────────┘
                      ▼
              ┌───────────────┐
              │   Auth Layer  │  ◄── per-connection session cache
              │   (plan 4)    │
              └───────┬───────┘
                      ▼
        ┌──────────────────────────┐
        │    Namespace Manager     │  ◄── cursor state, fan-out,
        │       (plan 7)           │      interceptors, DLQ
        │                          │
        │  PublishInterceptor chain │
        │  DeliveryInterceptor chain│
        └────────────┬─────────────┘
                     ▼
        ┌──────────────────────────┐
        │     Queue Engine         │  ◄── indexed log, offset-based
        │       (plan 6)           │      read/write, GC/compaction
        └────────────┬─────────────┘
                     ▼
        ┌──────────────────────────┐
        │     Storage Layer        │  ◄── append-only segments,
        │       (plan 5)           │      batched fsync, LRU fd mgmt,
        │                          │      parallel replay
        └──────────────────────────┘
```

## Core Abstraction: TopicLog

The data path is organized around the **TopicLog** — an append-only indexed log per topic. Three subsystems collaborate:

| Layer | Owns | Reads | Writes |
|---|---|---|---|
| **Storage** (plan 5) | On-disk segments | — | Appends with `TopicOffset`, replays in offset order, `list_topics()` for discovery |
| **Queue Engine** (plan 6) | In-memory indexed log (`VecDeque` + `base_offset`) | `consume_from(offset)` → `Option<PublishedMessage>` | `publish()` assigns offset, `compact()` removes old entries (in-memory only) |
| **Namespace Manager** (plan 7) | Cursor state per subscriber/group | Fetches messages via `consume_from` | Advances offsets on ack, computes min live offset for GC |

Key invariant: the queue engine is cursor-unaware. It stores the log and serves reads. The namespace manager owns all consumer/subscriber position tracking.

### Quota Tracking

Per-namespace `message_count` is maintained as an `AtomicU64` in `NamespaceState`. It is adjusted in three places:
- **Publish**: incremented by 1, decremented by `evicted_messages` (from `PublishResult`) when DropOldest backpressure evicts during publish
- **Ack/Compact**: decremented by `messages_freed` (from `CompactResult`) when compaction reclaims messages
- This dual-path accounting prevents the counter from drifting permanently upward.

### Disk Quota Tracking

Per-namespace disk bytes are tracked incrementally by `FsStorage` in a `RwLock<HashMap<String, AtomicU64>>`. The counter includes both `.seg` and `.idx` file sizes:

- **`init()`**: scans all segment and index files, sums sizes per namespace
- **`append()`**: atomically adds record bytes + index entry bytes (when written) to the namespace counter
- **`delete_segments()`**: stats files before deleting, atomically subtracts freed bytes

The counter is exposed through `StorageBackend::namespace_disk_bytes()` → `QueueEngine::namespace_disk_bytes()`. The namespace manager checks the counter against `NamespaceQuota.max_disk_bytes` before publishing. This is an approximate check (TOCTOU: concurrent publishes may briefly overshoot), consistent with how `max_total_messages` works.

See detailed cross-references in each plan's "Shared Abstraction: TopicLog" section.

## Delivery Model

Three consumption models coexist per topic:

- **Individual subscribers** (push, broadcast): each subscriber gets every message via independent server-managed offsets
- **Consumer groups** (push, load-balanced): group members share a single offset, round-robin delivery
- **Fetch / pull** (stateless): clients request batches of messages on-demand via `Fetch`/`FetchFrom` frames (TCP) or `GET /v1/{ns}/{topic}/fetch` (HTTP). No server-side state — clients manage their own offset progression. Ideal for batch/ETL consumers, large messages, and simple SDKs.

Push modes (subscribe) are managed by the namespace manager with delivery tasks and inflight tracking. Pull mode (fetch) reuses the same `consume_from()` read path but creates no server-side state. The queue engine and storage layer are agnostic to delivery mode.

### Delivery Mechanics (implemented)

Each subscriber/consumer group gets a dedicated `tokio::task` that loops:
1. Read current offset from state (brief Mutex lock)
2. Call `engine.consume_from(offset)` (no lock held)
3. If message available: run delivery interceptors → push to subscriber's `mpsc` channel → advance offset
4. If `None` and `offset < engine.base_offset()`: messages were evicted by DropOldest — **skip forward** to `base_offset`, log warning, emit `fafb_messages_skipped_total` metric
5. If caught up (`offset >= head_offset`): wait on `tokio::sync::Notify` (woken by publish)

Key properties:
- **No lock held during engine reads** — prevents contention between delivery and publish
- **Notify registered before consume** — prevents missed wake-ups
- **Skip-forward on eviction** — prevents stuck consumers when DropOldest evicts behind a subscriber's cursor (see Plan 12 Phase A)
- **Per-subscriber channels** — slow subscribers don't block others (bounded `mpsc` with backpressure via `try_send`)
- **Task lifecycle tracked** — `JoinHandle`s stored for shutdown and respawn detection
- **Seek/rewind** — subscribers can start from a specific offset (`SubscribeAt`) or timestamp (`SubscribeFrom`) instead of the head. The `SubscribeStart` enum (`Head`, `Offset`) controls starting position. Timestamps are resolved to offsets via `StorageBackend::offset_at_timestamp()` in the TCP session before calling `subscribe()`. The starting offset is clamped to `[oldest_available, head]`.

### Ack Flow

Clients ack by `MessageId` (matching the wire protocol `Frame::Ack`). The namespace manager:
1. Uses a reverse subscription index to determine if the client is an individual subscriber or group member (O(1) lookup)
2. Scans the client's inflight map for the matching `MessageId` → `TopicOffset`
3. Removes the inflight entry
4. Evicts expired inflight entries (default 300s timeout) to prevent compaction blocking
5. Computes `minimum_live_offset` across all subscribers/groups and calls `engine.compact()`
6. Decrements `message_count` by `CompactResult.messages_freed` (prevents namespace quotas from permanently drifting)

## Extension Points

The namespace manager exposes trait-based interceptor hooks:
- `PublishInterceptor` — modify/reject messages before queue entry (skipped for DLQ topics to avoid loops)
- `DeliveryInterceptor` — modify/filter messages before delivery
- Dead-letter routing — messages exceeding max redelivery go to a DLQ topic (`{topic}__dlq` by default)

## Connection Lifecycle

When a TCP client connects and sends a `Connect` frame:

1. **Auth** — `Session::process_pre_auth()` validates API key via `Authenticator`
2. **Namespace limit** — `manager.connect(ns, client_id)` atomically checks and increments the per-namespace connection count using a CAS loop (`fetch_update`). If the limit is reached, returns `ConnectAck { success: false }` without transitioning the session.
3. **Session transition** — On success, the session moves to `Authenticated` state and replies with `ConnectAck { success: true }`.

On disconnect, `manager.disconnect(client_id)` decrements the connection count and cleans up all subscriptions.

## Slow Consumer Handling

Both individual subscribers and consumer groups track slow consumer state:

- **Detection**: `consecutive_send_failures` counter incremented on `try_send` `Full` errors; reset to 0 on successful send. When failures exceed `slow_consumer_threshold`, the subscriber/group is marked `is_slow`.
- **DropForSlow policy** (default): Messages are skipped for slow consumers, `total_drops` incremented. The consumer catches up when it drains its channel.
- **DisconnectSlow policy**: After `total_drops` exceeds `slow_consumer_disconnect_after`, the subscriber is removed (individual) or the group is dissolved.

## Server Lifecycle (plans 10, 14)

### ServerLifecycle State Machine

The server's lifecycle is managed by `ServerLifecycle` (`src/server/state.rs`), a lock-free struct that tracks server state and broadcasts transitions:

```
Running ──► Draining ──► ShuttingDown
   │            │              │
   │            │              └── Force-close all connections,
   │            │                  flush engine/storage, exit
   │            │
   │            └── Stop accepting new connections,
   │                send GoingAway to capable clients,
   │                reject new subscribes, allow publishes/acks,
   │                wait for clients to disconnect voluntarily
   │
   └── Normal operation: accept connections,
       process all frame types
```

State transitions are forward-only via `AtomicU8` with `fetch_max`. The `watch::Sender<ServerState>` broadcasts state changes to all subscribers — servers, background tasks, and connection handlers all subscribe to this single channel. There is no separate shutdown boolean.

Key fields:
- `state: AtomicU8` — lock-free reads on every frame dispatch (hot path checks `is_draining()`)
- `initialized: AtomicBool` — set after `engine.init()` completes
- `active_connections: AtomicU64` — incremented on TCP accept, decremented on close
- `notify: watch::Sender<ServerState>` — single broadcast channel for all state transitions

### Startup

The `main.rs` entrypoint initializes subsystems in dependency order. HTTP starts early so health probes work during initialization:

1. **Tracing** — layered subscriber (stdout + optional rolling file via `LoggingConfig`)
2. **Config** — CLI args → TOML file → env overrides → validation
3. **ServerLifecycle** — created in `Running` state, `initialized = false`
4. **Storage** — `FsStorage::new()` (data dir created)
5. **Queue Engine** — `InMemoryQueueEngine::new(storage, EngineConfig { .. })`
6. **Auth** — `ApiKeyStore::from_config()` (immutable after startup)
7. **Namespace Manager** — `DefaultNamespaceManager::new(engine, ns_ids, quotas, ManagerConfig { .. })`
8. **HTTP Server** — starts immediately (binds port, serves health probes)
   - `/health/live` → 200 (immediately available)
   - `/health/ready` → 503 `"initializing"` (not yet ready)
   - `/v1/*` business routes → 503 via `require_initialized` middleware
9. **storage.init()** + **engine.init()** — topic replay (can take seconds for large datasets)
10. **lifecycle.set_initialized()** — `/health/ready` → 200 `"ready"`
11. **TCP Server** — starts accept loop (waits for `initialized == true` before accepting)
12. **Background tasks** — retention, compaction, log cleanup spawned with lifecycle subscribers

### Shutdown

Triggered by SIGINT, SIGTERM, or `POST /admin/drain`:

```
Signal received
    │
    ├── 1. engine.force_flush()
    │       └── Ensures all buffered writes are durable before clients disconnect
    │
    ├── 2. lifecycle.begin_drain(drain_timeout_secs)
    │       ├── State → Draining (records wall-clock timestamps)
    │       ├── TCP accept loop stops accepting new connections
    │       ├── /health/ready → 503 "draining"
    │       ├── Connection handlers react:
    │       │     ├── CAP_GOING_AWAY → send GoingAway frame
    │       │     └── No capability → send Error { ServerDraining }
    │       └── manager.set_inflight_timeout(accelerated)
    │             └── Reduces inflight timeout to min(30s, drain_timeout/2)
    │
    ├── 3. lifecycle.run_drain_wait(drain_timeout)
    │       ├── Wait for all connections to close OR timeout
    │       ├── Log "draining: N connections remaining" every 5s
    │       ├── Early exit if all connections close before timeout
    │       └── Record fafb_drain_duration_seconds metric
    │
    ├── 4. State → ShuttingDown
    │       ├── Force-close remaining connections
    │       ├── manager.shutdown() — abort delivery tasks
    │       └── engine.shutdown() — fsync storage
    │
    └── 5. Exit
          ├── If shutdown completes within shutdown_timeout_secs → exit 0
          └── If timeout → log error, exit 1
```

### Capability Negotiation

Protocol capabilities are negotiated during the Connect/ConnectAck handshake via a `u32` bitmask:

1. Client sends `Connect { ..., capabilities }` advertising what it supports
2. Server computes `agreed = client_caps & SERVER_CAPABILITIES`
3. Server sends `ConnectAck { ..., server_capabilities: agreed }`
4. Agreed capabilities stored in `AuthenticatedState.capabilities` for the session lifetime

This enables the server to make per-connection decisions (e.g., send GoingAway only to clients that understand it) without breaking legacy v1 clients that don't send capabilities.

Current capability bits: `CAP_GOING_AWAY` (bit 0), `CAP_RECONNECT` (bit 1).

### Namespace Manager Lifecycle

Consumer groups have a 60-second grace period when all members disconnect, retaining cursor state for seamless rejoin. Expired groups are lazily cleaned up on the next subscribe or publish.

## Timestamps

`MessageEnvelope.timestamp` is a `u64` representing milliseconds since Unix epoch. Timestamps are server-assigned at publish time and immutable. They appear in:

- **TCP `Deliver` frames** — `offset` (u64) and `timestamp_millis` (u64) fields, adding 16 bytes per delivery. Clients can use these for offset tracking, latency measurement, and future resume-after-reconnect.
- **HTTP publish response** — `timestamp_millis` field in the JSON response.
- **HTTP topic info** — `oldest_timestamp_millis` and `newest_timestamp_millis` fields (null when empty).
- **Delivery latency metric** — computed as `now_millis() - envelope.timestamp`.

## Retention & Cleanup

Per-namespace retention policies control how long segment files are kept on disk. Three dimensions can be configured independently — the most aggressive wins:

| Dimension | Config field | Behavior |
|-----------|-------------|----------|
| Age | `retention.max_age_secs` | Delete segments whose `mtime` is older than the cutoff |
| Size | `retention.max_bytes` | Delete oldest segments until total bytes per topic is within budget |
| Count | `retention.max_segments` | Keep at most N segment files per topic |

The `RetentionTask` runs as a background tokio task, sweeping every `storage.retention_check_interval_secs` (default 60s). It discovers topics by scanning the filesystem via `storage.list_topics()` — not the engine's in-memory map — so orphaned segments for evicted topics are also cleaned.

**Safety invariants:**

- **min_live_offset floor** — retention never deletes segments containing offsets that an active subscriber or consumer group still needs. `NamespaceManager::min_live_offset()` computes the lowest offset across all subscribers and groups for a topic.
- **Compact-before-delete** — the engine's in-memory buffer is compacted before segment files are removed. If a crash occurs after compact but before delete, orphaned segments are harmless (cleaned up next sweep). The reverse order would cause replay to reference deleted data.
- **Coexistence with `max_segments_per_topic`** — the global `StorageConfig.max_segments_per_topic` acts as a hard ceiling enforced inline during segment rotation. Per-namespace `retention.max_segments` is a policy enforced by the background task. Both can be active simultaneously.

## Segment Compaction

The **Compaction Task** is a background tokio task spawned in `main.rs` that periodically merges runs of consecutive small segment files into larger ones. This reduces fd pressure, speeds up segment listing, and makes cold reads more efficient (fewer segment boundary crossings).

The task runs on a configurable interval (`storage.compaction_interval_secs`, default 5 min) and for each topic:

1. Cleans up orphan `.tmp` files from any previously interrupted compaction
2. Scans segments via `retention::scan_segments()` (shared with the retention task)
3. Identifies the active writer segment (from the LRU writer cache, or falls back to the last segment)
4. Filters out the active writer and any segments at or above `min_live_offset` (consumer safety)
5. Finds runs of 2+ consecutive segments where each is smaller than `segment_merge_threshold_bytes` (default 16 MiB), capped so the merged output doesn't exceed `segment_max_bytes`
6. For each run: writes a merged `.seg.tmp` + `.idx.tmp` (with inline index entries), atomically renames both, deletes originals, and updates `segment_index`/`index_cache`/`namespace_disk_bytes` caches

**Error isolation:** Each merge run is independent — if one fails (I/O error, corrupt source), temp files are cleaned up, the error is logged, and remaining runs still proceed.

**Crash safety:** Merged segments are written to `.tmp` files first, then atomically renamed. If a crash occurs mid-write, orphan `.tmp` files are cleaned up on the next compaction sweep. If a crash occurs after rename but before deleting originals, duplicate records exist on disk — `replay_topic` handles these since records are sorted by offset.

**Coexistence with retention:** Compaction reduces file count; retention removes old data. They run independently on separate intervals and can be tuned or disabled separately.

## Observability

### Prometheus Metrics

Metrics are emitted via the `metrics` crate and exposed at `GET /v1/metrics` in Prometheus text format. Controlled by `metrics.enabled` config (default: true).

**Gauges:**
- `fafb_connections_active{namespace}` — current connections per namespace
- `fafb_queue_depth{namespace, topic}` — messages in buffer per topic
- `fafb_queue_memory_bytes{namespace, topic}` — bytes used per topic
- `fafb_global_memory_bytes` — total bytes across all topics
- `fafb_namespace_disk_bytes{namespace}` — total segment + index bytes on disk per namespace (emitted periodically by the cleanup task, not on every publish)

**Counters:**
- `fafb_messages_published_total{namespace, topic}` — publishes accepted
- `fafb_messages_delivered_total{namespace, topic, mode}` — deliveries sent (mode: individual/group)
- `fafb_messages_dropped_total{namespace, topic, reason}` — drops (reason: channel_full)
- `fafb_messages_evicted_total{namespace, topic}` — messages evicted by DropOldest backpressure
- `fafb_connections_total{namespace}` — connections accepted (lifetime)
- `fafb_connections_rejected_total{reason}` — rejected (reason: namespace_limit, auth_failed)
- `fafb_slow_consumers_total{namespace}` — slow consumer events (individual + group)
- `fafb_messages_skipped_total{namespace, topic}` — messages skipped by delivery task skip-forward (evicted offsets)
- `fafb_retention_segments_deleted_total{namespace, topic}` — segments removed by retention
- `fafb_segments_merged_total{namespace, topic}` — merge operations performed by compaction
- `fafb_quota_rejected_total{namespace, reason}` — namespace quota rejections (reason: max_messages, max_topics, max_disk_bytes, max_consumer_groups)
- `fafb_publish_rejected_total{reason}` — publish rejections (reason: memory_full, disk_full, queue_full, topic_bytes_exceeded)
- `fafb_auth_failures_total{reason}` — authentication failures (reason: unknown_key, namespace_mismatch)
- `fafb_fd_evictions_total` — LRU segment writer fd evictions
- `fafb_cold_batch_reads_total{namespace, topic}` — cold-path disk reads

**Histograms:**
- `fafb_delivery_latency_seconds{namespace, topic}` — publish-to-deliver latency
- `fafb_retention_sweep_duration_seconds` — time per retention sweep cycle
- `fafb_compaction_sweep_duration_seconds` — time per compaction sweep cycle

## Tiered Storage

The queue engine treats the in-memory `VecDeque` as a hot cache over a larger on-disk log. Segment files on disk are the authoritative record of all published messages; the in-memory buffer is a read-ahead window for fast consumers.

### Hot and Cold Read Paths

`consume_from(ns, topic, offset)` first checks the in-memory buffer (hot path). When the requested offset is below `base_offset` (evicted by `DropOldest` or compaction), the engine falls back to `StorageBackend::read_at` for single-message reads or `StorageBackend::read_cold_batch(ns, topic, start_offset, max_count)` for batched sequential reads (cold path). Both locate the correct segment via a cached segment index (`RwLock<HashMap<TopicKey, Vec<(TopicOffset, PathBuf)>>>`), populated on init, updated on rotation and deletion, which enables O(log n) binary search to find the target segment.

### Sparse Offset Index Files (`.idx`)

Each segment file has an accompanying `.idx` file containing sparse index entries. Every `index_interval` records (default 256), a 24-byte fixed-size entry is written:

```text
[8 bytes: u64 BE offset]
[8 bytes: u64 BE byte_position]       // position within .seg file
[8 bytes: u64 BE timestamp_millis]    // from MessageEnvelope.timestamp
```

The first record of every segment is always indexed (entry at byte position 0). For a 64MB segment with ~100-byte records at interval 256: ~2500 entries × 24 bytes = ~60KB — small enough to load entirely into memory.

`SegmentIndex` loads the `.idx` file and binary searches for the closest entry ≤ the target offset, then seeks the segment file to that byte position. This turns O(n) linear scans into O(log n + 256) for both `read_at` and `SegmentCursor::skip_to`. Loaded indexes are cached on `FsStorage` in a `RwLock<HashMap<PathBuf, Arc<SegmentIndex>>>` to avoid re-reading from disk on repeated cold reads. Cache entries are evicted when segments are deleted.

On startup, `init()` rebuilds missing `.idx` files by scanning segments with `RecordStream`. The `RecordStream::position()` method tracks cumulative byte offsets during reads, eliminating the need to re-serialize envelopes for byte position calculation.

### SegmentCursor for Sequential Cold Reads

A consumer catching up across many messages reads sequentially, not randomly. `SegmentCursor` keeps the file handle and `BufReader` position open across calls to `read_next()`, giving O(1) per-message cost. `SegmentCursor::skip_to(offset)` loads the segment's `.idx` file, binary searches for the nearest index entry, seeks to that byte position, then scans forward at most `index_interval` records — using a `pending` field for read-ahead buffering so partially-read segments are not re-scanned. When the cursor reaches the end of a segment it automatically opens the next one. This avoids the naive pattern of re-opening and re-scanning the segment file for every message.

Delivery tasks use a cold batch buffer (`VecDeque`, configurable via `ManagerConfig.cold_batch_size`, default 256) to amortize disk I/O. The shared delivery path is implemented by `cold_deliver_individual` and `cold_deliver_group` helper functions, which drain the batch buffer and push messages through the normal delivery interceptors and subscriber channels.

### Disk-Backed Mode

When `EngineConfig.disk_backed = true`, disk capacity gates publish acceptance instead of the global memory budget:

- Per-topic backpressure is forced to `DropOldest` (memory stays bounded via eviction).
- The `global_max_memory_bytes` check is skipped (memory is a cache, not the acceptance boundary).
- Publishes are rejected only when disk usage exceeds `disk_capacity_threshold_pct` (default 90%).
- `consume_from` transparently serves cold reads from disk when the offset is below the in-memory window.

When `disk_backed = false` (default), all existing behavior is unchanged.

### Segment Cleanup

A cleanup background task runs periodically (default 60s) and removes segment files that are no longer needed. It first calls `engine.compact()` to trim the in-memory buffer, then calls the unified `storage.delete_segments(before_offset, max_segments, safe_below)` to remove on-disk segments. This compact-before-delete ordering is crash-safe: if a crash occurs after compact but before delete, orphaned segments are harmless and cleaned up on the next sweep. The unified `delete_segments` method replaces the old separate `delete_segments` + `cleanup_segments` calls. It queries `NamespaceManager::min_live_offset(ns, topic)` to determine the lowest offset any active subscriber or consumer group still requires (`safe_below`). Segments whose highest offset falls below this floor are safe to delete. This ensures no consumer loses access to data it has not yet consumed.

The cleanup task also emits the `fafb_namespace_disk_bytes` gauge for each namespace after each sweep, providing periodic observability of per-namespace disk usage without adding overhead to the publish hot path.

## Logging

### Configuration

Logging is configured via the `[logging]` section in config.toml. All fields have defaults, and omitting the section entirely gives stdout-only INFO-level text output (backwards compatible).

- **`level`** — RUST_LOG-style filter (default `"info"`). Supports per-module levels: `"info,rust_fafb_q::storage::fs=debug"`. The `RUST_LOG` env var overrides this at startup.
- **`format`** — `"text"` (default) or `"json"` for structured output.
- **`stdout`** — write to stdout (default: true).
- **`file_enabled`** — write to rotating log files (default: false).
- **`file_dir`** / **`file_prefix`** / **`rotation`** — file output path, name prefix, and rotation period (`"daily"` or `"hourly"`).
- **`max_age_days`** (default 14) / **`max_total_size_mb`** (default 50) — log file retention limits enforced by a cleanup task.

### Subscriber Architecture

The tracing subscriber is assembled in `init_tracing()` (main.rs):

1. A `reload::Layer` wrapping an `EnvFilter` — enables dynamic level changes at runtime.
2. An optional stdout `fmt::Layer` (text or JSON).
3. An optional file `fmt::Layer` writing to a `tracing_appender::non_blocking` writer backed by a `RollingFileAppender`.

The non-blocking writer pushes log formatting to a background thread, keeping the hot path (publish/deliver) free of I/O overhead.

### Dynamic Log Level

`PUT /v1/admin/log-level` accepts a RUST_LOG-style filter string and hot-swaps the active filter via the reload handle. `GET /v1/admin/log-level` returns the current filter. Both require a valid API key with Query permission. See [api.md](api.md#admin-endpoints).

### Log Cleanup

When `file_enabled = true`, a cleanup task runs:
- **Once at startup** — cleans up stale logs from a previous crash.
- **Hourly** — deletes files older than `max_age_days`, then deletes oldest files until total size is under `max_total_size_mb`.

The cleanup uses `tokio::task::spawn_blocking` to avoid blocking the async runtime.

### Connection Spans

Each TCP connection is wrapped in `tracing::info_span!("conn", client, peer)`. All log events within a connection (auth, subscribe, publish, deliver, disconnect) inherit the span, enabling log correlation by client ID or peer address.

### Log Levels by Operation

| Level | What's logged |
|-------|---------------|
| `error` | Panics (via panic hook), server task failures, shutdown errors |
| `warn` | Quota rejections, resource pressure at 80%, publish failures, auth failures, slow consumers, codec errors, connection flapping, retention clamped, cold-path reads starting |
| `info` | Server startup/ready/shutdown, connection open/close, consumer group create/join/empty, segment cleanup, DLQ routing, compaction sweeps |
| `debug` | Auth success, namespace connect, subscribe/unsubscribe, segment rotation/deletion, LRU fd eviction, topic creation |
| `trace` | Per-message publish, storage append, per-message delivery, ACK processing |

## Crash Isolation

### Panic Hook

A global panic hook (set in `main()` after tracing init) logs all panics through tracing with the panic message and source location. This ensures panics in any tokio task — including spawned connection handlers — are captured in log files before unwinding.

### Background Task Respawn

The main monitoring loop uses `loop { select! { ... } }` instead of a one-shot `select!`. When a background task (retention, cleanup, compaction) panics:

1. The `JoinHandle` completes with `Err(JoinError)`.
2. The failure is logged at ERROR level with the task name.
3. The task is respawned after an exponential backoff delay (1s, 2s, 4s, 8s, ... capped at 60s).
4. A per-task failure counter tracks consecutive panics.

Clean exits (e.g., compaction task when merge is disabled) are logged at INFO and not respawned.

TCP and HTTP server failures still trigger shutdown — these are fatal since the server can't accept connections.

### Connection Isolation

Each TCP client connection runs in its own spawned tokio task. A panic or error in one connection:
- Aborts only that connection's task.
- The global panic hook logs it.
- The accept loop continues serving other clients.
- The semaphore permit (if connection-limited) is released via drop.

The session layer handles all namespace manager errors via `Result` match arms — no error from one client propagates to another.

## Resource Pressure Warnings

### 80% Threshold Warnings

Several resource limits have proactive 80% threshold warnings that fire once when usage crosses the threshold and reset when it drops back below:

| Resource | Where Checked | AtomicBool Location |
|----------|---------------|---------------------|
| `max_total_messages` per namespace | Publish path | `NamespaceState.messages_warning_fired` |
| `max_disk_bytes` per namespace | Publish path | `NamespaceState.disk_bytes_warning_fired` |
| `max_connections` per namespace | Connect path | `NamespaceState.connections_warning_fired` |
| `global_max_memory_bytes` | Publish path | `InMemoryQueueEngine.memory_warning_fired` |
| `disk_capacity_threshold_pct` | Flush task | `InMemoryQueueEngine.disk_warning_fired` |

### Other Operational Warnings

- **Cold-path reads**: `warn!` when a consumer first reads from disk (behind in-memory window). Uses `cold_read_warning_fired` on the engine.
- **Retention clamped**: `warn!` when a slow subscriber prevents segment cleanup.
- **High inflight**: `warn!` when unACKed messages cross 1000-message thresholds.
- **Connection flapping**: `warn!` when a single IP connects >50 times in 60 seconds (tracked per-window in the TCP accept loop).
- **Slow startup replay**: `warn!` when topic replay takes >5 seconds on init.

## Shared Error Codes

TCP and HTTP share a unified error code table (see [protocol.md](protocol.md#shared-error-codes)). Error codes are defined once in `src/error.rs`.
