Raft Distributed Consensus

/**
 * Raft Distributed Consensus State Machine (Ongaro & Ousterhout, 2014)
 * Implements monotonic currentTerm, votedFor persistence, randomized election timeouts (150..300ms),
 * §5.4.1 RequestVote log completeness checks, leader nextIndex[]/matchIndex[] tracking with backoff on rejection,
 * AppendEntries log replication with Figure 2 commitIndex safety (min(leaderCommit, prevLogIndex + entries.length)),
 * and §5.4.2 (Figure 8) leader quorum commit rule (only committing entries from currentTerm by counting replicas).
 */
export interface LogEntry { term: number; command: string; }
export interface LogMeta { lastLogTerm: number; lastLogIndex: number; }

export function isCandidateLogUpToDate(candidate: LogMeta, voter: LogMeta): boolean {
  if (candidate.lastLogTerm !== voter.lastLogTerm) {
    return candidate.lastLogTerm > voter.lastLogTerm;
  }
  return candidate.lastLogIndex >= voter.lastLogIndex;
}

export function evaluateRaftElectionQuorum(
  clusterSize: number,
  voterLogs: readonly LogMeta[],
  candidateLog: LogMeta
): { quorumReached: boolean; votesReceived: number; majorityRequired: number } {
  const majorityRequired = Math.floor(clusterSize / 2) + 1;
  let votesReceived = 1;
  for (const voter of voterLogs) {
    if (isCandidateLogUpToDate(candidateLog, voter)) votesReceived++;
  }
  return { quorumReached: votesReceived >= majorityRequired, votesReceived, majorityRequired };
}

export class RaftNode {
  public currentTerm = 0;
  public votedFor: number | null = null;
  public role: "follower" | "candidate" | "leader" = "follower";
  public log: LogEntry[] = [];
  public commitIndex = -1;
  public nextIndex = new Map<number, number>();
  public matchIndex = new Map<number, number>();
  public readonly nodeId: number;
  public readonly electionTimeoutMs: number;

  constructor(nodeId: number, seed = nodeId + 1) {
    this.nodeId = nodeId;
    const jitter = ((Math.imul(seed, 1664525) + 1013904223) >>> 0) % 151;
    this.electionTimeoutMs = 150 + jitter; // Randomized 150..300ms election timeout
  }

  lastLogMeta(): LogMeta {
    const lastIdx = this.log.length - 1;
    return { lastLogTerm: lastIdx >= 0 ? this.log[lastIdx].term : 0, lastLogIndex: lastIdx };
  }

  appendCommand(command: string): number {
    if (this.role !== "leader") throw new Error("Only leader can append client commands");
    this.log.push({ term: this.currentTerm, command });
    return this.log.length - 1;
  }

  requestVote(
    term: number,
    candidateId: number,
    lastLogIndex: number,
    lastLogTerm: number
  ): { term: number; voteGranted: boolean } {
    if (term > this.currentTerm) {
      this.currentTerm = term;
      this.role = "follower";
      this.votedFor = null;
    }
    if (term < this.currentTerm) {
      return { term: this.currentTerm, voteGranted: false };
    }
    const canVote = this.votedFor === null || this.votedFor === candidateId;
    const upToDate = isCandidateLogUpToDate(
      { lastLogTerm, lastLogIndex },
      this.lastLogMeta()
    );
    if (canVote && upToDate) {
      this.votedFor = candidateId;
      return { term: this.currentTerm, voteGranted: true };
    }
    return { term: this.currentTerm, voteGranted: false };
  }

  startElection(peers: readonly RaftNode[]): boolean {
    this.currentTerm += 1;
    this.role = "candidate";
    this.votedFor = this.nodeId;
    const meta = this.lastLogMeta();
    let votes = 1;
    const totalNodes = peers.length + 1;
    for (const peer of peers) {
      const reply = peer.requestVote(this.currentTerm, this.nodeId, meta.lastLogIndex, meta.lastLogTerm);
      if (reply.term > this.currentTerm) {
        this.currentTerm = reply.term;
        this.role = "follower";
        this.votedFor = null;
        return false;
      }
      if (reply.voteGranted) votes++;
    }
    if (votes >= Math.floor(totalNodes / 2) + 1) {
      this.role = "leader";
      this.nextIndex.clear();
      this.matchIndex.clear();
      for (const peer of peers) {
        this.nextIndex.set(peer.nodeId, this.log.length);
        this.matchIndex.set(peer.nodeId, -1);
      }
      return true;
    }
    return false;
  }

  replicateTo(peer: RaftNode): boolean {
    if (this.role !== "leader") return false;
    let nextIdx = this.nextIndex.get(peer.nodeId) ?? this.log.length;
    while (nextIdx >= 0) {
      const prevLogIndex = nextIdx - 1;
      const prevLogTerm = prevLogIndex >= 0 ? this.log[prevLogIndex].term : 0;
      const entries = this.log.slice(nextIdx);
      const reply = peer.appendEntries(
        this.currentTerm,
        this.nodeId,
        prevLogIndex,
        prevLogTerm,
        entries,
        this.commitIndex
      );
      if (reply.term > this.currentTerm) {
        this.currentTerm = reply.term;
        this.role = "follower";
        this.votedFor = null;
        return false;
      }
      if (reply.success) {
        const lastIdx = prevLogIndex + entries.length;
        this.nextIndex.set(peer.nodeId, lastIdx + 1);
        this.matchIndex.set(peer.nodeId, Math.max(this.matchIndex.get(peer.nodeId) ?? -1, lastIdx));
        return true;
      }
      nextIdx -= 1;
      this.nextIndex.set(peer.nodeId, Math.max(0, nextIdx));
    }
    return false;
  }

  advanceCommitIndex(clusterSize = this.matchIndex.size + 1): number {
    if (this.role !== "leader") return this.commitIndex;
    const majority = Math.floor(clusterSize / 2) + 1;
    for (let n = this.log.length - 1; n > this.commitIndex; n--) {
      // §5.4.2 (Figure 8): Leader only commits entries from its own currentTerm by counting replicas
      if (this.log[n].term !== this.currentTerm) continue;
      let count = 1; // Leader itself
      for (const mIdx of this.matchIndex.values()) {
        if (mIdx >= n) count++;
      }
      if (count >= majority) {
        this.commitIndex = n;
        break;
      }
    }
    return this.commitIndex;
  }

  appendEntries(
    term: number,
    _leaderId: number,
    prevLogIndex: number,
    prevLogTerm: number,
    entries: readonly LogEntry[],
    leaderCommit: number
  ): { term: number; success: boolean } {
    if (term < this.currentTerm) {
      return { term: this.currentTerm, success: false };
    }
    if (term > this.currentTerm) {
      this.votedFor = null;
    }
    this.currentTerm = term;
    this.role = "follower";

    if (prevLogIndex >= 0) {
      if (prevLogIndex >= this.log.length || this.log[prevLogIndex].term !== prevLogTerm) {
        return { term: this.currentTerm, success: false };
      }
    }
    let idx = prevLogIndex + 1;
    for (const entry of entries) {
      if (idx < this.log.length && this.log[idx].term !== entry.term) {
        this.log.length = idx;
      }
      if (idx >= this.log.length) {
        this.log.push({ ...entry });
      }
      idx++;
    }
    // Raft Figure 2 Receiver Step 5: min(leaderCommit, index of last new entry), monotonic forward only
    const indexOfLastNewEntry = prevLogIndex + entries.length;
    if (leaderCommit > this.commitIndex) {
      this.commitIndex = Math.max(
        this.commitIndex,
        Math.min(leaderCommit, indexOfLastNewEntry)
      );
    }
    return { term: this.currentTerm, success: true };
  }
}