federation

Documentation

Index

• Constants
• func Compare(a, b VClock) int
• func PeerCortexIDKey(peerName string) string
• func PeerCursorKey(peerName string) string
• func PeerSeenKey(peerName string) string
• type Config
  • func (c Config) EffectiveInterval() time.Duration
• type EventReplayer
• type PeerConfig
• type PeerState
• type State
  • func NewState(db *sql.DB) *State
  • func (s *State) Delete(key string) error
  • func (s *State) Get(key string) (string, error)
  • func (s *State) GetClock() (VClock, error)
  • func (s *State) GetPeerState(name, endpoint string) (PeerState, error)
  • func (s *State) Set(key, value string) error
  • func (s *State) SetClock(vc VClock) error
  • func (s *State) SetPeerCortexID(name, cortexID string) error
  • func (s *State) SetPeerCursor(name, eventID string) error
  • func (s *State) SetPeerSeen(name, timestamp string) error
• type Syncer
  • func NewSyncer(replayer EventReplayer, state *State, cfg Config) *Syncer
  • func (s *Syncer) Start()
  • func (s *Syncer) Stop()
• type VClock
  • func Merge(a, b VClock) VClock
  • func (vc VClock) Clone() VClock
  • func (vc VClock) Increment(peer string)

Constants

const (
	PeerModeSync   = "sync"
	PeerModePaused = "paused"
)

Peer mode constants — checked by the syncer to skip paused peers.

const DefaultInterval = 30 * time.Second

Functions

func Compare

func Compare(a, b VClock) int

Compare returns the causal relationship between two clocks:

-1 means a happened-before b
 0 means concurrent (no causal relationship — potential conflict)
+1 means b happened-before a

func PeerCortexIDKey

func PeerCortexIDKey(peerName string) string

PeerCortexIDKey returns the federation_state key under which a peer's verified cortex ULID is pinned after the first successful identity handshake. The syncer refuses to talk to a peer whose advertised ID has changed from what is stored here — see docs/design/cortex-uuid-plan.md.

func PeerCursorKey

func PeerCursorKey(peerName string) string

PeerCursorKey returns the federation_state key for a peer's last_event cursor.

func PeerSeenKey

func PeerSeenKey(peerName string) string

PeerSeenKey returns the federation_state key for a peer's last_seen time.

Types

type Config

type Config struct {
	Mode     string        // federation-level mode: sync | publish | subscribe
	Peers    []PeerConfig  `yaml:"peers,omitempty"`
	Interval time.Duration // parsed from string

	// SharedKey is the MCP bearer token this syncer attaches to every
	// outbound request as "Authorization: Bearer <SharedKey>". Empty
	// means open mode (no header sent). Populated at startup from
	// cortex.LoadAccessKey; never surfaced in YAML, logs, or events.
	SharedKey string `yaml:"-"`
}

Config holds federation settings from cortex.md.

func (Config) EffectiveInterval

func (c Config) EffectiveInterval() time.Duration

EffectiveInterval returns the configured interval or the default.

type EventReplayer

type EventReplayer interface {
	ReplayEvent(event.Event) error
	MergeClock(VClock) error
}

EventReplayer materializes a remote event on the local cortex.

type PeerConfig

type PeerConfig struct {
	Name     string `yaml:"name"`
	Endpoint string `yaml:"endpoint"`
	CA       string `yaml:"ca,omitempty"`   // path to CA certificate for TLS verification
	Mode     string `yaml:"mode,omitempty"` // sync | paused
}

PeerConfig is a peer entry as declared in cortex.md.

type PeerState

type PeerState struct {
	Name      string
	Endpoint  string
	LastSeen  string // RFC3339, "" if never reached
	LastEvent string // ULID of last synced event, "" if never synced
	CortexID  string // pinned ULID after first successful identity handshake
}

PeerState is the runtime state of a known peer.

type State

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

State provides read/write access to the federation_state table.

func NewState

func NewState(db *sql.DB) *State

func (*State) Delete

func (s *State) Delete(key string) error

Delete removes a single key from federation_state. Missing keys are not an error — Delete is used by `noema federation reset-peer` to clear pin / cursor / last_seen rows that may or may not exist depending on whether the peer has ever been contacted, so the operation has to be idempotent.

func (*State) Get

func (s *State) Get(key string) (string, error)

func (*State) GetClock

func (s *State) GetClock() (VClock, error)

GetClock loads the vector clock from federation_state.

func (*State) GetPeerState

func (s *State) GetPeerState(name, endpoint string) (PeerState, error)

GetPeerState loads the runtime state for a peer.

func (*State) Set

func (s *State) Set(key, value string) error

func (*State) SetClock

func (s *State) SetClock(vc VClock) error

SetClock persists the vector clock to federation_state.

func (*State) SetPeerCortexID

func (s *State) SetPeerCortexID(name, cortexID string) error

SetPeerCortexID pins a peer's verified cortex ULID. Should only be called after the cortex_identity handshake has succeeded once.

func (*State) SetPeerCursor

func (s *State) SetPeerCursor(name, eventID string) error

SetPeerCursor updates the last synced event cursor for a peer.

func (*State) SetPeerSeen

func (s *State) SetPeerSeen(name, timestamp string) error

SetPeerSeen updates the last seen time for a peer.

type Syncer

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

Syncer polls remote peers for new events and replays them locally.

func NewSyncer

func NewSyncer(replayer EventReplayer, state *State, cfg Config) *Syncer

func (*Syncer) Start

func (s *Syncer) Start()

func (*Syncer) Stop

func (s *Syncer) Stop()

type VClock

type VClock map[string]uint64

VClock is a vector clock: one counter per known peer.

func Merge

func Merge(a, b VClock) VClock

Merge returns a new VClock with the component-wise max of two clocks.

func (VClock) Clone

func (vc VClock) Clone() VClock

Clone returns a deep copy of the vector clock.

func (VClock) Increment

func (vc VClock) Increment(peer string)

Increment bumps the counter for the given peer.