Distributed Consensus Algorithms

// Phase 1: Prepare
// Proposer selects a proposal number n and sends Prepare(n) to a majority of acceptors

interface PrepareRequest {
  proposalNumber: number;
}

interface PrepareResponse {
  promise: boolean;           // Will the acceptor promise not to accept lower proposals?
  acceptedProposal?: number;  // Highest proposal number previously accepted
  acceptedValue?: string;     // Value from that proposal
}

// Phase 2: Accept
// If proposer receives promises from a majority:
// - If any acceptor already accepted a value, proposer must use the value
//   from the highest-numbered accepted proposal
// - Otherwise, proposer can use its own value
// Proposer sends Accept(n, value) to a majority of acceptors

interface AcceptRequest {
  proposalNumber: number;
  value: string;
}

interface AcceptResponse {
  accepted: boolean;
}

// Simplified Paxos Proposer
class PaxosProposer {
  private proposalNumber: number = 0;

  async propose(value: string, acceptors: Acceptor[]): Promise<string> {
    const majority = Math.floor(acceptors.length / 2) + 1;

    // Phase 1: Prepare
    this.proposalNumber++;
    const prepareResponses = await Promise.all(
      acceptors.map((a) => a.prepare({ proposalNumber: this.proposalNumber }))
    );

    const promises = prepareResponses.filter((r) => r.promise);
    if (promises.length < majority) {
      throw new Error("Failed to get majority promises");
    }

    // Use the value from the highest accepted proposal, or our own
    const highestAccepted = promises
      .filter((r) => r.acceptedProposal !== undefined)
      .sort((a, b) => (b.acceptedProposal || 0) - (a.acceptedProposal || 0))[0];

    const finalValue = highestAccepted?.acceptedValue || value;

    // Phase 2: Accept
    const acceptResponses = await Promise.all(
      acceptors.map((a) =>
        a.accept({ proposalNumber: this.proposalNumber, value: finalValue })
      )
    );

    const accepted = acceptResponses.filter((r) => r.accepted);
    if (accepted.length < majority) {
      throw new Error("Failed to get majority acceptance");
    }

    return finalValue; // Consensus reached!
  }
}

enum RaftState {
  FOLLOWER = "FOLLOWER",
  CANDIDATE = "CANDIDATE",
  LEADER = "LEADER",
}

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

interface RaftNode {
  id: string;
  state: RaftState;
  currentTerm: number;
  votedFor: string | null;
  log: LogEntry[];
  commitIndex: number;
  lastApplied: number;

  // Leader-only state
  nextIndex: Map<string, number>;  // For each follower: next log index to send
  matchIndex: Map<string, number>; // For each follower: highest replicated index
}

class RaftElection {
  private electionTimeout: number; // Random between 150-300ms
  private heartbeatInterval: number = 50; // Leader sends heartbeats

  startElection(node: RaftNode, peers: string[]): void {
    // 1. Increment current term
    node.currentTerm++;
    node.state = RaftState.CANDIDATE;
    node.votedFor = node.id; // Vote for self

    let votesReceived = 1; // Self-vote
    const majority = Math.floor(peers.length / 2) + 1;

    // 2. Send RequestVote to all peers
    for (const peerId of peers) {
      const response = this.sendRequestVote(peerId, {
        term: node.currentTerm,
        candidateId: node.id,
        lastLogIndex: node.log.length - 1,
        lastLogTerm: node.log.length > 0
          ? node.log[node.log.length - 1].term
          : 0,
      });

      if (response.voteGranted) {
        votesReceived++;
      }
      if (response.term > node.currentTerm) {
        // Discovered higher term: step down
        node.currentTerm = response.term;
        node.state = RaftState.FOLLOWER;
        return;
      }
    }

    // 3. Check if we won
    if (votesReceived >= majority) {
      node.state = RaftState.LEADER;
      this.initializeLeaderState(node, peers);
      this.sendHeartbeats(node, peers);
    }
  }

  // A node grants a vote if:
  // 1. The candidate's term >= the voter's term
  // 2. The voter hasn't voted for someone else in this term
  // 3. The candidate's log is at least as up-to-date as the voter's
  handleRequestVote(node: RaftNode, request: RequestVoteRequest): RequestVoteResponse {
    if (request.term < node.currentTerm) {
      return { term: node.currentTerm, voteGranted: false };
    }

    if (request.term > node.currentTerm) {
      node.currentTerm = request.term;
      node.state = RaftState.FOLLOWER;
      node.votedFor = null;
    }

    const logOk = request.lastLogTerm > this.getLastLogTerm(node) ||
      (request.lastLogTerm === this.getLastLogTerm(node) &&
       request.lastLogIndex >= node.log.length - 1);

    if ((node.votedFor === null || node.votedFor === request.candidateId) && logOk) {
      node.votedFor = request.candidateId;
      return { term: node.currentTerm, voteGranted: true };
    }

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

class RaftLogReplication {
  async replicateEntry(
    leader: RaftNode,
    command: string,
    peers: string[]
  ): Promise<boolean> {
    // 1. Append to leader's log
    const entry: LogEntry = {
      term: leader.currentTerm,
      index: leader.log.length,
      command,
    };
    leader.log.push(entry);

    // 2. Send AppendEntries to all followers
    let replicatedCount = 1; // Leader counts as one
    const majority = Math.floor(peers.length / 2) + 1;

    const results = await Promise.all(
      peers.map((peerId) =>
        this.sendAppendEntries(peerId, {
          term: leader.currentTerm,
          leaderId: leader.id,
          prevLogIndex: entry.index - 1,
          prevLogTerm: entry.index > 0 ? leader.log[entry.index - 1].term : 0,
          entries: [entry],
          leaderCommit: leader.commitIndex,
        })
      )
    );

    for (const result of results) {
      if (result.success) replicatedCount++;
    }

    // 3. If replicated to majority, commit
    if (replicatedCount >= majority) {
      leader.commitIndex = entry.index;
      return true; // Entry is committed and safe
    }

    return false;
  }
}