Raft Consensus Algorithm

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# Conceptual model of a replicated state machine
class ReplicatedStateMachine:
    def __init__(self):
        self.state = {}       # The current state (e.g., key-value store)
        self.log = []         # Ordered list of commands
        self.last_applied = 0 # Index of last command applied to state
    
    def apply_command(self, command: dict) -> any:
        """
        Apply a single command to the state machine.
        Commands are deterministic - same input = same output.
        """
        if command["type"] == "SET":
            self.state[command["key"]] = command["value"]
            return f"OK: set {command['key']}"
        elif command["type"] == "GET":
            return self.state.get(command["key"], "NOT_FOUND")
        elif command["type"] == "DELETE":
            if command["key"] in self.state:
                del self.state[command["key"]]
                return f"OK: deleted {command['key']}"
            return "NOT_FOUND"
    
    def apply_log_entries(self):
        """
        Apply all committed log entries that haven't been applied yet.
        This is called after consensus confirms entries are committed.
        """
        while self.last_applied < len(self.log):
            entry = self.log[self.last_applied]
            result = self.apply_command(entry["command"])
            self.last_applied += 1
            # In practice, we'd notify the client of the result here
 
    def append_to_log(self, entry: dict):
        """
        Append a new entry to the log.
        The entry contains: term, index, and command.
        """
        self.log.append(entry)
 
# Example: How identical logs lead to identical states
server1 = ReplicatedStateMachine()
server2 = ReplicatedStateMachine()
 
# If both servers receive the same log entries in the same order...
commands = [
    {"type": "SET", "key": "x", "value": 1},
    {"type": "SET", "key": "y", "value": 2},
    {"type": "SET", "key": "x", "value": 3},  # Overwrites x
]
 
for i, cmd in enumerate(commands):
    entry = {"term": 1, "index": i, "command": cmd}
    server1.append_to_log(entry)
    server2.append_to_log(entry)
 
server1.apply_log_entries()
server2.apply_log_entries()
 
# ...they will have identical final states
assert server1.state == server2.state  # Both: {"x": 3, "y": 2}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

# Raft's structure: Each server is in one of three states
from enum import Enum
 
class ServerState(Enum):
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"
 
class RaftServer:
    """
    A Raft server's state can be decomposed into:
    
    1. PERSISTENT state (survives crashes):
       - currentTerm: latest term server has seen
       - votedFor: candidateId that received vote in current term
       - log[]: log entries; each entry contains command and term
    
    2. VOLATILE state (rebuilt after crash):
       - commitIndex: index of highest log entry known to be committed
       - lastApplied: index of highest log entry applied to state machine
    
    3. VOLATILE state on LEADERS only:
       - nextIndex[]: for each server, index of next log entry to send
       - matchIndex[]: for each server, index of highest log entry replicated
    """
    
    def __init__(self, server_id: int, peer_ids: list):
        self.id = server_id
        self.peers = peer_ids
        
        # === PERSISTENT STATE ===
        self.current_term = 0
        self.voted_for = None
        self.log = []  # List of {term, command} entries
        
        # === VOLATILE STATE ===
        self.commit_index = 0
        self.last_applied = 0
        
        # === STATE ===
        self.state = ServerState.FOLLOWER
        
        # === LEADER-ONLY VOLATILE STATE ===
        self.next_index = {}   # Initialized on becoming leader
        self.match_index = {}  # Initialized on becoming leader
    
    def become_leader(self):
        """
        Called after winning an election.
        Initializes leader-specific state.
        """
        self.state = ServerState.LEADER
        # For each server, assume they're caught up, then discover otherwise
        for peer in self.peers:
            self.next_index[peer] = len(self.log) + 1
            self.match_index[peer] = 0
    
    def become_follower(self, term: int):
        """
        Called when discovering a higher term.
        This is how leaders step down gracefully.
        """
        self.state = ServerState.FOLLOWER
        self.current_term = term
        self.voted_for = None
        # Clear leader state
        self.next_index = {}
        self.match_index = {}
    
    def become_candidate(self):
        """
        Called when election timeout expires without
        hearing from a leader.
        """
        self.state = ServerState.CANDIDATE
        self.current_term += 1  # Increment term for this election
        self.voted_for = self.id  # Vote for self

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

"""
Terms in Raft serve multiple purposes:
 
1. LOGICAL CLOCK - Terms order events without physical time
2. ELECTION EPOCHS - Each election attempt starts a new term  
3. STALE DETECTION - Old messages/leaders are identified by old terms
4. TIE-BREAKING - Higher term always wins
 
Timeline example:
=================
 
Term 1:  [Server A is leader] --crash--> 
Term 2:  [Server B wins election, becomes leader] --crash-->
Term 3:  [Server C wins election, becomes leader] --> [operating normally]
 
Key invariant: At most one leader per term
"""
 
class TermManager:
    def __init__(self, initial_term: int = 0):
        self.term = initial_term
    
    def discover_higher_term(self, observed_term: int) -> bool:
        """
        Called when any RPC reveals a higher term.
        If observed term is higher, we:
        - Update our term
        - Revert to follower state
        - Clear any vote we've cast
        
        Returns True if our term was updated.
        """
        if observed_term > self.term:
            self.term = observed_term
            return True  # Caller should become follower
        return False
    
    def start_election(self) -> int:
        """
        Called when starting a new election.
        Increment term and return new value.
        """
        self.term += 1
        return self.term
    
    def is_valid_term(self, request_term: int) -> bool:
        """
        Check if an incoming request has a valid (not stale) term.
        Requests with term < current_term are rejected.
        """
        return request_term >= self.term
 
 
# Example: How terms prevent stale leaders
class RaftNode:
    def receive_append_entries(self, leader_term: int, leader_id: int, entries: list):
        """
        Handle AppendEntries RPC from (claimed) leader.
        """
        if leader_term < self.current_term:
            # This is a stale leader - reject!
            # Could be: old leader that was partitioned, 
            # or network delays delivered old message
            return {
                "success": False,
                "term": self.current_term  # Tell them to update
            }
        
        if leader_term > self.current_term:
            # They're ahead - we're stale
            # Step down and update our term
            self.become_follower(leader_term)
        
        # Now process the entries...
        # Reset election timer (we heard from valid leader)
        self.reset_election_timer()
        
        # ... log replication logic ...
        return {"success": True, "term": self.current_term}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

import random
import asyncio
from typing import Optional
 
class ElectionTimer:
    """
    Raft's election timer with randomized timeout.
    
    Key parameters:
    - HEARTBEAT_INTERVAL: How often leader sends heartbeats (e.g., 50ms)
    - ELECTION_TIMEOUT_MIN: Minimum timeout before starting election (e.g., 150ms)
    - ELECTION_TIMEOUT_MAX: Maximum timeout (e.g., 300ms)
    
    Rule of thumb: 
    ELECTION_TIMEOUT_MIN > HEARTBEAT_INTERVAL * 2
    to allow for network delays
    """
    
    HEARTBEAT_INTERVAL_MS = 50
    ELECTION_TIMEOUT_MIN_MS = 150
    ELECTION_TIMEOUT_MAX_MS = 300
    
    def __init__(self, on_timeout_callback):
        self.on_timeout = on_timeout_callback
        self.timer_task: Optional[asyncio.Task] = None
    
    def _random_timeout(self) -> float:
        """
        Generate random timeout in seconds.
        Using uniform distribution over the range.
        """
        timeout_ms = random.randint(
            self.ELECTION_TIMEOUT_MIN_MS,
            self.ELECTION_TIMEOUT_MAX_MS
        )
        return timeout_ms / 1000.0
    
    def reset(self):
        """
        Called when:
        1. Server receives valid AppendEntries from leader
        2. Server grants vote to a candidate
        
        Resets the timer with a new random timeout.
        This prevents unnecessary elections while leader is alive.
        """
        if self.timer_task:
            self.timer_task.cancel()
        
        timeout = self._random_timeout()
        self.timer_task = asyncio.create_task(
            self._run_timer(timeout)
        )
    
    async def _run_timer(self, timeout: float):
        """
        Wait for timeout, then trigger election.
        If reset() is called before timeout, this task is cancelled.
        """
        try:
            await asyncio.sleep(timeout)
            # Timeout expired without hearing from leader
            # Convert to candidate and start election
            await self.on_timeout()
        except asyncio.CancelledError:
            # Timer was reset - normal operation
            pass
 
 
# Simulation showing how randomized timeouts prevent split votes
def simulate_election_starts(num_servers: int, num_trials: int = 1000) -> dict:
    """
    Simulate how often servers would start elections 'simultaneously'.
    """
    simultaneous_starts = 0
    single_starter = 0
    threshold_ms = 10  # If within 10ms, consider 'simultaneous'
    
    for _ in range(num_trials):
        timeouts = [
            random.randint(150, 300) for _ in range(num_servers)
        ]
        min_timeout = min(timeouts)
        starters = sum(1 for t in timeouts if t - min_timeout < threshold_ms)
        
        if starters > 1:
            simultaneous_starts += 1
        else:
            single_starter += 1
    
    return {
        "simultaneous_starts": simultaneous_starts,
        "single_starter": single_starter,
        "probability_clean_election": single_starter / num_trials
    }
 
# With 5 servers: ~93% of elections have a clear first starter
# result = simulate_election_starts(5)  
# {'simultaneous_starts': 70, 'single_starter': 930, 'probability_clean_election': 0.93}