state

package
v0.1.0-alpha.8 Latest Latest
Warning

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

 Go to latest Published: Jul 18, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Overview

Package state holds the full replicated cluster state in memory (ADR-0004).

The Store is the Raft FSM's state: plain maps of proto messages keyed by ULID, a handful of secondary indexes on hot paths, one RWMutex, and a watch hub that lets control loops (scheduler, route builder, cert manager) react to changes.

Rules:

  • Mutating methods are called ONLY from FSM apply handlers (internal/daemon/raftstore). Everything else reads.
  • Read methods return deep clones; callers may mutate results freely.
  • Collections are small (thousands of objects): list methods do linear scans with filters unless a path is genuinely hot. Hot-path indexes: tokens by hash, users by email, domains by hostname, assignments by node. Do not add more indexes without a measured need.

Index

Constants

This section is empty.

Variables

View Source 
var ErrKVConflict = fmt.Errorf("state: kv version conflict")

ErrKVConflict is returned when a CAS expectation fails.

Functions

This section is empty.

Types

type Change 

type Change struct {
	Kind Kind
	ID   string
}

Change identifies one mutated object.

type Hub 

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

Hub fans out change notifications to subscribers without ever blocking the FSM apply path. Changes are coalesced per subscriber (a set keyed by Kind+ID); the subscriber gets a level-triggered signal and drains the pending set when it is ready. Slow subscribers lose granularity, never correctness — control loops re-read state anyway.

type Kind 

type Kind string

Kind identifies a resource collection for watch subscriptions. 

const (
	KindOrg            Kind = "org"
	KindUser           Kind = "user"
	KindProject        Kind = "project"
	KindProjectMember  Kind = "project_member"
	KindApp            Kind = "app"
	KindEnvironment    Kind = "environment"
	KindEnvVar         Kind = "env_var"
	KindRelease        Kind = "release"
	KindDeployment     Kind = "deployment"
	KindBuild          Kind = "build"
	KindNode           Kind = "node"
	KindJoinToken      Kind = "join_token"
	KindAssignment     Kind = "assignment"
	KindToken          Kind = "token"
	KindDomain         Kind = "domain"
	KindKV             Kind = "kv"
	KindDNSProvider    Kind = "dns_provider"
	KindVolume         Kind = "volume"
	KindVolumeSnapshot Kind = "volume_snapshot"
	KindBackup         Kind = "backup"
	KindNetworkAlloc   Kind = "network_alloc"
	KindServiceVIP     Kind = "service_vip"
	KindJob            Kind = "job"
	KindAlertRule      Kind = "alert_rule"
	KindNotifyChannel  Kind = "notify_channel"
	KindEvent          Kind = "event"
	KindClusterKey     Kind = "cluster_key"
)

type Store 

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

Store is the in-memory cluster state. Zero value is not usable; call New.

func New 

func New() *Store

New returns an empty Store.

func (*Store) App 

func (s *Store) App(id string) (*zatterav1.App, bool)

func (*Store) AppByName 

func (s *Store) AppByName(projectID, name string) (*zatterav1.App, bool)

func (*Store) AppendAudit 

func (s *Store) AppendAudit(entries []*zatterav1.AuditEntry)

func (*Store) AppendEvents 

func (s *Store) AppendEvents(events []*zatterav1.Event)

func (*Store) Assignment 

func (s *Store) Assignment(id string) (*zatterav1.Assignment, bool)

func (*Store) AuditSince 

func (s *Store) AuditSince(sinceMs int64) []*zatterav1.AuditEntry

AuditSince returns every audit entry created at or after sinceMs, in append order (oldest first). The archiver walks the ring forward from its cursor with this; unlike QueryAudit it is uncapped, because skipping entries would silently lose them from the archive.

func (*Store) BackupConfig 

func (s *Store) BackupConfig() (*zatterav1.BackupConfig, bool)

func (*Store) Build 

func (s *Store) Build(id string) (*zatterav1.Build, bool)

func (*Store) ClusterKeyMaterial 

func (s *Store) ClusterKeyMaterial() (*zatterav1.ClusterKeyMaterial, bool)

func (*Store) ClusterVersionRange 

func (s *Store) ClusterVersionRange() (min, max string, anyUnknown bool)

ClusterVersionRange returns the oldest and newest comparable binary versions across all nodes, and whether any node's version could not be parsed (T-93). The minimum gates upgrade ordering and version-skew warnings; an unknown version is reported rather than silently treated as old.

func (*Store) DeleteAlertRule 

func (s *Store) DeleteAlertRule(id string)

func (*Store) DeleteApp 

func (s *Store) DeleteApp(id string)

func (*Store) DeleteAssignments 

func (s *Store) DeleteAssignments(ids []string)

func (*Store) DeleteDNSProvider 

func (s *Store) DeleteDNSProvider(id string)

func (*Store) DeleteDomain 

func (s *Store) DeleteDomain(id string)

func (*Store) DeleteEnvironment 

func (s *Store) DeleteEnvironment(id string)

func (*Store) DeleteKV 

func (s *Store) DeleteKV(key string, expectedVersion int64) error

func (*Store) DeleteNode 

func (s *Store) DeleteNode(id string)

func (*Store) DeleteNotificationChannel 

func (s *Store) DeleteNotificationChannel(id string)

func (*Store) DeleteProject 

func (s *Store) DeleteProject(id string)

func (*Store) DeleteProjectMember 

func (s *Store) DeleteProjectMember(projectID, userID string)

func (*Store) DeleteRelease 

func (s *Store) DeleteRelease(id string)

DeleteRelease removes a release (retention GC, T-38).

func (*Store) DeleteToken 

func (s *Store) DeleteToken(id string)

func (*Store) DeleteUser 

func (s *Store) DeleteUser(id string)

func (*Store) DeleteVolume 

func (s *Store) DeleteVolume(id string)

func (*Store) DeleteVolumeSnapshot 

func (s *Store) DeleteVolumeSnapshot(id string)

func (*Store) Deployment 

func (s *Store) Deployment(id string) (*zatterav1.Deployment, bool)

func (*Store) Domain 

func (s *Store) Domain(id string) (*zatterav1.Domain, bool)

func (*Store) DomainByHostname 

func (s *Store) DomainByHostname(hostname string) (*zatterav1.Domain, bool)

DomainByHostname returns ONE domain for a hostname — enough for the ACME on-demand policy ("may this host get a certificate?"), which is per-hostname and indifferent to which path route asked. When a hostname carries several path routes the choice among them is arbitrary, so anything that cares about the specific route must use DomainsByHostname or DomainByRoute (T-104).

func (*Store) DomainByRoute 

func (s *Store) DomainByRoute(hostname, pathPrefix string) (*zatterav1.Domain, bool)

DomainByRoute looks up the exact (hostname, path prefix) pair — the identity of a route, and what uniqueness is enforced on.

func (*Store) DomainsByHostname 

func (s *Store) DomainsByHostname(hostname string) []*zatterav1.Domain

DomainsByHostname returns every domain registered for a hostname, ordered by path prefix (longest first) so callers see the most specific route first.

func (*Store) EnvVars 

func (s *Store) EnvVars(envID string) map[string]*zatterav1.EncryptedValue

func (*Store) Environment 

func (s *Store) Environment(id string) (*zatterav1.Environment, bool)

func (*Store) EnvironmentByName 

func (s *Store) EnvironmentByName(appID, name string) (*zatterav1.Environment, bool)

func (*Store) EventsSince 

func (s *Store) EventsSince(sinceMs int64) []*zatterav1.Event

EventsSince is AuditSince for the event ring.

func (*Store) Job 

func (s *Store) Job(id string) (*zatterav1.Job, bool)

func (*Store) KV 

func (s *Store) KV(key string) (value []byte, version int64, expiresAtUnixMs int64, ok bool)

KV returns value, version, expiry (unix ms) for a key.

func (*Store) ListAlertRules 

func (s *Store) ListAlertRules() []*zatterav1.AlertRule

func (*Store) ListApps 

func (s *Store) ListApps(projectID string) []*zatterav1.App

func (*Store) ListAssignments 

func (s *Store) ListAssignments(envID string) []*zatterav1.Assignment

ListAssignments filters by environment (or everything when empty).

func (*Store) ListAssignmentsByNode 

func (s *Store) ListAssignmentsByNode(nodeID string) []*zatterav1.Assignment

func (*Store) ListBackupRecords 

func (s *Store) ListBackupRecords() []*zatterav1.BackupRecord

func (*Store) ListBuilds 

func (s *Store) ListBuilds(appID string) []*zatterav1.Build

func (*Store) ListDNSProviders 

func (s *Store) ListDNSProviders() []*zatterav1.DNSProviderConfig

func (*Store) ListDeployments 

func (s *Store) ListDeployments(envID string) []*zatterav1.Deployment

func (*Store) ListDomains 

func (s *Store) ListDomains(projectID string) []*zatterav1.Domain

func (*Store) ListEnvironments 

func (s *Store) ListEnvironments(projectID, appID string) []*zatterav1.Environment

ListEnvironments filters by app (or project when appID == "" and projectID != ""); both empty lists everything.

func (*Store) ListEvents 

func (s *Store) ListEvents(limit int) []*zatterav1.Event

func (*Store) ListJobs 

func (s *Store) ListJobs(projectID, envID string) []*zatterav1.Job

func (*Store) ListJoinTokens 

func (s *Store) ListJoinTokens() []*zatterav1.JoinToken

func (*Store) ListKVPrefix 

func (s *Store) ListKVPrefix(prefix string) []string

func (*Store) ListMembershipsOfUser 

func (s *Store) ListMembershipsOfUser(userID string) []*zatterav1.ProjectMember

ListMembershipsOfUser returns all project memberships of one user.

func (*Store) ListNetworkAllocations 

func (s *Store) ListNetworkAllocations() []*internalv1.NetworkAllocation

func (*Store) ListNodes 

func (s *Store) ListNodes() []*zatterav1.Node

func (*Store) ListNotificationChannels 

func (s *Store) ListNotificationChannels() []*zatterav1.NotificationChannel

func (*Store) ListProjectMembers 

func (s *Store) ListProjectMembers(projectID string) []*zatterav1.ProjectMember

func (*Store) ListProjects 

func (s *Store) ListProjects() []*zatterav1.Project

func (*Store) ListReleases 

func (s *Store) ListReleases(envID string) []*zatterav1.Release

ListReleases returns an environment's releases sorted by version descending.

func (*Store) ListServiceVIPs 

func (s *Store) ListServiceVIPs() map[string]string

func (*Store) ListTokens 

func (s *Store) ListTokens(userID string) []*zatterav1.Token

func (*Store) ListUsers 

func (s *Store) ListUsers() []*zatterav1.User

func (*Store) ListVolumeSnapshots 

func (s *Store) ListVolumeSnapshots(volumeID string) []*zatterav1.VolumeSnapshot

func (*Store) ListVolumes 

func (s *Store) ListVolumes(projectID string) []*zatterav1.Volume

func (*Store) MarkApplied 

func (s *Store) MarkApplied(requestID string, raftIndex uint64) bool

MarkApplied records a request id; returns false if it was already applied.

func (*Store) NetworkAllocation 

func (s *Store) NetworkAllocation(projectID, envID, nodeID string) (string, bool)

func (*Store) NextReleaseVersion 

func (s *Store) NextReleaseVersion(envID string) uint64

NextReleaseVersion returns 1 + the highest release version of the env.

func (*Store) Node 

func (s *Store) Node(id string) (*zatterav1.Node, bool)

func (*Store) Org 

func (s *Store) Org() (*zatterav1.Org, bool)

func (*Store) Project 

func (s *Store) Project(id string) (*zatterav1.Project, bool)

func (*Store) ProjectByName 

func (s *Store) ProjectByName(name string) (*zatterav1.Project, bool)

ProjectByName resolves a project by its unique name.

func (*Store) ProjectMember 

func (s *Store) ProjectMember(projectID, userID string) (*zatterav1.ProjectMember, bool)

func (*Store) PutAlertRule 

func (s *Store) PutAlertRule(r *zatterav1.AlertRule)

func (*Store) PutApp 

func (s *Store) PutApp(a *zatterav1.App)

func (*Store) PutAssignment 

func (s *Store) PutAssignment(a *zatterav1.Assignment)

func (*Store) PutAssignments 

func (s *Store) PutAssignments(as []*zatterav1.Assignment)

PutAssignments applies a batch in one lock acquisition.

func (*Store) PutBackupRecord 

func (s *Store) PutBackupRecord(r *zatterav1.BackupRecord)

func (*Store) PutBuild 

func (s *Store) PutBuild(b *zatterav1.Build)

func (*Store) PutDNSProvider 

func (s *Store) PutDNSProvider(p *zatterav1.DNSProviderConfig)

func (*Store) PutDeployment 

func (s *Store) PutDeployment(d *zatterav1.Deployment)

func (*Store) PutDomain 

func (s *Store) PutDomain(d *zatterav1.Domain)

func (*Store) PutEnvironment 

func (s *Store) PutEnvironment(e *zatterav1.Environment)

func (*Store) PutJob 

func (s *Store) PutJob(j *zatterav1.Job)

func (*Store) PutJoinToken 

func (s *Store) PutJoinToken(t *zatterav1.JoinToken)

func (*Store) PutKV 

func (s *Store) PutKV(key string, value []byte, expectedVersion int64, expiresAtUnixMs int64) (int64, error)

PutKV stores a key. expectedVersion: -1 unconditional, 0 requires absent, >0 requires the current version to match. Returns the new version.

func (*Store) PutNode 

func (s *Store) PutNode(n *zatterav1.Node)

func (*Store) PutNotificationChannel 

func (s *Store) PutNotificationChannel(c *zatterav1.NotificationChannel)

func (*Store) PutProject 

func (s *Store) PutProject(p *zatterav1.Project)

func (*Store) PutProjectMember 

func (s *Store) PutProjectMember(m *zatterav1.ProjectMember)

func (*Store) PutRelease 

func (s *Store) PutRelease(r *zatterav1.Release)

func (*Store) PutToken 

func (s *Store) PutToken(t *zatterav1.Token)

func (*Store) PutUser 

func (s *Store) PutUser(u *zatterav1.User)

func (*Store) PutVolume 

func (s *Store) PutVolume(v *zatterav1.Volume)

func (*Store) PutVolumeSnapshot 

func (s *Store) PutVolumeSnapshot(snap *zatterav1.VolumeSnapshot)

func (*Store) QueryAudit 

func (s *Store) QueryAudit(filter func(*zatterav1.AuditEntry) bool, limit int) []*zatterav1.AuditEntry

QueryAudit returns the newest entries matching the filter, newest first.

func (*Store) QueryEvents 

func (s *Store) QueryEvents(filter func(*zatterav1.Event) bool, limit int) []*zatterav1.Event

QueryEvents returns the newest events matching the filter, newest first — the same ordering contract as QueryAudit, so the two read consistently. (ListEvents keeps its append-order tail for the alert engine, which replays events chronologically.)

func (*Store) Release 

func (s *Store) Release(id string) (*zatterav1.Release, bool)

func (*Store) RestoreProto 

func (s *Store) RestoreProto(snap *internalv1.Snapshot)

RestoreProto replaces the entire store content with the snapshot's. Watch subscribers receive a single wildcard notification per kind.

func (*Store) ServiceVIP 

func (s *Store) ServiceVIP(envID string) (string, bool)

func (*Store) SetAssignmentObserved 

func (s *Store) SetAssignmentObserved(nodeID string, observed map[string]*zatterav1.AssignmentObserved)

SetAssignmentObserved merges an observed status batch for one node. Unknown assignment ids are skipped (they may have been deleted since the agent reported).

func (*Store) SetBackupConfig 

func (s *Store) SetBackupConfig(c *zatterav1.BackupConfig)

func (*Store) SetClusterKeyMaterial 

func (s *Store) SetClusterKeyMaterial(m *zatterav1.ClusterKeyMaterial)

func (*Store) SetEnvVars 

func (s *Store) SetEnvVars(envID string, set map[string]*zatterav1.EncryptedValue, unset []string)

SetEnvVars applies a set/unset batch to an environment's variables.

func (*Store) SetNetworkAllocation 

func (s *Store) SetNetworkAllocation(projectID, envID, nodeID, subnetCIDR string)

func (*Store) SetOrg 

func (s *Store) SetOrg(o *zatterav1.Org)

func (*Store) SetServiceVIP 

func (s *Store) SetServiceVIP(envID, vip string)

func (*Store) SetVolumeLease 

func (s *Store) SetVolumeLease(volumeID string, lease *zatterav1.VolumeLease)

func (*Store) SnapshotProto 

func (s *Store) SnapshotProto(fsmIndex uint64) *internalv1.Snapshot

SnapshotProto serializes the entire store into a Snapshot message (ADR-0004). Called by the Raft FSM under snapshotting.

func (*Store) Token 

func (s *Store) Token(id string) (*zatterav1.Token, bool)

func (*Store) TokenByHash 

func (s *Store) TokenByHash(hash string) (*zatterav1.Token, bool)

TokenByHash is the auth hot path.

func (*Store) TouchTokens 

func (s *Store) TouchTokens(lastUsed map[string]int64)

TouchTokens updates last_used_at in batch (from a periodic flush).

func (*Store) User 

func (s *Store) User(id string) (*zatterav1.User, bool)

func (*Store) UserByEmail 

func (s *Store) UserByEmail(email string) (*zatterav1.User, bool)

func (*Store) Version 

func (s *Store) Version() uint64

Version returns the monotonic mutation counter.

func (*Store) Volume 

func (s *Store) Volume(id string) (*zatterav1.Volume, bool)

func (*Store) VolumeByName 

func (s *Store) VolumeByName(projectID, envID, name string) (*zatterav1.Volume, bool)

func (*Store) Watch 

func (s *Store) Watch(kinds ...Kind) *Subscription

Watch subscribes to change notifications for the given kinds (all kinds if empty). See Hub for delivery semantics.

type Subscription 

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

Subscription receives coalesced change notifications.

func (*Subscription) Close 

func (s *Subscription) Close()

Close unsubscribes. Safe to call multiple times.

func (*Subscription) Drain 

func (s *Subscription) Drain() []Change

Drain returns and clears the pending change set.

func (*Subscription) Notify 

func (s *Subscription) Notify() <-chan struct{}

Notify returns a channel that receives a value whenever there are pending changes. It is level-triggered with capacity 1: after receiving, call Drain to collect what changed, then wait again.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.