raft

package module
v0.3.11 Latest Latest
Warning

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

Go to latest
Published: Oct 5, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

README

raft

CI

Raft consensus for Go, built for state machines that keep their own data on disk.

go get github.com/acneism/raft

Features

  • Deterministic core. No IO, no goroutines, no clock; elections with PreVote and CheckQuorum, pipelined replication, snapshots.
  • Parallel leader fsync. AppendEntries leave before the leader's own fsync, so a round takes the longer of the two fsyncs, not their sum.
  • Segmented log. CRC-chained records, logical compaction to any index, crash-safe HardState; Linux and Windows.
  • Transport. Framed TCP with a binary codec, mutual TLS, resumable snapshot transfer; peers negotiate the protocol version, so a cluster can be upgraded one node at a time. Failed dials, rejected connections and TLS errors go to the node's Logger, at most once per 30 seconds for each peer. With ClusterID set, a node refuses connections from nodes of another cluster; nodes without a ClusterID are still accepted, with a warning, so it can be turned on one node at a time.
  • Node. Propose returns the index and term without waiting for IO, Wait always completes, leadership events are never lost, and replay after a restart starts at the state machine's durable index. Status reports the log bounds, the latest snapshot and, on the leader, the replication progress of every member.
  • Snapshots on demand. The log is compacted behind the state machine's durable index without a snapshot; a snapshot is taken only when a follower falls behind the compacted log. KeepSnapshots sets how many of the newest stay on disk, 2 by default, and a snapshot whose files are gone is taken again.
  • Linearizable reads. ReadIndex confirms leadership with one heartbeat round instead of a log write, batches concurrent reads and works on followers. With LeaseReads the leader answers from its lease without a round, assuming node clocks run at rates that differ by at most MaxClockDrift.
  • Leadership transfer. TransferLeadership brings the target up to date and hands leadership over without waiting for an election timeout. Catching up and the election get an election timeout each, and a failed transfer says which of them ran out or that another node won.
  • Membership changes. A new node joins as a learner, catches up from the log or a snapshot and is promoted to voter; members are removed one change at a time. The configuration, with member addresses, travels in the log.
  • Simulation. A deterministic cluster simulator with network, disk and clock faults checks the Raft safety invariants after every step and the client history for linearizability with Porcupine.

Usage

Implement a state machine:

type StateMachine interface {
	Apply(entries []raft.Entry) error
	DurableIndex() uint64
	Snapshot(dir string) (raft.SnapshotMeta, error)
	Restore(src node.SnapshotSource) error
}

DurableIndex is the last index the state machine has on disk: the node replays the log after it on restart and compacts the log up to it, keeping TrailingEntries more. It may lag Apply, for example until a background fsync: the node reads it after every applied batch and once per election timeout. Snapshot writes the applied state into dir and returns its index, which must not be below DurableIndex.

Run a node:

n, err := node.Open(node.Config{
	ID:           "n1",
	Dir:          "data/n1",
	Peers:        map[raft.NodeID]string{"n1": "10.0.0.1:7000", "n2": "10.0.0.2:7000", "n3": "10.0.0.3:7000"},
	StateMachine: fsm,
	PreVote:      true,
	CheckQuorum:  true,
})
if err != nil {
	return err
}
defer n.Close()

p, err := n.Propose(command)
if err != nil {
	return err
}
err = n.Wait(ctx, p)

Wait returns nil once the entry is applied, node.ErrLost if another entry took its index, node.ErrUnknown if leadership was lost first, and node.ErrClosed after Close. node.HasState(dir) tells a restart from a first start before Open, for example to refuse to bootstrap over existing data.

Read linearizably on any node:

index, err := n.ReadIndex(ctx)
if err != nil {
	return err
}
if err := n.WaitApplied(ctx, index); err != nil {
	return err
}
value := fsm.Get(key)

ReadIndex returns node.ErrNotLeader when no leader could confirm the read; retrying is safe.

Add a node: start it with Join: true and Peers listing at least itself, then on the leader:

err := n.AddLearner(ctx, "n4", "10.0.0.4:7000")
err = n.Promote(ctx, "n4")
err = n.Remove(ctx, "n1")

The members dial the new node once they add it, and the node learns their addresses from the connections it accepts before it has a configuration. Promote waits until the learner has caught up.

A removed node stops by itself, and Err returns node.ErrRemoved: the members that applied the removal drop its connections and tell it on its next dial. They remember removed nodes until they restart, so a node that was down during its removal and comes back after that is refused but not told; stop it yourself.

Demo

A key-value node over HTTP. Run a three-node cluster in one process:

go run ./cmd/node --all --peers 1=127.0.0.1:8001,2=127.0.0.1:8002,3=127.0.0.1:8003
curl -X PUT http://localhost:9001/kv/greeting -d 'hello'
curl -X POST http://localhost:9001/kv/greeting/append -d ', world'
curl 'http://localhost:9001/kv/greeting?consistent=1'

Or one node per process: the same --peers everywhere and --id 1, --id 2, --id 3. The HTTP API listens on the Raft port plus 1000. A plain GET reads the local state machine; ?consistent=1 confirms the read with ReadIndex first.

--lease-reads serves consistent reads from the leader's lease.

POST /members/{id}?addr=host:port, POST /members/{id}/promote and DELETE /members/{id} change the membership on the leader; start a new node with --join. POST /transfer/{id} hands leadership over.

Testing

go test ./...
go test ./sim -run TestSim$ -timeout 60m -args -sim.steps=10000000 -sim.seeds=8
go test ./cmd/node -run TestKillCycles -timeout 60m -args -kill.cycles=200
go test ./sim -run TestSimLinearizable$ -timeout 60m -args -sim.lin.steps=10000000 -sim.seeds=8
go test ./cmd/node -run TestLinearizability -timeout 60m -args -lin.duration=5m
go test ./cmd/node -run TestMembership -timeout 60m -args -member.duration=5m

License

Apache License 2.0, see LICENSE.

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrConfChangePending = errors.New("raft: a configuration change is in progress")
	ErrConfChangeInvalid = errors.New("raft: invalid configuration change")
)
View Source
var (
	ErrNotLeader                      = errors.New("raft: not leader")
	ErrCompacted                      = errors.New("raft: requested index is compacted")
	ErrUnavailable                    = errors.New("raft: requested entry is unavailable")
	ErrSnapshotTemporarilyUnavailable = errors.New("raft: snapshot is temporarily unavailable")
)
View Source
var ErrTransferTarget = errors.New("raft: leadership transfer target is not a voter")

Functions

func AppendConfState added in v0.3.0

func AppendConfState(b []byte, cs ConfState) []byte

Types

type ConfChange added in v0.3.0

type ConfChange struct {
	Type ConfChangeType
	Node NodeID
	Addr string
}

type ConfChangeType added in v0.3.0

type ConfChangeType uint8
const (
	ConfAddLearner ConfChangeType = iota + 1
	ConfPromote
	ConfRemove
)

type ConfState

type ConfState struct {
	Voters   []NodeID
	Learners []NodeID
	Addrs    map[NodeID]string
}

func DecodeConfState added in v0.3.0

func DecodeConfState(b []byte) (ConfState, []byte, error)

func (ConfState) Apply added in v0.3.0

func (cs ConfState) Apply(cc ConfChange) (ConfState, error)

func (ConfState) Clone added in v0.3.0

func (cs ConfState) Clone() ConfState

func (ConfState) IsLearner added in v0.3.0

func (cs ConfState) IsLearner(id NodeID) bool

func (ConfState) IsVoter added in v0.3.0

func (cs ConfState) IsVoter(id NodeID) bool

type Config

type Config struct {
	ID                       NodeID
	ElectionTick             int
	HeartbeatTick            int
	Storage                  Storage
	Applied                  uint64
	MaxSizePerMsg            uint64
	MaxInflightMsgs          int
	MaxInflightBytes         uint64
	MaxCommittedSizePerReady uint64
	PreVote                  bool
	CheckQuorum              bool
	LeaseTicks               int
	Rand                     *rand.Rand
}

type Core

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

func New

func New(cfg Config) (*Core, error)

func (*Core) Advance

func (c *Core) Advance(rd Ready)

func (*Core) AdvanceApply added in v0.2.1

func (c *Core) AdvanceApply(rd Ready)

func (*Core) AdvancePersist added in v0.2.1

func (c *Core) AdvancePersist(rd Ready)

func (*Core) ConfState added in v0.3.0

func (c *Core) ConfState() ConfState

func (*Core) DropLease added in v0.3.0

func (c *Core) DropLease()

func (*Core) HasReady

func (c *Core) HasReady() bool

func (*Core) Match added in v0.3.0

func (c *Core) Match(id NodeID) (uint64, bool)

func (*Core) Progress added in v0.3.6

func (c *Core) Progress() map[NodeID]PeerProgress

func (*Core) Propose

func (c *Core) Propose(data []byte) (index, term uint64, err error)

func (*Core) ProposeConfChange added in v0.3.0

func (c *Core) ProposeConfChange(cc ConfChange) (index, term uint64, err error)

func (*Core) ReadIndex added in v0.3.0

func (c *Core) ReadIndex() (uint64, error)

func (*Core) Ready

func (c *Core) Ready() Ready

func (*Core) ReportSnapshot

func (c *Core) ReportSnapshot(id NodeID, failed bool)

func (*Core) ReportUnreachable

func (c *Core) ReportUnreachable(id NodeID)

func (*Core) Status

func (c *Core) Status() Status

func (*Core) Step

func (c *Core) Step(m Message) error

func (*Core) Tick

func (c *Core) Tick()

func (*Core) TransferLeadership added in v0.3.0

func (c *Core) TransferLeadership(to NodeID) error

type Entry

type Entry struct {
	Index uint64
	Term  uint64
	Type  EntryType
	Data  []byte
}

type EntryType

type EntryType uint8
const (
	EntryNormal EntryType = iota
	EntryNoop
	EntryConfChange
)

type HardState

type HardState struct {
	Term   uint64
	Vote   NodeID
	Commit uint64
}

func (HardState) IsEmpty

func (h HardState) IsEmpty() bool

type MemoryStorage

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

func NewMemoryStorage

func NewMemoryStorage(cs ConfState) *MemoryStorage

func (*MemoryStorage) Append

func (s *MemoryStorage) Append(ents []Entry) error

func (*MemoryStorage) ApplySnapshot

func (s *MemoryStorage) ApplySnapshot(snap SnapshotMeta) error

func (*MemoryStorage) Compact

func (s *MemoryStorage) Compact(i uint64) error

func (*MemoryStorage) CreateSnapshot

func (s *MemoryStorage) CreateSnapshot(i uint64, cs ConfState) (SnapshotMeta, error)

func (*MemoryStorage) Entries

func (s *MemoryStorage) Entries(lo, hi, maxSize uint64) ([]Entry, error)

func (*MemoryStorage) FirstIndex

func (s *MemoryStorage) FirstIndex() (uint64, error)

func (*MemoryStorage) InitialState

func (s *MemoryStorage) InitialState() (HardState, ConfState, error)

func (*MemoryStorage) LastIndex

func (s *MemoryStorage) LastIndex() (uint64, error)

func (*MemoryStorage) SetConf added in v0.3.0

func (s *MemoryStorage) SetConf(cs ConfState)

func (*MemoryStorage) SetHardState

func (s *MemoryStorage) SetHardState(hs HardState)

func (*MemoryStorage) Snapshot

func (s *MemoryStorage) Snapshot() (SnapshotMeta, error)

func (*MemoryStorage) Term

func (s *MemoryStorage) Term(i uint64) (uint64, error)

type Message

type Message struct {
	Type       MessageType
	From       NodeID
	To         NodeID
	Term       uint64
	Index      uint64
	LogTerm    uint64
	Commit     uint64
	Entries    []Entry
	Snapshot   *SnapshotMeta
	Reject     bool
	RejectHint uint64
	Transfer   bool
}

type MessageType

type MessageType uint8
const (
	MsgApp MessageType = iota + 1
	MsgAppResp
	MsgVote
	MsgVoteResp
	MsgPreVote
	MsgPreVoteResp
	MsgHeartbeat
	MsgHeartbeatResp
	MsgSnap
	MsgReadIndex
	MsgReadIndexResp
	MsgTimeoutNow
)

func (MessageType) String

func (t MessageType) String() string

type NodeID

type NodeID string
const None NodeID = ""

type PeerProgress added in v0.3.6

type PeerProgress struct {
	Match, Next     uint64
	State           ProgressState
	PendingSnapshot uint64
	Learner         bool
	RecentActive    bool
	Paused          bool
}

type Progress

type Progress struct {
	Match, Next     uint64
	State           ProgressState
	PendingSnapshot uint64
	RecentActive    bool

	IsLearner bool
	// contains filtered or unexported fields
}

type ProgressState

type ProgressState uint8
const (
	ProgressProbe ProgressState = iota
	ProgressReplicate
	ProgressSnapshot
)

func (ProgressState) String added in v0.3.6

func (s ProgressState) String() string

type ReadState added in v0.3.0

type ReadState struct {
	ID     uint64
	Index  uint64
	Failed bool
}

type Ready

type Ready struct {
	SoftState            *SoftState
	HardState            HardState
	Entries              []Entry
	Snapshot             *SnapshotMeta
	Committed            []Entry
	Messages             []Message
	MessagesAfterPersist []Message
	ReadStates           []ReadState
	MustSync             bool
}

type SnapshotMeta

type SnapshotMeta struct {
	Index uint64
	Term  uint64
	Conf  ConfState
}

type SoftState

type SoftState struct {
	Lead  NodeID
	State StateType
}

type StateType

type StateType uint8
const (
	StateFollower StateType = iota
	StateCandidate
	StateLeader
	StatePreCandidate
)

func (StateType) String

func (s StateType) String() string

type Status

type Status struct {
	ID NodeID
	HardState
	SoftState
	Applied        uint64
	LastIndex      uint64
	LeaderReady    bool
	LeadTransferee NodeID
	TransferSent   bool
}

type Storage

type Storage interface {
	InitialState() (HardState, ConfState, error)
	Entries(lo, hi, maxSize uint64) ([]Entry, error)
	Term(i uint64) (uint64, error)
	LastIndex() (uint64, error)
	FirstIndex() (uint64, error)
	Snapshot() (SnapshotMeta, error)
}

Directories

Path Synopsis
cmd
node command
internal
fsx

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL