rules

Documentation

Overview

Package rules is the per-tenant rules-distribution substrate (seam P3a): a pluggable Source that delivers a typed ruleset per tenant and streams changes, plus a fail-safe local Cache that keeps a last-known-good snapshot so consumers evaluate locally — never a per-operation call to a rules service.

It is the shared machinery beneath ility evaluators such as feature flags (a Source[featureflags.FlagSet]) and tag validation (a Source of tag definitions): a synced per-tenant snapshot plus an evaluator on top. The Watch contract deliberately mirrors cells.RoutingTable.Watch — the proven read-mostly fan-out pattern already in the SDK.

Mechanism, not policy: this package distributes and caches whatever ruleset type a consumer parameterises it with; it does not interpret rules. The rules service that authors them, and the evaluator that applies them, live elsewhere (an evaluator in a public package such as featureflags; the authoring service in a separate repo, bound to a Source the same way an external adapter implements the authz.Authorizer seam).

The package depends only on the standard library and the SDK's health seam. Heavy transports — fsnotify, OPA, a Kubernetes ConfigMap bridge, Kafka — are separate adapters built just-in-time, never in core, so the root module stays dependency-light.

Index

• Constants
• Variables
• type Cache
  • func NewCache[T any](name string, src Source[T]) *Cache[T]
  • func (c *Cache[T]) Check(_ context.Context) error
  • func (c *Cache[T]) Get(tenant string) (T, bool)
  • func (c *Cache[T]) Name() string
  • func (c *Cache[T]) Ready() bool
  • func (c *Cache[T]) Run(ctx context.Context) error
• type Event
• type FileSource
  • func NewFileSource[T any](path string, interval time.Duration) *FileSource[T]
  • func (f *FileSource[T]) Load() error
  • func (f *FileSource[T]) Run(ctx context.Context) error
  • func (f *FileSource[T]) Snapshot(ctx context.Context) (map[string]T, uint64, error)
• type Snapshotter
• type Source
• type StaticSource
  • func NewStaticSource[T any]() *StaticSource[T]
  • func (s *StaticSource[T]) Delete(tenant string)
  • func (s *StaticSource[T]) Get(_ context.Context, tenant string) (T, error)
  • func (s *StaticSource[T]) Replace(next map[string]T)
  • func (s *StaticSource[T]) Set(tenant string, v T)
  • func (s *StaticSource[T]) Snapshot(_ context.Context) (map[string]T, uint64, error)
  • func (s *StaticSource[T]) Watch(ctx context.Context) (<-chan Event[T], error)

Constants

const DefaultPollInterval = 5 * time.Second

DefaultPollInterval is the modtime poll cadence used by NewFileSource when the caller passes a non-positive interval.

Variables

var ErrNotFound = errors.New("rules: no ruleset for tenant")

ErrNotFound is returned by Source.Get when a tenant has no ruleset. Callers treat it as "use the default", not as an error.

Types

type Cache

type Cache[T any] struct {
	// contains filtered or unexported fields
}

Cache is the consumer-side fail-safe snapshot over any Source. It does the initial load (via Snapshotter when the source supports it), then subscribes to Source.Watch and keeps a last-known-good copy of every tenant's ruleset in memory. Consumers read from that copy, so evaluation is local and a source outage degrades to stale-but-available data rather than a failed request — the same decoupling the change feed gets from the outbox.

A Cache is NOT ready until its initial load succeeds (or, for a source that is not a Snapshotter, until the Watch subscription is established). It implements [health.Check] so readiness wires into the server's /readyz and gRPC health endpoints: a service does not report ready until its rules are loaded.

The zero value is not usable; construct with NewCache.

func NewCache

func NewCache[T any](name string, src Source[T]) *Cache[T]

NewCache returns a cache over src. name identifies it in readiness output (e.g. "feature-flags"). Call Cache.Run once to start syncing.

func (*Cache[T]) Check

func (c *Cache[T]) Check(_ context.Context) error

Check implements health.Check: nil once the cache has loaded, otherwise an error so /readyz reports the service as not ready.

func (*Cache[T]) Get

func (c *Cache[T]) Get(tenant string) (T, bool)

Get returns the cached ruleset for tenant and true, or the zero value and false when the tenant has no ruleset (the caller falls back to a default). It reads from memory and never blocks on the source.

func (*Cache[T]) Name

func (c *Cache[T]) Name() string

Name implements health.Check.

func (*Cache[T]) Ready

func (c *Cache[T]) Ready() bool

Ready reports whether the initial load has completed. Before the first successful load it is false; consumers may still call Get (it returns not-found and they use defaults), but readiness gates traffic.

func (*Cache[T]) Run

func (c *Cache[T]) Run(ctx context.Context) error

Run subscribes to the source, loads the initial snapshot, then applies Watch events until ctx is cancelled. It blocks; run it in a goroutine. Run returns the context error on cancellation, or a non-nil error if the initial Watch subscription fails.

Watch is opened BEFORE the snapshot so no change is missed in the gap; events that predate the snapshot are buffered and skipped by revision, so the cache converges correctly. A failed initial Snapshot is non-fatal: the cache stays not-ready and becomes ready on the first event that arrives, so a rules source that is down at startup degrades to "not ready yet" rather than crashing the consumer.

type Event

type Event[T any] struct {
	// Tenant is the affected tenant. The empty string denotes the global/default
	// ruleset that applies when a tenant has no entry of its own.
	Tenant string
	// Value is the tenant's new ruleset (the zero value of T when Deleted).
	Value T
	// Deleted reports that the tenant's ruleset was removed; consumers fall back
	// to the default ruleset.
	Deleted bool
	// Revision is the source-global monotonic revision at which this change
	// applied. It only ever increases.
	Revision uint64
}

Event is a single change observed on a Source.Watch stream.

type FileSource

type FileSource[T any] struct {
	*StaticSource[T]
	// contains filtered or unexported fields
}

FileSource is a zero-dependency Source backed by a JSON file that maps tenant IDs to rulesets, e.g.

{"tenant-a": {<ruleset>}, "tenant-b": {<ruleset>}, "": {<default ruleset>}}

It polls the file's modification time on an interval and reloads on change — a "file-watch" without an fsnotify dependency. This suits rules delivered as a Kubernetes ConfigMap mounted into the pod (the kubelet rewrites the file on update, bumping its modtime). An fsnotify-backed source, or a real ConfigMap API bridge, is a separate adapter built when needed.

FileSource embeds a StaticSource for in-memory storage and watcher fan-out; the poll loop reconciles the file's contents into it via Replace, so a Cache over a FileSource gets the same Get/Watch/Snapshot behaviour as over any other source.

The zero value is not usable; construct with NewFileSource.

func NewFileSource

func NewFileSource[T any](path string, interval time.Duration) *FileSource[T]

NewFileSource returns a file-backed source reading path. A non-positive interval defaults to DefaultPollInterval. Call FileSource.Run to start polling; call FileSource.Load once first if you need the data present synchronously before serving.

func (*FileSource[T]) Load

func (f *FileSource[T]) Load() error

Load reads the file and reconciles it into the in-memory store if the file's modtime has changed since the last load (or on the first call). It is safe to call repeatedly. A read or parse error is returned and leaves the last good state intact — fail-safe: a momentarily unreadable or malformed file does not wipe loaded rules.

func (*FileSource[T]) Run

func (f *FileSource[T]) Run(ctx context.Context) error

Run loads once, then polls for modtime changes on the configured interval and reloads on change, until ctx is cancelled. It blocks; run it in a goroutine. Poll errors (a transiently missing or malformed file) are swallowed so a bad write does not stop the loop or clear loaded rules; the next good poll recovers. Run returns the context error on cancellation.

func (*FileSource[T]) Snapshot

func (f *FileSource[T]) Snapshot(ctx context.Context) (map[string]T, uint64, error)

Snapshot ensures the file has been read at least once, then returns the in-memory snapshot. It overrides the embedded StaticSource.Snapshot so a Cache's initial load reflects the file (and only reports ready once the file has been read). If the file cannot be read and nothing has loaded yet, the read error propagates and the cache stays not-ready; if a prior good load exists, that stale-but-good snapshot is returned (fail-safe).

type Snapshotter

type Snapshotter[T any] interface {
	// Snapshot returns every tenant's ruleset and the revision the snapshot was
	// taken at. The returned map is owned by the caller.
	Snapshot(ctx context.Context) (map[string]T, uint64, error)
}

Snapshotter is an optional Source capability: a bulk read of every tenant's ruleset at a revision. Cache uses it for the initial load so the cache is ready with complete data before it serves, then switches to Watch for incremental updates (the standard list-then-watch pattern). Sources that can enumerate their tenants — StaticSource, FileSource, a ConfigMap bridge — implement it; a pure event stream need not, in which case the Cache becomes ready once its Watch subscription is established.

type Source

type Source[T any] interface {
	// Get returns the current ruleset for tenant, or ErrNotFound when the tenant
	// has no explicit ruleset.
	Get(ctx context.Context, tenant string) (T, error)

	// Watch returns a channel of changes. The channel is closed when ctx is
	// cancelled. Implementations may coalesce rapid updates but must always
	// deliver the latest state for any changed tenant. Mirrors
	// cells.RoutingTable.Watch.
	Watch(ctx context.Context) (<-chan Event[T], error)
}

Source delivers a typed ruleset T per tenant and streams changes. It is the pluggable transport seam (P3a): the in-memory StaticSource is the dev/test default and FileSource is the zero-dependency file-backed default; a ConfigMap bridge, an OPA-backed source, or a Kafka-backed source implement the same interface in adapters.

Implementations must be safe for concurrent use.

type StaticSource

type StaticSource[T any] struct {
	// contains filtered or unexported fields
}

StaticSource is the in-memory Source — the dev/test default. It is concurrency-safe and supports live updates via Set, Delete, and Replace, each of which broadcasts to watchers so a Cache over it stays current. Production transports (FileSource, a ConfigMap bridge, OPA) implement Source in adapters; tests and OSS dev defaults use this.

The zero value is not usable; construct with NewStaticSource.

func NewStaticSource

func NewStaticSource[T any]() *StaticSource[T]

NewStaticSource returns an empty in-memory source.

func (*StaticSource[T]) Delete

func (s *StaticSource[T]) Delete(tenant string)

Delete removes the ruleset for tenant (a no-op if absent) and broadcasts a delete event so consumers revert the tenant to the default ruleset.

func (*StaticSource[T]) Get

func (s *StaticSource[T]) Get(_ context.Context, tenant string) (T, error)

Get implements Source.

func (*StaticSource[T]) Replace

func (s *StaticSource[T]) Replace(next map[string]T)

Replace atomically swaps the entire tenant→ruleset map: tenants in next are upserted and tenants present before but absent from next are deleted. Each resulting change is broadcast. Useful for a source that reloads a whole document (e.g. a file or ConfigMap) and reconciles to it.

func (*StaticSource[T]) Set

func (s *StaticSource[T]) Set(tenant string, v T)

Set upserts the ruleset for tenant and broadcasts the change.

func (*StaticSource[T]) Snapshot

func (s *StaticSource[T]) Snapshot(_ context.Context) (map[string]T, uint64, error)

Snapshot implements Snapshotter.

func (*StaticSource[T]) Watch

func (s *StaticSource[T]) Watch(ctx context.Context) (<-chan Event[T], error)

Watch implements Source. The returned channel is closed when ctx is cancelled.