Distributed Consensus: Raft & Paxos

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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

// PRODUCTION RAFT REPLICATED STATE MACHINE ENGINE CORE
export type NodeRole = 'FOLLOWER' | 'CANDIDATE' | 'LEADER';

export interface LogEntry<T> {
  index: number;
  term: number;
  command: T;
}

export interface RequestVoteArgs {
  term: number;           // Candidate's term
  candidateId: string;    // Candidate requesting vote
  lastLogIndex: number;   // Index of candidate's last log entry
  lastLogTerm: number;    // Term of candidate's last log entry
}

export interface RequestVoteReply {
  term: number;           // CurrentTerm, for candidate to update itself
  voteGranted: boolean;   // True means candidate received vote
}

export interface AppendEntriesArgs<T> {
  term: number;           // Leader's term
  leaderId: string;       // So follower can redirect clients
  prevLogIndex: number;   // Index of log entry immediately preceding new ones
  prevLogTerm: number;    // Term of prevLogIndex entry
  entries: LogEntry<T>[]; // Log entries to store (empty for heartbeat)
  leaderCommit: number;   // Leader's commitIndex
}

export interface AppendEntriesReply {
  term: number;           // CurrentTerm, for leader to update itself
  success: boolean;       // True if follower contained entry matching prevLogIndex/prevLogTerm
  matchIndex: number;     // Follower's actual log index for fast backoff
}

export class RaftNode<T> {
  public id: string;
  public peers: string[];
  public currentTerm: number = 0;
  public votedFor: string | null = null;
  public role: NodeRole = 'FOLLOWER';
  public log: LogEntry<T>[] = [];
  public commitIndex: number = 0;
  public lastApplied: number = 0;

  // Leader-only state (reinitialized after election)
  public nextIndex: Map<string, number> = new Map();
  public matchIndex: Map<string, number> = new Map();

  constructor(id: string, peers: string[]) {
    this.id = id;
    this.peers = peers;
  }

  // Raft Safety Invariant: RequestVote RPC handler
  public handleRequestVote(args: RequestVoteArgs): RequestVoteReply {
    // 1. Rule: If term < currentTerm, reject vote immediately
    if (args.term < this.currentTerm) {
      return { term: this.currentTerm, voteGranted: false };
    }

    // If newer term discovered, convert to follower
    if (args.term > this.currentTerm) {
      this.currentTerm = args.term;
      this.role = 'FOLLOWER';
      this.votedFor = null;
    }

    // 2. Election Safety: Candidate log must be at least as up-to-date as receiver's log
    const lastLog = this.log[this.log.length - 1];
    const myLastTerm = lastLog ? lastLog.term : 0;
    const myLastIndex = lastLog ? lastLog.index : 0;

    const logIsUpToDate =
      args.lastLogTerm > myLastTerm ||
      (args.lastLogTerm === myLastTerm && args.lastLogIndex >= myLastIndex);

    const canVote = (this.votedFor === null || this.votedFor === args.candidateId) && logIsUpToDate;

    if (canVote) {
      this.votedFor = args.candidateId;
      return { term: this.currentTerm, voteGranted: true };
    }

    return { term: this.currentTerm, voteGranted: false };
  }

  // Raft Safety Invariant: AppendEntries RPC handler (Log Replication + Heartbeat)
  public handleAppendEntries(args: AppendEntriesArgs<T>): AppendEntriesReply {
    // 1. Reply false if term < currentTerm
    if (args.term < this.currentTerm) {
      return { term: this.currentTerm, success: false, matchIndex: this.commitIndex };
    }

    // Recognize valid leader
    if (args.term > this.currentTerm || this.role === 'CANDIDATE') {
      this.currentTerm = args.term;
      this.role = 'FOLLOWER';
      this.votedFor = null;
    }

    // 2. Log Matching Invariant: Check if entry exists at prevLogIndex with prevLogTerm
    if (args.prevLogIndex > 0) {
      const prevEntry = this.log[args.prevLogIndex - 1];
      if (!prevEntry || prevEntry.term !== args.prevLogTerm) {
        return { term: this.currentTerm, success: false, matchIndex: this.log.length };
      }
    }

    // 3. Truncate any conflicting entries and append new entries
    for (let i = 0; i < args.entries.length; i++) {
      const entry = args.entries[i];
      const targetIdx = args.prevLogIndex + i;
      if (this.log[targetIdx] && this.log[targetIdx].term !== entry.term) {
        this.log = this.log.slice(0, targetIdx); // Truncate conflicting suffix
      }
      if (!this.log[targetIdx]) {
        this.log.push(entry);
      }
    }

    // 4. Update commitIndex: min(leaderCommit, index of last new entry)
    if (args.leaderCommit > this.commitIndex) {
      this.commitIndex = Math.min(args.leaderCommit, this.log.length);
    }

    return { term: this.currentTerm, success: true, matchIndex: this.log.length };
  }
}