coord

package
v0.7.1 Latest Latest
Warning

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

Go to latest
Published: Sep 8, 2026 License: AGPL-3.0 Imports: 13 Imported by: 0

Documentation

Overview

Package coord is mlfs's multi-node coordination plane (L2.8). It embeds (or connects to) NATS JetStream and uses it as the LOW-LATENCY accelerator for ownership coordination: lease liveness lives in a KV bucket with native TTL + Watch, so refresh and ownership-discovery never hit PostgreSQL on the hot path. PostgreSQL remains the system of record and the authoritative fence — the generation a write is fenced on always comes from the SQL ownership row (see internal/meta/fence.go). NATS only decides who is alive and notifies peers fast; if NATS is wiped or partitioned the worst case is a slower handoff or a blocked write, never a double-owner. See docs/L2.8_IMPLEMENTATION.md.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ChangeMsg

type ChangeMsg struct {
	LSN     int64           `json:"l"`
	Ino     uint64          `json:"i"`
	Op      uint8           `json:"o"`
	Payload json.RawMessage `json:"p"`
}

ChangeMsg is one fan-out entry: the inode, op code (meta.Change*), and the op's raw fs_changelog payload (parent/name/etc.) for entry invalidation.

type Changefeed

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

Changefeed publishes and subscribes change events over NATS.

func NewChangefeed

func NewChangefeed(n *NATS) *Changefeed

NewChangefeed builds a change feed over the coordination NATS connection.

func (*Changefeed) RunPublisher

func (c *Changefeed) RunPublisher(ctx context.Context, scanner ChangelogScanner, isLeader func() bool, interval time.Duration)

RunPublisher tails the changelog and fans out new entries while this node is the GC leader (so exactly one node publishes). It advances its LSN cursor past whatever exists at startup so it only forwards changes from here on. Returns when ctx is cancelled.

func (*Changefeed) Subscribe

func (c *Changefeed) Subscribe(ctx context.Context, handler func(ChangeMsg)) error

Subscribe delivers fan-out change events to handler until ctx is cancelled.

type ChangelogScanner

type ChangelogScanner interface {
	ScanChangelog(ctx context.Context, after int64, fn func(meta.ChangeEntry) error) error
}

ChangelogScanner is the durable-rail source. *meta.Engine satisfies it.

type EmbeddedConfig

type EmbeddedConfig struct {
	NodeName string // server + cluster identity
	StoreDir string // JetStream file storage directory

	// Cluster wiring. Leave ClusterName empty for a standalone single node
	// (in-process, no network listener — ideal for tests and single-mount dev).
	// Set ClusterName + ListenHost/ListenPort + Routes to form a multi-node
	// JetStream cluster across mlfs daemons.
	ClusterName string
	ListenHost  string
	ListenPort  int      // client port; 0 picks a free port
	ClusterPort int      // route port
	Routes      []string // nats-route URLs of peer nodes

	ReadyTimeout time.Duration // default 10s
}

EmbeddedConfig configures an in-process NATS+JetStream node.

type Leader

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

Leader is a single-winner election over a TTL'd JetStream KV key. Exactly one live node holds the key at a time; the holder renews it (resetting the TTL), and if it dies the key expires and a peer claims it. Used to gate cluster-wide singletons — GC, changelog trim, and the lock reaper — so they run on one node only (closing the plan's GC-leader-lock residual #29).

func NewLeader

func NewLeader(ctx context.Context, n *NATS, key, nodeID string, ttl, renew time.Duration, replicas int) (*Leader, error)

NewLeader provisions the leader bucket and returns an election for key (e.g. "gc"). ttl is the leadership lease; renew (<= ttl/2) is how often the holder reasserts it.

func (*Leader) IsLeader

func (l *Leader) IsLeader() bool

IsLeader reports whether this node currently holds leadership.

func (*Leader) Run

func (l *Leader) Run(ctx context.Context)

Run campaigns for and maintains leadership until ctx is cancelled, releasing the key on exit if held.

func (*Leader) SetMetrics

func (l *Leader) SetMetrics(m LeaderMetrics)

SetMetrics installs the leadership-transition counter hook (nil clears it). Call before Run.

type LeaderMetrics

type LeaderMetrics interface {
	LeaderTransition(leader bool)
}

LeaderMetrics is the election's optional metrics hook (the daemon's *metrics.Registry satisfies it). LeaderTransition fires only on an actual role change, with leader=true when this node just became leader.

type LeaseMetrics

type LeaseMetrics interface {
	LeaseAcquired()
	LeaseLost()
}

LeaseMetrics is the owner's optional metrics hook (the daemon's *metrics.Registry satisfies it). LeaseAcquired fires when this node takes a scope's lease; LeaseLost fires when it drops one to a peer steal.

type LockReaper

type LockReaper interface {
	ListLockSIDs(ctx context.Context) ([]uint64, error)
	ClearLocksForSID(ctx context.Context, sid uint64) error
}

LockReaper is the meta side the reaper needs: list lock-holding sids and clear one sid's rows. *meta.Engine satisfies it.

type NATS

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

NATS holds a JetStream handle plus, when embedded, the in-process server that backs it. Close tears down the connection and (if owned) the server.

func Connect

func Connect(url string, opts ...nats.Option) (*NATS, error)

Connect attaches to an EXTERNAL NATS/JetStream cluster (no embedded server).

func StartEmbedded

func StartEmbedded(cfg EmbeddedConfig) (*NATS, error)

StartEmbedded boots an in-process NATS server with JetStream and returns a handle over an in-process (or local) connection. Standalone mode (no ClusterName) uses DontListen + an in-process pipe — zero network surface.

func (*NATS) Close

func (n *NATS) Close()

Close drains the connection and shuts down the embedded server if owned.

func (*NATS) Conn

func (n *NATS) Conn() *nats.Conn

Conn returns the underlying NATS connection.

func (*NATS) JS

func (n *NATS) JS() jetstream.JetStream

JS returns the JetStream context.

type NATSOwner

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

NATSOwner implements meta.OwnerView. It keeps lease LIVENESS in a JetStream KV bucket (fast refresh + Watch-based discovery) and takes the authoritative fence GENERATION from the SQL ownership row via OwnershipStore. This is the low-latency path the plan calls for: the per-refresh and per-discovery traffic is NATS, not PostgreSQL; SQL is consulted only to mint/bump a generation on a real acquire or steal, and as the periodic durability backstop.

func NewNATSOwner

func NewNATSOwner(ctx context.Context, n *NATS, sql OwnershipStore, nodeID string, ttl, refresh time.Duration, replicas int) (*NATSOwner, error)

NewNATSOwner provisions the liveness bucket and returns an owner for nodeID. ttl is the lease lifetime (bucket TTL); refresh is the NATS renew interval (<= ttl/2). replicas sets KV replication for clustered deployments (1 for single-node).

func (*NATSOwner) Enabled

func (o *NATSOwner) Enabled() bool

Enabled reports that multi-node fencing is active.

func (*NATSOwner) Ensure

func (o *NATSOwner) Ensure(ctx context.Context, scope string) (int64, bool, error)

Ensure acquires-or-confirms this node's lease on scope. Order of checks, cheapest first: in-memory cache → NATS liveness (is it free / ours / a live peer's?) → SQL AcquireOrResume (authoritative generation) only when free.

func (*NATSOwner) Run

func (o *NATSOwner) Run(ctx context.Context)

Run renews held leases until ctx is cancelled. Every tick refreshes NATS liveness (fast); every Nth tick also bumps the SQL lease_expires backstop so the authoritative row never looks falsely dead if NATS state is lost. A scope observed owned by a peer (we were stolen) is dropped.

func (*NATSOwner) SetMetrics

func (o *NATSOwner) SetMetrics(m LeaseMetrics)

SetMetrics installs the lease-transition counter hook (nil clears it). Call before Run.

func (*NATSOwner) SetReleaseHook

func (o *NATSOwner) SetReleaseHook(fn func(ctx context.Context, scope string) error)

SetReleaseHook installs a callback invoked before this node releases a scope during a handoff — used to flush node-local dirty state (e.g. the write-back cache) so the incoming owner can read it from the shared backing store.

func (*NATSOwner) StartHandoffResponder

func (o *NATSOwner) StartHandoffResponder(ctx context.Context) error

StartHandoffResponder subscribes this node as a handoff responder until ctx is cancelled. Safe to call once after construction.

type OwnershipStore

type OwnershipStore interface {
	AcquireOrResume(ctx context.Context, scopeKey, node string, ttlMs int64) (meta.Lease, bool, error)
	Refresh(ctx context.Context, scopeKey, node string, generation, ttlMs int64) (bool, error)
	ReleaseOwnership(ctx context.Context, scopeKey, node string, generation int64) (bool, error)
}

OwnershipStore is the authoritative (SQL) side the NATS owner defers to for the fence generation. *meta.Engine satisfies it.

type Registry

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

Registry tracks live mounts via a TTL'd JetStream KV bucket: each node heartbeats its lock-session id (sid), and the entry auto-expires if the node dies. The liveness reaper diffs the live sids against the lock tables to clear a dead peer's stale flock/plock rows — the multi-node replacement for the single-node wholesale ClearAllLocks.

func NewRegistry

func NewRegistry(ctx context.Context, n *NATS, nodeID string, sid uint64, ttl, beat time.Duration, replicas int) (*Registry, error)

NewRegistry provisions the node bucket and returns a registry for nodeID/sid.

func (*Registry) Heartbeat

func (r *Registry) Heartbeat(ctx context.Context) error

Heartbeat publishes this node's liveness once.

func (*Registry) LiveSIDs

func (r *Registry) LiveSIDs(ctx context.Context) (map[uint64]struct{}, error)

LiveSIDs returns the lock-session ids of all currently-live nodes (entries that have not expired).

func (*Registry) ReapDeadLocks

func (r *Registry) ReapDeadLocks(ctx context.Context, m LockReaper) (int, error)

ReapDeadLocks clears flock/plock rows whose owning sid is not in the live set. Run it on the GC leader only. Returns the number of dead sessions reaped.

func (*Registry) Run

func (r *Registry) Run(ctx context.Context)

Run heartbeats on a ticker until ctx is cancelled, then removes the entry.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL