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) Refute(id NodeID) bool
  • 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 records an observer-side state update for the given node. Behavior depends on whether `state` represents a real transition:

• state == current: no-op. LastSeen refreshes (the peer answered a heartbeat probe, so its observability window resets), but incarnation, the membership version vector, and registered observers are all left alone.
• state != current: incarnation bumps, version vector advances, and observers fire — this is a real membership state change and gossip-merge / SSE consumers need to know.

Returns true if the node exists.

Why no-op on same state: pre-fix every successful heartbeat probe called Mark(peer, Alive) and unconditionally bumped both the node's incarnation AND the membership version counter. Operators running a 5-node cluster for a few hours saw incarnations climb to ~2,378 and MembershipVersion past 4,800 — the value tracked "elapsed probes," not "state changes." SWIM defines incarnation as a sequence number OWNED by the node itself, and the membership-version vector should signal "something real happened" so gossip knows when to fan out a delta. Both must be quiet during steady-state liveness churn.

Self-refute — the one observer-side path that legitimately needs to bump incarnation regardless of the local state value — lives in `Refute()` so the divergent semantic is explicit at the call site.

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) Refute

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

Refute is the SWIM self-refute primitive: when this node receives gossip that it is Suspect/Dead at some incarnation N, it MUST publish a higher incarnation (N+1) with state=Alive so the next gossip tick disseminates the refutation cluster-wide (peers using "higher incarnation wins" adopt the new value).

Unlike `Mark()`, Refute always bumps incarnation — the local state is already Alive (a node never thinks itself dead), so a transition-only rule would silently no-op and the refutation would fail to propagate. The dedicated method makes the divergent semantic obvious at the only call site that needs it (`refuteIfSuspected` in pkg/backend/dist_memory.go).

Returns true if the node exists. State is unconditionally set to NodeAlive; the typical caller already knows the local node is alive but the explicit assignment guards against odd states the local view might be holding.

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.