Building a Distributed Consensus Engine from Scratch: Raft Election and Replication Invariants

interface LogEntry {
  term: number;
  command: string;
}

interface AppendEntriesPayload {
  term: number;
  leaderId: number;
  prevLogIndex: number;
  prevLogTerm: number;
  entries: LogEntry[];
  leaderCommit: number;
}

export class RaftConsensusEngine {
  private nodeId: number;
  private currentTerm: number = 0;
  private votedFor: number | null = null;
  private log: LogEntry[] = [];
  
  private commitIndex: number = 0;
  private lastApplied: number = 0;
  
  private state: "Leader" | "Follower" | "Candidate" = "Follower";
  private leaderId: number | null = null;

  constructor(nodeId: number) {
    this.nodeId = nodeId;
    this.initializeLog();
  }

  private initializeLog() {
    // Index 0 sentinel boundary node log
    this.log.push({ term: 0, command: "SENTINEL" });
  }

  // 1. Inbound AppendEntries RPC handler (Heartbeats & Replication)
  public handleAppendEntries(payload: AppendEntriesPayload): { term: number; success: boolean } {
    // Rule 1: Reply false if term is outdated
    if (payload.term < this.currentTerm) {
      return { term: this.currentTerm, success: false };
    }

    // Update term and demote self if payload term is higher
    if (payload.term > this.currentTerm) {
      this.currentTerm = payload.term;
      this.state = "Follower";
      this.votedFor = null;
    }

    this.leaderId = payload.leaderId;

    // Rule 2: Reply false if log doesn't contain entry at prevLogIndex matching prevLogTerm
    if (this.log.length <= payload.prevLogIndex || this.log[payload.prevLogIndex].term !== payload.prevLogTerm) {
      return { term: this.currentTerm, success: false };
    }

    // Rule 3: Write new entries, overwrite if existing conflicts
    let entryIdx = payload.prevLogIndex + 1;
    for (const newEntry of payload.entries) {
      if (this.log[entryIdx]) {
        if (this.log[entryIdx].term !== newEntry.term) {
          // Conflict found! Prune log from this point
          this.log = this.log.slice(0, entryIdx);
          this.log.push(newEntry);
        }
      } else {
        this.log.push(newEntry);
      }
      entryIdx++;
    }

    // Rule 4: Align local commitIndex coordinates with leader commit bounds
    if (payload.leaderCommit > this.commitIndex) {
      this.commitIndex = Math.min(payload.leaderCommit, this.log.length - 1);
      this.applyToStateMachine();
    }

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

  // 2. Election Trigger: Broadcast RequestVote RPCs
  public initiateElection(clusterSize: number, broadcastVotes: () => Promise<boolean>) {
    this.state = "Candidate";
    this.currentTerm += 1;
    this.votedFor = this.nodeId; // vote for self
    
    console.log(`Node ${this.nodeId} starting election for Term ${this.currentTerm}`);
    
    // Broadcast RequestVote payloads to peer nodes
    broadcastVotes().then(granted => {
      if (granted && this.state === "Candidate") {
        this.state = "Leader";
        this.leaderId = this.nodeId;
        console.log(`Node ${this.nodeId} elected Leader for Term ${this.currentTerm}`);
        this.startHeartbeatBroadcast();
      }
    });
  }

  private startHeartbeatBroadcast() {
    // Periodically emit empty AppendEntries payloads to preserve lease ownership
    console.log(`Leader Node ${this.nodeId} broadcasting Term ${this.currentTerm} heartbeat.`);
  }

  private applyToStateMachine() {
    while (this.commitIndex > this.lastApplied) {
      this.lastApplied++;
      const committedCommand = this.log[this.lastApplied].command;
      console.log(`Applying committed index ${this.lastApplied} ("${committedCommand}") to Node ${this.nodeId} State Machine.`);
    }
  }
}