zongzi

package module
v0.0.23 Latest Latest
Warning

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

 Go to latest Published: Nov 7, 2026 License: Apache-2.0 Imports: 29 Imported by: 3

README

Zongzi

chibi style zongzi rice pyramid

A cluster coordinator for Dragonboat

Go Reference Go Report Card Go Coverage

This package provides a centralized coordination layer for Dragonboat multi-group Raft consensus clusters.

Usage

The Zongzi Agent simplifies multi-host operations using an internal API that automatically coordinates the necessary multi-host actions required to achieve the desired cluster state.

  1. Call zongzi.(*Agent).CreateReplica from any host in the cluster
  2. The desired replica state will be stored in the registry
  3. The responsible host controller will start the replica on the desired host

The cluster state registry is replicated to every host in the cluster so every host always has an eventually consistent snapshot of the cluster topology for command/query forwarding. Changes to the cluster can be proposed via internal API so cluster changes are as simple as writing to the registry and the host controllers will reconcile the difference.

There is no need to import the dragonboat package or interact with the dragonboat host directly.

Components
  • Registry
    • Stores desired state of all hosts, shards and replicas in the cluster
    • Cluster state stored in raft shard zero (shardID: 0)
    • All cluster state changes pass through the Zongzi agent
  • Message Bus
    • Internal gPRC
    • Facilitates cluster bootstrap
    • Forwards proposals and queries between nodes
  • Host Controller
    • Manages replicas on all hosts (start, stop, recover, delete, etc)
    • Responds automatically to changes in cluster state registry
  • Shard Controller
    • Reads placement policies from shard tags
    • Creates and destroys replicas to reconcile cluster state
    • You can replace the default shard controller with your own custom version
  • Host Client
    • Used to make proposals and queries to replicas on specific hosts
    • gRPC over message bus
  • Shard Client
    • Intelligently routes proposals and queries to active shard replicas
    • Selects nearest replica based on ping
    • Load balances stale reads across replicas

Architectural Constraints

Zongzi imposes some additional architectural constraints on top of Dragonboat:

  1. Although Dragonboat statemachine reads accept and return interface{}, all queries and responses sent through Zongzi must be expressed as []byte, just like command proposals. This serialization overhead is necessary for request forwarding because empty interfaces are not serializable. See ADR: Message Bus

  2. Although Dragonboat replica IDs are unique per shard, Zongzi replica IDs are unique per cluster. Having independent replica ids simplifies many replica operations which may have previously required both a shard id and replica id to be passed together up and down the callstack. The loss of address space (uint64 * uint64 vs uint64) is not expected to be a concern as 18.4 quintillion is still an astronomically large number. Having a materialized view of the global cluster state replicated to every host in the cluster makes it simple and efficient to dereference a replicaID to the correct host and shard. Dragonboat can't do this alone because its cluster state is decentralized.

  3. Any host may have at most one active replica of any shard. This aligns with many Dragonboat host operations like (*NodeHost).StopShard which assumes no more than one active replica per shard per host.

  4. Dragonboat sessions are not supported (they didn't work with on disk state machines anyways). Proposal deduplication is delegated to the developer. See ADR: Raft Sessions

  5. Initializing a Zongzi cluster requires at least 3 peer agents. Single host clusters are not supported.

Setup

Requirements for make gen

  1. Install protoc
  2. Install protoc-gen-go

Project Status

This library is in alpha (v0.0.x). Expect regular breaking changes to interfaces and serious defects outside the green path. When the API is stable and the code is well tested, beta releases will begin (v0.1.0). General availability will come after the library has been thoroughly tested in at least one production deployment for an extended period of time (v1.0.0).

Documentation

Index

Constants

View Source 
const (
	DefaultGossipAddress = "127.0.0.1:17001"
	DefaultRaftAddress   = "127.0.0.1:17002"
	DefaultApiAddress    = "127.0.0.1:17003"

	ShardID = 0
)
View Source 
const (
	LogLevelCritical = logger.CRITICAL
	LogLevelError    = logger.ERROR
	LogLevelWarning  = logger.WARNING
	LogLevelInfo     = logger.INFO
	LogLevelDebug    = logger.DEBUG
)
View Source 
const (
	AgentStatus_Initializing = AgentStatus("initializing")
	AgentStatus_Joining      = AgentStatus("joining")
	AgentStatus_Pending      = AgentStatus("pending")
	AgentStatus_Ready        = AgentStatus("ready")
	AgentStatus_Rejoining    = AgentStatus("rejoining")
	AgentStatus_Stopped      = AgentStatus("stopped")

	HostStatus_Active     = HostStatus("active")
	HostStatus_Gone       = HostStatus("gone")
	HostStatus_Missing    = HostStatus("missing")
	HostStatus_New        = HostStatus("new")
	HostStatus_Recovering = HostStatus("recovering")

	ShardStatus_Active      = ShardStatus("active")
	ShardStatus_Closed      = ShardStatus("closed")
	ShardStatus_Closing     = ShardStatus("closing")
	ShardStatus_New         = ShardStatus("new")
	ShardStatus_Unavailable = ShardStatus("unavailable")

	ReplicaStatus_Active        = ReplicaStatus("active")
	ReplicaStatus_Closed        = ReplicaStatus("closed")
	ReplicaStatus_Closing       = ReplicaStatus("closing")
	ReplicaStatus_Bootstrapping = ReplicaStatus("bootstrapping")
	ReplicaStatus_Joining       = ReplicaStatus("joining")
	ReplicaStatus_New           = ReplicaStatus("new")
)

Variables

View Source 
var (
	// HostMemory256 can be used to set max log memory usage to 256 MB
	HostMemory256 = config.GetTinyMemLogDBConfig()

	// HostMem1024 can be used to set max log memory usage to 1 GB
	HostMemory1024 = config.GetSmallMemLogDBConfig()

	// HostMem4096 can be used to set max log memory usage to 4 GB
	HostMemory4096 = config.GetMediumMemLogDBConfig()

	// HostMem8192 can be used to set max log memory usage to 8 GB (default)
	HostMemory8192 = config.GetLargeMemLogDBConfig()
)
View Source 
var (
	DefaultHostConfig = HostConfig{
		NodeHostDir:    "/var/lib/zongzi/raft",
		RaftAddress:    DefaultRaftAddress,
		RTTMillisecond: 10,
		WALDir:         "/var/lib/zongzi/wal",
		Expert: config.ExpertConfig{
			LogDBFactory: tan.Factory,
			LogDB:        config.GetMediumMemLogDBConfig(),
		},
	}
	DefaultReplicaConfig = ReplicaConfig{
		CheckQuorum:             true,
		CompactionOverhead:      10000,
		ElectionRTT:             100,
		EntryCompressionType:    config.Snappy,
		HeartbeatRTT:            10,
		OrderedConfigChange:     true,
		Quiesce:                 false,
		SnapshotCompressionType: config.Snappy,
		SnapshotEntries:         10000,
		WaitReady:               true,
	}
)
View Source 
var (
	GetLogger = logger.GetLogger

	GetDefaultEngineConfig = config.GetDefaultEngineConfig
)
View Source 
var (
	ErrAborted       = dragonboat.ErrAborted
	ErrCanceled      = dragonboat.ErrCanceled
	ErrRejected      = dragonboat.ErrRejected
	ErrShardClosed   = dragonboat.ErrShardClosed
	ErrShardNotReady = dragonboat.ErrShardNotReady
	ErrTimeout       = dragonboat.ErrTimeout

	ErrAgentNotReady     = fmt.Errorf("Agent not ready")
	ErrHostNotFound      = fmt.Errorf(`Host not found`)
	ErrIDOutOfRange      = fmt.Errorf(`ID out of range`)
	ErrInvalidFactory    = fmt.Errorf(`Invalid Factory`)
	ErrInvalidGossipAddr = fmt.Errorf(`Invalid gossip address`)
	ErrReplicaNotActive  = fmt.Errorf("Replica not active")
	ErrReplicaNotAllowed = fmt.Errorf("Replica not allowed")
	ErrReplicaNotFound   = fmt.Errorf("Replica not found")
	ErrShardExists       = fmt.Errorf(`Shard already exists`)
	ErrShardNotFound     = fmt.Errorf(`Shard not found`)

	ErrStreamConnectMissing   = fmt.Errorf(`First message of stream should be connect message`)
	ErrStreamConnectDuplicate = fmt.Errorf(`Connect message already received`)

	ErrInvalidNumberOfArguments = fmt.Errorf(`Invalid number of arguments`)

	// ErrClusterNameInvalid indicates that the clusterName is invalid
	// Base36 supports only lowercase alphanumeric characters
	ErrClusterNameInvalid = fmt.Errorf("Invalid cluster name (base36 maxlen 12)")
	ClusterNameRegex      = `^[a-z0-9]{1,12}$`

	// ErrNotifyCommitDisabled is logged when non-linearizable writes are requested but disabled.
	// Set property `NotifyCommit` to `true` in `HostConfig` to add support for non-linearizable writes.
	ErrNotifyCommitDisabled = fmt.Errorf("Attempted to make a non-linearizable write while NotifyCommit is disabled")
)

Functions

func ReleaseResult added in v0.0.10 

func ReleaseResult(r *Result)

func SetLogLevel 

func SetLogLevel(level LogLevel)

SetLogLevel sets log level for all zongzi and dragonboat loggers.

Recommend LogLevelWarning for production.

func SetLogLevelDebug 

func SetLogLevelDebug()

SetLogLevelDebug sets a debug log level for most loggers. but filters out loggers having tons of debug output.

func SetLogLevelProduction 

func SetLogLevelProduction()

SetLogLevelProduction sets a good log level for production (gossip logger is a bit noisy).

Types

type Agent 

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

func NewAgent 

func NewAgent(clusterName string, peers []string, opts ...AgentOption) (a *Agent, err error)

func (*Agent) Client added in v0.0.5 

func (a *Agent) Client(shardID uint64, opts ...ClientOption) (c ShardClient)

Client returns a client for a specific shard. It will send writes to the nearest member and send reads to the nearest replica (by ping).

func (*Agent) ClientByName added in v0.0.21 

func (a *Agent) ClientByName(name string, opts ...ClientOption) (c ShardClient)

ClientByName returns a client for a specific shard.

func (*Agent) ClusterName added in v0.0.22 

func (a *Agent) ClusterName() string

func (*Agent) HostID added in v0.0.3 

func (a *Agent) HostID() (id string)

hostID returns host ID if host is initialized, otherwise empty string.

func (*Agent) ReplicaAwait added in v0.0.20 

func (a *Agent) ReplicaAwait(ctx context.Context, d time.Duration, shardID uint64) (err error)

ReplicaAwait waits for local shard replica to be ready

func (*Agent) ShardCreate added in v0.0.5 

func (a *Agent) ShardCreate(ctx context.Context, uri string, opts ...ShardOption) (shard Shard, created bool, err error)

ShardCreate creates a new shard. If shard name option is provided and shard exists, found shard is returned.

func (*Agent) ShardDelete added in v0.0.9 

func (a *Agent) ShardDelete(ctx context.Context, id uint64) (err error)

ShardDelete deletes a shard.

func (*Agent) ShardFind added in v0.0.10 

func (a *Agent) ShardFind(ctx context.Context, id uint64) (shard Shard, err error)

ShardFind returns a shard by id.

func (*Agent) ShardUpdate added in v0.0.10 

func (a *Agent) ShardUpdate(ctx context.Context, id uint64, opts ...ShardOption) (shard Shard, err error)

ShardUpdate creates a new shard. If shard name option is provided and shard exists, found shard is returned.

func (*Agent) Start 

func (a *Agent) Start(ctx context.Context) (err error)

Start starts the agent, bootstrapping the cluster if required.

func (*Agent) State added in v0.0.10 

func (a *Agent) State(ctx context.Context, fn func(*State)) (err error)

State executes a callback function passing a snapshot of the cluster state. 

err := agent.State(ctx, func(s *State) error {
    log.Println(s.Index())
    return nil
})

Linear reads are enabled by default to achieve "Read Your Writes" consistency following a proposal. Pass optional argument _stale_ as true to disable linearizable reads (for higher performance). State will always provide snapshot isolation, even for stale reads.

State will block indefinitely if the prime shard is unavailable. This may prevent the agent from stopping gracefully. Pass a timeout context to avoid blocking indefinitely.

State is thread safe and will not block writes.

State read will be stale (non-linearizable) when ctx is nil.

func (*Agent) StateLocal added in v0.0.20 

func (a *Agent) StateLocal(fn func(*State)) (err error)

func (*Agent) StateMachineRegister added in v0.0.10 

func (a *Agent) StateMachineRegister(uri string, factory any, config ...ReplicaConfig) (err error)

StateMachineRegister registers a shard type. Call before Starting agent.

func (*Agent) Status added in v0.0.3 

func (a *Agent) Status() AgentStatus

Status returns the agent status.

func (*Agent) Stop 

func (a *Agent) Stop()

Stop stops the agent.

type AgentOption 

type AgentOption func(*Agent) error

func WithAddrApi added in v0.0.20 

func WithAddrApi(advertiseAddress string, bindAddress ...string) AgentOption

func WithAddrGossip added in v0.0.20 

func WithAddrGossip(advertiseAddress string, bindAddress ...string) AgentOption

func WithAddrRaft added in v0.0.20 

func WithAddrRaft(raftAddress string, listenAddress ...string) AgentOption

func WithDirRaft added in v0.0.20 

func WithDirRaft(dir string) AgentOption

func WithDirWAL added in v0.0.20 

func WithDirWAL(dir string) AgentOption

func WithHostConfig 

func WithHostConfig(cfg HostConfig) AgentOption

func WithHostMemoryLimit added in v0.0.10 

func WithHostMemoryLimit(limit HostMemory) AgentOption

WithHostMemory sets the maximum memory alloted to the raft log 

zongzi.WithHostMemory(zongzi.HostMemory256)

func WithHostTags added in v0.0.5 

func WithHostTags(tags ...string) AgentOption

func WithLogLevel added in v0.0.22 

func WithLogLevel(level logger.LogLevel) AgentOption

func WithRaftEventListener added in v0.0.6 

func WithRaftEventListener(listener RaftEventListener) AgentOption

func WithReplicaConfig 

func WithReplicaConfig(cfg ReplicaConfig) AgentOption

func WithShardController added in v0.0.9 

func WithShardController(c Controller) AgentOption

func WithSystemEventListener added in v0.0.6 

func WithSystemEventListener(listener SystemEventListener) AgentOption

type AgentStatus 

type AgentStatus string

type ClientOption added in v0.0.10 

type ClientOption func(*client) error

func WithRetries added in v0.0.9 

func WithRetries(retries int) ClientOption

func WithWriteToLeader added in v0.0.12 

func WithWriteToLeader() ClientOption

type ConnectionInfo added in v0.0.23 

type ConnectionInfo = raftio.ConnectionInfo

type Controller added in v0.0.10 

type Controller interface {
	Reconcile(*State, Shard, Controls) error
}

type Controls added in v0.0.10 

type Controls interface {
	Create(hostID string, shardID uint64, isNonVoting bool) (id uint64, err error)
	Delete(replicaID uint64) error
}

type EngineConfig added in v0.0.18 

type EngineConfig = config.EngineConfig

type Entry 

type Entry = statemachine.Entry

type EntryInfo added in v0.0.23 

type EntryInfo = raftio.EntryInfo

type GossipConfig 

type GossipConfig = config.GossipConfig

type Host 

type Host struct {
	ID      string            `json:"id"`
	Created uint64            `json:"created"`
	Updated uint64            `json:"updated"`
	Status  HostStatus        `json:"status"`
	Tags    map[string]string `json:"tags"`

	ApiAddress  string   `json:"apiAddress"`
	RaftAddress string   `json:"raftAddress"`
	ShardTypes  []string `json:"shardTypes"`
}

type HostConfig 

type HostConfig = config.NodeHostConfig

type HostMemory added in v0.0.10 

type HostMemory = LogDBConfig

type HostStatus 

type HostStatus string

type LeaderInfo 

type LeaderInfo = raftio.LeaderInfo

type LogDBConfig added in v0.0.10 

type LogDBConfig = config.LogDBConfig

type LogLevel 

type LogLevel = logger.LogLevel

type Logger added in v0.0.8 

type Logger = logger.ILogger

type NodeInfo added in v0.0.23 

type NodeInfo = raftio.NodeInfo

type RaftEventListener added in v0.0.6 

type RaftEventListener = raftio.IRaftEventListener

type Replica 

type Replica struct {
	ID      uint64            `json:"id"`
	Created uint64            `json:"created"`
	Updated uint64            `json:"updated"`
	Status  ReplicaStatus     `json:"status"`
	Tags    map[string]string `json:"tags"`

	HostID      string `json:"hostID"`
	IsNonVoting bool   `json:"isNonVoting"`
	IsWitness   bool   `json:"isWitness"`
	ShardID     uint64 `json:"shardID"`
}

type ReplicaConfig 

type ReplicaConfig = config.Config

type ReplicaStatus 

type ReplicaStatus string

type Result 

type Result = statemachine.Result

func GetResult 

func GetResult() *Result

GetResult can be used to efficiently retrieve an empty Result from a global pool. It is recommended to use this method to instantiate Result objects returned by Lookup or sent over Watch channels as they will be automatically returned to the pool to reduce allocation overhead.

type Shard 

type Shard struct {
	ID      uint64            `json:"id"`
	Created uint64            `json:"created"`
	Updated uint64            `json:"updated"`
	Status  ShardStatus       `json:"status"`
	Tags    map[string]string `json:"tags"`

	Leader uint64 `json:"leader"`
	Name   string `json:"name"`
	Term   uint64 `json:"term"`
	Type   string `json:"type"`
}

type ShardClient added in v0.0.9 

type ShardClient interface {
	Apply(ctx context.Context, cmd []byte) (value uint64, data []byte, err error)
	Commit(ctx context.Context, cmd []byte) (err error)
	Index(ctx context.Context) (err error)
	Leader() (uint64, uint64)
	Read(ctx context.Context, query []byte, stale bool) (value uint64, data []byte, err error)
	Watch(ctx context.Context, query []byte, results chan<- *Result, stale bool) (err error)
	Stream(ctx context.Context, in <-chan []byte, out chan<- *Result, stale bool) (err error)
}

ShardClient can be used to interact with a shard regardless of its placement in the cluster Requests will be forwarded to the appropriate host based on ping

type ShardOption added in v0.0.6 

type ShardOption func(*Shard) error

func WithName added in v0.0.6 

func WithName(name string) ShardOption

func WithPlacementCover added in v0.0.8 

func WithPlacementCover(tagKeys ...string) ShardOption

func WithPlacementMembers added in v0.0.6 

func WithPlacementMembers(n int, tags ...string) ShardOption

func WithPlacementReplicas added in v0.0.6 

func WithPlacementReplicas(group string, n int, tags ...string) ShardOption

func WithPlacementVary added in v0.0.6 

func WithPlacementVary(tagKeys ...string) ShardOption

func WithTag added in v0.0.10 

func WithTag(k string, v any) ShardOption

func WithTags added in v0.0.23 

func WithTags(m map[string]string) ShardOption

type ShardStatus 

type ShardStatus string

type SnapshotFile 

type SnapshotFile = statemachine.SnapshotFile

type SnapshotFileCollection 

type SnapshotFileCollection = statemachine.ISnapshotFileCollection

type SnapshotInfo added in v0.0.23 

type SnapshotInfo = raftio.SnapshotInfo

type State 

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

State represents internal state. All public methods are read only.

func (*State) Host added in v0.0.5 

func (fsm *State) Host(id string) (h Host, ok bool)

Host returns the host with the specified ID or ok false if not found.

func (*State) HostID added in v0.0.8 

func (fsm *State) HostID() string

HostID returns the ID of the last created Host

func (*State) HostIterate added in v0.0.2 

func (fsm *State) HostIterate(fn func(h Host) bool)

HostIterate executes a callback for every host in the cluster. Return true to continue iterating, false to stop. 

var hostCount int
agent.Read(func(s *zongzi.State) {
    s.HostIterate(func(h Host) bool {
        hostCount++
        return true
    })
})

func (*State) HostIterateByShardType added in v0.0.5 

func (fsm *State) HostIterateByShardType(shardType string, fn func(h Host) bool)

HostIterateByShardType executes a callback for every host in the cluster supporting to the provided shard type, ordered by host id ascending. Return true to continue iterating, false to stop.

func (*State) HostIterateByTag added in v0.0.5 

func (fsm *State) HostIterateByTag(tag, val string, fn func(h Host) bool)

HostIterateByTag executes a callback for every host in the cluster matching the specified tag, ordered by host id ascending. Return true to continue iterating, false to stop.

func (*State) Index 

func (fsm *State) Index() (val uint64)

func (*State) Replica added in v0.0.5 

func (fsm *State) Replica(id uint64) (r Replica, ok bool)

Replica returns the replica with the specified id or ok false if not found.

func (*State) ReplicaID added in v0.0.8 

func (fsm *State) ReplicaID() uint64

ReplicaID returns the ID of the last created Replica

func (*State) ReplicaIterate added in v0.0.2 

func (fsm *State) ReplicaIterate(fn func(r Replica) bool)

ReplicaIterate executes a callback for every replica in the cluster ordered by replica id ascending. Return true to continue iterating, false to stop.

func (*State) ReplicaIterateByHostID added in v0.0.2 

func (fsm *State) ReplicaIterateByHostID(hostID string, fn func(r Replica) bool)

ReplicaIterateByHostID executes a callback for every replica in the cluster belonging to the provided host id, ordered by replica id ascending. Return true to continue iterating, false to stop.

func (*State) ReplicaIterateByShardID added in v0.0.2 

func (fsm *State) ReplicaIterateByShardID(shardID uint64, fn func(r Replica) bool)

ReplicaIterateByShardID executes a callback for every replica in the cluster belonging to the provided shard id, ordered by replica id ascending. Return true to continue iterating, false to stop.

func (*State) ReplicaIterateByTag added in v0.0.5 

func (fsm *State) ReplicaIterateByTag(tag, val string, fn func(r Replica) bool)

ReplicaIterateByTag executes a callback for every host in the cluster matching the specified tag, ordered by host id ascending. Return true to continue iterating, false to stop.

func (*State) Save added in v0.0.2 

func (fsm *State) Save(w io.Writer) error

func (*State) Shard added in v0.0.5 

func (fsm *State) Shard(id uint64) (s Shard, ok bool)

Shard returns the shard with the specified id or ok false if not found.

func (*State) ShardFindByName added in v0.0.6 

func (fsm *State) ShardFindByName(name string) (s Shard, ok bool)

ShardFindByName returns the shard with the specified name or ok false if not found.

func (*State) ShardID added in v0.0.8 

func (fsm *State) ShardID() uint64

ShardID returns the ID of the last created Shard

func (*State) ShardIterate added in v0.0.2 

func (fsm *State) ShardIterate(fn func(s Shard) bool)

ShardIterate executes a callback for every shard in the cluster ordered by shard id ascending. Return true to continue iterating, false to stop.

func (*State) ShardIterateByTag added in v0.0.5 

func (fsm *State) ShardIterateByTag(tag, val string, fn func(r Shard) bool)

ShardIterateByTag executes a callback for every shard in the cluster matching the specified tag, ordered by shard id ascending. Return true to continue iterating, false to stop.

func (*State) ShardIterateUpdatedAfter added in v0.0.6 

func (fsm *State) ShardIterateUpdatedAfter(index uint64, fn func(r Shard) bool)

ShardIterateUpdatedAfter executes a callback for every shard in the cluster having an updated index greater than the supplied index

func (*State) ShardMembers added in v0.0.4 

func (fsm *State) ShardMembers(id uint64) map[uint64]string

ShardMembers returns a map of shard members (replicaID: hostID)

type StateMachine 

type StateMachine interface {
	Update(entries []Entry) []Entry
	Query(ctx context.Context, query []byte) *Result
	Watch(ctx context.Context, query []byte, result chan<- *Result)
	Stream(ctx context.Context, in <-chan []byte, out chan<- *Result)
	PrepareSnapshot() (cursor any, err error)
	SaveSnapshot(cursor any, w io.Writer, c SnapshotFileCollection, close <-chan struct{}) error
	RecoverFromSnapshot(r io.Reader, f []SnapshotFile, close <-chan struct{}) error
	Close() error
}

StateMachine is a deterministic finite state machine. Snapshots are requested during log compaction to ensure that the in-memory state can be recovered following a restart. If you expect a dataset larger than memory, a persistent state machine may be more appropriate.

Lookup may be called concurrently with Update and SaveSnapshot. It is the caller's responsibility to ensure that snapshots are generated using snapshot isolation. This can be achieved using Multi Version Concurrency Control (MVCC). A simple mutex can also be used if blocking writes during read is acceptable.

type StateMachineFactory 

type StateMachineFactory = func(shardID uint64, replicaID uint64) StateMachine

StateMachineFactory is a function that returns a StateMachine

type StateMachinePersistent added in v0.0.9 

type StateMachinePersistent interface {
	Open(stopc <-chan struct{}) (index uint64, err error)
	Update(entries []Entry) []Entry
	Query(ctx context.Context, query []byte) *Result
	Watch(ctx context.Context, query []byte, result chan<- *Result)
	Stream(ctx context.Context, in <-chan []byte, out chan<- *Result)
	PrepareSnapshot() (cursor any, err error)
	SaveSnapshot(cursor any, w io.Writer, close <-chan struct{}) error
	RecoverFromSnapshot(r io.Reader, close <-chan struct{}) error
	Sync() error
	Close() error
}

StateMachinePersistent is a StateMachine where the state is persisted to a medium (such as disk) that can survive restart. During compaction, calls to Snapshot are replaced with calls to Sync which effectively flushes state to the persistent medium. SaveSnapshot and RecoverFromSnapshot are used to replicate full on-disk state to new replicas.

type StateMachinePersistentFactory added in v0.0.9 

type StateMachinePersistentFactory = func(shardID uint64, replicaID uint64) StateMachinePersistent

StateMachinePersistentFactory is a function that returns a StateMachinePersistent

type SystemEventListener added in v0.0.6 

type SystemEventListener = raftio.ISystemEventListener

Source Files

View all Source files

Directories

Path Synopsis

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.
⌂ 换网址