distributed-systems-consensus-raft

RAFT consensus algorithm including leader election, log replication, safety guarantees, and implementation patterns

RAFT Consensus

Scope: RAFT algorithm, leader election, log replication, safety, implementation patterns Lines: ~380 Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)

When to Use This Skill

Activate this skill when:

Core Concepts

RAFT Overview

Goal: Maintain replicated log across cluster with strong consistency

Key properties:

Node states:

Follower → Candidate → Leader
   ↑                      ↓
   └──────────────────────┘

Terms (Logical Clock)

Term: Monotonically increasing number representing election cycle

Term 1:        Term 2:        Term 3:
Leader A ──X   Election  →    Leader B ────→

Each term has at most one leader
Terms prevent conflicts from split-brain

Leader Election

Election Process

1. Follower times out waiting for leader heartbeat
2. Increment term, become Candidate
3. Vote for self, request votes from all nodes
4. If majority votes received → become Leader
5. If another leader discovered → become Follower
6. If election timeout → start new election

Election Rules

class RaftNode:
    def __init__(self):
        self.state = "follower"
        self.current_term = 0
        self.voted_for = None
        self.election_timeout = random.randint(150, 300)  # ms

    def on_election_timeout(self):
        """Start election"""
        self.current_term += 1
        self.state = "candidate"
        self.voted_for = self.node_id
        self.votes_received = {self.node_id}

        # Request votes from all nodes
        for node in self.cluster_nodes:
            self.send_request_vote(node, self.current_term, self.node_id,
                                  self.last_log_index, self.last_log_term)

    def handle_request_vote(self, term, candidate_id, last_log_index, last_log_term):
        """Handle vote request"""
        # Reject if term is old
        if term < self.current_term:
            return False

        # Update term if newer
        if term > self.current_term:
            self.current_term = term
            self.voted_for = None
            self.state = "follower"

        # Grant vote if:
        # 1. Haven't voted this term
        # 2. Candidate's log is at least as up-to-date
        if self.voted_for is None or self.voted_for == candidate_id:
            if self.is_log_up_to_date(last_log_index, last_log_term):
                self.voted_for = candidate_id
                return True

        return False

    def is_log_up_to_date(self, candidate_last_log_index, candidate_last_log_term):
        """Check if candidate's log is at least as up-to-date"""
        my_last_log_term = self.log[-1].term if self.log else 0
        my_last_log_index = len(self.log) - 1

        # Candidate's last log term is more recent
        if candidate_last_log_term > my_last_log_term:
            return True

        # Same term, candidate's log is at least as long
        if candidate_last_log_term == my_last_log_term:
            return candidate_last_log_index >= my_last_log_index

        return False

Log Replication

Replication Process

Client → Leader: Command
Leader:
  1. Append to local log
  2. Send AppendEntries to all followers
  3. Wait for majority acknowledgment
  4. Apply to state machine
  5. Respond to client
  6. Eventually notify followers to apply

AppendEntries RPC

class LogEntry:
    def __init__(self, term, command):
        self.term = term
        self.command = command

class RaftNode:
    def __init__(self):
        self.log = []  # List of LogEntry
        self.commit_index = 0  # Index of highest committed entry
        self.last_applied = 0  # Index of highest applied entry

        # Leader state (reinitialized after election)
        self.next_index = {}   # For each server, index of next log entry to send
        self.match_index = {}  # For each server, index of highest log entry known to be replicated

    def append_entries(self, term, leader_id, prev_log_index, prev_log_term,
                      entries, leader_commit):
        """Handle AppendEntries RPC (heartbeat or log replication)"""

        # Reply false if term < current_term
        if term < self.current_term:
            return False

        # Update term and revert to follower if necessary
        if term > self.current_term:
            self.current_term = term
            self.state = "follower"
            self.voted_for = None

        # Reset election timeout (received heartbeat from valid leader)
        self.reset_election_timeout()

        # Reply false if log doesn't contain entry at prev_log_index matching prev_log_term
        if prev_log_index >= 0:
            if prev_log_index >= len(self.log) or \
               self.log[prev_log_index].term != prev_log_term:
                return False

        # Delete conflicting entries and append new ones
        log_index = prev_log_index + 1
        for i, entry in enumerate(entries):
            if log_index + i < len(self.log):
                if self.log[log_index + i].term != entry.term:
                    # Delete this and all following entries
                    self.log = self.log[:log_index + i]
                    self.log.append(entry)
            else:
                self.log.append(entry)

        # Update commit index
        if leader_commit > self.commit_index:
            self.commit_index = min(leader_commit, len(self.log) - 1)

        # Apply newly committed entries
        self.apply_committed_entries()

        return True

    def apply_committed_entries(self):
        """Apply committed log entries to state machine"""
        while self.last_applied < self.commit_index:
            self.last_applied += 1
            entry = self.log[self.last_applied]
            self.state_machine.apply(entry.command)

Leader Logic

class RaftLeader(RaftNode):
    def replicate_log(self):
        """Send AppendEntries to all followers"""
        for follower_id in self.cluster_nodes:
            if follower_id == self.node_id:
                continue

            # Get entries to send to this follower
            next_index = self.next_index[follower_id]
            prev_log_index = next_index - 1
            prev_log_term = self.log[prev_log_index].term if prev_log_index >= 0 else 0
            entries = self.log[next_index:]

            success = self.send_append_entries(
                follower_id,
                self.current_term,
                self.node_id,
                prev_log_index,
                prev_log_term,
                entries,
                self.commit_index
            )

            if success:
                # Update next_index and match_index
                self.next_index[follower_id] = next_index + len(entries)
                self.match_index[follower_id] = self.next_index[follower_id] - 1

                # Update commit_index if majority replicated
                self.update_commit_index()
            else:
                # Decrement next_index and retry
                self.next_index[follower_id] = max(0, self.next_index[follower_id] - 1)

    def update_commit_index(self):
        """Advance commit_index if majority of servers have replicated entry"""
        for n in range(self.commit_index + 1, len(self.log)):
            # Only commit entries from current term
            if self.log[n].term != self.current_term:
                continue

            # Count replicas
            replicas = 1  # Leader has it
            for follower_id in self.cluster_nodes:
                if follower_id == self.node_id:
                    continue
                if self.match_index[follower_id] >= n:
                    replicas += 1

            # Commit if majority
            if replicas > len(self.cluster_nodes) / 2:
                self.commit_index = n
                self.apply_committed_entries()

Safety Guarantees

Election Safety

Guarantee: At most one leader per term

How: Node votes for at most one candidate per term, majority quorum required

Leader Append-Only

Guarantee: Leader never overwrites or deletes entries in its log

How: Leader only appends new entries

Log Matching

Guarantee: If two logs contain entry with same index and term, all preceding entries are identical

How: AppendEntries consistency check (prev_log_index, prev_log_term)

Leader Completeness

Guarantee: If entry is committed in term T, it will be present in all leader logs for terms > T

How: Candidate with incomplete log won't get majority votes

State Machine Safety

Guarantee: If server has applied log entry at index, no other server will apply different entry at same index

How: Follows from other properties

Real-World Implementation (etcd)

Using etcd

# Start etcd cluster (3 nodes)
etcd --name node1 --initial-cluster node1=http://10.0.1.1:2380,node2=http://10.0.1.2:2380,node3=http://10.0.1.3:2380

Python client:

import etcd3

etcd = etcd3.client(host='localhost', port=2379)

# Write (goes through RAFT consensus)
etcd.put('/config/setting', 'value')

# Read (consistent read from leader)
value, metadata = etcd.get('/config/setting')

# Watch for changes
watch_id = etcd.add_watch_callback('/config/', lambda event: print(event))

# Lease (TTL)
lease = etcd.lease(ttl=60)
etcd.put('/ephemeral/key', 'value', lease=lease)

# Transactions (atomic operations)
etcd.transaction(
    compare=[etcd.transactions.value('/counter') == b'5'],
    success=[etcd.transactions.put('/counter', '6')],
    failure=[etcd.transactions.get('/counter')]
)

Go client (native etcd client):

package main

import (
    "context"
    "time"
    clientv3 "go.etcd.io/etcd/client/v3"
)

func main() {
    cli, _ := clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    })
    defer cli.Close()

    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()

    // Put
    cli.Put(ctx, "/config/setting", "value")

    // Get
    resp, _ := cli.Get(ctx, "/config/setting")
    for _, kv := range resp.Kvs {
        fmt.Printf("%s: %s\n", kv.Key, kv.Value)
    }

    // Watch
    watchChan := cli.Watch(context.Background(), "/config/", clientv3.WithPrefix())
    for watchResp := range watchChan {
        for _, event := range watchResp.Events {
            fmt.Printf("Event: %s %s\n", event.Type, event.Kv.Key)
        }
    }
}

Performance Considerations

Cluster Size

3 nodes:  Tolerates 1 failure (quorum = 2)
5 nodes:  Tolerates 2 failures (quorum = 3)
7 nodes:  Tolerates 3 failures (quorum = 4)

More nodes:
✅ Higher availability
❌ Slower writes (more nodes to replicate to)
❌ More network traffic

Recommendation: 3 or 5 nodes for most use cases

Latency

Write latency = Network RTT + Disk sync time

Typically:
- Same datacenter: 1-10ms
- Cross-region: 50-300ms

Optimization:
- Batch writes
- Use faster disks (SSD)
- Tune election timeout

Common Patterns

Pattern 1: Configuration Management

class ConfigManager:
    """Use RAFT for distributed configuration"""

    def __init__(self, etcd_client):
        self.etcd = etcd_client

    def set_config(self, key, value):
        """Set configuration value (strongly consistent)"""
        self.etcd.put(f'/config/{key}', value)

    def get_config(self, key):
        """Get configuration value"""
        value, _ = self.etcd.get(f'/config/{key}')
        return value

    def watch_config(self, key, callback):
        """Watch for configuration changes"""
        self.etcd.add_watch_callback(f'/config/{key}', callback)

Pattern 2: Leader Election

import etcd3
import time

class LeaderElection:
    """Distributed leader election using etcd"""

    def __init__(self, etcd_client, name):
        self.etcd = etcd_client
        self.name = name
        self.lease = None

    def campaign(self):
        """Attempt to become leader"""
        self.lease = self.etcd.lease(ttl=60)

        # Try to create election key with our name
        success, _ = self.etcd.transaction(
            compare=[
                self.etcd.transactions.version('/election/leader') == 0
            ],
            success=[
                self.etcd.transactions.put('/election/leader', self.name, lease=self.lease)
            ],
            failure=[]
        )

        if success:
            print(f"{self.name} became leader")
            return True
        return False

    def resign(self):
        """Resign from leadership"""
        if self.lease:
            self.lease.revoke()

    def keep_alive(self):
        """Keep leadership alive"""
        if self.lease:
            self.lease.refresh()

Testing RAFT

Partition Testing

def test_network_partition():
    """Test RAFT behavior during network partition"""
    cluster = RaftCluster(nodes=5)

    # Normal operation
    cluster.write('key1', 'value1')
    assert cluster.read('key1') == 'value1'

    # Partition: [1, 2] | [3, 4, 5]
    cluster.partition([1, 2], [3, 4, 5])

    # Majority side (3, 4, 5) should elect new leader and continue
    cluster.write_to_node(3, 'key2', 'value2')  # Should succeed
    assert cluster.read_from_node(3, 'key2') == 'value2'

    # Minority side (1, 2) should refuse writes
    with pytest.raises(NoQuorumError):
        cluster.write_to_node(1, 'key3', 'value3')

    # Heal partition
    cluster.heal_partition()

    # Minority nodes should sync with majority
    time.sleep(1)
    assert cluster.read_from_node(1, 'key2') == 'value2'

Level 3 Resources

This skill includes executable scripts, comprehensive references, and examples for hands-on RAFT consensus work.

Quick Start

# Navigate to resources
cd skills/distributed-systems/consensus-raft/resources/scripts

# Start test cluster (Docker-based)
./test_etcd_cluster.sh --nodes 3

# Benchmark consensus performance
./benchmark_consensus.py --operations 5000 --concurrency 50

# Visualize RAFT state machine
./visualize_raft_state.py --type state-machine --format mermaid

Resources Structure

consensus-raft/resources/
├── REFERENCE.md           # RAFT algorithm deep-dive, etcd internals, proofs
├── scripts/
│   ├── README.md         # Scripts documentation
│   ├── test_etcd_cluster.sh      # Docker-based etcd cluster setup
│   ├── benchmark_consensus.py    # Performance benchmarking
│   └── visualize_raft_state.py   # State machine visualizations
└── examples/
    ├── python/           # Python RAFT client examples
    └── go/               # Go etcd integration examples

Available Scripts

test_etcd_cluster.sh

Automated Docker-based etcd RAFT cluster for testing:

# 5-node cluster
./test_etcd_cluster.sh --nodes 5

# JSON output for CI/CD
./test_etcd_cluster.sh --nodes 3 --json

# Cleanup
./test_etcd_cluster.sh --cleanup

benchmark_consensus.py

Measure consensus latency and throughput:

# Basic benchmark
./benchmark_consensus.py --operations 10000

# High concurrency
./benchmark_consensus.py --operations 50000 --concurrency 200 --json

Metrics: min/max/mean/p95/p99 latency, throughput, success rates

visualize_raft_state.py

Generate RAFT diagrams:

# State machine diagram
./visualize_raft_state.py --type state-machine --format mermaid

# Log replication sequence
./visualize_raft_state.py --type log-replication --format ascii

# Live cluster status
./visualize_raft_state.py --type cluster-status --endpoints localhost:2379

Reference Material

REFERENCE.md includes:

Example Workflow

# 1. Setup cluster
./test_etcd_cluster.sh --nodes 3

# 2. Run benchmarks
./benchmark_consensus.py --operations 5000 --json > results.json

# 3. Analyze results
cat results.json | jq '.results.put.latency.p99_ms'

# 4. Generate diagrams
./visualize_raft_state.py --type state-machine --output state.mmd

# 5. Test fault tolerance (in separate terminal)
docker stop etcd-node2  # Observe leader election

# 6. Cleanup
./test_etcd_cluster.sh --cleanup

See: resources/scripts/README.md for complete documentation

Related Skills

Last Updated: 2025-10-27