# Cluster Mode Options

## Overview

This document explores options for making rstmdb distributed and highly available.

## Requirements Analysis

| Requirement | Priority | Notes |
|------------|----------|-------|
| High Availability | High | Survive node failures |
| Strong Consistency | High | State machines need deterministic ordering |
| Read Scalability | Medium | Multiple replicas can serve reads |
| Write Scalability | Medium | Sharding for high write throughput |
| Operational Simplicity | High | Easy to deploy and manage |

## Option 1: Single-Leader with WAL Streaming

```
┌─────────────────┐     WAL Stream     ┌─────────────────┐
│     Primary     │ ─────────────────► │    Replica 1    │
│   (read/write)  │                    │   (read-only)   │
└─────────────────┘                    └─────────────────┘
        │
        │              WAL Stream     ┌─────────────────┐
        └────────────────────────────►│    Replica 2    │
                                      │   (read-only)   │
                                      └─────────────────┘
```

### How It Works
- Primary node handles all writes
- WAL entries streamed to replicas in real-time
- Replicas apply WAL entries to stay in sync
- Clients can read from replicas (eventual consistency) or primary (strong)
- External failover (manual or via orchestrator like Patroni)

### Implementation

```rust
// Primary: Stream WAL to replicas
pub struct ReplicationManager {
    replicas: Vec<ReplicaConnection>,
    wal: Arc<Wal>,
}

impl ReplicationManager {
    async fn stream_to_replicas(&self) {
        let mut last_offset = 0;
        loop {
            let entries = self.wal.read_from(last_offset, Some(100));
            for replica in &self.replicas {
                replica.send_entries(&entries).await;
            }
            last_offset = entries.last().map(|e| e.offset).unwrap_or(last_offset);
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    }
}

// Replica: Apply streamed WAL
pub struct ReplicaEngine {
    engine: StateMachineEngine,
    primary_addr: SocketAddr,
}

impl ReplicaEngine {
    async fn follow_primary(&mut self) {
        let stream = connect_to_primary(self.primary_addr).await;
        while let Some(entries) = stream.recv().await {
            for entry in entries {
                self.engine.apply_wal_entry(entry);
            }
        }
    }
}
```

### Pros
- Simple to implement and understand
- Low latency writes (no consensus overhead)
- Easy to reason about consistency
- Can use existing WAL infrastructure

### Cons
- Single point of failure for writes
- Manual/external failover required
- Replicas may lag behind primary
- No automatic leader election

### Use Cases
- Small to medium deployments
- When operational simplicity is key
- When you have existing failover infrastructure (K8s, Patroni)

---

## Option 2: Raft Consensus

```
                    ┌─────────────────┐
                    │     Leader      │
                    │  (read/write)   │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
              ▼              ▼              ▼
     ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
     │  Follower   │ │  Follower   │ │  Follower   │
     │  (read*)    │ │  (read*)    │ │  (read*)    │
     └─────────────┘ └─────────────┘ └─────────────┘

     * Reads from followers may be stale unless using read index
```

### How It Works
- Nodes elect a leader via Raft protocol
- All writes go through leader
- Leader replicates to followers before acknowledging
- Automatic leader election on failure
- Reads can go to leader (strong) or followers (stale)

### Implementation

```rust
// Using existing Raft library (e.g., openraft, tikv/raft-rs)
use openraft::{Config, Raft, RaftStorage};

pub struct RstmdbRaft {
    raft: Raft<TypeConfig>,
    engine: Arc<StateMachineEngine>,
}

// Raft log entry = WAL entry
#[derive(Serialize, Deserialize)]
pub enum RaftCommand {
    PutMachine { name: String, version: u32, def: Value },
    CreateInstance { id: String, machine: String, version: u32, ctx: Value },
    ApplyEvent { instance_id: String, event: String, payload: Value },
    DeleteInstance { id: String },
}

// State machine applies Raft commands
impl RaftStateMachine for RstmdbStateMachine {
    async fn apply(&mut self, entry: Entry<TypeConfig>) -> Result<Response> {
        match entry.payload {
            EntryPayload::Normal(cmd) => {
                match cmd {
                    RaftCommand::ApplyEvent { instance_id, event, payload } => {
                        self.engine.apply_event(&instance_id, &event, payload, ...)
                    }
                    // ... other commands
                }
            }
            _ => Ok(Response::Empty),
        }
    }
}

// Client request handling
impl RstmdbRaft {
    pub async fn apply_event(&self, params: ApplyEventParams) -> Result<ApplyEventResult> {
        let cmd = RaftCommand::ApplyEvent {
            instance_id: params.instance_id,
            event: params.event,
            payload: params.payload,
        };

        // Propose to Raft - will be replicated before returning
        let response = self.raft.client_write(cmd).await?;
        Ok(response)
    }
}
```

### Configuration

```yaml
cluster:
  mode: raft
  node_id: 1
  peers:
    - id: 1
      addr: "node1:7401"
    - id: 2
      addr: "node2:7401"
    - id: 3
      addr: "node3:7401"

  raft:
    election_timeout_ms: 1000
    heartbeat_interval_ms: 100
    snapshot_threshold: 10000
```

### Pros
- Automatic leader election and failover
- Strong consistency (linearizable writes)
- Well-understood protocol
- Good library support (openraft, raft-rs)

### Cons
- Higher write latency (quorum required)
- Odd number of nodes required (3, 5, 7)
- All data on all nodes (no sharding)
- More complex than single-leader

### Use Cases
- Mission-critical deployments
- When automatic failover is required
- Small to medium data sizes (fits on one node)

---

## Option 3: Raft per Shard (TiKV-style)

```
┌─────────────────────────────────────────────────────────────┐
│                      Placement Driver                        │
│            (metadata, shard assignment, balancing)           │
└─────────────────────────────────────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
        ▼                     ▼                     ▼
┌───────────────┐     ┌───────────────┐     ┌───────────────┐
│   Shard 1     │     │   Shard 2     │     │   Shard 3     │
│ (instances    │     │ (instances    │     │ (instances    │
│  a-f, Raft)   │     │  g-m, Raft)   │     │  n-z, Raft)   │
└───────────────┘     └───────────────┘     └───────────────┘
      │                     │                     │
   3 replicas            3 replicas            3 replicas
```

### How It Works
- Data partitioned into shards by instance_id hash/range
- Each shard is a Raft group with 3+ replicas
- Placement Driver manages shard assignment and rebalancing
- Clients route requests to correct shard
- Can scale horizontally by adding shards

### Implementation

```rust
// Shard assignment
pub struct ShardRouter {
    pd_client: PlacementDriverClient,
    shard_cache: DashMap<ShardId, Vec<NodeAddr>>,
}

impl ShardRouter {
    fn get_shard(&self, instance_id: &str) -> ShardId {
        // Consistent hashing or range-based
        let hash = xxhash(instance_id);
        ShardId(hash % self.num_shards)
    }

    async fn route_request(&self, instance_id: &str, request: Request) -> Response {
        let shard_id = self.get_shard(instance_id);
        let nodes = self.get_shard_nodes(shard_id).await;

        // Try leader first, then followers
        for node in nodes {
            match node.send(request.clone()).await {
                Ok(resp) => return resp,
                Err(e) if e.is_not_leader() => continue,
                Err(e) => return Err(e),
            }
        }
    }
}

// Placement Driver
pub struct PlacementDriver {
    shards: DashMap<ShardId, ShardInfo>,
    nodes: DashMap<NodeId, NodeInfo>,
}

impl PlacementDriver {
    // Balance shards across nodes
    async fn rebalance(&self) {
        // Move shards from overloaded to underloaded nodes
    }

    // Handle node failure
    async fn handle_node_down(&self, node_id: NodeId) {
        // Re-replicate shards that lost a replica
    }
}
```

### Pros
- Horizontal write scalability
- Fault tolerance at shard level
- Can handle very large datasets
- Automatic rebalancing

### Cons
- Significantly more complex
- Cross-shard operations are hard
- Requires Placement Driver (another component)
- Higher operational overhead

### Use Cases
- Large-scale deployments
- High write throughput requirements
- Data too large for single node

---

## Option 4: Leaderless (Dynamo-style)

```
        ┌─────────────────┐
        │     Client      │
        └────────┬────────┘
                 │ W=2, R=2
    ┌────────────┼────────────┐
    │            │            │
    ▼            ▼            ▼
┌───────┐   ┌───────┐   ┌───────┐
│ Node1 │   │ Node2 │   │ Node3 │
└───────┘   └───────┘   └───────┘
```

### How It Works
- No leader, any node can accept writes
- Writes sent to W nodes, reads from R nodes
- If W + R > N, guaranteed to read latest write
- Conflict resolution via vector clocks or LWW

### Why NOT Recommended for rstmdb

State machines require **deterministic event ordering**:
- Event A then B ≠ Event B then A
- Concurrent writes to same instance cause conflicts
- Conflict resolution is non-trivial for state machines

```
Node 1: instance-1 in state "created", receives PAY → "paid"
Node 2: instance-1 in state "created", receives CANCEL → "cancelled"

Conflict! Which state wins? Can't merge state machine states.
```

### Verdict
**Not recommended** for rstmdb due to state machine semantics.

---

## Option 5: Hybrid - Single Leader + Read Replicas + Sharding

```
                    ┌─────────────────┐
                    │   Coordinator   │
                    │ (routes by key) │
                    └────────┬────────┘
                             │
         ┌───────────────────┼───────────────────┐
         │                   │                   │
         ▼                   ▼                   ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│    Shard 1      │ │    Shard 2      │ │    Shard 3      │
│   Primary +     │ │   Primary +     │ │   Primary +     │
│   2 Replicas    │ │   2 Replicas    │ │   2 Replicas    │
└─────────────────┘ └─────────────────┘ └─────────────────┘
```

### How It Works
- Data sharded by instance_id
- Each shard has one primary + replicas
- WAL streaming within each shard
- External failover per shard (or simple Raft per shard)

### Pros
- Simpler than full Raft-per-shard
- Scales writes via sharding
- Familiar single-leader model per shard

### Cons
- External failover management
- Coordinator is a bottleneck (can be stateless though)

---

## Comparison Matrix

| Option | Consistency | Write Scale | Read Scale | Complexity | Auto-Failover |
|--------|-------------|-------------|------------|------------|---------------|
| Single-Leader + WAL | Strong | 1 node | N replicas | Low | No |
| Raft Consensus | Strong | 1 leader | N followers | Medium | Yes |
| Raft per Shard | Strong | N shards | N×3 nodes | High | Yes |
| Leaderless | Eventual | N nodes | N nodes | Medium | N/A |
| Hybrid Sharding | Strong | N shards | N×replicas | Medium | Optional |

## Recommendation

### Phase 1: Start Simple
**Single-Leader with WAL Streaming**
- Implement WAL streaming to replicas
- Use Kubernetes or external tool for failover
- Good for most use cases

### Phase 2: Add Consensus
**Raft Consensus (single group)**
- Integrate openraft or raft-rs
- Automatic leader election
- 3-5 node clusters

### Phase 3: Scale Out (if needed)
**Raft per Shard**
- Only if single-node can't handle load
- Significant engineering investment

---

## Implementation Roadmap

### Phase 1: WAL Streaming (2-3 weeks)

```
Week 1:
- [ ] Replication protocol design
- [ ] WAL streaming from primary
- [ ] Replica connection handling

Week 2:
- [ ] Replica WAL application
- [ ] Catch-up from snapshot
- [ ] Health checks and monitoring

Week 3:
- [ ] Read routing (primary vs replica)
- [ ] Lag monitoring
- [ ] Documentation and testing
```

### Phase 2: Raft Consensus (4-6 weeks)

```
Week 1-2:
- [ ] Integrate openraft
- [ ] Raft storage backend using existing WAL
- [ ] Command serialization

Week 3-4:
- [ ] Leader election and failover
- [ ] Cluster membership changes
- [ ] Snapshot and log compaction

Week 5-6:
- [ ] Client request routing
- [ ] Read index for consistent reads
- [ ] Testing and hardening
```

### Configuration Example

```yaml
# config.yaml for cluster mode
cluster:
  enabled: true
  mode: raft  # or "streaming"

  # Raft mode settings
  raft:
    node_id: 1
    peers:
      - { id: 1, addr: "node1:7401" }
      - { id: 2, addr: "node2:7401" }
      - { id: 3, addr: "node3:7401" }
    election_timeout_ms: 1000
    heartbeat_interval_ms: 100

  # Streaming mode settings
  streaming:
    role: primary  # or "replica"
    primary_addr: "primary:7401"  # for replicas
    replicas:  # for primary
      - "replica1:7401"
      - "replica2:7401"
```

## Questions to Decide

1. **What's the target scale?**
   - < 1M instances → Single Raft group is fine
   - > 1M instances → Consider sharding

2. **What's the consistency requirement?**
   - Strong → Raft or single-leader with sync replication
   - Eventual → WAL streaming with async replication

3. **Who manages failover?**
   - Automatic → Raft
   - External → WAL streaming + orchestrator

4. **What's the latency budget?**
   - < 10ms → Single-leader, local writes
   - < 100ms → Raft with 3 nodes in same region
   - Cross-region → Multi-region Raft (higher latency)
