raft

Documentation

Index

• func IsInternalError(err error) bool
• func MakeInternalError(err error) error
• type AbortInstruction
• type ApplyInstruction
• type Driver
  • func NewDriver(nodeID NodeID, stateCh <-chan Instruction, msgCh chan<- *raftpb.Message) *Driver
  • func (d *Driver) Apply(state State, entry *raftpb.Entry) error
  • func (d *Driver) ApplyLog(state State, log *Log) error
  • func (d *Driver) Drive(state State) error
• type Index
• type Instruction
• type InternalError
  • func (e *InternalError) Error() string
  • func (e *InternalError) Unwrap() error
• type Log
  • func NewLog(engine storage.Engine, sync bool) (*Log, error)
  • func (l *Log) Append(term Term, command []byte) (Index, error)
  • func (l *Log) Commit(index Index) error
  • func (l *Log) CommitIndex() (Index, Term)
  • func (l *Log) Get(index Index) (*raftpb.Entry, error)
  • func (l *Log) GetTerm() (_ Term, votedFor NodeID, _ error)
  • func (l *Log) Has(index Index, term Term) (bool, error)
  • func (l *Log) LastIndex() (Index, Term)
  • func (l *Log) Scan(start, end Index) (itertools.Iterator[*raftpb.Entry], error)
  • func (l *Log) SetTerm(term Term, votedFor NodeID) error
  • func (l *Log) Splice(entries []*raftpb.Entry) (Index, error)
  • func (l *Log) Status() (*storagepb.EngineStatus, error)
• type Node
  • func NewNode(id NodeID, peers sets.Set[NodeID], log *Log, state State, ...) (*Node, error)
  • func (n *Node) ID() NodeID
  • func (n *Node) Role() Role
  • func (n *Node) Step(m *raftpb.Message) error
  • func (n *Node) Tick() error
• type NodeID
• type Notify
• type NotifyInstruction
• type Query
  • func (q *Query) Ready() bool
• type QueryInstruction
• type RequestID
• type Role
• type Server
  • func NewServer(id NodeID, log *Log, state State, peerConns map[NodeID]*grpc.ClientConn) (*Server, error)
  • func (s *Server) Mutate(ctx context.Context, command []byte) ([]byte, error)
  • func (s *Server) Query(ctx context.Context, command []byte) ([]byte, error)
  • func (s *Server) SendMessages(stream raftpb.Raft_SendMessagesServer) error
  • func (s *Server) Serve(ctx context.Context) error
  • func (s *Server) Status(ctx context.Context) (*raftpb.Status, error)
• type State
• type StatusInstruction
• type Term
• type VoteInstruction

Functions

func IsInternalError

func IsInternalError(err error) bool

func MakeInternalError

func MakeInternalError(err error) error

Types

type AbortInstruction

type AbortInstruction struct{}

AbortInstruction is the instruction to abort any pending operations.

type ApplyInstruction

type ApplyInstruction struct {
	Entry *raftpb.Entry
}

ApplyInstruction is the instruction to apply a log entry.

type Driver

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

Driver is a driver for driving the raft state machine. Taking operations from the stateCh and sending results via the nodeCh.

func NewDriver

func NewDriver(nodeID NodeID, stateCh <-chan Instruction, msgCh chan<- *raftpb.Message) *Driver

func (*Driver) Apply

func (d *Driver) Apply(state State, entry *raftpb.Entry) error

func (*Driver) ApplyLog

func (d *Driver) ApplyLog(state State, log *Log) error

func (*Driver) Drive

func (d *Driver) Drive(state State) error

type Index

type Index = uint64

type Instruction

type Instruction interface {
	// contains filtered or unexported methods
}

Instruction is the instructions sent to driver.

type InternalError

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

InternalError is an error that is used to wrap internal errors. It is used to distinguish internal errors from user errors.

func (*InternalError) Error

func (e *InternalError) Error() string

func (*InternalError) Unwrap

func (e *InternalError) Unwrap() error

type Log

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

Log is a persistent log of entries. It is the source of truth for the state machine. Each entry is identified by its index and term.

func NewLog

func NewLog(engine storage.Engine, sync bool) (*Log, error)

func (*Log) Append

func (l *Log) Append(term Term, command []byte) (Index, error)

func (*Log) Commit

func (l *Log) Commit(index Index) error

func (*Log) CommitIndex

func (l *Log) CommitIndex() (Index, Term)

func (*Log) Get

func (l *Log) Get(index Index) (*raftpb.Entry, error)

func (*Log) GetTerm

func (l *Log) GetTerm() (_ Term, votedFor NodeID, _ error)

func (*Log) Has

func (l *Log) Has(index Index, term Term) (bool, error)

func (*Log) LastIndex

func (l *Log) LastIndex() (Index, Term)

func (*Log) Scan

func (l *Log) Scan(start, end Index) (itertools.Iterator[*raftpb.Entry], error)

func (*Log) SetTerm

func (l *Log) SetTerm(term Term, votedFor NodeID) error

func (*Log) Splice

func (l *Log) Splice(entries []*raftpb.Entry) (Index, error)

Splice splices a set log entries into the log and returns the index of the last entry.

The entries must be contiguous, and the first entry's index must be at most lastIndex+1 and at least commitIndex+1. When splicing, the log will overwrite any existing entries and truncate any existing entries after the last spliced entry.

func (*Log) Status

func (l *Log) Status() (*storagepb.EngineStatus, error)

type Node

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

func NewNode

func NewNode(
	id NodeID,
	peers sets.Set[NodeID],
	log *Log,
	state State,
	msgCh chan<- *raftpb.Message,
) (*Node, error)

func (*Node) ID

func (n *Node) ID() NodeID

func (*Node) Role

func (n *Node) Role() Role

func (*Node) Step

func (n *Node) Step(m *raftpb.Message) error

func (*Node) Tick

func (n *Node) Tick() error

type NodeID

type NodeID = uint64

type Notify

type Notify struct {
	ID     RequestID
	NodeID NodeID
	Index  Index
}

type NotifyInstruction

type NotifyInstruction struct {
	ID     RequestID
	NodeID NodeID
	Index  Index
}

NotifyInstruction is the instruction to notify the given node with the result of applying the entry at the given index.

type Query

type Query struct {
	ID      RequestID
	NodeID  NodeID
	Command []byte
	Term    Term
	Index   Index
	Quorum  int
	Votes   *hashset.Set[NodeID]
}

func (*Query) Ready

func (q *Query) Ready() bool

type QueryInstruction

type QueryInstruction struct {
	ID      RequestID
	NodeID  NodeID
	Command RequestID
	Term    Term
	Index   Index
	Quorum  int
}

QueryInstruction is the instruction to query the state machine when the given term and index has been confirmed by vote.

type RequestID

type RequestID = []byte

type Role

type Role uint8

const (
	Follower Role = iota
	Candidate
	Leader
)

type Server

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

func NewServer

func NewServer(
	id NodeID,
	log *Log,
	state State,
	peerConns map[NodeID]*grpc.ClientConn,
) (*Server, error)

func (*Server) Mutate

func (s *Server) Mutate(ctx context.Context, command []byte) ([]byte, error)

func (*Server) Query

func (s *Server) Query(ctx context.Context, command []byte) ([]byte, error)

func (*Server) SendMessages

func (s *Server) SendMessages(stream raftpb.Raft_SendMessagesServer) error

func (*Server) Serve

func (s *Server) Serve(ctx context.Context) error

func (*Server) Status

func (s *Server) Status(ctx context.Context) (*raftpb.Status, error)

type State

type State interface {
	// AppliedIndex returns the last applied index.
	AppliedIndex() Index
	// Apply applies a log to the state machine. Apply should
	// be deterministic and idempotent.
	//
	// If InternalError is returned, the raft node will be terminated.
	// Any other error is considered applied and returned to the caller.
	Apply(entry *raftpb.Entry) ([]byte, error)
	// Query queries the state machine with the given command.
	Query(command []byte) ([]byte, error)
}

State represents a raft state machine. The caller of raft is responsible for implementing the state machine.

type StatusInstruction

type StatusInstruction struct {
	ID     RequestID
	NodeID NodeID
	Status *raftpb.Status
}

StatusQuery is the instruction to extend the given server status and return it the the given node.

type Term

type Term = uint64

type VoteInstruction

type VoteInstruction struct {
	Term   Term
	Index  Index
	NodeID NodeID
}

VoteInstruction is the instruction to vote for the given term and index.