raft

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Apr 9, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type DiscoveredPeer 

type DiscoveredPeer struct {
	ID           string
	Address      string
	Region       string
	Version      string
	Capabilities core.NodeCapabilities
	LastSeen     time.Time
	LastGossip   time.Time
	GossipCount  int
	IsStatic     bool // Configured vs discovered
}

DiscoveredPeer represents a peer discovered through gossip/mDNS

type Discovery 

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

Discovery handles peer discovery via gossip and mDNS The jackals howl to find each other in the Necropolis

func NewDiscovery 

func NewDiscovery(config core.RaftConfig, logger *slog.Logger) (*Discovery, error)

NewDiscovery creates a new discovery service

func (*Discovery) GetPeers 

func (d *Discovery) GetPeers() []DiscoveredPeer

GetPeers returns all known peers

func (*Discovery) RegisterPeerCallback 

func (d *Discovery) RegisterPeerCallback(onDiscovered func(core.RaftPeer), onLost func(string))

RegisterPeerCallback sets the callback for discovered peers

func (*Discovery) Start 

func (d *Discovery) Start() error

Start starts the discovery service

func (*Discovery) Stop 

func (d *Discovery) Stop() error

Stop stops the discovery service

type DistributionStats 

type DistributionStats struct {
	TotalSouls       int            `json:"total_souls"`
	TotalNodes       int            `json:"total_nodes"`
	HealthyNodes     int            `json:"healthy_nodes"`
	Strategy         string         `json:"strategy"`
	Revision         uint64         `json:"revision"`
	NodeDistribution map[string]int `json:"node_distribution"`
}

DistributionStats holds distribution statistics

type Distributor 

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

Distributor handles assignment of souls (monitors) to nodes The Pharaoh decides which priest judges which soul

func NewDistributor 

func NewDistributor(nodeID, region string, strategy core.DistributionStrategy) *Distributor

NewDistributor creates a new distributor

func (*Distributor) AddNode 

func (d *Distributor) AddNode(node *core.NodeInfo)

AddNode adds a node to the distribution pool

func (*Distributor) AddSoul 

func (d *Distributor) AddSoul(soul *core.Soul)

AddSoul adds a soul to be distributed

func (*Distributor) GetAssignment 

func (d *Distributor) GetAssignment(soulID string) *core.SoulAssignment

GetAssignment returns the assignment for a soul

func (*Distributor) GetMyAssignments 

func (d *Distributor) GetMyAssignments() []core.SoulAssignment

GetMyAssignments returns assignments for the local node

func (*Distributor) GetNodeAssignments 

func (d *Distributor) GetNodeAssignments(nodeID string) []core.SoulAssignment

GetNodeAssignments returns all assignments for a node

func (*Distributor) GetPlan 

func (d *Distributor) GetPlan() core.DistributionPlan

GetPlan returns the current distribution plan

func (*Distributor) GetStats 

func (d *Distributor) GetStats() DistributionStats

GetStats returns distribution statistics

func (*Distributor) IsResponsible 

func (d *Distributor) IsResponsible(soulID string) bool

IsResponsible checks if this node is responsible for a soul

func (*Distributor) Recompute 

func (d *Distributor) Recompute() (core.DistributionPlan, error)

Recompute recomputes the distribution plan

func (*Distributor) RemoveNode 

func (d *Distributor) RemoveNode(nodeID string)

RemoveNode removes a node from the distribution pool

func (*Distributor) RemoveSoul 

func (d *Distributor) RemoveSoul(soulID string)

RemoveSoul removes a soul

func (*Distributor) SetOnRebalanceCallback 

func (d *Distributor) SetOnRebalanceCallback(cb func(core.DistributionPlan))

SetOnRebalanceCallback sets the callback for rebalance events

func (*Distributor) SetStrategy 

func (d *Distributor) SetStrategy(strategy core.DistributionStrategy)

SetStrategy changes the distribution strategy

func (*Distributor) UpdateNode 

func (d *Distributor) UpdateNode(node *core.NodeInfo)

UpdateNode updates node information

func (*Distributor) ValidatePlan 

func (d *Distributor) ValidatePlan(plan core.DistributionPlan) error

ValidatePlan validates a distribution plan

type FSM 

type FSM interface {
	Apply(log *core.RaftLogEntry) interface{}
	Snapshot() (core.FSMCommand, error)
	Restore(snapshot []byte) error
}

FSM is the finite state machine interface

type GossipMessage 

type GossipMessage struct {
	Type      string           `json:"type"`
	NodeID    string           `json:"node_id"`
	Address   string           `json:"address"`
	Region    string           `json:"region"`
	Version   string           `json:"version"`
	Peers     []GossipPeerInfo `json:"peers"`
	Timestamp int64            `json:"timestamp"`
}

GossipMessage is sent between peers during gossip

type GossipPeerInfo 

type GossipPeerInfo struct {
	ID       string `json:"id"`
	Address  string `json:"address"`
	Region   string `json:"region"`
	LastSeen int64  `json:"last_seen"`
}

GossipPeerInfo is information about a peer in gossip

type InMemoryLogStore 

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

InMemoryLogStore is an in-memory log store for testing

func NewInMemoryLogStore 

func NewInMemoryLogStore() *InMemoryLogStore

NewInMemoryLogStore creates a new in-memory log store

func (*InMemoryLogStore) DeleteRange 

func (s *InMemoryLogStore) DeleteRange(min, max uint64) error

DeleteRange deletes log entries in a range

func (*InMemoryLogStore) FirstIndex 

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

FirstIndex returns the first log index

func (*InMemoryLogStore) GetLog 

func (s *InMemoryLogStore) GetLog(index uint64, log *core.RaftLogEntry) error

GetLog retrieves a log entry

func (*InMemoryLogStore) LastIndex 

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

LastIndex returns the last log index

func (*InMemoryLogStore) StoreLog 

func (s *InMemoryLogStore) StoreLog(log *core.RaftLogEntry) error

StoreLog stores a log entry

func (*InMemoryLogStore) StoreLogs 

func (s *InMemoryLogStore) StoreLogs(logs []core.RaftLogEntry) error

StoreLogs stores multiple log entries

type InMemorySnapshot 

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

InMemorySnapshot represents a stored snapshot

type InMemorySnapshotSink 

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

InMemorySnapshotSink implements SnapshotSink

func (*InMemorySnapshotSink) Cancel 

func (s *InMemorySnapshotSink) Cancel() error

Cancel cancels the snapshot

func (*InMemorySnapshotSink) Close 

func (s *InMemorySnapshotSink) Close() error

Close closes the snapshot sink

func (*InMemorySnapshotSink) ID 

func (s *InMemorySnapshotSink) ID() string

ID returns the snapshot ID

func (*InMemorySnapshotSink) Write 

func (s *InMemorySnapshotSink) Write(p []byte) (n int, err error)

Write writes data to the snapshot

type InMemorySnapshotSource 

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

InMemorySnapshotSource implements SnapshotSource

func (*InMemorySnapshotSource) Close 

func (s *InMemorySnapshotSource) Close() error

Close closes the snapshot source

func (*InMemorySnapshotSource) Read 

func (s *InMemorySnapshotSource) Read(p []byte) (n int, err error)

Read reads data from the snapshot

type InMemorySnapshotStore 

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

InMemorySnapshotStore is an in-memory snapshot store for testing

func NewInMemorySnapshotStore 

func NewInMemorySnapshotStore() *InMemorySnapshotStore

NewInMemorySnapshotStore creates a new in-memory snapshot store

func (*InMemorySnapshotStore) Create 

func (s *InMemorySnapshotStore) Create(version, index, term uint64, configuration []byte) (SnapshotSink, error)

Create creates a new snapshot

func (*InMemorySnapshotStore) List 

func (s *InMemorySnapshotStore) List() ([]SnapshotMeta, error)

List returns all snapshots

func (*InMemorySnapshotStore) Open 

func (s *InMemorySnapshotStore) Open(id string) (SnapshotSource, error)

Open opens a snapshot for reading

type InMemoryStorage 

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

InMemoryStorage is a simple in-memory storage for testing

func NewInMemoryStorage 

func NewInMemoryStorage() *InMemoryStorage

NewInMemoryStorage creates a new in-memory storage

func (*InMemoryStorage) Delete 

func (s *InMemoryStorage) Delete(key string) error

Delete removes a key

func (*InMemoryStorage) DeletePrefix 

func (s *InMemoryStorage) DeletePrefix(prefix string) error

DeletePrefix removes all keys with the given prefix

func (*InMemoryStorage) Get 

func (s *InMemoryStorage) Get(key string) ([]byte, error)

Get retrieves a value

func (*InMemoryStorage) List 

func (s *InMemoryStorage) List(prefix string) ([]string, error)

List returns all keys with the given prefix

func (*InMemoryStorage) Set 

func (s *InMemoryStorage) Set(key string, value []byte) error

Set stores a value

type LogStore 

type LogStore interface {
	FirstIndex() (uint64, error)
	LastIndex() (uint64, error)
	GetLog(index uint64, log *core.RaftLogEntry) error
	StoreLog(log *core.RaftLogEntry) error
	StoreLogs(logs []core.RaftLogEntry) error
	DeleteRange(min, max uint64) error
}

LogStore is the interface for log storage

type MDNSClient 

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

MDNSClient handles mDNS service discovery

func (*MDNSClient) Query 

func (c *MDNSClient) Query(service string) []*MDNSService

func (*MDNSClient) Start 

func (c *MDNSClient) Start() error

func (*MDNSClient) Stop 

func (c *MDNSClient) Stop()

type MDNSServer 

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

MDNSServer handles mDNS service advertising

func (*MDNSServer) Start 

func (s *MDNSServer) Start() error

func (*MDNSServer) Stop 

func (s *MDNSServer) Stop()

type MDNSService 

type MDNSService struct {
	Name string
	Host string
	Port int
	IPs  []net.IP
	TXT  []string
}

MDNSService represents a discovered mDNS service

type Node 

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

Node represents a Raft consensus node The Pharaoh's throne in the Necropolis

func NewNode 

func NewNode(config core.RaftConfig, storage LogStore, snapshot SnapshotStore, fsm FSM, logger *slog.Logger) (*Node, error)

NewNode creates a new Raft node

func (*Node) AddPeer 

func (n *Node) AddPeer(peer core.RaftPeer) error

AddPeer adds a peer to the cluster using joint consensus This is safe for production use and prevents split-brain scenarios

func (*Node) Apply 

func (n *Node) Apply(cmd core.FSMCommand, timeout time.Duration) (uint64, uint64, interface{}, error)

Apply applies a command to the FSM through Raft

func (*Node) CurrentTerm 

func (n *Node) CurrentTerm() uint64

CurrentTerm returns the current term (public getter)

func (*Node) Done 

func (n *Node) Done() <-chan struct{}

Done returns the shutdown channel

func (*Node) GetPeers 

func (n *Node) GetPeers() []core.RaftPeer

GetPeers returns the current peers

func (*Node) GetState 

func (n *Node) GetState() core.ClusterState

GetState returns the current cluster state

func (*Node) IsLeader 

func (n *Node) IsLeader() bool

IsLeader returns true if this node is the leader

func (*Node) Leader 

func (n *Node) Leader() string

Leader returns the current leader ID

func (*Node) LeaderID 

func (n *Node) LeaderID() string

LeaderID returns the current leader ID (public getter)

func (*Node) Peers 

func (n *Node) Peers() map[string]*Peer

Peers returns a copy of the peers map

func (*Node) RemovePeer 

func (n *Node) RemovePeer(peerID string) error

RemovePeer removes a peer from the cluster using joint consensus This is safe for production use and prevents quorum loss

func (*Node) SetTransport 

func (n *Node) SetTransport(transport Transport)

SetTransport sets the transport for the node

func (*Node) Shutdown 

func (n *Node) Shutdown()

Shutdown initiates graceful shutdown (alias for Stop)

func (*Node) Start 

func (n *Node) Start() error

Start starts the Raft node

func (*Node) State 

func (n *Node) State() core.RaftState

State returns the current state

func (*Node) Stop 

func (n *Node) Stop() error

Stop gracefully stops the Raft node

func (*Node) Term 

func (n *Node) Term() uint64

Term returns the current term

type Peer 

type Peer struct {
	ID      string
	Address string
	Region  string
	Role    core.RaftRole
	// contains filtered or unexported fields
}

Peer represents a remote Raft node

type RPCHandler 

type RPCHandler func(cmd interface{}, respCh chan interface{})

RPCHandler handles incoming RPCs

type SnapshotMeta 

type SnapshotMeta struct {
	ID      string
	Index   uint64
	Term    uint64
	Size    int64
	Version uint64
}

SnapshotMeta holds metadata about a snapshot

type SnapshotSink 

type SnapshotSink interface {
	Write(p []byte) (n int, err error)
	Close() error
	ID() string
	Cancel() error
}

SnapshotSink is where snapshots are written

type SnapshotSource 

type SnapshotSource interface {
	Read(p []byte) (n int, err error)
	Close() error
}

SnapshotSource is where snapshots are read from

type SnapshotStore 

type SnapshotStore interface {
	Create(version, index, term uint64, configuration []byte) (SnapshotSink, error)
	List() ([]SnapshotMeta, error)
	Open(id string) (SnapshotSource, error)
}

SnapshotStore is the interface for snapshot storage

type Storage 

type Storage interface {
	Get(key string) ([]byte, error)
	Set(key string, value []byte) error
	Delete(key string) error
	DeletePrefix(prefix string) error
	List(prefix string) ([]string, error)
}

Storage is the interface for key-value storage

type StorageFSM 

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

StorageFSM implements the FSM interface using CobaltDB The sacred records inscribed on the tablets of the Necropolis

func NewStorageFSM 

func NewStorageFSM(store Storage) *StorageFSM

NewStorageFSM creates a new FSM backed by storage

func (*StorageFSM) Apply 

func (f *StorageFSM) Apply(log *core.RaftLogEntry) interface{}

Apply applies a Raft log entry to the FSM

func (*StorageFSM) LastApplied 

func (f *StorageFSM) LastApplied() uint64

LastApplied returns the last applied index

func (*StorageFSM) Restore 

func (f *StorageFSM) Restore(snapshot []byte) error

Restore restores the FSM from a snapshot

func (*StorageFSM) Snapshot 

func (f *StorageFSM) Snapshot() (core.FSMCommand, error)

Snapshot creates a snapshot of the FSM

type TCPTransport 

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

TCPTransport implements Transport over TCP with optional TLS

func NewTCPTransport 

func NewTCPTransport(bindAddr, advertiseAddr string, tlsConfig *tls.Config, logger *slog.Logger) (*TCPTransport, error)

NewTCPTransport creates a new TCP transport

func (*TCPTransport) AddPeerConnection 

func (t *TCPTransport) AddPeerConnection(peerID string, address string) error

AddPeerConnection adds or updates a connection to a peer

func (*TCPTransport) LocalAddr 

func (t *TCPTransport) LocalAddr() string

LocalAddr returns the local address

func (*TCPTransport) RegisterHandler 

func (t *TCPTransport) RegisterHandler(method string, handler RPCHandler)

RegisterHandler registers an RPC handler

func (*TCPTransport) RegisterPeer added in v0.0.2 

func (t *TCPTransport) RegisterPeer(peerID, address string)

RegisterPeer registers a peer address for future connections

func (*TCPTransport) SendAppendEntries 

func (t *TCPTransport) SendAppendEntries(peerID string, req *core.AppendEntriesRequest) (*core.AppendEntriesResponse, error)

SendAppendEntries sends an AppendEntries RPC

func (*TCPTransport) SendHeartbeat 

func (t *TCPTransport) SendHeartbeat(peerID string, req *core.HeartbeatRequest) (*core.HeartbeatResponse, error)

SendHeartbeat sends a heartbeat RPC

func (*TCPTransport) SendInstallSnapshot 

func (t *TCPTransport) SendInstallSnapshot(peerID string, req *core.InstallSnapshotRequest) (*core.InstallSnapshotResponse, error)

SendInstallSnapshot sends an InstallSnapshot RPC

func (*TCPTransport) SendPreVote added in v0.0.2 

func (t *TCPTransport) SendPreVote(peerID string, req *core.PreVoteRequest) (*core.PreVoteResponse, error)

SendPreVote sends a PreVote RPC

func (*TCPTransport) SendRequestVote 

func (t *TCPTransport) SendRequestVote(peerID string, req *core.RequestVoteRequest) (*core.RequestVoteResponse, error)

SendRequestVote sends a RequestVote RPC

func (*TCPTransport) Start 

func (t *TCPTransport) Start() error

Start starts listening for incoming connections

func (*TCPTransport) Stop 

func (t *TCPTransport) Stop() error

Stop stops the transport

func (*TCPTransport) UnregisterPeer added in v0.0.2 

func (t *TCPTransport) UnregisterPeer(peerID string)

UnregisterPeer removes a peer and closes its connection

type Transport 

type Transport interface {
	Start() error
	Stop() error
	SendAppendEntries(peerID string, req *core.AppendEntriesRequest) (*core.AppendEntriesResponse, error)
	SendRequestVote(peerID string, req *core.RequestVoteRequest) (*core.RequestVoteResponse, error)
	SendPreVote(peerID string, req *core.PreVoteRequest) (*core.PreVoteResponse, error)
	SendInstallSnapshot(peerID string, req *core.InstallSnapshotRequest) (*core.InstallSnapshotResponse, error)
	SendHeartbeat(peerID string, req *core.HeartbeatRequest) (*core.HeartbeatResponse, error)
	LocalAddr() string
}

Transport handles network communication

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.