# Clustering: Primary-Backup Replication with Client-Assisted Failover

## Summary

Enable fafb-q to run as a multi-node cluster using primary-backup replication. A single leader handles all reads and writes. Followers replicate via semi-synchronous streaming for zero data loss. Clients are cluster-aware — they know about all nodes, receive `NotLeader` redirects, and handle failover transparently. No Raft, no external consensus library. The cluster module is fully isolated behind a Cargo feature gate — core code never knows it exists.

**Design priorities:** Availability first, then durability. A single-node deployment (the default) has zero clustering overhead. Any cluster continues operating as long as at least one node is alive.

## Guiding Principles

- **Availability over consensus.** No quorum requirement. The leader serves writes without waiting for a majority. Semi-sync replication guarantees data is on at least 2 nodes before ack, but a single surviving node can always serve.
- **Leader-follower, not leaderless.** A single leader handles all writes and reads. Followers exist for durability and failover readiness. No conflict resolution, no multi-writer complexity.
- **Clients are cluster-aware.** Clients know about all nodes. The protocol includes `NotLeader` redirects so clients find the leader. No transparent server-side proxy — the client handles routing.
- **Clustering is a plugin, not a branch.** The `src/cluster/` module is feature-gated. Core modules (`DefaultNamespaceManager`, `InMemoryQueueEngine`, `FsStorage`, codec, auth, retention, DLQ, wildcards) never import from `src/cluster/`. A developer working on subscriptions, fsync, or retention never encounters clustering code.
- **No external consensus dependencies.** No Raft library. The replication protocol is purpose-built: single leader, unidirectional stream, semi-sync ack from 1 follower. Simple, fast, auditable.
- **2-node clusters are first-class.** The common deployment is 2 nodes. The design doesn't require 3+ nodes for correctness.
- **Cluster membership is static config.** Nodes are configured with a static peer list. All nodes must have identical peer lists (validated on handshake). No gossip, no service discovery.

## Cluster Architecture

```
              Client (nodes=A,B)
                    │
        ┌───────────┴───────────┐
        ▼                       ▼
   ┌─────────┐            ┌─────────┐
   │ Node A  │◄──────────►│ Node B  │
   │(follower)│ heartbeat  │ (leader)│
   │         │ + replicate │         │
   └────┬────┘            └────┬────┘
        │                      │
   ┌────▼────┐            ┌────▼────┐
   │ Storage │            │ Storage │
   └─────────┘            └─────────┘
```

Each node runs the full fafb-q stack (TCP server, HTTP server, queue engine, storage). The `ClusteredManager` wraps the `DefaultNamespaceManager` on each node — it does not modify it.

### Modularity: Wrapper Architecture

```
TCP/HTTP Session<M: NamespaceManager>
         │
         ▼
┌── cluster.enabled = false ──────────────────────┐
│ DefaultNamespaceManager<Q>  (direct, zero cost)  │
└──────────────────────────────────────────────────┘

┌── cluster.enabled = true ───────────────────────┐
│ ClusteredManager<M: NamespaceManager>            │
│   │                                              │
│   ├─ publish() → validate → replicate →          │
│   │              semi-sync ack → apply           │
│   ├─ subscribe() → delegate to inner (local)     │
│   ├─ fetch() → delegate to inner (local)         │
│   ├─ ack() → delegate to inner + replicate       │
│   │          offset                              │
│   └─ NotLeader redirect for follower nodes       │
│                        │                         │
│            DefaultNamespaceManager<Q> (unchanged) │
└──────────────────────────────────────────────────┘
```

**Key invariant:** `DefaultNamespaceManager`, `InMemoryQueueEngine`, `FsStorage`, and all existing modules never import `src/cluster/`. The `ClusteredManager` imports them. Dependencies flow inward, never outward.

**Wiring in `main.rs`** — the only place clustering touches core startup:

```rust
if config.cluster.enabled {
    let inner = DefaultNamespaceManager::new(engine.clone(), ...);
    let clustered = ClusteredManager::new(inner, engine, cluster_config).await;
    start_servers(clustered, ...).await;
} else {
    let manager = DefaultNamespaceManager::new(engine, ...);
    start_servers(manager, ...).await;  // current path, zero change
}
```

## Node Communication

### Heartbeat Protocol

Leader sends heartbeats to all followers every `heartbeat_interval_ms` (default: 1000ms). The heartbeat serves triple duty: liveness detection, replication progress tracking, and time skew monitoring.

**Heartbeat contents (leader → follower):**
- Leader's term (monotonic counter incremented on each promotion)
- Current head offsets per topic
- Leader's corrected timestamp
- Replication head sequence number

**Heartbeat response (follower → leader):**
- Follower's last applied replication sequence
- Follower's corrected timestamp

### Peer List Validation

On every node-to-node connection handshake, nodes exchange their full peer lists. If the sets don't match, the connection is rejected:

```
Node 3 connects to Node 1:
  → Handshake: "I am Node 3, my peers are [Node 1, Node 3]"
  → Node 1: "REJECT — my peers are [Node 1, Node 2, Node 3], you're missing Node 2"
  → Node 3 logs FATAL, exits immediately
  → Node 3 never opens TCP/HTTP ports for clients
```

**Rule:** Every node must have an identical peer set (order doesn't matter). This prevents misconfigured nodes from joining and causing subtle replication gaps or split-brain scenarios.

### Leader Election

**Fresh cluster startup:**
- Nodes connect to all configured peers
- If no existing leader found, the node with the lowest `node_id` becomes initial leader
- Other nodes join as followers

**Joining a running cluster:**
- New/restarted node connects to peers, discovers existing leader
- Joins as follower, begins catch-up

**Term counter:**
- Monotonic `u64`, persisted to disk (`cluster.data_dir/term`)
- Incremented on every promotion
- Included in every heartbeat and replication entry
- A node that sees a higher term immediately demotes to follower
- Prevents stale leaders on recovery

## Replication Protocol

### Semi-Synchronous Streaming

The leader streams mutations to followers over a persistent TCP connection on the cluster port. The leader acks the client only after at least 1 follower confirms — zero data loss guarantee.

**What gets replicated:**
- `Publish` — full `MessageEnvelope` with leader-assigned offset
- `CommitOffset` — consumer group offset advances
- `TopicCreated` — new topic materialized on first publish

**What does NOT get replicated:**
- Config (each node loads its own)
- Subscriptions (node-local, rebuilt on client connect)
- Inflight tracking (node-local delivery state)
- Retention/compaction actions (each node manages its own segments independently)

### Replication Flow (Publish Hot Path)

```
Client → Publish → Leader
                     │
         ┌───────────┤ (parallel)
         ▼           ▼
    Local write   Stream to followers
         │           │
         │      First follower acks
         │           │
         └─────┬─────┘
               ▼
         max(local, first_ack) complete
               │
               ▼
         PublishAck → Client
```

1. Client sends `Publish` to leader
2. Leader assigns offset, writes locally AND streams `ReplicationEntry::Publish` to all followers — in parallel
3. Leader waits for: local write complete AND at least 1 follower ack
4. Latency = `max(local_write, fastest_follower_ack)` — on LAN, often zero added latency
5. Leader sends `PublishAck` to client
6. Remaining followers catch up asynchronously

### Replication Wire Format

```rust
/// Streamed from leader to followers over the cluster TCP connection.
/// Length-prefixed, same framing pattern as client protocol.
enum ReplicationEntry {
    Publish {
        term: u64,
        namespace: NamespaceId,
        topic: TopicName,
        envelope: MessageEnvelope,  // includes leader-assigned offset
    },
    CommitOffset {
        term: u64,
        namespace: NamespaceId,
        topic: TopicName,
        group: GroupName,
        offset: TopicOffset,
    },
    TopicCreated {
        term: u64,
        namespace: NamespaceId,
        topic: TopicName,
    },
}

// TopicCreated is idempotent — if the topic already exists locally
// (e.g., from a prior snapshot or replay), the follower silently ignores it.

/// Follower → Leader acknowledgment
enum ReplicationAck {
    Progress {
        last_applied_sequence: u64,  // monotonic sequence number across all entries
    },
}
```

### Sequence Numbering

Each `ReplicationEntry` gets a monotonic sequence number (separate from per-topic offsets). This is the single cursor the follower reports back. The leader tracks each follower's position for catch-up eligibility and promotion decisions.

### Consumer Group Offset Replication

`ack()` in the `NamespaceManager` trait takes a `MessageId`. The existing `DefaultNamespaceManager` resolves the `MessageId` to a topic offset and updates the consumer group's committed offset internally. The `ClusteredManager` replicates this by:

1. Delegating `ack()` to the inner manager (local state updated immediately)
2. After the inner ack resolves, reading the group's `next_offset` (the next offset to deliver). This is the replication cursor — on failover, the new leader resumes delivery from this offset, which may re-deliver messages that were inflight (not yet acked) on the old leader. This is consistent with at-least-once delivery semantics.
3. Batching these offset updates — every 100 acks or every 1 second (whichever comes first), a single `CommitOffset` replication entry is sent with the group's latest `next_offset`

**What gets replicated:** Consumer group offsets only. These are the durable cursors that survive failover.

**What does NOT get replicated:** Individual (non-group) subscriber offsets. These are per-connection state, tied to a specific client's TCP session. When a client reconnects after failover, it re-subscribes (with `SubscribeAt` to resume from its last known offset). Individual subscriber position is the client's responsibility.

**Trade-off:** Batching means up to 1 second of consumer group offset progress may be lost on failover, causing bounded re-delivery of already-acked messages. This is consistent with at-least-once delivery semantics.

### Degraded Mode

If all followers are `CatchingUp` or `Disconnected`, the leader operates in degraded mode — accepts writes without follower ack, logs a warning. Availability is never sacrificed for replication.

## Failover

### Leader State Machine

```
                    ┌──────────────────────┐
                    │     Standalone        │
                    │  (no cluster config)  │
                    └──────────────────────┘

    ┌───────────┐       ┌───────────┐       ┌──────────────┐
    │  Follower  │◄─────│  Joining   │      │ Leader       │
    │            │      │ (startup)  │      │ Suspect      │
    │ replicates │      └─────┬─────┘      │ (no clients  │
    │ from leader│            │             │  corroborate)│
    └─────┬──────┘            │             └──────┬───────┘
          │                   ▼                    │
          │         ┌─────────────────┐            │
          │         │  Determine role  │            │
          │         │  (leader exists? │            │
          │         │   lowest ID?)    │            │
          │         └────┬───────┬────┘            │
          │              │       │                  │
          │         leader  follower                │
          │              │       │                  │
          │              ▼       └──────────────────┤
          │         ┌─────────┐                     │
          │         │ Leader   │◄────────────────────┘
          │         │          │  (promote: client corroboration
          │         │ accepts  │   OR no_clients_timeout expires)
          │         │ writes & │
          │         │ reads    │
          └────────►│          │  (old leader recovers,
                    └─────────┘   sees higher term → demotes)
```

### Client-Assisted Failover

**Fast path (clients active):**

```
T+0s     Leader stops sending heartbeats
T+5s     Follower(s) detect missed heartbeats (failover_timeout_ms)
         Follower enters "Leader Suspect" state
         - Responds ClusterUnavailable to new client connections
         - Counts client connection attempts as corroboration

T+5-10s  Client(s) can't reach old leader, try follower(s)
         Follower receives client connection → corroboration
         Most-caught-up follower promotes:
           - Increments term, persists to disk
           - Transitions to Leader state
           - Begins accepting writes/reads
           - Sends heartbeats to other followers with new term
```

**Slow path (no clients):**

```
T+0s     Leader stops sending heartbeats
T+5s     Follower enters Leader Suspect state
T+35s    failover_no_clients_timeout_ms expires (default 30s)
         Most-caught-up follower auto-promotes
```

### Multi-Follower Promotion (3+ nodes)

When multiple followers detect leader failure simultaneously:

1. Each enters Leader Suspect state
2. Each follower broadcasts a `PromotionCandidate { node_id, last_applied_sequence }` message to all peers on the cluster port
3. Followers collect responses for a bounded window (`failover_timeout_ms / 2`, e.g., 2.5 seconds)
4. After the window:
   - If a follower has heard from all other followers: the one with highest `last_applied_sequence` promotes. Tie: lowest `node_id` wins.
   - If a follower has NOT heard from all peers (partition): it promotes only if it is the highest-sequence among responding peers AND has received client corroboration. If no client corroboration, it waits for `failover_no_clients_timeout_ms`.
5. The promoting follower increments term, persists, begins serving
6. Losing followers see the new leader's heartbeat with higher term → stay follower

**Ordering:** Client corroboration is checked after position exchange, not before. This prevents a less-caught-up follower from promoting just because it received a client first.

**Wire format:**

```rust
/// Sent on the peer protocol during Leader Suspect state
enum PeerMessage {
    // ... existing heartbeat/replication messages ...
    PromotionCandidate {
        node_id: NodeId,
        last_applied_sequence: u64,
        term: u64,  // current term before promotion
    },
}
```

### Graceful Leadership Transfer (Rolling Upgrades)

```
POST /admin/drain on the leader:

1. Leader picks transfer target (most caught-up follower)
2. Leader sends LeadershipTransfer signal to target (no term in message)
3. Target increments its own term (current_term + 1), persists to disk, promotes
4. Target begins sending heartbeats with new term
5. Old leader receives heartbeat with higher term → self-demotes to follower
6. Old leader sends GoingAway to connected clients
7. Clients reconnect → hit old leader → NotLeader { new leader addr }
   OR connect to new leader directly
```

The target node is solely responsible for computing the new term. The leader simply signals "you should promote" — it does not dictate the term value.

If no eligible follower (InSync state) exists, drain is rejected with an error.

### Split-Brain Prevention

- **Term counter:** Higher term always wins. Stale leader self-demotes on seeing a higher term.
- **Client corroboration:** Follower needs external evidence (client connection attempts) before promoting, reducing false promotion during network partitions.
- **Peer list validation:** Misconfigured nodes can't join the cluster.
- **Time sync:** Lease/timeout decisions use corrected timestamps.

**Known limitation — 2-node network partition with clients on both sides:**

In a 2-node cluster, if the network partitions such that both nodes are alive but can't reach each other, AND clients can reach both nodes independently:
- The leader keeps serving (it's still alive)
- The follower receives client corroboration (clients can't reach leader from the follower's network side) and promotes
- Two leaders exist simultaneously during the partition

**Resolution on partition heal:**
1. The two nodes reconnect and exchange heartbeats
2. The node with the lower term sees a higher term → self-demotes to follower
3. The demoted node truncates any writes it accepted that aren't on the surviving leader (same as the "old leader recovery" flow — these writes were accepted during a split, so one side must yield)
4. Clients connected to the demoted node receive `NotLeader` on their next operation and reconnect to the true leader

**Practical impact:** This window requires a simultaneous network partition AND clients on both sides of the partition — a rare scenario. The data loss is bounded to writes accepted by the demoted side during the partition. For most deployments (same rack, same network), this is not a concern. For deployments where this matters, use 3+ nodes where the multi-follower promotion protocol provides stronger guarantees.

### Old Leader Recovery

1. Node starts, loads persisted term from disk
2. Connects to peers, discovers a leader with higher term
3. Immediately joins as follower
4. Leader streams missed `ReplicationEntry` items to catch up
5. If the old leader has unreplicated messages (written locally but never confirmed by a follower before crash), these are truncated — they were never acked to clients (semi-sync guarantee), so no data loss
6. Once caught up, follower reaches `InSync` state and is eligible for future promotion

## Catch-Up & Snapshot Transfer

### Strategy 1 — Stream Replay (Small Gaps)

```
Follower: "My last applied sequence is 50,000"
Leader:   "My head is at sequence 52,347"
Gap:      2,347 entries — fits in replication buffer

Leader streams ReplicationEntries 50,001 → 52,347
Follower applies each to local engine
Follower enters InSync state
```

### Strategy 2 — Snapshot Transfer (Large Gaps or New Node)

```
Follower: "My last applied sequence is 10,000"
Leader:   "My head is at 500,000, buffer starts at 420,000"
Gap exceeds buffer → snapshot required

1. Leader generates snapshot manifest:
   - Segment files per namespace/topic (filename, size, checksum)
   - Topic metadata (head offsets, consumer group positions)
   - Current term, current sequence number

2. Leader sends manifest to follower

3. Follower diffs against local segments:
   - Already has (checksum match): skip
   - Missing: request from leader
   - Extra (orphaned from pre-crash): delete

4. Leader streams missing segment files
   - Immutable (rolled) segments: full file transfer
   - Active segment: captured up to the snapshot's sequence number
     (the manifest records the exact byte length to transfer)
   - Streamed, not loaded into memory — handles GBs of data

5. Follower writes segments, rebuilds engine state via engine.init()

6. Stream replay picks up from the snapshot's sequence number
   (entries written to the active segment after the snapshot point
   are delivered via normal replication catch-up)

7. Once fully caught up, follower enters normal replication
```

### Replication Buffer

```rust
struct ReplicationBuffer {
    entries: VecDeque<(SequenceNum, ReplicationEntry)>,
    max_entries: usize,       // default: 100,000
    max_bytes: usize,         // default: 256MB
    current_bytes: usize,
    oldest_sequence: u64,
    head_sequence: u64,
}
```

When a follower's gap is within `oldest_sequence..head_sequence`, stream replay is used. Otherwise, snapshot transfer.

### Follower States

```rust
enum FollowerState {
    Joining,          // handshake in progress
    CatchingUp,       // receiving replay or snapshot
    InSync,           // fully caught up, eligible for promotion
    Disconnected,     // heartbeat lost
}
```

Only `InSync` followers count for semi-sync replication acks and are eligible for promotion.

## Client Protocol Changes

### New Frame Types

```rust
/// Server → Client: You connected to a follower, here's the leader
Frame::NotLeader {
    leader_addr: String,  // "host:port" of current leader
}

/// Server → Client: Cluster is in failover, no leader currently
Frame::ClusterUnavailable {
    retry_after_ms: u32,  // suggested wait before retry
    message: String,      // human-readable reason
}
```

### New Capability Bit

```rust
CAP_CLUSTER = 0x80  // bit 7
```

Clients that set `CAP_CLUSTER` in their `Connect` frame understand `NotLeader` and `ClusterUnavailable`. Clients without this bit receive a standard `Error::Unavailable` when connecting to a follower — backwards compatible.

### Client Connection Flow

```
Client configured with nodes = ["192.168.1.1:9876", "192.168.1.2:9876"]

1. Try nodes in order (cached leader first if known):
   a. Connect to node
   b. Send Connect { ..., capabilities: CAP_CLUSTER | ... }
   c. Receive response:

      ConnectAck { success: true }
        → Leader found. Proceed normally.

      NotLeader { leader_addr }
        → Cache leader_addr, connect to it.

      ClusterUnavailable { retry_after_ms }
        → Failover in progress. Wait, try next node.

      Connection refused / timeout
        → Node is down. Try next node.

2. On connection drop (leader crashed):
   a. Clear cached leader
   b. Retry from step 1
   c. Exponential backoff with jitter, capped at 5s
   d. Max retries configurable (default: 10)

3. On GoingAway (graceful drain):
   a. Reconnect — try cached leader first
   b. If cached leader also draining (rolling upgrade), fall through to node list
```

### Follower Behavior on Client Connect

- Client has `CAP_CLUSTER` + leader known: respond `NotLeader { leader_addr }`
- Client has `CAP_CLUSTER` + in failover: respond `ClusterUnavailable { retry_after_ms }`
- Client lacks `CAP_CLUSTER`: respond `Error { code: Unavailable }`

**Client-assisted failover signal:** A cluster-aware client connecting to a follower in Leader Suspect state is itself the corroboration. No explicit "I couldn't reach the leader" message needed.

### Client Spec Documentation Requirements

- All frame types and their semantics
- Connection lifecycle state machine (connecting → authenticating → ready → reconnecting)
- Retry/backoff behavior with exact defaults
- Interaction between `NotLeader` redirect and `GoingAway` during rolling upgrades
- Example flows: normal connect, redirect, failover, rolling upgrade

## FafbClient (Rust Client Library)

### ClusterClientConfig

```rust
pub struct ClusterClientConfig {
    pub nodes: Vec<String>,           // ["192.168.1.1:9876", "192.168.1.2:9876"]
    pub namespace: NamespaceId,
    pub api_key: ApiKey,
    pub timeout: Duration,
    pub max_frame_size: u32,
    pub tls: ClientTlsConfig,
    pub max_retries: u32,             // default: 10
    pub retry_backoff_base_ms: u64,   // default: 100
    pub retry_backoff_max_ms: u64,    // default: 5000
}
```

### ClusterClient

```rust
pub struct ClusterClient {
    config: ClusterClientConfig,
    inner: FafbClient,              // current connection
    cached_leader: Option<String>,  // last known leader addr
}

impl ClusterClient {
    pub async fn connect(config: ClusterClientConfig) -> Result<Self>;
    pub async fn publish(&mut self, topic, payload, headers) -> Result<PublishAck>;
    pub async fn subscribe(&mut self, topic, group) -> Result<SubscribeAck>;
    // ... other methods delegate to inner with reconnect logic
}
```

- On connection drop, automatically retries the node list
- Tries cached leader first, then falls through to configured nodes
- `NotLeader` redirect handled internally — caller never sees it
- `ClusterUnavailable` triggers retry with backoff
- Existing `FafbClient` unchanged for single-node use

## CLI Changes

```bash
# Single node (unchanged)
fafb-q pub --server 192.168.1.1:9876 ...

# Cluster-aware
fafb-q pub --nodes 192.168.1.1:9876,192.168.1.2:9876 ...
```

- `--nodes` accepts comma-separated addresses
- `--server` still works for single-node (backwards compatible)
- If both `--nodes` and `--server` provided: error
- Profile config supports `nodes`:

```toml
[profiles.production]
nodes = ["192.168.1.1:9876", "192.168.1.2:9876"]
namespace = "myapp"
api_key = "secret"
```

## Time Synchronization

### Application-Level NTP

Each node syncs to a configurable NTP server to maintain consistent time across the cluster.

**On startup:**
1. Query configured `time_server` (default: `pool.ntp.org`)
2. Compute `offset = ntp_time - local_clock`
3. Store offset in memory
4. All cluster time operations use `corrected_time() → local_clock + offset`
5. If NTP fails: log `WARN`, set offset to 0, retry in background

**Periodic re-sync:**
- Background task re-queries every `time_sync_interval_secs` (default: 300)
- On success: apply delta gradually over ~10 seconds (no timestamp regression)
- On failure: keep current offset, log warning, retry next interval
- Never a hard gate — node always starts and keeps running

**Skew monitoring:**
- Heartbeats include corrected timestamps
- If skew between nodes exceeds `clock_skew_warn_threshold_ms` (default: 500ms), log warning
- Exposed as metric: `fafb_cluster_clock_skew_ms{peer}`

**Injection mechanism — preserving module isolation:**

`now_millis()` in `src/types.rs` reads from a global `AtomicI64` offset (initialized to 0). The cluster module sets this offset at startup and on each re-sync. Core code never imports `src/cluster/` — it just reads the atomic:

```rust
// src/types.rs — no cluster import needed
static CLOCK_OFFSET_MS: AtomicI64 = AtomicI64::new(0);

pub fn set_clock_offset(offset_ms: i64) {
    CLOCK_OFFSET_MS.store(offset_ms, Ordering::Relaxed);
}

pub fn now_millis() -> u64 {
    let raw = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as i64;
    (raw + CLOCK_OFFSET_MS.load(Ordering::Relaxed)) as u64
}
```

In non-cluster mode, offset stays 0 — `now_millis()` behaves identically to today.

**What uses corrected time:**
- Message timestamps (via `now_millis()` which includes the offset)
- Heartbeat timestamps
- Failover timeout calculations

**What does NOT use corrected time:**
- Tokio timers / `sleep` / `timeout` (monotonic clock, unaffected)
- Retention age calculations (self-consistent — compare corrected timestamps against corrected timestamps)

## Configuration

```toml
[cluster]
enabled = false                          # master switch (Tier 3 — restart only)
node_id = 1                              # unique ID for this node (u64, > 0)
bind = "0.0.0.0:9877"                   # node-to-node communication port
peers = ["192.168.1.2:9877"]            # all other nodes in the cluster

# Heartbeat & failover
heartbeat_interval_ms = 1000             # leader → follower heartbeat frequency
failover_timeout_ms = 5000               # missed heartbeats before leader suspect
failover_no_clients_timeout_ms = 30000   # auto-promote when no client corroboration

# Replication
replication_buffer_max_entries = 100000  # stream replay buffer (entry count)
replication_buffer_max_bytes = 268435456 # stream replay buffer (256MB)

# Time sync
time_server = "pool.ntp.org"             # NTP server, empty = disabled
time_sync_interval_secs = 300            # re-sync frequency
clock_skew_warn_threshold_ms = 500       # skew warning threshold

# Security
cluster_secret = ""                      # shared secret for node-to-node auth (≥32 chars)
```

### Validation Rules

| Rule | Error |
|---|---|
| `enabled` requires `node_id > 0` | "cluster.node_id must be > 0" |
| `enabled` requires `peers` non-empty | "cluster.peers must contain at least one peer" |
| `enabled` requires `cluster_secret` ≥ 32 chars | "cluster.cluster_secret must be at least 32 characters" |
| `node_id` must not appear in `peers` | "cluster.peers must not include this node's own address" |
| `failover_timeout_ms` ≥ `heartbeat_interval_ms * 3` | "failover_timeout must be at least 3x heartbeat_interval" |
| `failover_no_clients_timeout_ms` ≥ `failover_timeout_ms` | "failover_no_clients_timeout must be ≥ failover_timeout" |
| `enabled` but `cluster` feature not compiled | "cluster.enabled requires --features cluster" |

### Config Tiers

- **Tier 3 (restart only):** `enabled`, `node_id`, `bind`, `peers`, `cluster_secret`
- **Tier 1 (safe hot-reload):** all timeout/buffer/time-sync settings

### No Per-Namespace Replication

Replication is all-or-nothing. If clustering is enabled, all namespaces and topics are replicated. No selective replication complexity.

## Module Structure

```
src/cluster/                          # #[cfg(feature = "cluster")]
├── mod.rs                            # Public API: ClusteredManager, re-exports
├── manager.rs                        # ClusteredManager<M> — NamespaceManager wrapper
├── replication.rs                    # ReplicationStream, ReplicationBuffer, semi-sync ack
├── heartbeat.rs                      # Heartbeat sender/receiver, follower state tracking
├── failover.rs                       # Leader suspect, client corroboration, promotion
├── peer.rs                           # PeerPool, node-to-node TCP, handshake, peer validation
├── snapshot.rs                       # Manifest generation, incremental segment sync
├── time_sync.rs                      # NTP client, offset tracking, corrected_time()
└── types.rs                          # ReplicationEntry, ReplicationAck, FollowerState, term
```

### Cargo Feature

```toml
[features]
default = []
cluster = []   # No external dependencies
```

No openraft. No external consensus library. Pure Rust.

## Core Changes (Minimal, Additive)

### 1. Split `DefaultNamespaceManager::publish()`

```rust
impl<Q: QueueEngine> DefaultNamespaceManager<Q> {
    /// Validate topic, check quotas, run interceptors. Does NOT write.
    pub async fn validate_publish(&self, ns, topic, msg) -> Result<MessageEnvelope> { ... }

    /// Write to engine, update state, notify delivery tasks.
    /// Returns full PublishResult (offset, evicted_messages, topic_created).
    pub async fn apply_publish(&self, ns, topic, msg) -> Result<PublishResult> { ... }

    // Existing trait method — calls both, extracts offset. Zero behavior change.
    // The NamespaceManager trait returns Result<TopicOffset>, matching the current signature.
    pub async fn publish(&self, ns, topic, msg) -> Result<TopicOffset> {
        let msg = self.validate_publish(ns, topic, msg).await?;
        let result = self.apply_publish(ns, topic, msg).await?;
        Ok(result.offset)
    }
}
```

Note: `apply_publish()` returns `PublishResult` (needed by `ClusteredManager` for `topic_created` to drive `TopicCreated` replication). The trait-level `publish()` continues to return `TopicOffset` — no trait signature change.

### 2. Add `QueueEngine::publish_with_offset()`

```rust
pub trait QueueEngine: Send + Sync {
    fn publish_with_offset(&self, ns, topic, msg, offset)
        -> impl Future<Output = Result<PublishResult>> + Send
    {
        async { Err(Error::Internal("publish_with_offset requires cluster feature".into())) }
    }
}
```

Returns an error instead of panicking — consistent with the project's crash isolation principle. The `InMemoryQueueEngine` implementation (behind `#[cfg(feature = "cluster")]`) provides the real logic.

### 3. Add `NamespaceManager::commit_offset()`

```rust
pub trait NamespaceManager: Send + Sync {
    /// Apply a replicated consumer group offset. Called on followers
    /// when receiving CommitOffset replication entries.
    /// Creates the consumer group if it doesn't exist locally.
    /// Sets the group's next_offset to the given offset.
    async fn commit_offset(&self, ns, topic, group, offset) -> Result<()> {
        Ok(()) // no-op default for non-clustered mode
    }
}
```

`DefaultNamespaceManager` provides the real implementation: it creates the consumer group state if absent and sets `next_offset` to the replicated value. This is NOT behind `#[cfg(feature = "cluster")]` — it's a simple method on the existing manager. The cluster module calls it; the manager doesn't know why.

### 4. New frame types in codec

```rust
Frame::NotLeader { leader_addr: String },
Frame::ClusterUnavailable { retry_after_ms: u32, message: String },

pub const CAP_CLUSTER: u64 = 0x80;
```

### 5. `main.rs` branching

Only touch point between core code and clustering.

## Interaction with Existing Subsystems

| Subsystem | Impact | Changes |
|---|---|---|
| **QueueEngine** | Followers write via `publish_with_offset()` | Add method (default returns error, impl behind feature) |
| **DefaultNamespaceManager** | Split publish, add commit_offset | Additive only, zero behavior change |
| **StorageBackend / FsStorage** | Each node has own storage | No change |
| **Auth** | Same on all nodes, loaded from local config | No change |
| **Retention / Compaction** | Runs independently per node. `publish_with_offset()` always appends to the current active segment regardless of offset value — it does not seek into historical segments. Followers receiving replicated data for ranges their retention already deleted skip gracefully (the data is already gone, and was retained long enough per policy). | No change to retention logic. `publish_with_offset()` appends forward-only. |
| **Graceful Deploy** | Leader transfers leadership before drain | Drain blocks until transfer |
| **Connection Handler** | Calls `M: NamespaceManager` | No change |
| **HTTP Server** | Calls manager methods | New `/health/cluster` and `/admin/cluster/*` routes |
| **Codec / Frame** | Two new frames, one new capability bit | Additive |
| **CLI** | Cluster-aware node addressing | Add `--nodes` flag, `nodes` in profile config |
| **FafbClient** | Cluster-aware client | New `ClusterClient` with redirect/failover handling |
| **Wildcard / DLQ / Retention** | Node-local on committed data | No change |

## Metrics

| Metric | Type | Description |
|---|---|---|
| `fafb_cluster_role` | gauge | 1 = leader, 0 = follower |
| `fafb_cluster_term` | gauge | current term |
| `fafb_cluster_replication_sequence` | gauge | head sequence number |
| `fafb_cluster_peer_state{peer}` | gauge | 0=disconnected, 1=joining, 2=catching_up, 3=in_sync |
| `fafb_cluster_peer_lag{peer}` | gauge | entries behind leader |
| `fafb_cluster_peer_rtt_ms{peer}` | histogram | heartbeat RTT per peer |
| `fafb_cluster_clock_skew_ms{peer}` | gauge | time skew per peer |
| `fafb_cluster_replication_ack_ms` | histogram | leader write to follower ack |
| `fafb_cluster_failover_total` | counter | leader promotions |
| `fafb_cluster_snapshot_total` | counter | snapshot transfers |
| `fafb_cluster_redirect_total` | counter | NotLeader redirects sent |
| `fafb_cluster_unavailable_total` | counter | ClusterUnavailable responses |
| `fafb_cluster_degraded_writes_total` | counter | writes without follower ack |

## Health & Admin Endpoints

### Health

```
GET /health/cluster → 200 (healthy) or 503 (failover/catching up)
```

```json
{
  "node_id": 1,
  "role": "leader",
  "term": 42,
  "replication_sequence": 123456,
  "time_sync": { "offset_ms": -12, "last_sync": "2026-03-28T14:30:00Z", "status": "synced" },
  "peers": [
    { "node_id": 2, "addr": "192.168.1.2:9877", "state": "in_sync", "lag": 0, "rtt_ms": 1.2, "clock_skew_ms": 3 }
  ]
}
```

### Admin

```
GET  /admin/cluster/status            — full cluster status
POST /admin/cluster/transfer-leader   — { "target_node_id": 2 }
POST /admin/cluster/promote           — force-promote this follower (emergency)
```

No `add-peer` or `remove-peer` — peer list is static config.

### Structured Logging

Every log line in cluster mode includes `node_id`:

```
INFO node_id=1 role=leader peer connected node_id=2
WARN node_id=2 role=follower heartbeat missed count=3
INFO node_id=2 role=follower promoted to leader term=43 reason="client corroboration"
```

## Single-Node Mode Guarantee

When `cluster.enabled = false` (default):
- No cluster code executes
- No extra threads, heartbeats, or replication overhead
- `DefaultNamespaceManager` used directly — not wrapped
- `cargo build` without `--features cluster` doesn't compile `src/cluster/`
- Benchmark: publish throughput within 1% of current baseline

## Testing Strategy

### Test Harness

`tests/cluster_common/mod.rs` — `TestCluster` struct managing N in-process nodes with unique ports.

Key methods: `start(n)`, `kill_node(id)`, `restart_node(id)`, `leader_id()`, `await_leader()`, `client_for(id)`, `cluster_client()`, `partition(a, b)`, `heal_partition(a, b)`, `await_in_sync(id)`.

All cluster tests gated behind `#[cfg(feature = "cluster")]`.

### Test Categories

**Unit tests:** ReplicationEntry serialization, buffer eviction, term persistence, failover state machine, peer list validation, catch-up strategy selection, time sync offset calculation.

**Cluster formation:** 2-node leader election, 3-node formation, mismatched peer list rejection, single-node zero overhead.

**Replication:** Publish replicates to followers, 10K message consistency, consumer group offset replication, topic auto-create replication, semi-sync ack guarantee.

**Failover:** Leader crash → client-assisted promotion, no-clients auto-promote, publish to new leader, old leader auto-rejoin, unreplicated tail truncation, most-caught-up wins (3+ nodes).

**Client protocol:** NotLeader redirect, ClusterUnavailable during failover, legacy client compatibility, GoingAway + reconnect, rolling upgrade zero failures.

**Catch-up & snapshot:** Stream replay for small gaps, snapshot for large gaps, new node full snapshot, follower not eligible during catch-up.

**Degraded mode:** All followers disconnected → leader accepts writes with warning, follower reconnects → degraded clears.

**Time sync:** Unreachable time server → warning + local clock, skew detection.

**Benchmarks:** Single-node within 1% of baseline, 2-node semi-sync overhead measurement, failover time measurement.

## Out of Scope

- **Topic partitioning** — splitting a topic across nodes for throughput scaling. This plan covers replication only.
- **Multi-datacenter replication** — WAN replication needs async replication with higher latency tolerance.
- **Dynamic membership** — `add-peer` / `remove-peer` API. Peer list is static config for now.
- **Gossip-based discovery** — nodes use static peer list. DNS/Consul/Kubernetes discovery is a deployment concern.
- **Read replicas** — followers don't serve reads. All traffic goes to leader.
- **Per-namespace replication settings** — replication is all-or-nothing when clustering is enabled.

## Current State

| Capability | Implementation | Status |
|---|---|---|
| `NamespaceManager` trait | Generic `M: NamespaceManager` throughout TCP/HTTP sessions, dependency injection ready | **Done** — `ClusteredManager<M>` can wrap without touching callers |
| `QueueEngine` trait | Generic `Q: QueueEngine` with `publish()`, `consume_from()`, `init()` | **Done** — needs `publish_with_offset()` addition |
| `FafbClient` | TCP/TLS client with connect, auth, publish, subscribe, fetch, ack | **Done** — `ClusterClient` wraps this |
| Codec capabilities | Bitmask negotiation at connect (`CAP_GOING_AWAY` through `CAP_MESSAGE_TTL`) | **Done** — needs `CAP_CLUSTER` (bit 7) |
| `ServerLifecycle` | `Running → Draining → ShuttingDown` state machine, `GoingAway` frame | **Done** — failover hooks into drain path |
| `FsStorage` | Segment files with sparse index, LRU fd cache, compaction | **Done** — each node manages own storage independently |
| Auth / hot-reload | `ArcSwap`-based auth store, `SharedConfig` for lock-free reads | **Done** — each node loads own config |
| `now_millis()` | `SystemTime`-based timestamp in `src/types.rs` | **Done** — needs `AtomicI64` clock offset for NTP |
| Graceful drain | `POST /admin/drain` → `GoingAway` → connection close | **Done** — needs leadership transfer before drain |
| CLI profiles | `~/.fafb-q/config.toml` with `[profiles.*]` sections, `--server` flag | **Done** — needs `--nodes` flag |

## Implementation Phases

### Phase 24.1 — Foundation: Types, Config, Feature Gate & Test Harness

**Goal:** Establish the `src/cluster/` module behind a Cargo feature gate. Define all cluster-specific types. Add `[cluster]` config with full validation. Build the `TestCluster` test harness that all subsequent phases depend on. Add the minimal hooks to core modules that clustering needs — without importing `src/cluster/`. All changes are additive — zero behavior change for single-node mode.

**Subsystems touched:** `Cargo.toml`, `src/config.rs`, `src/types.rs`, `src/codec/frame.rs`, `src/namespace/`, `src/queue/`, `tests/`. **Not touched:** `src/storage/`, `src/auth/`, `src/tls.rs`, `src/retention.rs`, `src/server/`.

- [x] Add `cluster` feature to `Cargo.toml`
  - [x] `[features]` section: `cluster = []` (no external dependencies)
  - [x] Verify `cargo build` without `--features cluster` does not compile `src/cluster/`
- [x] Create `src/cluster/mod.rs` with `#[cfg(feature = "cluster")]` gate and `src/cluster/types.rs`
  - [x] Module files for `manager`, `replication`, `heartbeat`, `failover`, `peer`, `snapshot`, `time_sync` are created in later phases when their implementations begin — no empty stubs
- [x] Define `ReplicationEntry` enum in `src/cluster/types.rs`
  - [x] `Publish { term, namespace, topic, envelope }` variant
  - [x] `CommitOffset { term, namespace, topic, group, offset }` variant
  - [x] `TopicCreated { term, namespace, topic }` variant
  - [x] Derive `Serialize`, `Deserialize` for wire format
- [x] Define `ReplicationAck` enum
  - [x] `Progress { last_applied_sequence: u64 }` variant
- [x] Define `FollowerState` enum: `Joining`, `CatchingUp`, `InSync`, `Disconnected`
- [x] Define `PeerMessage` enum for peer protocol
  - [x] `PromotionCandidate { node_id, last_applied_sequence, term }` variant
- [x] Add `Term` type (wrapper around `u64`) with persistence helpers (read/write to `cluster.data_dir/term`)
- [x] Add `[cluster]` config section to `src/config.rs`
  - [x] `ClusterConfig` struct with all fields: `enabled`, `node_id`, `bind`, `peers`, `heartbeat_interval_ms`, `failover_timeout_ms`, `failover_no_clients_timeout_ms`, `replication_buffer_max_entries`, `replication_buffer_max_bytes`, `time_server`, `time_sync_interval_secs`, `clock_skew_warn_threshold_ms`, `cluster_secret`
  - [x] `Default` impl with documented defaults (heartbeat 1000ms, failover 5000ms, no-clients 30000ms, buffer 100K entries / 256MB, time server `pool.ntp.org`, sync 300s, skew warn 500ms)
  - [x] Wire into `ServerConfig` as `cluster: ClusterConfig`
- [x] Add config validation rules in `ServerConfig::validate()`
  - [x] `enabled` requires `node_id > 0`
  - [x] `enabled` requires `peers` non-empty
  - [x] `enabled` requires `cluster_secret` >= 32 chars
  - [x] `node_id` must not appear in `peers` — *Note: validated as bind-address string match; equivalent addresses like 0.0.0.0 vs concrete IP not caught at config time — peer handshake provides runtime defense*
  - [x] `failover_timeout_ms >= heartbeat_interval_ms * 3`
  - [x] `failover_no_clients_timeout_ms >= failover_timeout_ms`
  - [x] `enabled` but `cluster` feature not compiled: startup error
- [x] Classify config fields by hot-reload tier
  - [x] Tier 3 (restart only): `enabled`, `node_id`, `bind`, `peers`, `cluster_secret`
  - [x] Tier 1 (safe hot-reload): all timeout/buffer/time-sync settings
- [x] Update `config/config.example.toml` with `[cluster]` section (commented out, all defaults shown)
- [x] Split `DefaultNamespaceManager::publish()` into two steps
  - [x] `validate_publish(&self, ns, topic, msg) -> Result<MessageEnvelope>` — validate topic, check quotas, run interceptors, does NOT write
  - [x] `apply_publish(&self, ns, topic, msg) -> Result<PublishResult>` — write to engine, update state, notify delivery tasks, return full `PublishResult { offset, evicted_messages, topic_created }`
  - [x] Existing `publish()` calls both, extracts `offset` — zero behavior change
  - [x] Verify all existing publish tests still pass
- [x] Add `publish_with_offset()` to `QueueEngine` trait
  - [x] Default impl returns `Err(Error::InvalidOperation("publish_with_offset requires cluster feature"))` — not a panic (crash isolation principle)
  - [x] `InMemoryQueueEngine` implementation behind `#[cfg(feature = "cluster")]`: uses `Topic::publish_at_offset()` for atomic offset assignment and append
- [x] Add `commit_offset()` to `NamespaceManager` trait
  - [x] `async fn commit_offset(&self, ns, topic, group, offset) -> Result<()>` with default no-op
  - [x] `DefaultNamespaceManager` implementation: creates consumer group state if absent, sets `next_offset` to the given value
  - [x] NOT behind `#[cfg(feature = "cluster")]` — it's a simple method on the existing manager
- [x] Add new frame types to `src/codec/frame.rs`
  - [x] `Frame::NotLeader { leader_addr: String }` — server → client redirect (0x1A)
  - [x] `Frame::ClusterUnavailable { retry_after_ms: u32, message: String }` — server → client (0x1B)
  - [x] Assign frame type bytes, add encode/decode
  - [x] `CAP_CLUSTER: Capabilities = 1 << 7` (bit 7) capability constant
  - [x] Do NOT add `CAP_CLUSTER` to `SERVER_CAPABILITIES` yet — that happens in Phase 24.6
- [x] Add `clock_offset` injection point in `src/types.rs`
  - [x] `static CLOCK_OFFSET_MS: AtomicI64 = AtomicI64::new(0)`
  - [x] `pub fn set_clock_offset(offset_ms: i64)` — stores to atomic
  - [x] Update `now_millis()` to add `CLOCK_OFFSET_MS.load(Ordering::Relaxed)`
  - [x] In non-cluster mode, offset stays 0 — `now_millis()` behaves identically to today
- ~~Create `tests/cluster_common/mod.rs` test harness~~ — *Deferred to Phase 24.2: TestCluster requires PeerPool which doesn't exist yet. Creating todo!() stubs provides no value.*
- [x] Unit tests for config validation (all 7 rules above)
- [x] Unit tests for `ReplicationEntry` serialization round-trip
- [x] Unit tests for `Term` persistence (write to temp dir, read back)
- [x] Unit tests for `publish_with_offset` (writes at given offset and is consumable; non-cluster build returns error)
- [x] Unit tests for `commit_offset` (creates group and sets offset; updates existing group offset)
- [x] Unit tests for new frame encode/decode round-trip (`NotLeader`, `ClusterUnavailable`)
- [x] Unit test: `set_clock_offset` shifts `now_millis()` by the expected amount

**Verification:** `src/cluster/` compiles behind feature gate. Config validates all cluster rules. Types serialize/deserialize correctly. Core hooks (`validate_publish`, `apply_publish`, `publish_with_offset`, `commit_offset`, new frames, clock offset) all work. No core module imports `src/cluster/`. `cargo build` (no cluster feature) unchanged. `publish()` behavior unchanged. All existing tests pass. 809 lib tests pass with `--features cluster`, 796 without.

---

### Phase 24.2 — Peer Communication & Heartbeat

**Goal:** Nodes can establish authenticated TCP connections to each other, validate peer lists, and exchange heartbeats. Leader tracks follower states (`Joining → CatchingUp → InSync → Disconnected`). No replication data yet — just the communication layer and liveness detection.

**Subsystems touched:** `src/cluster/peer.rs` (new), `src/cluster/heartbeat.rs` (new). **Not touched:** all core modules, `src/server/`, `src/client.rs`.

- [ ] Implement `PeerPool` in `src/cluster/peer.rs`
  - [ ] Manages persistent TCP connections to all configured peers
  - [ ] Reconnects with exponential backoff on disconnect
  - [ ] Connection lifecycle: connect → handshake → authenticated → ready
- [ ] Implement node-to-node handshake protocol
  - [ ] Exchange: `node_id`, full peer list, `cluster_secret` (HMAC or challenge-response)
  - [ ] Validate peer lists match exactly (order-independent) — reject on mismatch with FATAL log
  - [ ] Mismatched node causes immediate disconnect, node exits (never opens TCP/HTTP for clients)
- [ ] Implement cluster port listener (`cluster.bind`)
  - [ ] Accept incoming peer connections
  - [ ] Same handshake validation on inbound connections
- [ ] Implement heartbeat sender (leader → followers) in `src/cluster/heartbeat.rs`
  - [ ] Sends every `heartbeat_interval_ms`
  - [ ] Heartbeat payload: leader term, head offsets per topic, corrected timestamp, replication head sequence
- [ ] Implement heartbeat receiver (follower)
  - [ ] Responds with: `last_applied_sequence`, corrected timestamp
  - [ ] Tracks missed heartbeats count
- [ ] Implement follower state machine on leader
  - [ ] Track each follower: `FollowerState`, `last_applied_sequence`, RTT
  - [ ] Transitions: `Joining → CatchingUp → InSync` (when fully caught up), `* → Disconnected` (heartbeat lost)
  - [ ] Only `InSync` followers count for semi-sync ack
- [ ] Implement clock skew detection
  - [ ] Compare corrected timestamps in heartbeat exchange
  - [ ] Log warning if skew > `clock_skew_warn_threshold_ms`
  - [ ] Emit `fafb_cluster_clock_skew_ms{peer}` metric
- [ ] Unit tests for peer list validation (match, mismatch, reordered)
- [ ] Unit tests for heartbeat serialization/deserialization
- [ ] Unit tests for follower state transitions
- [ ] Integration test: 2-node cluster formation — nodes connect, exchange heartbeats, follower reaches `InSync`
- [ ] Integration test: mismatched peer list — connection rejected

**Verification:** Nodes establish authenticated connections, validate peer lists, exchange heartbeats. Leader tracks follower liveness. Clock skew detected and logged. No replication data flows yet.

---

### Phase 24.3 — Time Synchronization

**Goal:** Implement application-level NTP sync so cluster nodes use corrected timestamps for message ordering, heartbeat timing, and failover decisions. This runs early because heartbeats (Phase 24.2) already exchange corrected timestamps — without real NTP, those timestamps are just local clock values. Graceful degradation when NTP is unreachable.

**Subsystems touched:** `src/cluster/time_sync.rs` (new). **Not touched:** all core modules (clock offset injection already wired in Phase 24.1).

- [ ] Implement NTP client in `src/cluster/time_sync.rs`
  - [ ] Query configured `time_server` (default `pool.ntp.org`)
  - [ ] Compute `offset = ntp_time - local_clock`
  - [ ] Store offset via `set_clock_offset()` from `src/types.rs`
- [ ] Implement startup sync
  - [ ] Query NTP on cluster startup before accepting traffic
  - [ ] If NTP fails: log `WARN`, set offset to 0, continue startup (never a hard gate)
  - [ ] Retry in background
- [ ] Implement periodic re-sync
  - [ ] Background task re-queries every `time_sync_interval_secs` (default 300)
  - [ ] On success: apply delta gradually over ~10 seconds (no timestamp regression)
  - [ ] On failure: keep current offset, log warning, retry next interval
- [ ] Implement skew monitoring via heartbeats
  - [ ] Heartbeats already include corrected timestamps (Phase 24.2)
  - [ ] Compare peer timestamps: if skew > `clock_skew_warn_threshold_ms`, log warning
  - [ ] Emit `fafb_cluster_clock_skew_ms{peer}` metric (already wired in Phase 24.2)
- [ ] Verify non-cluster mode unaffected
  - [ ] `CLOCK_OFFSET_MS` stays 0 — `now_millis()` identical to today
  - [ ] No NTP queries, no background tasks
- [ ] Unit tests for offset calculation (positive/negative offsets)
- [ ] Unit tests for gradual offset application (no timestamp regression)
- [ ] Integration test: NTP unreachable → warning logged, node starts with local clock
- [ ] Integration test: skew detection between nodes → warning logged

**Verification:** Cluster nodes use corrected timestamps. NTP failure never blocks startup. Skew detected and reported. Non-cluster mode has zero overhead.

---

### Phase 24.4a — Replication Buffer & Streaming

**Goal:** Leader streams publish, offset commit, and topic creation events to followers in real-time. Followers apply replicated entries to their local engine. Replication buffer supports stream replay for small catch-up gaps. No semi-sync ack yet — leader streams fire-and-forget (ack semantics added in Phase 24.4b).

**Subsystems touched:** `src/cluster/replication.rs` (new). **Not touched:** all core modules, `src/server/`, `src/client.rs`.

- [ ] Implement `ReplicationBuffer` in `src/cluster/replication.rs`
  - [ ] `VecDeque<(SequenceNum, ReplicationEntry)>` with bounded size/bytes
  - [ ] `max_entries` (default 100K), `max_bytes` (default 256MB)
  - [ ] Track `oldest_sequence`, `head_sequence`, `current_bytes`
  - [ ] Evict oldest entries when limits exceeded
- [ ] Implement `ReplicationStream` — leader side
  - [ ] Persistent TCP stream to each follower on cluster port
  - [ ] Length-prefixed framing (same pattern as client protocol)
  - [ ] Streams `ReplicationEntry` items with monotonic sequence numbers
  - [ ] Handles backpressure: if a follower is slow, buffer entries up to limit
- [ ] Implement follower apply logic
  - [ ] Receive `ReplicationEntry::Publish` → call `engine.publish_with_offset()` on local engine
  - [ ] Receive `ReplicationEntry::CommitOffset` → call `manager.commit_offset()` on local manager
  - [ ] Receive `ReplicationEntry::TopicCreated` → idempotent topic creation on local engine (skip if exists)
  - [ ] Send `ReplicationAck::Progress { last_applied_sequence }` after each batch
- [ ] Implement consumer group offset replication batching
  - [ ] Batch offset updates: every 100 acks or every 1 second (whichever first)
  - [ ] Single `CommitOffset` entry with the group's latest `next_offset`
  - [ ] Trade-off: up to 1 second of offset progress lost on failover (at-least-once)
- [ ] Implement stream replay for catch-up
  - [ ] When follower reconnects with `last_applied_sequence` within replication buffer range: stream entries from that point
  - [ ] Follower transitions: `Disconnected → CatchingUp → InSync`
- [ ] Unit tests for `ReplicationBuffer` (add, evict, range query, size tracking)
- [ ] Unit tests for `ReplicationEntry` wire format (encode/decode round-trip)
- [ ] Unit tests for offset batching (batch triggers at count and time thresholds)
- [ ] Integration test: publish on leader → verify message appears on follower via engine
- [ ] Integration test: 10K message replication consistency (leader and follower have identical data)
- [ ] Integration test: consumer group offset replicates to follower
- [ ] Integration test: topic auto-create replicates to follower

**Verification:** Messages replicate from leader to followers. Buffer evicts oldest entries correctly. Stream replay handles small gaps. Follower data matches leader. All existing tests pass.

---

### Phase 24.4b — Semi-Sync Ack & Degraded Mode

**Goal:** Leader acks the client only after at least 1 `InSync` follower confirms the write — zero data loss guarantee. When no followers are available, leader enters degraded mode and accepts writes without follower ack to preserve availability.

**Subsystems touched:** `src/cluster/replication.rs`. **Not touched:** all core modules, `src/server/`, `src/client.rs`.

- [ ] Implement semi-synchronous ack on leader
  - [ ] On publish: write locally AND stream to followers — in parallel
  - [ ] Wait for: local write complete AND at least 1 `InSync` follower ack
  - [ ] Latency = `max(local_write, fastest_follower_ack)`
  - [ ] Only then send `PublishAck` to client
- [ ] Implement degraded mode
  - [ ] If all followers are `CatchingUp` or `Disconnected`: accept writes without follower ack
  - [ ] Log warning on entering/exiting degraded mode
  - [ ] Emit `fafb_cluster_degraded_writes_total` counter metric
- [ ] Unit tests for semi-sync ack (1 follower ack triggers client ack, 0 followers → degraded)
- [ ] Unit tests for degraded mode entry/exit logic
- [ ] Integration test: semi-sync ack — publish latency includes follower ack time
- [ ] Integration test: all followers disconnected → leader enters degraded mode, still accepts writes
- [ ] Integration test: follower reconnects → degraded mode clears, semi-sync resumes

**Verification:** Semi-sync ack guarantees data on >= 2 nodes before client ack. Degraded mode preserves availability. Transition between modes is clean. All existing tests pass.

---

### Phase 24.5a — Term Persistence & Initial Leader Election

**Goal:** Implement the term counter (monotonic, persisted) and initial leader election. On fresh cluster startup, the lowest `node_id` becomes leader. On joining a running cluster, nodes discover the existing leader and join as followers.

**Subsystems touched:** `src/cluster/failover.rs` (new). **Not touched:** all core modules, `src/server/`, `src/client.rs`.

- [ ] Implement term persistence in `src/cluster/failover.rs`
  - [ ] Monotonic `u64` stored at `cluster.data_dir/term`
  - [ ] Read on startup, increment on every promotion, persist to disk before accepting writes
  - [ ] All heartbeats and replication entries include term
  - [ ] Node seeing higher term immediately self-demotes to follower
- [ ] Implement fresh cluster startup / leader election
  - [ ] Nodes connect to all configured peers
  - [ ] If no existing leader found: lowest `node_id` becomes initial leader
  - [ ] Other nodes join as followers
- [ ] Implement joining a running cluster
  - [ ] New/restarted node connects to peers, discovers existing leader via heartbeat term
  - [ ] Joins as follower, begins catch-up (stream replay or snapshot per Phase 24.7)
- [ ] Unit tests for term persistence (increment, persist, reload across restarts)
- [ ] Unit tests for election (lowest ID wins with no existing leader, higher term always wins)
- [ ] Integration test: 2-node fresh cluster → lowest ID becomes leader
- [ ] Integration test: 3-node fresh cluster → lowest ID becomes leader, others follow
- [ ] Integration test: node seeing higher term self-demotes to follower

**Verification:** Term counter persists across restarts. Fresh cluster elects leader deterministically. Nodes joining a running cluster discover leader correctly. Higher term always wins.

---

### Phase 24.5b — Failure Detection & Promotion

**Goal:** Implement leader failure detection via missed heartbeats, client-assisted promotion (fast path), no-clients auto-promote (slow path), and multi-follower promotion protocol for 3+ node clusters.

**Subsystems touched:** `src/cluster/failover.rs`. **Not touched:** all core modules, `src/server/` (session integration is Phase 24.6), `src/client.rs`.

- [ ] Implement Leader Suspect state
  - [ ] Follower triggers after `failover_timeout_ms` with no heartbeat
  - [ ] Responds `ClusterUnavailable` to new peer/client connections
  - [ ] Counts client connection attempts as corroboration
- [ ] Implement client-assisted promotion (fast path)
  - [ ] Cluster-aware client connects to follower in Leader Suspect → counts as corroboration
  - [ ] Most-caught-up follower promotes: increment term, persist, transition to Leader
  - [ ] Begin accepting writes/reads, send heartbeats with new term
- [ ] Implement no-clients auto-promote (slow path)
  - [ ] `failover_no_clients_timeout_ms` expires with no client corroboration
  - [ ] Most-caught-up follower auto-promotes
- [ ] Implement multi-follower promotion protocol (3+ nodes)
  - [ ] Broadcast `PromotionCandidate { node_id, last_applied_sequence }` to all peers
  - [ ] Collect responses for bounded window (`failover_timeout_ms / 2`)
  - [ ] Highest `last_applied_sequence` promotes (tie: lowest `node_id`)
  - [ ] Partition handling: promote only with client corroboration or wait for `no_clients_timeout`
  - [ ] Client corroboration checked AFTER position exchange (prevent less-caught-up from winning)
- [ ] Implement split-brain prevention safeguards
  - [ ] Term counter: higher term always wins (from Phase 24.5a)
  - [ ] Client corroboration reduces false promotion during network partitions
  - [ ] Peer list validation prevents misconfigured nodes (from Phase 24.2)
  - [ ] On partition heal: lower-term node demotes, truncates divergent writes
- [ ] Unit tests for failover state machine transitions (Leader Suspect → Leader, Leader Suspect → Follower on higher term)
- [ ] Unit tests for multi-follower promotion (highest sequence wins, lowest ID tiebreak, partition with/without corroboration)
- [ ] Integration test: leader crash → client-assisted promotion (2-node)
- [ ] Integration test: leader crash → no-clients auto-promote (2-node)
- [ ] Integration test: publish to new leader after failover succeeds
- [ ] Integration test: most-caught-up follower wins promotion (3-node)

**Verification:** Failover completes within `failover_timeout_ms + corroboration window`. Client-assisted path promotes faster than no-clients path. Multi-follower protocol picks the most-caught-up node. Split-brain prevented by term counter.

---

### Phase 24.5c — Graceful Transfer & Old Leader Recovery

**Goal:** Implement graceful leadership transfer for rolling upgrades (via `/admin/drain`) and old leader recovery with unreplicated tail truncation. These are the operational workflows — not failure paths.

**Subsystems touched:** `src/cluster/failover.rs`, `src/server/http/` (admin drain interaction). **Not touched:** core engine/storage/namespace modules.

- [ ] Implement graceful leadership transfer (rolling upgrades)
  - [ ] `POST /admin/drain` on leader picks most caught-up `InSync` follower
  - [ ] Leader sends `LeadershipTransfer` signal to target (no term in message)
  - [ ] Target increments own term, persists, promotes
  - [ ] Old leader receives heartbeat with higher term → self-demotes
  - [ ] Old leader sends `GoingAway` to connected clients
  - [ ] Reject drain if no eligible `InSync` follower exists
  - [ ] Drain blocks until leadership transfer completes (then proceeds with normal drain flow)
- [ ] Implement old leader recovery
  - [ ] Node starts, loads persisted term
  - [ ] Connects to peers, discovers leader with higher term → joins as follower
  - [ ] Truncate unreplicated messages (written locally but never confirmed by follower pre-crash — these were never acked to clients per semi-sync guarantee)
  - [ ] Leader streams missed entries for catch-up
  - [ ] Once caught up, follower reaches `InSync` state
- [ ] Unit tests for graceful transfer protocol (target computes new term, old leader demotes)
- [ ] Unit tests for tail truncation (identifies unreplicated messages, truncates only those)
- [ ] Integration test: graceful leadership transfer via `/admin/drain`
- [ ] Integration test: drain rejected when no `InSync` follower exists
- [ ] Integration test: old leader auto-rejoins as follower after recovery
- [ ] Integration test: unreplicated tail truncated on old leader recovery

**Verification:** Graceful transfer enables zero-downtime rolling upgrades. Old leader recovers cleanly with tail truncation. Drain blocks until transfer completes. All existing tests pass.

---

### Phase 24.6 — ClusteredManager & Server Wiring

**Goal:** Implement `ClusteredManager<M>` — the `NamespaceManager` wrapper that intercepts publish/ack/subscribe calls, handles replication on the leader, applies replicated state on followers, and sends `NotLeader` redirects. Wire into `main.rs` branching. This is the integration phase that connects all cluster internals to the server.

**Subsystems touched:** `src/cluster/manager.rs` (new), `src/server/tcp/` (session `NotLeader`/`ClusterUnavailable` handling), `src/server/http/` (follower redirect), `src/main.rs` (branching). **Not touched:** `src/storage/`, `src/auth/`, `src/retention.rs`, `src/wildcard.rs`, `src/dlq.rs`.

- [ ] Implement `ClusteredManager<M: NamespaceManager>` in `src/cluster/manager.rs`
  - [ ] Wraps `DefaultNamespaceManager` — delegates all read operations directly
  - [ ] Implements `NamespaceManager` trait
- [ ] Implement leader-side `publish()` in `ClusteredManager`
  - [ ] Call `inner.validate_publish()` — reject invalid publishes before replication
  - [ ] Create `ReplicationEntry::Publish` with assigned offset
  - [ ] Stream to followers, wait for semi-sync ack
  - [ ] Call `inner.apply_publish()` — write locally
  - [ ] Return offset to caller (session/HTTP handler)
- [ ] Implement leader-side `ack()` in `ClusteredManager`
  - [ ] Delegate `ack()` to inner manager (local state updated immediately)
  - [ ] Read group's `next_offset` after ack
  - [ ] Batch into `CommitOffset` replication entries (100 acks or 1 second)
- [ ] Implement follower-side behavior
  - [ ] `publish()` → return `NotLeader` error (session translates to `Frame::NotLeader`)
  - [ ] `subscribe()` → delegate to inner (subscriptions are node-local)
  - [ ] `fetch()` → delegate to inner (reads from local replicated data)
  - [ ] `ack()` → delegate to inner (local inflight tracking)
- [ ] Implement `NotLeader` redirect in TCP session
  - [ ] When `ClusteredManager` returns `NotLeader` error, session sends `Frame::NotLeader { leader_addr }`
  - [ ] Only sent to clients with `CAP_CLUSTER` — others get `Error { code: Unavailable }`
- [ ] Implement `ClusterUnavailable` response in TCP session
  - [ ] During Leader Suspect state: respond to new connections with `Frame::ClusterUnavailable`
  - [ ] Include `retry_after_ms` hint
- [ ] Add `CAP_CLUSTER` to `SERVER_CAPABILITIES` when cluster feature enabled
- [ ] Wire `ClusteredManager` into `main.rs`
  - [ ] `if config.cluster.enabled { ClusteredManager::new(inner, engine, cluster_config) } else { DefaultNamespaceManager }`
  - [ ] Start cluster peer connections, heartbeat, replication before TCP accept loop
  - [ ] TCP accept loop waits on cluster initialization (leader elected or follower synced)
- [ ] Implement follower HTTP behavior
  - [ ] Publish routes return 307 redirect to leader (or 503 during failover)
  - [ ] Query/fetch routes serve from local data
- [ ] Integration test: publish on leader → message available on both leader and follower
- [ ] Integration test: publish on follower → `NotLeader` redirect to leader
- [ ] Integration test: publish on follower without `CAP_CLUSTER` → `Error::Unavailable`
- [ ] Integration test: subscribe on follower → receives replicated messages
- [ ] Integration test: HTTP publish on follower → 307 redirect

**Verification:** `ClusteredManager` correctly routes operations. Leader handles writes with replication. Followers redirect or serve local data. `main.rs` branching works for both modes. All existing single-node tests pass.

---

### Phase 24.7 — Catch-Up & Snapshot Transfer

**Goal:** New or long-offline followers can fully synchronize with the leader. Small gaps use stream replay (from replication buffer, already working in Phase 24.4a). Large gaps or new nodes use snapshot transfer (segment file sync). Followers are not eligible for promotion until fully caught up.

**Subsystems touched:** `src/cluster/snapshot.rs` (new), `src/cluster/replication.rs` (catch-up strategy selection). **Not touched:** all core modules, `src/server/`, `src/client.rs`.

- [ ] Implement catch-up strategy selection
  - [ ] If follower's `last_applied_sequence` is within replication buffer range → stream replay (Phase 24.4a already handles this)
  - [ ] If gap exceeds buffer → snapshot required
- [ ] Implement snapshot manifest generation on leader
  - [ ] List all segment files per namespace/topic (filename, size, checksum)
  - [ ] Include topic metadata: head offsets, consumer group positions
  - [ ] Include current term and sequence number
  - [ ] Capture active segment up to exact byte length at snapshot point
- [ ] Implement snapshot transfer protocol
  - [ ] Leader sends manifest to follower
  - [ ] Follower diffs against local segments:
    - [ ] Already has (checksum match): skip
    - [ ] Missing: request from leader
    - [ ] Extra (orphaned from pre-crash): delete
  - [ ] Leader streams missing segment files (immutable: full file, active: up to snapshot point)
  - [ ] Streamed, not loaded into memory — handles GBs of data
- [ ] Implement follower snapshot apply
  - [ ] Write received segments to local storage
  - [ ] Rebuild engine state via `engine.init()`
  - [ ] Resume stream replay from snapshot's sequence number
  - [ ] Entries written to active segment after snapshot point delivered via normal replication
- [ ] Implement catch-up → `InSync` transition
  - [ ] Follower enters `InSync` when `last_applied_sequence == leader.head_sequence`
  - [ ] Only `InSync` followers count for semi-sync ack and promotion eligibility
- [ ] Unit tests for catch-up strategy selection (buffer hit vs snapshot needed)
- [ ] Unit tests for manifest generation (correct file list, checksums, metadata)
- [ ] Unit tests for diff logic (skip matching, request missing, delete orphaned)
- [ ] Integration test: stream replay for small gap (follower disconnects briefly, catches up via buffer)
- [ ] Integration test: snapshot transfer for large gap (follower far behind, gets full snapshot)
- [ ] Integration test: new node joins cluster — receives full snapshot, reaches `InSync`
- [ ] Integration test: follower not eligible for promotion during catch-up

**Verification:** Followers synchronize fully via stream replay or snapshot transfer. New nodes can join a running cluster. Catch-up followers are excluded from promotion. All existing tests pass.

---

### Phase 24.8 — ClusterClient & CLI

**Goal:** Implement the `ClusterClient` Rust client library with automatic leader discovery, `NotLeader` redirect handling, failover retry with backoff, and `GoingAway` reconnection. Add `--nodes` CLI flag for cluster-aware commands.

**Subsystems touched:** `src/client.rs` (new `ClusterClient`), `src/cli/` (`--nodes` flag). **Not touched:** all server-side modules.

- [ ] Implement `ClusterClientConfig` in `src/client.rs`
  - [ ] `nodes: Vec<String>` — all cluster node addresses
  - [ ] `max_retries: u32` (default 10)
  - [ ] `retry_backoff_base_ms: u64` (default 100)
  - [ ] `retry_backoff_max_ms: u64` (default 5000)
  - [ ] Plus existing fields: `namespace`, `api_key`, `timeout`, `max_frame_size`, `tls`
- [ ] Implement `ClusterClient` struct
  - [ ] Wraps `FafbClient` for current connection
  - [ ] `cached_leader: Option<String>` — last known leader address
  - [ ] `connect()` — try nodes in order (cached leader first)
- [ ] Implement connection flow
  - [ ] Send `Connect { capabilities: CAP_CLUSTER | ... }`
  - [ ] Handle `ConnectAck { success: true }` → leader found
  - [ ] Handle `NotLeader { leader_addr }` → cache and connect to leader
  - [ ] Handle `ClusterUnavailable { retry_after_ms }` → wait, try next node
  - [ ] Handle connection refused / timeout → try next node
- [ ] Implement automatic reconnection
  - [ ] On connection drop: clear cached leader, retry from node list
  - [ ] Exponential backoff with jitter, capped at `retry_backoff_max_ms`
  - [ ] Max retries configurable
- [ ] Implement `GoingAway` handling
  - [ ] Reconnect — try cached leader first
  - [ ] If cached leader also draining (rolling upgrade): fall through to node list
- [ ] Delegate operations (`publish`, `subscribe`, `fetch`, `ack`) to inner `FafbClient` with reconnect logic
  - [ ] On `NotLeader` during operation: transparent reconnect and retry
  - [ ] Caller never sees `NotLeader`
- [ ] Existing `FafbClient` unchanged for single-node use
- [ ] Add `--nodes` flag to CLI commands (`pub`, `sub`, `fetch`, `info`, `drain`, `health`, `bench`)
  - [ ] Accepts comma-separated addresses
  - [ ] `--server` still works for single-node (backwards compatible)
  - [ ] Error if both `--nodes` and `--server` provided
- [ ] Add `nodes` to CLI profile config
  - [ ] `[profiles.production] nodes = ["addr1", "addr2"]`
  - [ ] Profile `nodes` used when `--nodes` not specified and `--server` not specified
- [ ] Unit tests for connection flow state machine (all response types)
- [ ] Unit tests for reconnect backoff (exponential, jitter, cap)
- [ ] Integration test: `ClusterClient` connects, gets `NotLeader`, follows redirect
- [ ] Integration test: leader crash → `ClusterClient` auto-reconnects to new leader
- [ ] Integration test: `ClusterUnavailable` during failover → client waits and retries
- [ ] Integration test: `GoingAway` during rolling upgrade → client reconnects
- [ ] Integration test: legacy client (no `CAP_CLUSTER`) → `Error::Unavailable` from follower
- [ ] Integration test: rolling upgrade — zero publish failures across leadership transfer

**Verification:** `ClusterClient` transparently handles leader discovery, redirects, and failover. CLI cluster-aware with `--nodes`. Existing single-node `FafbClient` and `--server` unchanged. All existing tests pass.

---

### Phase 24.9 — Health, Admin, Metrics & Observability

**Goal:** Expose cluster state via health endpoints, admin endpoints for operational control, comprehensive metrics for monitoring, and structured logging. All gated behind cluster feature.

**Subsystems touched:** `src/server/http/` (new routes), `src/cluster/` (metrics emission points throughout). **Not touched:** core engine/storage/namespace modules.

- [ ] Implement `GET /health/cluster` endpoint
  - [ ] 200 when healthy (leader serving or follower `InSync`)
  - [ ] 503 during failover or catch-up
  - [ ] JSON body: `node_id`, `role`, `term`, `replication_sequence`, `time_sync` status, `peers` array with state/lag/RTT/skew
- [ ] Implement admin endpoints
  - [ ] `GET /admin/cluster/status` — full cluster status (same as health but always 200, for dashboards)
  - [ ] `POST /admin/cluster/transfer-leader` — `{ "target_node_id": N }` — initiate graceful transfer
  - [ ] `POST /admin/cluster/promote` — force-promote this follower (emergency use)
  - [ ] All gated by `Authorization: Bearer <admin_api_key>`
- [ ] Register all cluster metrics
  - [ ] `fafb_cluster_role` (gauge): 1 = leader, 0 = follower
  - [ ] `fafb_cluster_term` (gauge): current term
  - [ ] `fafb_cluster_replication_sequence` (gauge): head sequence number
  - [ ] `fafb_cluster_peer_state{peer}` (gauge): 0=disconnected, 1=joining, 2=catching_up, 3=in_sync
  - [ ] `fafb_cluster_peer_lag{peer}` (gauge): entries behind leader
  - [ ] `fafb_cluster_peer_rtt_ms{peer}` (histogram): heartbeat RTT per peer
  - [ ] `fafb_cluster_replication_ack_ms` (histogram): leader write to follower ack latency
  - [ ] `fafb_cluster_failover_total` (counter): leader promotions
  - [ ] `fafb_cluster_snapshot_total` (counter): snapshot transfers
  - [ ] `fafb_cluster_redirect_total` (counter): NotLeader redirects sent
  - [ ] `fafb_cluster_unavailable_total` (counter): ClusterUnavailable responses
  - [ ] `fafb_cluster_degraded_writes_total` (counter): writes without follower ack
- [ ] Implement structured logging with `node_id` in all cluster log lines
  - [ ] `INFO node_id=1 role=leader peer connected node_id=2`
  - [ ] `WARN node_id=2 role=follower heartbeat missed count=3`
  - [ ] `INFO node_id=2 role=follower promoted to leader term=43 reason="client corroboration"`
- [ ] Integration test: `/health/cluster` returns correct state for leader and follower
- [ ] Integration test: `/admin/cluster/transfer-leader` triggers graceful transfer
- [ ] Integration test: `/admin/cluster/promote` force-promotes a follower
- [ ] Integration test: metrics emitted during publish replication and failover

**Verification:** Operators can monitor cluster health via endpoints and metrics. Admin operations work. Structured logging enables debugging. All endpoints gated by auth.

---

### Phase 24.9a — API Key Replication

**Goal:** Replicate runtime API key mutations (create, revoke, update) from the leader to followers via the existing replication stream. After this phase, a key created on the leader is immediately usable on any follower after failover, and followers' `_system/keys.toml` files stay in sync.

**Prerequisites:**
- Phase 24.4 complete (replication stream with `ReplicationEntry`)
- Plan 4 Phase 4.1 complete (`ApiKeyStore` with `ManagedKey`, mutation methods, `keys.toml`)
- Plan 25 Phase I complete (admin key management endpoints)

**Design:**

Key management via admin API is **leader-only**. If a client sends a key management request to a follower's admin API, the follower returns a redirect (same pattern as publish — `307 Temporary Redirect` to the leader's HTTP address). This is consistent with the existing "leader handles all writes" principle.

Three new `ReplicationEntry` variants:

```rust
enum ReplicationEntry {
    // ... existing variants ...
    KeyCreated {
        name: String,
        namespace: NamespaceId,
        api_key: ApiKey,
        permissions: PermissionSet,
        created_at: u64,
        expires_at: Option<u64>,
    },
    KeyRevoked {
        namespace: NamespaceId,
        name: String,
    },
    KeyUpdated {
        namespace: NamespaceId,
        name: String,
        permissions: PermissionSet,
    },
}
```

**Semi-sync guarantee for key mutations:**

Key mutations use the same semi-sync replication guarantee as publishes — the leader waits for at least one follower to ack the `ReplicationEntry` before returning success to the admin API client. Key creation/revocation is low-frequency and high-importance, so the added latency (one follower RTT) is acceptable and prevents the scenario where a leader crashes right after key creation but before replication, causing the admin to see success while the key is lost.

If no followers are available (degraded mode), key mutations still succeed locally (same as degraded publishes) with a warning logged.

**Leader flow:**
1. Admin API creates/revokes/updates a key via `ApiKeyStore` (existing path)
2. After successful local mutation, leader creates the corresponding `ReplicationEntry` variant
3. Entry is appended to the replication buffer and streamed to followers
4. Leader waits for at least one follower ack (semi-sync) before returning success to admin API

**Follower flow:**
1. Receive `ReplicationEntry::KeyCreated` → call `ApiKeyStore::apply_replicated_key_create()` on local store, persist to local `_system/keys.toml`
2. Receive `ReplicationEntry::KeyRevoked` → call `ApiKeyStore::apply_replicated_key_revoke()`, persist
3. Receive `ReplicationEntry::KeyUpdated` → call `ApiKeyStore::apply_replicated_key_update()`, persist

**Startup after failover:**
- The new leader's `_system/keys.toml` is already up to date (replicated before the old leader went down)
- If a key mutation was in-flight during failover (leader wrote locally but entry not yet acked by follower), the new leader's store is missing that key — same as unreplicated publishes, which are already handled by tail truncation. The admin must re-create the key.

**Apply methods on ApiKeyStore:**

```rust
impl ApiKeyStore {
    /// Apply a replicated key creation from the leader. Idempotent.
    pub fn apply_replicated_key_create(&self, key: ManagedKey, data_dir: &Path) -> Result<()>;

    /// Apply a replicated key revocation from the leader. Idempotent.
    pub fn apply_replicated_key_revoke(&self, namespace: &NamespaceId, name: &str, data_dir: &Path) -> Result<()>;

    /// Apply a replicated permission update from the leader. Idempotent.
    pub fn apply_replicated_key_update(&self, namespace: &NamespaceId, name: &str, permissions: PermissionSet, data_dir: &Path) -> Result<()>;
}
```

All apply methods are idempotent — applying a create for an already-existing key is a no-op (handles replay during catch-up).

**Subsystems touched:** `src/cluster/types.rs` (new `ReplicationEntry` variants), `src/cluster/replication.rs` (encode/decode new variants), `src/cluster/manager.rs` (intercept key mutations on leader, apply on follower), `src/auth/apikey.rs` (apply methods), `src/server/http/admin.rs` (follower redirect for key endpoints). **Not touched:** `src/storage/`, `src/namespace/`, `src/queue/`, `src/codec/`.

**Tasks:**

- [ ] Add `ReplicationEntry::KeyCreated`, `KeyRevoked`, `KeyUpdated` variants to `src/cluster/types.rs`
- [ ] Implement encode/decode for new `ReplicationEntry` variants in `src/cluster/replication.rs`
- [ ] Implement `ApiKeyStore::apply_replicated_key_create()` — idempotent create + persist
- [ ] Implement `ApiKeyStore::apply_replicated_key_revoke()` — idempotent revoke + persist
- [ ] Implement `ApiKeyStore::apply_replicated_key_update()` — idempotent update + persist
- [ ] Update `ClusteredManager` (or admin HTTP handler) to emit `ReplicationEntry::KeyCreated` after successful `create_key()`
- [ ] Update `ClusteredManager` to emit `ReplicationEntry::KeyRevoked` after successful `revoke_key()`
- [ ] Update `ClusteredManager` to emit `ReplicationEntry::KeyUpdated` after successful `update_key_permissions()`
- [ ] Update follower replication apply loop to handle new `ReplicationEntry` variants
- [ ] Add follower redirect (307) for key management admin endpoints when node is not leader
- [ ] Unit test: `ReplicationEntry::KeyCreated` round-trip encode/decode
- [ ] Unit test: `ReplicationEntry::KeyRevoked` round-trip encode/decode
- [ ] Unit test: `ReplicationEntry::KeyUpdated` round-trip encode/decode
- [ ] Unit test: `apply_replicated_key_create` idempotent (applying twice is a no-op)
- [ ] Unit test: `apply_replicated_key_revoke` idempotent (revoking nonexistent key is a no-op)
- [ ] Integration test: create key on leader → key usable on follower after replication
- [ ] Integration test: revoke key on leader → key rejected on follower after replication
- [ ] Integration test: update key permissions on leader → follower reflects new permissions
- [ ] Integration test: key management on follower returns 307 redirect to leader
- [ ] Integration test: leader failover → new leader's key store has all replicated keys
- [ ] Integration test: catch-up replay includes key mutation entries
- [ ] Integration test: key creation waits for follower ack before returning success (semi-sync)
- [ ] Integration test: key creation in degraded mode (no followers) succeeds with warning

**Verification:** Runtime API keys replicate from leader to followers. Key mutations are leader-only with follower redirects. Follower `_system/keys.toml` stays in sync. Idempotent apply handles replay. All existing cluster tests pass unchanged.

---

### Phase 24.10 — Documentation, Benchmarks & Single-Node Guarantee

**Goal:** Document the clustering feature end-to-end. Verify single-node mode has zero overhead. Run benchmarks to measure replication latency. Update all project docs.

- [ ] Update `docs/protocol.md` with cluster frame types
  - [ ] `NotLeader` frame format and semantics
  - [ ] `ClusterUnavailable` frame format and semantics
  - [ ] `CAP_CLUSTER` capability bit
  - [ ] Client connection flow with cluster awareness
- [ ] Update `docs/api.md` with cluster endpoints
  - [ ] `GET /health/cluster`
  - [ ] `GET /admin/cluster/status`
  - [ ] `POST /admin/cluster/transfer-leader`
  - [ ] `POST /admin/cluster/promote`
  - [ ] Follower HTTP behavior (307 redirect on publish, local reads)
- [ ] Update `docs/client_sdk_spec.md`
  - [ ] `ClusterClientConfig` and `ClusterClient` API
  - [ ] Connection lifecycle state machine (connecting → authenticating → ready → reconnecting)
  - [ ] Retry/backoff behavior with exact defaults
  - [ ] Interaction between `NotLeader` redirect and `GoingAway` during rolling upgrades
  - [ ] Example flows: normal connect, redirect, failover, rolling upgrade
- [ ] Create `docs/clustering.md` — operator guide
  - [ ] Deployment topologies (2-node, 3-node)
  - [ ] Configuration reference
  - [ ] Failover behavior and timing
  - [ ] Rolling upgrade procedure
  - [ ] Monitoring and alerting recommendations
  - [ ] Known limitations (2-node split-brain window)
  - [ ] API key replication behavior (leader-only mutations, follower redirects, `_system/keys.toml` sync, semi-sync guarantee)
- [ ] Update `config/config.example.toml` with documented `[cluster]` section
- [ ] Update `README.md` with clustering overview
- [ ] Update `CLAUDE.md` with clustering module map and gotchas
- [ ] Run single-node benchmark: publish throughput within 1% of baseline
  - [ ] `cargo build --release` (no cluster feature) — zero overhead
  - [ ] `cargo build --release --features cluster` with `cluster.enabled = false` — zero overhead
- [ ] Run 2-node benchmark: measure semi-sync replication overhead
  - [ ] Publish latency (p50, p95, p99) with replication vs single-node
  - [ ] Throughput impact
- [ ] Run failover timing benchmark
  - [ ] Time from leader crash to new leader accepting writes
  - [ ] With client corroboration vs no-clients auto-promote

**Verification:** All documentation reflects clustering behavior. Single-node mode benchmarks at <= 1% overhead. Benchmarks establish baseline replication latency.
