raft

Documentation

Index

• func AddEventView[V encoding.BinaryMarshaler, VPtr sm.Unmarshallable[V], ...](ctx context.Context, builder *Builder, shardID uint64) eventstream.EventView[V, E]
• func AddShard[Q any, R any, E sm.Marshallable, EPtr sm.Unmarshallable[E]](ctx context.Context, to *Builder, shardID uint64, ...) sm.Handle[Q, R, E]
• type Builder
  • func NewBuilder(cfg *RaftConfig) *Builder
  • func (b *Builder) Build(ctx context.Context) *Cluster
  • func (b *Builder) WithControlClient(client *http.Client) *Builder
• type Cluster
  • func (c *Cluster) AddMember(ctx context.Context, req *connect.Request[raftpb.AddMemberRequest]) (*connect.Response[raftpb.AddMemberResponse], error)
  • func (c *Cluster) Join(ctx context.Context, controlAddress string) error
  • func (c *Cluster) Ping(ctx context.Context, req *connect.Request[ftlv1.PingRequest]) (*connect.Response[ftlv1.PingResponse], error)
  • func (c *Cluster) Start(ctx context.Context) error
  • func (c *Cluster) Stop(ctx context.Context)
• type RaftConfig
• type RaftEventView
  • func (s *RaftEventView[V, VPrt, E]) Changes(ctx context.Context) (iter.Seq[V], error)
  • func (s *RaftEventView[V, VPrt, E]) Publish(ctx context.Context, event E) error
  • func (s *RaftEventView[V, VPrt, E]) View(ctx context.Context) (V, error)
• type RaftStreamEvent
• type ShardHandle
  • func (s *ShardHandle[Q, R, E]) Publish(ctx context.Context, msg E) error
  • func (s *ShardHandle[Q, R, E]) Query(ctx context.Context, query Q) (R, error)
  • func (s *ShardHandle[Q, R, E]) StateIter(ctx context.Context, query Q) (iter.Seq[R], error)
• type UnitQuery

Functions

func AddEventView

func AddEventView[
	V encoding.BinaryMarshaler,
	VPtr sm.Unmarshallable[V],
	E RaftStreamEvent[V, VPtr],
	EPtr sm.Unmarshallable[E],
](ctx context.Context, builder *Builder, shardID uint64) eventstream.EventView[V, E]

AddEventView to the Builder

func AddShard

func AddShard[Q any, R any, E sm.Marshallable, EPtr sm.Unmarshallable[E]](
	ctx context.Context,
	to *Builder,
	shardID uint64,
	statemachine sm.Snapshotting[Q, R, E],
) sm.Handle[Q, R, E]

AddShard adds a shard to the cluster Builder.

Types

type Builder

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

Builder for a Raft Cluster.

func NewBuilder

func NewBuilder(cfg *RaftConfig) *Builder

func (*Builder) Build

func (b *Builder) Build(ctx context.Context) *Cluster

func (*Builder) WithControlClient

func (b *Builder) WithControlClient(client *http.Client) *Builder

WithControlClient sets the http client used to communicate with the control plane.

type Cluster

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

Cluster of dragonboat nodes.

func (*Cluster) AddMember

func (c *Cluster) AddMember(ctx context.Context, req *connect.Request[raftpb.AddMemberRequest]) (*connect.Response[raftpb.AddMemberResponse], error)

AddMember to the cluster. This needs to be called on an existing running cluster member, before the new member is started.

func (*Cluster) Join

func (c *Cluster) Join(ctx context.Context, controlAddress string) error

Join the cluster as a new member. Blocks until the cluster instance is ready.

func (*Cluster) Ping

func (c *Cluster) Ping(ctx context.Context, req *connect.Request[ftlv1.PingRequest]) (*connect.Response[ftlv1.PingResponse], error)

Ping the cluster.

func (*Cluster) Start

func (c *Cluster) Start(ctx context.Context) error

Start the cluster. Blocks until the cluster instance is ready.

func (*Cluster) Stop

func (c *Cluster) Stop(ctx context.Context)

Stop the node host and all shards. After this call, all the shard handlers created with this cluster are invalid.

type RaftConfig

type RaftConfig struct {
	InitialMembers    []string          `help:"Initial members"`
	ReplicaID         uint64            `help:"Node ID" required:""`
	DataDir           string            `help:"Data directory" required:""`
	Address           string            `help:"Address to advertise to other nodes" required:""`
	ListenAddress     string            `help:"Address to listen for incoming traffic. If empty, Address will be used."`
	ControlBind       *url.URL          `help:"Address to listen for control traffic. If empty, no control listener will be started."`
	ShardReadyTimeout time.Duration     `help:"Timeout for shard to be ready" default:"5s"`
	Retry             retry.RetryConfig `help:"Connection retry configuration" prefix:"retry-" embed:""`
	ChangesInterval   time.Duration     `help:"Interval for changes to be checked" default:"10ms"`
	ChangesTimeout    time.Duration     `help:"Timeout for changes to be checked" default:"1s"`

	// Raft configuration
	RTT                time.Duration `help:"Estimated average round trip time between nodes" default:"200ms"`
	ElectionRTT        uint64        `help:"Election RTT as a multiple of RTT" default:"10"`
	HeartbeatRTT       uint64        `help:"Heartbeat RTT as a multiple of RTT" default:"1"`
	SnapshotEntries    uint64        `help:"Snapshot entries" default:"10"`
	CompactionOverhead uint64        `help:"Compaction overhead" default:"100"`
}

type RaftEventView

type RaftEventView[V encoding.BinaryMarshaler, VPrt sm.Unmarshallable[V], E RaftStreamEvent[V, VPrt]] struct {
	// contains filtered or unexported fields
}

func (*RaftEventView[V, VPrt, E]) Changes

func (s *RaftEventView[V, VPrt, E]) Changes(ctx context.Context) (iter.Seq[V], error)

func (*RaftEventView[V, VPrt, E]) Publish

func (s *RaftEventView[V, VPrt, E]) Publish(ctx context.Context, event E) error

func (*RaftEventView[V, VPrt, E]) View

func (s *RaftEventView[V, VPrt, E]) View(ctx context.Context) (V, error)

type RaftStreamEvent

type RaftStreamEvent[View encoding.BinaryMarshaler, VPtr sm.Unmarshallable[View]] interface {
	encoding.BinaryMarshaler
	eventstream.Event[View]
}

type ShardHandle

type ShardHandle[Q any, R any, E sm.Marshallable] struct {
	// contains filtered or unexported fields
}

ShardHandle is a handle to a shard in the cluster. It is the interface to update and query the state of a shard.

E is the event type. Q is the query type. R is the query response type.

func (*ShardHandle[Q, R, E]) Publish

func (s *ShardHandle[Q, R, E]) Publish(ctx context.Context, msg E) error

Publish an event to the shard.

func (*ShardHandle[Q, R, E]) Query

func (s *ShardHandle[Q, R, E]) Query(ctx context.Context, query Q) (R, error)

Query the state of the shard.

func (*ShardHandle[Q, R, E]) StateIter

func (s *ShardHandle[Q, R, E]) StateIter(ctx context.Context, query Q) (iter.Seq[R], error)

StateIter returns an iterator that will return the result of the query when the shard state changes.

This can only be called when the cluster is running.

Note, that this is not guaranteed to receive an event for every change, but will always receive the latest state of the shard.

type UnitQuery

type UnitQuery struct{}