cluster

Documentation

Overview

Package cluster contains primitives for node identity, membership tracking and consistent hashing used by distributed backends.

Index

• Variables
• type Membership
  • func NewMembership(ring *Ring) *Membership
  • func (m *Membership) List() []*Node
  • func (m *Membership) Mark(id NodeID, state NodeState) bool
  • func (m *Membership) OnStateChange(fn StateChangeObserver)
  • func (m *Membership) Remove(id NodeID) bool
  • func (m *Membership) Ring() *Ring
  • func (m *Membership) Upsert(n *Node)
  • func (m *Membership) Version() uint64
• type MembershipVersion
  • func (mv *MembershipVersion) Get() uint64
  • func (mv *MembershipVersion) Next() uint64
• type Node
  • func NewNode(id, addr string) *Node
  • func (n *Node) Validate() error
• type NodeID
• type NodeState
  • func (s NodeState) String() string
• type Ring
  • func NewRing(opts ...RingOption) *Ring
  • func (r *Ring) Build(nodes []*Node)
  • func (r *Ring) Lookup(key string) []NodeID
  • func (r *Ring) Replication() int
  • func (r *Ring) VNodeHashes() []string
  • func (r *Ring) VirtualNodesPerNode() int
• type RingOption
  • func WithReplication(n int) RingOption
  • func WithVirtualNodes(n int) RingOption
• type StateChangeObserver

Variables

var ErrInvalidAddress = ewrap.New("invalid node address")

ErrInvalidAddress is returned when the node address is invalid.

Types

type Membership

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

Membership tracks current cluster nodes (static MVP, future: gossip/swim).

func NewMembership

func NewMembership(ring *Ring) *Membership

NewMembership creates a new membership container bound to a ring.

func (*Membership) List

func (m *Membership) List() []*Node

List returns current nodes snapshot.

func (*Membership) Mark

func (m *Membership) Mark(id NodeID, state NodeState) bool

Mark updates node state + incarnation and refreshes LastSeen. Returns true if node exists.

func (*Membership) OnStateChange

func (m *Membership) OnStateChange(fn StateChangeObserver)

OnStateChange registers a callback invoked after every membership mutation (Upsert, Mark, Remove). Registration is append-only and not safe for concurrent use with mutations — call this once at construction before the cache starts driving heartbeats / gossip.

The callback runs OUTSIDE the membership lock so a slow observer does not block the SWIM heartbeat loop. Observers fire in registration order; one panicking observer would skip every observer registered after it, so observer authors must recover in their own code (the package itself does not wrap with recover() because that would mask programming bugs).

func (*Membership) Remove

func (m *Membership) Remove(id NodeID) bool

Remove deletes a node from membership and rebuilds the ring. Returns true if removed.

func (*Membership) Ring

func (m *Membership) Ring() *Ring

Ring returns the underlying ring reference.

func (*Membership) Upsert

func (m *Membership) Upsert(n *Node)

Upsert adds or updates a node and rebuilds ring.

func (*Membership) Version

func (m *Membership) Version() uint64

Version returns current membership version.

type MembershipVersion

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

MembershipVersion tracks a monotonically increasing version for membership changes. Used to expose a cheap cluster epoch for clients/metrics.

func (*MembershipVersion) Get

func (mv *MembershipVersion) Get() uint64

Get returns current version.

func (*MembershipVersion) Next

func (mv *MembershipVersion) Next() uint64

Next increments and returns the next version.

type Node

type Node struct {
	ID          NodeID
	Address     string // host:port for intra-cluster RPC
	State       NodeState
	Incarnation uint64
	LastSeen    time.Time
}

Node holds identity & state.

func NewNode

func NewNode(id, addr string) *Node

NewNode creates a node from address (host:port). If id empty, derive a short hex id using xxhash64.

func (*Node) Validate

func (n *Node) Validate() error

Validate basic fields.

type NodeID

type NodeID string

NodeID is a stable identifier for a node.

type NodeState

type NodeState int

NodeState represents membership state of a node.

const (
	NodeAlive NodeState = iota
	NodeSuspect
	NodeDead
)

Node state enumeration.

func (NodeState) String

func (s NodeState) String() string

type Ring

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

Ring implements a consistent hashing ring with virtual nodes.

func NewRing

func NewRing(opts ...RingOption) *Ring

NewRing creates a ring with defaults overridden by options. NewRing constructs a new Ring applying provided options.

func (*Ring) Build

func (r *Ring) Build(nodes []*Node)

Build rebuilds ring from nodes set (copy-on-write). Build rebuilds the ring using the supplied node list.

func (*Ring) Lookup

func (r *Ring) Lookup(key string) []NodeID

Lookup returns primary + (replication-1) replicas. Lookup returns the primary owner and (replication-1) replicas for a key.

func (*Ring) Replication

func (r *Ring) Replication() int

Replication returns replication factor.

func (*Ring) VNodeHashes

func (r *Ring) VNodeHashes() []string

VNodeHashes returns a copy of vnode hash values as hex strings (debug only).

func (*Ring) VirtualNodesPerNode

func (r *Ring) VirtualNodesPerNode() int

VirtualNodesPerNode returns configured virtual nodes per physical node.

type RingOption

type RingOption func(*Ring)

RingOption configures ring.

func WithReplication

func WithReplication(n int) RingOption

WithReplication sets the replication factor (number of owners per key).

func WithVirtualNodes

func WithVirtualNodes(n int) RingOption

WithVirtualNodes sets the number of virtual nodes per physical node.

type StateChangeObserver

type StateChangeObserver func(id NodeID, state NodeState, version uint64)

StateChangeObserver is invoked AFTER a membership mutation (Upsert / Mark / Remove) commits, with the membership lock already released. Observers MUST NOT call back into the Membership they were registered on (deadlock); they SHOULD do minimal work (publish to a channel, increment a counter) and return promptly, since multiple observers run sequentially on the goroutine that performed the mutation.

Phase C SSE: the cache binary registers one observer that publishes a `members` event onto the in-process event bus, so SSE subscribers see state transitions without re-deriving them by polling.