Distributed consensus không cần PhD

Lần đầu mình mở Raft paper ra, thấy hình state machine với mũi tên tứ tung là đóng tab ngay. Mấy tuần sau deadline ép, mình mới ngồi lại đọc từng trang — core idea chỉ gói gọn trong vài câu. Phần còn lại là edge cases.

Mình sẽ build một simplified Raft bằng Go: leader election và log replication. Không dùng library, không abstract gì cả. Mục tiêu là anh em hiểu consensus hoạt động như thế nào — không phải thuộc lòng specification.

Vấn đề consensus trông như thế nào khi mọi thứ vỡ

Có N nodes trong một cluster. Anh em muốn chúng agree on một sequence of values — “transaction này đã committed”, “config đã thay đổi”. Nhưng network drop messages, nodes crash bất cứ lúc nào.

Không có consensus protocol thì:

• Node A xử lý request X, node B không nhận được message → hai node có state khác nhau
• Node A crash giữa chừng → ai là source of truth?
• Network partition → hai nhóm nodes tự ra quyết định riêng, reconnect với conflicting history

Raft giải quyết bằng một rule đơn giản: chỉ có một leader được phép quyết định, và mọi quyết định phải được majority (quorum) xác nhận trước khi coi là committed.

State machine của một Raft node

Ba states, chuyển qua lại theo term:

type State int

const (
    Follower  State = iota
    Candidate
    Leader
)

type LogEntry struct {
    Term    int
    Command interface{}
}

type RaftNode struct {
    mu          sync.Mutex
    id          int
    peers       []int
    state       State
    currentTerm int
    votedFor    int // -1 nếu chưa vote trong term này

    log         []LogEntry
    commitIndex int // index cao nhất đã được commit
    lastApplied int // index cao nhất đã apply vào state machine

    // Leader-only
    nextIndex  []int // với mỗi follower: index tiếp theo cần gửi
    matchIndex []int // với mỗi follower: index cao nhất đã replicate thành công

    electionTimer  *time.Timer
    heartbeatTimer *time.Timer

    voteCh      chan RequestVoteArgs
    appendCh    chan AppendEntriesArgs
    stopCh      chan struct{}
}

Transitions:

• Follower → Candidate: election timeout hết mà không nhận heartbeat từ leader
• Candidate → Leader: nhận đủ votes từ majority
• Candidate / Leader → Follower: nhận message với term cao hơn

Election timeout và cách trigger election

Follower chờ heartbeat từ leader. Timeout mà không có gì → assume leader đã chết, bắt đầu election.

Timeout phải randomized. Nếu tất cả cùng timeout một lúc thì ai cũng self-vote, không ai thắng — split vote mãi mãi.

func randomTimeout() time.Duration {
    // 150-300ms là range trong Raft paper
    ms := 150 + rand.Intn(150)
    return time.Duration(ms) * time.Millisecond
}

func (n *RaftNode) runElectionTimer() {
    for {
        select {
        case <-n.stopCh:
            return
        default:
        }

        timeout := randomTimeout()
        timer := time.NewTimer(timeout)

        select {
        case <-timer.C:
            n.mu.Lock()
            if n.state != Leader {
                // Timeout mà vẫn là Follower hoặc Candidate
                // → bắt đầu election mới
                go n.startElection()
            }
            n.mu.Unlock()
        case <-n.stopCh:
            timer.Stop()
            return
        }
    }
}

func (n *RaftNode) resetElectionTimer() {
    // Gọi khi nhận heartbeat hoặc vote cho ai đó
    // → restart timeout từ đầu
    n.electionTimer.Reset(randomTimeout())
}

RequestVote: nodes vote cho nhau như thế nào

Node bắt đầu election: increment term, chuyển thành Candidate, tự vote cho mình, gửi RequestVote tới tất cả peers.

type RequestVoteArgs struct {
    Term         int
    CandidateID  int
    LastLogIndex int // để kiểm tra log up-to-date
    LastLogTerm  int
}

type RequestVoteReply struct {
    Term        int
    VoteGranted bool
}

func (n *RaftNode) startElection() {
    n.mu.Lock()
    n.state = Candidate
    n.currentTerm++
    n.votedFor = n.id
    term := n.currentTerm
    lastLogIndex := len(n.log) - 1
    lastLogTerm := -1
    if lastLogIndex >= 0 {
        lastLogTerm = n.log[lastLogIndex].Term
    }
    n.mu.Unlock()

    votes := 1 // tự vote cho mình
    var mu sync.Mutex
    var wg sync.WaitGroup

    for _, peer := range n.peers {
        wg.Add(1)
        go func(peerID int) {
            defer wg.Done()
            args := RequestVoteArgs{
                Term:         term,
                CandidateID:  n.id,
                LastLogIndex: lastLogIndex,
                LastLogTerm:  lastLogTerm,
            }
            var reply RequestVoteReply
            if n.sendRequestVote(peerID, args, &reply) {
                n.mu.Lock()
                if reply.Term > n.currentTerm {
                    // Có term cao hơn → mình outdated, step down
                    n.currentTerm = reply.Term
                    n.state = Follower
                    n.votedFor = -1
                    n.mu.Unlock()
                    return
                }
                n.mu.Unlock()

                if reply.VoteGranted {
                    mu.Lock()
                    votes++
                    quorum := len(n.peers)/2 + 1
                    if votes >= quorum {
                        n.becomeLeader()
                    }
                    mu.Unlock()
                }
            }
        }(peer)
    }
    wg.Wait()
}

Phía nhận vote:

func (n *RaftNode) handleRequestVote(args RequestVoteArgs) RequestVoteReply {
    n.mu.Lock()
    defer n.mu.Unlock()

    reply := RequestVoteReply{Term: n.currentTerm}

    if args.Term < n.currentTerm {
        // Candidate bị outdated → từ chối
        reply.VoteGranted = false
        return reply
    }

    if args.Term > n.currentTerm {
        // Nhận term mới hơn → step down và cập nhật
        n.currentTerm = args.Term
        n.state = Follower
        n.votedFor = -1
    }

    alreadyVoted := n.votedFor != -1 && n.votedFor != args.CandidateID
    if alreadyVoted {
        reply.VoteGranted = false
        return reply
    }

    // Kiểm tra log up-to-date: candidate's log phải ít nhất bằng mình
    myLastIndex := len(n.log) - 1
    myLastTerm := -1
    if myLastIndex >= 0 {
        myLastTerm = n.log[myLastIndex].Term
    }

    logOK := args.LastLogTerm > myLastTerm ||
        (args.LastLogTerm == myLastTerm && args.LastLogIndex >= myLastIndex)

    if logOK {
        n.votedFor = args.CandidateID
        n.resetElectionTimer()
        reply.VoteGranted = true
    }

    return reply
}

Điều kiện logOK là chỗ anh em hay miss: mình chỉ vote cho candidate nếu log của nó “at least as up-to-date” như log của mình. Đây là cách đảm bảo leader mới không bị thiếu thông tin về những gì đã committed.

Tại sao quorum là N/2 + 1

Đây là cái mình thấy elegant nhất trong toàn bộ Raft.

Với 5 nodes, quorum là 3. Bất kỳ hai quorums nào cũng overlap ít nhất 1 node:

Quorum A: {1, 2, 3}
Quorum B: {3, 4, 5}
Overlap:  {3}        ← node 3 biết cả hai sides

Nếu một entry đã committed (ack bởi quorum A), thì bất kỳ leader mới nào được bầu lên (cần quorum B để win) cũng sẽ có ít nhất một node biết về entry đó. Node đó sẽ không vote cho candidate nào mà log thiếu entry đó.

Nếu quorum là N/2 thay vì N/2 + 1 thì hai quorums có thể không overlap → conflicting commits → split brain.

AppendEntries: heartbeat và log replication

AppendEntries là RPC làm hai việc:

1. Gửi heartbeat định kỳ để followers biết leader còn sống
2. Replicate log entries từ leader xuống followers

type AppendEntriesArgs struct {
    Term         int
    LeaderID     int
    PrevLogIndex int      // index của entry ngay trước entries mới
    PrevLogTerm  int      // term của entry đó
    Entries      []LogEntry
    LeaderCommit int      // commitIndex của leader
}

type AppendEntriesReply struct {
    Term    int
    Success bool
}

func (n *RaftNode) handleAppendEntries(args AppendEntriesArgs) AppendEntriesReply {
    n.mu.Lock()
    defer n.mu.Unlock()

    reply := AppendEntriesReply{Term: n.currentTerm}

    if args.Term < n.currentTerm {
        reply.Success = false
        return reply
    }

    // Nhận message từ current/newer leader → step down nếu cần, reset timer
    if args.Term > n.currentTerm {
        n.currentTerm = args.Term
        n.votedFor = -1
    }
    n.state = Follower
    n.resetElectionTimer()

    // Kiểm tra consistency: entry tại PrevLogIndex phải match PrevLogTerm
    if args.PrevLogIndex >= 0 {
        if args.PrevLogIndex >= len(n.log) {
            // Log của mình ngắn hơn → thiếu entries
            reply.Success = false
            return reply
        }
        if n.log[args.PrevLogIndex].Term != args.PrevLogTerm {
            // Conflict tại PrevLogIndex → truncate và retry
            reply.Success = false
            return reply
        }
    }

    // Append entries mới, overwrite nếu có conflict
    for i, entry := range args.Entries {
        idx := args.PrevLogIndex + 1 + i
        if idx < len(n.log) {
            if n.log[idx].Term != entry.Term {
                // Conflict → truncate từ đây
                n.log = n.log[:idx]
            }
        }
        if idx >= len(n.log) {
            n.log = append(n.log, entry)
        }
    }

    // Cập nhật commitIndex theo leader
    if args.LeaderCommit > n.commitIndex {
        n.commitIndex = min(args.LeaderCommit, len(n.log)-1)
    }

    reply.Success = true
    return reply
}

Leader gửi heartbeat định kỳ:

func (n *RaftNode) runHeartbeat() {
    ticker := time.NewTicker(50 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            n.mu.Lock()
            if n.state != Leader {
                n.mu.Unlock()
                return
            }
            n.mu.Unlock()
            n.broadcastAppendEntries()
        case <-n.stopCh:
            return
        }
    }
}

func (n *RaftNode) broadcastAppendEntries() {
    for _, peer := range n.peers {
        go n.sendAppendEntriesToPeer(peer)
    }
}

Commit index: khi nào entry được coi là committed

Leader chỉ advance commitIndex khi majority đã ack:

func (n *RaftNode) sendAppendEntriesToPeer(peerID int) {
    n.mu.Lock()
    if n.state != Leader {
        n.mu.Unlock()
        return
    }

    nextIdx := n.nextIndex[peerID]
    prevLogIndex := nextIdx - 1
    prevLogTerm := -1
    if prevLogIndex >= 0 && prevLogIndex < len(n.log) {
        prevLogTerm = n.log[prevLogIndex].Term
    }

    entries := n.log[nextIdx:]
    args := AppendEntriesArgs{
        Term:         n.currentTerm,
        LeaderID:     n.id,
        PrevLogIndex: prevLogIndex,
        PrevLogTerm:  prevLogTerm,
        Entries:      append([]LogEntry{}, entries...), // copy để tránh race
        LeaderCommit: n.commitIndex,
    }
    n.mu.Unlock()

    var reply AppendEntriesReply
    if !n.sendAppendEntries(peerID, args, &reply) {
        return
    }

    n.mu.Lock()
    defer n.mu.Unlock()

    if reply.Term > n.currentTerm {
        n.currentTerm = reply.Term
        n.state = Follower
        n.votedFor = -1
        return
    }

    if reply.Success {
        n.matchIndex[peerID] = prevLogIndex + len(entries)
        n.nextIndex[peerID] = n.matchIndex[peerID] + 1

        // Check nếu có entry mới có thể commit
        n.maybeAdvanceCommitIndex()
    } else {
        // Follower thiếu entries → decrement nextIndex và retry
        if n.nextIndex[peerID] > 0 {
            n.nextIndex[peerID]--
        }
    }
}

func (n *RaftNode) maybeAdvanceCommitIndex() {
    // Tìm N lớn nhất mà majority matchIndex[i] >= N
    // và log[N].Term == currentTerm
    for N := len(n.log) - 1; N > n.commitIndex; N-- {
        if n.log[N].Term != n.currentTerm {
            // Chỉ commit entries của current term
            // (safety property của Raft)
            continue
        }
        count := 1 // leader tự tính
        for _, peer := range n.peers {
            if n.matchIndex[peer] >= N {
                count++
            }
        }
        quorum := len(n.peers)/2 + 1
        if count >= quorum {
            n.commitIndex = N
            break
        }
    }
}

Chú ý n.log[N].Term != n.currentTerm: leader chỉ commit entries của current term. Đây là subtlety hay bị miss — không thể commit entry từ previous term trực tiếp. Phải commit một entry của current term, và entry đó kéo theo tất cả entries trước đó theo Log Matching Property.

Kết

• Đừng đọc paper trước — build trước, paper sau. Code ép anh em phải giải quyết từng edge case thay vì đọc lướt qua và tưởng mình hiểu.
• Quorum overlap là toàn bộ safety guarantee của Raft. Election restriction, log matching, commit rule — tất cả đều serve một invariant duy nhất này.
• Muốn production-grade thì xem etcd/raft: pure Go, không có network layer, anh em plug in transport riêng. Code readable và có comment đối chiếu từng bước với paper.