raft

package
v0.0.0-...-e8d51d3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 9, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AppendEntriesArgs 

type AppendEntriesArgs struct {
	Term     int
	LeaderId int

	PrevLogIndex int
	PrevLogTerm  int
	Entries      []LogEntry
	LeaderCommit int
}

type AppendEntriesReply 

type AppendEntriesReply struct {
	Term    int
	Success bool

	// Conflict-index optimisation fields.
	ConflictIndex int
	ConflictTerm  int
}

type CMState 

type CMState int
const (
	Follower CMState = iota
	Candidate
	Leader
	Dead
)

func (CMState) String 

func (s CMState) String() string

type CommitEntry 

type CommitEntry struct {
	Command any
	Index   int
	Term    int
}

CommitEntry is the data reported by Raft to the commit channel.

type ConfigChangeEntry 

type ConfigChangeEntry struct {
	Type   ConfigChangeType
	NodeId int
}

type ConfigChangeType 

type ConfigChangeType int
const (
	AddNode ConfigChangeType = iota
	RemoveNode
)

type ConsensusModule 

type ConsensusModule struct {
	// contains filtered or unexported fields
}

ConsensusModule (CM) implements a single node of Raft consensus.

func NewConsensusModule 

func NewConsensusModule(
	id int,
	peerIds []int,
	server *Server,
	storage Storage,
	ready <-chan any,
	commitChan chan<- CommitEntry,
	logger *slog.Logger,
) *ConsensusModule

func (*ConsensusModule) AddPeer 

func (cm *ConsensusModule) AddPeer(nodeId int) bool

AddPeer proposes adding nodeId to the cluster. Must be called on the leader. Returns false if not leader or a config change is already pending.

func (*ConsensusModule) AppendEntries 

func (cm *ConsensusModule) AppendEntries(args AppendEntriesArgs, reply *AppendEntriesReply) error

func (*ConsensusModule) InstallSnapshot 

func (cm *ConsensusModule) InstallSnapshot(lastIndex, lastTerm int, snapshotData []byte)

InstallSnapshot is called by the application layer (or test harness) to compact the log. snapshotData is an opaque blob that fully encodes the application state as of lastIndex/lastTerm.

func (*ConsensusModule) InstallSnapshotRPC 

func (cm *ConsensusModule) InstallSnapshotRPC(args InstallSnapshotArgs, reply *InstallSnapshotReply) error

InstallSnapshotRPC is the RPC handler called on a follower by the leader when the follower's nextIndex has fallen behind the leader's snapshot point.

func (*ConsensusModule) RemovePeer 

func (cm *ConsensusModule) RemovePeer(nodeId int) bool

RemovePeer proposes removing nodeId from the cluster. Must be called on the leader.

func (*ConsensusModule) Report 

func (cm *ConsensusModule) Report() (id int, term int, isLeader bool)

func (*ConsensusModule) RequestVote 

func (cm *ConsensusModule) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) error

func (*ConsensusModule) SnapshotDone 

func (cm *ConsensusModule) SnapshotDone() <-chan struct{}

SnapshotDone returns a channel that is closed when the CM is stopped. Use it to terminate goroutines that drain SnapshotReady().

func (*ConsensusModule) SnapshotReady 

func (cm *ConsensusModule) SnapshotReady() <-chan SnapshotEntry

SnapshotReady returns the channel on which snapshot notifications arrive. The application should drain this channel and restore its state machine. The channel is never closed; watch SnapshotDone() to know when to stop.

func (*ConsensusModule) Stop 

func (cm *ConsensusModule) Stop()

func (*ConsensusModule) Submit 

func (cm *ConsensusModule) Submit(command any) SubmitResult

Submit submits a new command to the CM.

type FileStorage 

type FileStorage struct {
	// contains filtered or unexported fields
}

FileStorage is a durable Storage implementation that writes each key to its own file under a directory. Writes are atomic: data is first flushed to a temp file in the same directory, then renamed over the target, so a crash mid-write never leaves a partially-written value behind.

func NewFileStorage 

func NewFileStorage(dir string) (*FileStorage, error)

NewFileStorage creates (or opens) a FileStorage rooted at dir. The directory is created with 0700 permissions if it does not exist.

func (*FileStorage) Get 

func (fs *FileStorage) Get(key string) ([]byte, bool)

Get retrieves the value for key. Returns (nil, false) if the key has never been Set; panics on unexpected I/O errors.

func (*FileStorage) HasData 

func (fs *FileStorage) HasData() bool

HasData returns true if the storage directory contains at least one .dat file, meaning at least one Set has been persisted to disk.

func (*FileStorage) Set 

func (fs *FileStorage) Set(key string, value []byte)

Set writes value for key atomically. Panics (like MapStorage) on I/O error so callers don't have to check — a storage failure is fatal for Raft anyway.

type Harness 

type Harness struct {
	// contains filtered or unexported fields
}

func NewHarness 

func NewHarness(t *testing.T, n int) *Harness

NewHarness creates a new test Harness, initialized with n servers connected to each other.

func (*Harness) CheckCommitted 

func (h *Harness) CheckCommitted(cmd int) (nc int, index int)

CheckCommitted verifies that all connected servers have cmd committed with the same index. It also verifies that all commands *before* cmd in the commit sequence match. For this to work properly, all commands submitted to Raft should be unique positive ints. Returns the number of servers that have this command committed, and its log index.

func (*Harness) CheckCommittedAtLeastN 

func (h *Harness) CheckCommittedAtLeastN(cmd int, n int)

CheckCommittedAtLeastN verifies that cmd was committed by at least n connected servers. Use this when a newly-joined server may have replayed historical entries, making the exact count uncertain.

func (*Harness) CheckCommittedIgnoringSnapshot 

func (h *Harness) CheckCommittedIgnoringSnapshot(cmd int) (nc int, index int)

CheckCommittedIgnoringSnapshot is like CheckCommitted but does NOT require that all connected servers have the same length commits slice. This is needed after a snapshot: a restarted server will have had its log replaced by the snapshot and will only report the post-snapshot commits, while servers that were never restarted show the full history.

Instead of a length equality check it simply looks for cmd in each connected server's commits slice and verifies that wherever it appears it has the same index.

func (*Harness) CheckCommittedN 

func (h *Harness) CheckCommittedN(cmd int, n int)

CheckCommittedN verifies that cmd was committed by exactly n connected servers.

func (*Harness) CheckNoLeader 

func (h *Harness) CheckNoLeader()

CheckNoLeader checks that no connected server considers itself the leader.

func (*Harness) CheckNoSnapshotDelivered 

func (h *Harness) CheckNoSnapshotDelivered(id int)

CheckNoSnapshotDelivered asserts that server id has received no snapshots yet.

func (*Harness) CheckNotCommitted 

func (h *Harness) CheckNotCommitted(cmd int)

CheckNotCommitted verifies that no command equal to cmd has been committed by any of the active servers yet.

func (*Harness) CheckSingleLeader 

func (h *Harness) CheckSingleLeader() (int, int)

CheckSingleLeader checks that only a single server thinks it's the leader. Returns the leader's id and term. It retries several times if no leader is identified yet.

func (*Harness) CheckSnapshotDelivered 

func (h *Harness) CheckSnapshotDelivered(id int, wantIndex int) SnapshotEntry

snapshot whose Index equals wantIndex, then returns that snapshot. It fails the test if no such snapshot arrives within ~5 seconds.

func (*Harness) CrashPeer 

func (h *Harness) CrashPeer(id int)

CrashPeer "crashes" a server by disconnecting it from all peers and then asking it to shut down. We're not going to use the same server instance again, but its storage is retained.

func (*Harness) DisconnectPeer 

func (h *Harness) DisconnectPeer(id int)

DisconnectPeer disconnects a server from all other servers in the cluster.

func (*Harness) PeerDontDropCalls 

func (h *Harness) PeerDontDropCalls(id int)

PeerDontDropCalls instructs peer `id` to stop dropping calls.

func (*Harness) PeerDropCallsAfterN 

func (h *Harness) PeerDropCallsAfterN(id int, n int)

PeerDropCallsAfterN instructs peer `id` to drop calls after the next `n` are made.

func (*Harness) ReconnectPeer 

func (h *Harness) ReconnectPeer(id int)

ReconnectPeer connects a server to all other servers in the cluster.

func (*Harness) RestartPeer 

func (h *Harness) RestartPeer(id int)

RestartPeer "restarts" a server by creating a new Server instance and giving it the appropriate storage, reconnecting it to peers.

func (*Harness) Shutdown 

func (h *Harness) Shutdown()

Shutdown shuts down all the servers in the harness and waits for them to stop running.

func (*Harness) SubmitToServer 

func (h *Harness) SubmitToServer(serverId int, cmd any) int

SubmitToServer submits the command to serverId.

func (*Harness) WaitForStableCommits 

func (h *Harness) WaitForStableCommits(n int)

WaitForStableCommits polls until every connected server has committed at least n integer entries, or until ~2 s elapses. It does not fail on its own — callers should use CheckCommittedN / CheckCommittedAtLeastN after returning to verify consistency. Useful under RAFT_UNRELIABLE_RPC where commit-index heartbeats can be delayed or dropped.

type InstallSnapshotArgs 

type InstallSnapshotArgs struct {
	Term              int
	LeaderId          int
	LastIncludedIndex int
	LastIncludedTerm  int
	Data              []byte
}

type InstallSnapshotReply 

type InstallSnapshotReply struct {
	Term int
}

type LogEntry 

type LogEntry struct {
	Command any
	Term    int
}

type MapStorage 

type MapStorage struct {
	// contains filtered or unexported fields
}

MapStorage is a simple in-memory implementation of Storage for testing.

func NewMapStorage 

func NewMapStorage() *MapStorage

func (*MapStorage) Get 

func (ms *MapStorage) Get(key string) ([]byte, bool)

func (*MapStorage) HasData 

func (ms *MapStorage) HasData() bool

func (*MapStorage) Set 

func (ms *MapStorage) Set(key string, value []byte)

type RPCProxy 

type RPCProxy struct {
	// contains filtered or unexported fields
}

func NewProxy 

func NewProxy(cm *ConsensusModule) *RPCProxy

func (*RPCProxy) AppendEntries 

func (rpp *RPCProxy) AppendEntries(args AppendEntriesArgs, reply *AppendEntriesReply) error

func (*RPCProxy) Call 

func (rpp *RPCProxy) Call(peer *rpc.Client, method string, args any, reply any) error

func (*RPCProxy) DontDropCalls 

func (rpp *RPCProxy) DontDropCalls()

func (*RPCProxy) DropCallsAfterN 

func (rpp *RPCProxy) DropCallsAfterN(n int)

func (*RPCProxy) InstallSnapshotRPC 

func (rpp *RPCProxy) InstallSnapshotRPC(args InstallSnapshotArgs, reply *InstallSnapshotReply) error

func (*RPCProxy) RequestVote 

func (rpp *RPCProxy) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) error

type RequestVoteArgs 

type RequestVoteArgs struct {
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

type RequestVoteReply 

type RequestVoteReply struct {
	Term        int
	VoteGranted bool
}

type Server 

type Server struct {
	// contains filtered or unexported fields
}

func NewServer 

func NewServer(serverId int, peerIds []int, storage Storage, ready <-chan any, commitChan chan<- CommitEntry, logger *slog.Logger) *Server

func (*Server) AddPeer 

func (s *Server) AddPeer(nodeId int, addr net.Addr) bool

AddPeer adds nodeId to the cluster via a ConfigChange log entry. addr is the RPC address of the new node; this server will connect to it.

func (*Server) Call 

func (s *Server) Call(id int, serviceMethod string, args any, reply any) error

func (*Server) ConnectToPeer 

func (s *Server) ConnectToPeer(peerId int, addr net.Addr) error

func (*Server) DisconnectAll 

func (s *Server) DisconnectAll()

func (*Server) DisconnectPeer 

func (s *Server) DisconnectPeer(peerId int) error

func (*Server) GetListenAddr 

func (s *Server) GetListenAddr() net.Addr

func (*Server) InstallSnapshot 

func (s *Server) InstallSnapshot(lastIndex, lastTerm int, data []byte)

InstallSnapshot tells the ConsensusModule to compact the log up to lastIndex / lastTerm using the provided application snapshot data. This is the application-initiated path (as opposed to the leader-push path via InstallSnapshotRPC).

func (*Server) IsLeader 

func (s *Server) IsLeader() bool

func (*Server) Proxy 

func (s *Server) Proxy() *RPCProxy

func (*Server) RemovePeer 

func (s *Server) RemovePeer(nodeId int) bool

RemovePeer removes nodeId from the cluster via a ConfigChange log entry.

func (*Server) Serve 

func (s *Server) Serve()

func (*Server) Shutdown 

func (s *Server) Shutdown()

func (*Server) SnapshotDone 

func (s *Server) SnapshotDone() <-chan any

SnapshotDone returns a channel that is closed when this Server is shut down via Shutdown(). It is the same channel that terminates the RPC accept loop, so callers are guaranteed to observe the close at most once after Shutdown returns. Use it to terminate goroutines that drain SnapshotReady().

func (*Server) SnapshotReady 

func (s *Server) SnapshotReady() <-chan SnapshotEntry

SnapshotReady returns the channel on which the ConsensusModule delivers SnapshotEntry values when a snapshot is installed by the leader. The application must drain this channel and restore its state machine.

func (*Server) Submit 

func (s *Server) Submit(cmd any) SubmitResult

type SnapshotEntry 

type SnapshotEntry struct {
	Data  []byte
	Index int
	Term  int
}

SnapshotEntry is sent on the commit channel when a snapshot is installed. The application must restore its state from Data and discard all previously applied entries up through Index.

type Storage 

type Storage interface {
	Set(key string, value []byte)

	Get(key string) ([]byte, bool)

	// HasData returns true iff any Sets were made on this Storage.
	HasData() bool
}

Storage is an interface implemented by stable storage providers.

type SubmitResult 

type SubmitResult struct {
	Index      int
	IsLeader   bool
	LeaderHint int
}

SubmitResult is returned by Submit.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.