raft-log-replication-recovery

Answer

The implementation consists of a complete Raft consensus protocol in Go 1.26, structured as a single RaftNode with the following core subsystems:

Core Implementation (raft.go)

1. Leader Election with Randomized Timeouts

func (rn *RaftNode) randomElectionTimeout() time.Duration {
    return time.Duration(150+rand.Intn(150)) * time.Millisecond
}

Each follower runs an election loop with a random timeout (150-300ms). When a timeout fires, the node increments its term, votes for itself, and sends RequestVote RPCs to all peers. A candidate becomes leader if it receives votes from a majority (including itself). Upon receiving a RequestVote, a follower grants its vote only if the candidate's log is at least as up-to-date (lastLogTerm > currentTerm || (lastLogTerm == currentTerm && lastLogIndex >= currentIndex)).

2. AppendEntries RPC Handling & Log Healing

// In handleAppendEntries — conflict detection with term-based optimization
if rn.log[args.PrevLogIndex].Term != args.PrevLogTerm {
    conflictTerm := rn.log[args.PrevLogIndex].Term
    firstIndex := args.PrevLogIndex
    for i := args.PrevLogIndex; i > 0; i-- {
        if rn.log[i].Term == conflictTerm {
            firstIndex = i
        } else { break }
    }
    reply.ConflictTerm = conflictTerm
    reply.ConflictIndex = firstIndex
    return reply
}

When a follower rejects an AppendEntries due to log mismatch (term mismatch at PrevLogIndex), the reply carries the conflicting term and the first index at which that term appears. The leader uses this to skip past entire conflicting terms rather than decrementing one-by-one.

3. Leader Reconciliation (Log Inconsistency Healing)

// In sendHeartbeats — on rejection, fast-forward nextIndex
if reply.ConflictTerm != 0 {
    lastIndexInTerm := 0
    for i := len(rn.log) - 1; i > 0; i-- {
        if rn.log[i].Term == reply.ConflictTerm {
            lastIndexInTerm = rn.log[i].Index
            break
        }
    }
    if lastIndexInTerm > 0 {
        rn.nextIndex[peer] = lastIndexInTerm + 1  // skip to end of our term
    } else {
        rn.nextIndex[peer] = reply.ConflictIndex   // we lack this term entirely
    }
}

4. Commitment Safety

func (rn *RaftNode) updateCommitIndexLocked() {
    for n := len(rn.log) - 1; n > rn.commitIndex; n-- {
        // Only commit entries from current term (Raft safety rule §5.4.2)
        if rn.log[n].Term == rn.currentTerm {
            rn.commitIndex = n
        }
        break
    }
}

The leader only advances commitIndex for entries from its own term, preventing stale leaders from committing entries from previous terms.

5. Simulated Network for Testing (TestCluster) A complete 5-node test cluster with configurable connected[from][to] network matrix, allowing partition/heal cycles to be injected deterministically.