heartbeat.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"time"
)

const heartbeatInterval = 50 * time.Millisecond

func (n *Node) runHeartbeat() {
	ticker := time.NewTicker(heartbeatInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			n.mu.Lock()
			if n.role != Leader {
				n.mu.Unlock()
				return
			}
			peers := n.config.Peers
			term := n.currentTerm
			commitIndex := n.commitIndex
			id := n.id
			n.mu.Unlock()

			// send heartbeat to all peers in parallel, including any missing entries
			for _, peer := range peers {
				if peer == id {
					continue
				}
				go func(peer string) {
					n.mu.Lock()
					next := n.nextIndex[peer]
					var entries []LogEntry
					if next > 0 && next <= len(n.log) {
						entries = n.log[next-1:]
					}
					n.mu.Unlock()
					n.sendAppendEntries(peer, term, commitIndex, entries)
				}(peer)
			}

		case <-n.stopHeartbeat:
			return
		}
	}
}

func (n *Node) sendAppendEntries(peer string, term int, commitIndex int, entries []LogEntry) {
	body := AppendEntriesRequest{
		Term:        term,
		LeaderID:    n.id,
		CommitIndex: commitIndex,
		Entries:     entries,
	}

	data, err := json.Marshal(body)
	if err != nil {
		return
	}

	resp, err := http.Post(fmt.Sprintf("http://%s/append-entries", peer), "application/json", bytes.NewReader(data))
	if err != nil {
		return
	}
	defer resp.Body.Close()

	var result AppendEntriesResponse
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return
	}

	// if peer has a higher term we are stale, step down
	n.mu.Lock()
	defer n.mu.Unlock()
	if result.Term > n.currentTerm {
		n.currentTerm = result.Term
		n.stepDown()
		return
	}

	// update matchIndex and nextIndex for this peer
	if result.Success && len(entries) > 0 {
		lastEntry := entries[len(entries)-1]
		n.matchIndex[peer] = lastEntry.Index
		n.nextIndex[peer] = lastEntry.Index + 1
		n.maybeCommit()
	}
}

// checks if any new entries can be committed based on matchIndex. an entry is committed once a majority of nodes have it
func (n *Node) maybeCommit() {
	quorum := n.config.Quorum()

	for i := len(n.log) - 1; i >= 0; i-- {
		entry := n.log[i]
		if entry.Index <= n.commitIndex {
			break
		}
		if entry.Term != n.currentTerm {
			continue
		}

		// count how many nodes have this entry
		count := 1 // count ourselves
		for _, peer := range n.config.Peers {
			if peer == n.id {
				continue
			}
			if n.matchIndex[peer] >= entry.Index {
				count++
			}
		}

		if count >= quorum {
			// commit everything up to this index
			for j := 0; j <= i; j++ {
				if n.log[j].Index > n.commitIndex {
					n.applyEntry(n.log[j])
					n.commitIndex = n.log[j].Index
				}
			}
			break
		}
	}
}

func (n *Node) stepDown() {
	if n.role == Leader {
		close(n.stopHeartbeat)
		n.stopHeartbeat = make(chan struct{})
	}
	n.role = Follower
	n.currentLeader = ""
}