# 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.
