diff --git a/internal/querysignalstest/harness.go b/internal/querysignalstest/harness.go new file mode 100644 index 000000000..d04941edd --- /dev/null +++ b/internal/querysignalstest/harness.go @@ -0,0 +1,39 @@ +// Package querysignalstest supplies shared fixtures for signal tests. +package querysignalstest + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/internal/datastore/revisions" + "github.com/authzed/spicedb/pkg/datastore" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/query/signals" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +// Owner creates an isolated signal owner using the supplied providers. +func Owner(t testing.TB, ds ...signal.Descriptor) *signals.Owner { + t.Helper() + r, err := signals.NewRegistry(ds...) + require.NoError(t, err) + o, err := signals.NewOwner(r, signals.Options{}) + require.NoError(t, err) + return o +} + +// Stored creates an empty stored-schema fixture with the supplied hash. +func Stored(hash string) *datastore.ReadOnlyStoredSchema { + return datastore.NewReadOnlyStoredSchema( + &core.StoredSchema{VersionOneof: &core.StoredSchema_V1{V1: &core.StoredSchema_V1StoredSchema{SchemaHash: hash}}}, + ) +} + +// View binds a test owner to a schema hash and transaction revision. +func View(t testing.TB, o *signals.Owner, hash string, revision uint64) *signals.View { + t.Helper() + v, err := o.Bind(Stored(hash), signal.Scope{SchemaHash: hash, Revision: revisions.NewForTransactionID(revision)}) + require.NoError(t, err) + return v +} diff --git a/internal/querysignalstest/schema.go b/internal/querysignalstest/schema.go new file mode 100644 index 000000000..2dd0030ab --- /dev/null +++ b/internal/querysignalstest/schema.go @@ -0,0 +1,44 @@ +package querysignalstest + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/pkg/datastore" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/schemadsl/compiler" + "github.com/authzed/spicedb/pkg/schemadsl/input" +) + +// Schema compiles schema text into a stored-schema fixture with the supplied hash. +func Schema(t testing.TB, hash, text string) *datastore.ReadOnlyStoredSchema { + t.Helper() + c, err := compiler.Compile( + compiler.InputSchema{Source: input.Source("signal-test"), SchemaString: text}, + compiler.AllowUnprefixedObjectType(), + ) + require.NoError(t, err) + namespaces := map[string]*core.NamespaceDefinition{} + + for _, d := range c.ObjectDefinitions { + namespaces[d.Name] = d + } + + caveats := map[string]*core.CaveatDefinition{} + + for _, d := range c.CaveatDefinitions { + caveats[d.Name] = d + } + + return datastore.NewReadOnlyStoredSchema(&core.StoredSchema{ + VersionOneof: &core.StoredSchema_V1{ + V1: &core.StoredSchema_V1StoredSchema{ + SchemaHash: hash, + SchemaText: text, + NamespaceDefinitions: namespaces, + CaveatDefinitions: caveats, + }, + }, + }) +} diff --git a/pkg/query/signals/README.md b/pkg/query/signals/README.md new file mode 100644 index 000000000..8a5c451cd --- /dev/null +++ b/pkg/query/signals/README.md @@ -0,0 +1,76 @@ +# Planning signals + +Signals store observations that can inform query-planner choices. Each provider +owns a named definition, typed population key, observation and estimate, and an +accumulator. This package supplies registration, bounded storage and scoped reads; +it does not change query execution or enable a particular optimization. + +## Registration and scope + +Construct descriptors with `signal.NewSource` or `signal.NewDerived`, register +all dependencies, and create one `Owner` for the datalayer using `NewOwner`. +Registration rejects duplicate names, incompatible bindings and dependency cycles. +Freeze a registry before sharing it concurrently. Provider callbacks must follow +the accumulator contract: clones and returned values are independent snapshots. + +Bind the owner to a loaded stored schema and the request's selected datastore +revision. A provider explicitly chooses `RevisionBound` or `CrossRevision`: + +- Revision-bound evidence stays at that exact revision, including a quantized + revision selected for checks. Neighboring revisions do not share that evidence. +- Cross-revision evidence can be reused within the same schema. Its provider owns + aging and availability; storage does not infer how much the data changed. +- On a schema change, only cross-revision providers with an explicit compatibility + callback can transfer evidence. `signal.Unchanged` compares their declared + dependencies against the schema diff. A transfer may retain, discount or reset + evidence. It preserves observation times but starts a distinct publication epoch. + +For example, checks of `document:report#view@user:alice` at revision R may record +an observation under a population key for the `document#view` subtree. A second +subject can contribute to the same population if the provider's key excludes the +subject. A revision-bound lookup at R+1 is missing; a cross-revision provider may +return historical evidence. Adding an unrelated schema definition can preserve +that historical evidence, while changing a declared dependency resets it. + +Consumers use `signal.Observe(view, input, key, observation)` and +`signal.Lookup(view, reference, key)`. Check `estimate.IsAvailable()` before using +the value. Missing, expired and disabled evidence are distinct from a measured +zero. `LookupFirst` tries compatible references in order and reports which one +supplied the first available estimate; a lookup error stops that search. + +`view.ForPlan()` memoizes each population's first independent snapshot for one +planning run. Further observations or store eviction cannot change that snapshot. +Different populations are read independently, not as one transactional snapshot. + +## Ownership and bounds + +Defaults are 16 MiB total estimated resident storage, 256 KiB per entry, four +schemas, 64 exact revisions per schema, and 64 weak owner handles per stored +schema. Options can override these limits. Storage accounts for retained schemas, +entry and key storage, dependency footprints, and accumulator-reported bytes. +Providers must include all their retained state in `Size()`. + +Keys support scalar values, strings, and arrays or structs recursively containing +those values. Struct fields must be exported. Pointers, interfaces, channels and +other reference graphs are rejected, as are nonreflexive values such as NaN. +Retained keys and footprint strings are copied so a substring cannot retain an +unaccounted backing buffer. Footprint slices are copied before retention. + +Oversized or evicted evidence becomes unavailable; it does not affect permission +semantics. The schema cache holds only bounded weak owner references. Owners and +schemas do not leak learning state between independent datalayers. Legacy readers +can use an isolated local view instead of a shared stored-schema attachment. + +Derived providers consume accepted source increments in dependency order, not +cumulative snapshots. Source publication is serialized through derived delivery. +They cannot be written directly through the public observation API. + +## Tests + +Package tests use synthetic providers and stored-schema fixtures to cover scope +isolation, publication, snapshots, eviction, schema compatibility and concurrency. +The shared `internal/querysignalstest` helpers provide owner and schema +builders for provider-specific tests. Numeric distribution helpers are tested in +`signal` without depending on a built-in provider. + +Time-sensitive tests use `testing/synctest` to control time without a custom clock. diff --git a/pkg/query/signals/key.go b/pkg/query/signals/key.go new file mode 100644 index 000000000..ab36308fc --- /dev/null +++ b/pkg/query/signals/key.go @@ -0,0 +1,144 @@ +package signals + +import ( + "errors" + "reflect" + "strings" + + "github.com/ccoveille/go-safecast/v2" + + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +// Value-only keys make retention predictable: no mutable object graph can hide +// behind a pointer or interface. Exported fields let the store own string data. +func supportedKey(t reflect.Type) bool { + switch t.Kind() { + case reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, + reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr, + reflect.Float32, reflect.Float64, reflect.Complex64, reflect.Complex128, reflect.String: + return true + case reflect.Array: + return supportedKey(t.Elem()) + case reflect.Struct: + for i := range t.NumField() { + f := t.Field(i) + if !f.IsExported() || !supportedKey(f.Type) { + return false + } + } + + return true + default: + return false + } +} + +func validateKey(key any, expected reflect.Type) error { + if reflect.TypeOf(key) != expected || !reflect.ValueOf(key).Comparable() { + return errors.New("invalid signal key type") + } + + // NaN, including a NaN inside an array or struct, cannot be found or deleted + // from a map. Reject it before lookup as well as before admission. + if !reflect.ValueOf(key).Equal(reflect.ValueOf(key)) { + return errors.New("nonreflexive signal key") + } + + return nil +} + +func ownedKey(value reflect.Value) reflect.Value { + switch value.Kind() { + case reflect.String: + snapshot := reflect.New(value.Type()).Elem() + snapshot.SetString(strings.Clone(value.String())) + return snapshot + case reflect.Struct: + snapshot := reflect.New(value.Type()).Elem() + + for i := range value.NumField() { + snapshot.Field(i).Set(ownedKey(value.Field(i))) + } + + return snapshot + case reflect.Array: + snapshot := reflect.New(value.Type()).Elem() + + for i := range value.Len() { + snapshot.Index(i).Set(ownedKey(value.Index(i))) + } + + return snapshot + default: + return value + } +} + +// String headers and inline fields are in Type.Size; only backing bytes are extra. +func keyStringBytes(value reflect.Value) int64 { + switch value.Kind() { + case reflect.String: + return int64(value.Len()) + case reflect.Struct: + var size int64 + + for i := range value.NumField() { + size += keyStringBytes(value.Field(i)) + } + + return size + case reflect.Array: + var size int64 + + for i := range value.Len() { + size += keyStringBytes(value.Index(i)) + } + + return size + default: + return 0 + } +} + +func ownedFootprint(f signal.Footprint) signal.Footprint { + result := signal.Footprint{ + Elements: make([]signal.Element, len(f.Elements)), + Caveats: make([]string, len(f.Caveats)), + } + + for i, e := range f.Elements { + result.Elements[i] = signal.Element{ + Namespace: strings.Clone(e.Namespace), + Relation: strings.Clone(e.Relation), + } + } + + for i, name := range f.Caveats { + result.Caveats[i] = strings.Clone(name) + } + + return result +} + +func entryCost(key entryKey, state signal.State, footprint signal.Footprint) int64 { + // Include map/entry overhead, retained names, inline key storage, string + // backing bytes and the dependency slices as well as provider-owned state. + size := int64(256) + size += safecast.MustConvert[int64](len(key.schema)) + size += safecast.MustConvert[int64](len(key.name)) + size += safecast.MustConvert[int64](len(key.revision)) + size += safecast.MustConvert[int64](reflect.TypeOf(key.key).Size()) + keyStringBytes(reflect.ValueOf(key.key)) + size += safecast.MustConvert[int64](len(footprint.Elements)) * safecast.MustConvert[int64](reflect.TypeFor[signal.Element]().Size()) + size += safecast.MustConvert[int64](len(footprint.Caveats)) * safecast.MustConvert[int64](reflect.TypeFor[string]().Size()) + + for _, e := range footprint.Elements { + size += safecast.MustConvert[int64](len(e.Namespace)) + safecast.MustConvert[int64](len(e.Relation)) + } + + for _, name := range footprint.Caveats { + size += safecast.MustConvert[int64](len(name)) + } + + return size + state.Size() +} diff --git a/pkg/query/signals/lifecycle_test.go b/pkg/query/signals/lifecycle_test.go new file mode 100644 index 000000000..f593754ee --- /dev/null +++ b/pkg/query/signals/lifecycle_test.go @@ -0,0 +1,248 @@ +package signals_test + +import ( + "errors" + "reflect" + "testing" + "testing/synctest" + "time" + + "github.com/ccoveille/go-safecast/v2" + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/internal/querysignalstest" + "github.com/authzed/spicedb/pkg/query/signals" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +func TestCompositePopulationsRemainIndependent(t *testing.T) { + type key struct { + Names [2]string + Mode uint8 + } + + d := signal.NewSource[key, int, int]("sum", signal.SourceConfig[key, int, int]{ + New: func() signal.Accumulator[int, int] { return &sum{} }, + }) + o := querysignalstest.Owner(t, d) + v := querysignalstest.View(t, o, "schema", 1) + a := key{Names: [2]string{"doc", "view"}, Mode: 1} + b := a + b.Mode = 2 + + require.NoError(t, signal.Observe(v, signal.Input[key, int]{Name: "sum"}, a, 7)) + e, err := signal.Lookup(v, signal.Reference[key, int]{Name: "sum"}, a) + + require.NoError(t, err) + require.Equal(t, 7, e.Value) + e, err = signal.Lookup(v, signal.Reference[key, int]{Name: "sum"}, b) + + require.NoError(t, err) + require.False(t, e.IsAvailable()) + _, _, _, err = v.Read("sum", []string{"bad"}, reflect.TypeFor[key](), reflect.TypeFor[int]()) + + require.ErrorContains(t, err, "key") +} + +func TestEvidenceProfileAndLegacyIsolation(t *testing.T) { + r, err := signals.NewRegistry(definition("sum")) + + require.NoError(t, err) + require.True(t, r.Contains("sum")) + require.False(t, r.Contains("missing")) + _, err = signals.NewOwner(nil, signals.Options{}) + + require.Error(t, err) + _, err = signals.NewOwner(r, signals.Options{MaxSchemas: -1}) + + require.Error(t, err) + _, err = signals.NewOwner(r, signals.Options{Enabled: []string{"missing"}}) + + require.Error(t, err) + disabled, err := signals.NewOwner(r, signals.Options{Enabled: []string{}}) + + require.NoError(t, err) + v := view(t, disabled, "schema", 1) + + require.False(t, v.Enabled("sum")) + require.Same(t, r, v.Registry()) + require.NoError(t, signal.Observe(v, sumInput, "x", 1)) + e, err := signal.Lookup(v, sumRef, "x") + + require.NoError(t, err) + require.Equal(t, signal.Unsupported, e.Availability) + enabled, err := signals.NewOwner(r, signals.Options{Enabled: []string{"sum"}}) + + require.NoError(t, err) + a := enabled.Local(signal.Scope{}) + b := enabled.Local(signal.Scope{}) + + require.NoError(t, signal.Observe(a, sumInput, "x", 7)) + e, err = signal.Lookup(b, sumRef, "x") + + require.NoError(t, err) + require.False(t, e.IsAvailable()) +} + +func TestStorageBudgetsEvictWholePopulations(t *testing.T) { + t.Run("schema count", func(t *testing.T) { + o := owner(t, signals.Options{MaxSchemas: 1}) + a := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(a, sumInput, "x", 1)) + b := view(t, o, "b", 1) + + require.NoError(t, signal.Observe(b, sumInput, "x", 2)) + e, err := signal.Lookup(a, sumRef, "x") + + require.NoError(t, err) + require.False(t, e.IsAvailable()) + }) + + t.Run("total bytes", func(t *testing.T) { + budget := stored("a").EstimatedSize() + 256 + 320 + o := owner(t, signals.Options{MaxBytes: budget}) + a := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(a, sumInput, "first", 1)) + require.NoError(t, signal.Observe(a, sumInput, "second", 2)) + first, err := signal.Lookup(a, sumRef, "first") + + require.NoError(t, err) + require.False(t, first.IsAvailable()) + second, err := signal.Lookup(a, sumRef, "second") + + require.NoError(t, err) + require.Equal(t, 2, second.Value) + require.LessOrEqual(t, o.ResidentBytes(), budget) + }) + + t.Run("oversized schema", func(t *testing.T) { + o := owner(t, signals.Options{MaxBytes: 1}) + a := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(a, sumInput, "x", 1)) + e, err := signal.Lookup(a, sumRef, "x") + + require.NoError(t, err) + require.False(t, e.IsAvailable()) + require.LessOrEqual(t, o.ResidentBytes(), int64(1)) + }) +} + +type qualityEstimate struct{ count int } + +func (e qualityEstimate) SignalSamples() uint64 { return safecast.MustConvert[uint64](e.count) } +func (e qualityEstimate) SignalWeight() float64 { return float64(e.count) } + +type growingAccumulator struct{ count int } + +func (a *growingAccumulator) Add(value int) error { + if value < 0 { + return errors.New("negative observation") + } + + a.count += value + return nil +} + +func (a *growingAccumulator) Value(time.Time) (qualityEstimate, signal.Availability) { + return qualityEstimate{count: a.count}, signal.Available +} + +func (a *growingAccumulator) Size() int64 { return 8 + safecast.MustConvert[int64](a.count)*16 } + +func (a *growingAccumulator) Clone() signal.Accumulator[int, qualityEstimate] { + return &growingAccumulator{count: a.count} +} + +func (a *growingAccumulator) DiscountEvidence(factor float64) { + a.count = int(float64(a.count) * factor) +} + +func TestPublicationMetadataAndGrowingEntry(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + observed := time.Now() + var updates []signal.UpdateID + d := signal.NewSource[string, int, qualityEstimate]("size", signal.SourceConfig[string, int, qualityEstimate]{ + New: func() signal.Accumulator[int, qualityEstimate] { return &growingAccumulator{} }, + Accept: func(k string, _ int) bool { return k != "ignored" }, + Stamp: func(value int, id signal.UpdateID, at time.Time) int { + updates = append(updates, id) + require.Equal(t, observed, at) + return value + }, + ObservationTime: func(int) time.Time { return observed }, + }) + r, err := signals.NewRegistry(d) + + require.NoError(t, err) + o, err := signals.NewOwner(r, signals.Options{MaxEntryBytes: 512}) + + require.NoError(t, err) + v := view(t, o, "a", 1) + input := signal.Input[string, int]{Name: "size"} + ref := signal.Reference[string, qualityEstimate]{Name: "size"} + + require.NoError(t, signal.Observe(v, input, "ignored", 1)) + require.NoError(t, signal.Observe(v, input, "x", 4)) + require.ErrorContains(t, signal.Observe(v, input, "x", -1), "negative") + plan := v.ForPlan() + before, err := signal.Lookup(plan, ref, "x") + + require.NoError(t, err) + require.Equal(t, uint64(4), before.Evidence.Samples) + require.InDelta(t, 4.0, before.Evidence.EffectiveWeight, 1e-9) + time.Sleep(time.Minute) + + require.NoError(t, signal.Observe(v, input, "x", 64)) + fresh, err := signal.Lookup(v, ref, "x") + + require.NoError(t, err) + require.False(t, fresh.IsAvailable()) + after, err := signal.Lookup(plan, ref, "x") + + require.NoError(t, err) + require.Equal(t, before, after) + require.Len(t, updates, 2) + require.Equal(t, updates[0].Epoch, updates[1].Epoch) + require.Equal(t, updates[0].Sequence+1, updates[1].Sequence) + }) +} + +func TestSchemaTransferDecisions(t *testing.T) { + for _, tc := range []struct { + name string + action signal.TransferAction + factor float64 + keep bool + }{ + {"retain", signal.Retain, 1, true}, + {"discount", signal.Discount, .5, true}, + {"reset", signal.Reset, 0, false}, + {"invalid factor", signal.Discount, 1, false}, + } { + t.Run(tc.name, func(t *testing.T) { + d := signal.NewSource[string, int, qualityEstimate]("weighted", signal.SourceConfig[string, int, qualityEstimate]{ + Policy: signal.CrossRevision, + New: func() signal.Accumulator[int, qualityEstimate] { return &growingAccumulator{} }, + Transfer: func(signal.SchemaChange, signal.Footprint) signal.TransferDecision { + return signal.TransferDecision{Action: tc.action, Factor: tc.factor} + }, + }) + o := querysignalstest.Owner(t, d) + a := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(a, signal.Input[string, int]{Name: "weighted"}, "x", 8)) + b := view(t, o, "b", 1) + e, err := signal.Lookup(b, signal.Reference[string, qualityEstimate]{Name: "weighted"}, "x") + + require.NoError(t, err) + require.Equal(t, tc.keep, e.IsAvailable()) + + if tc.keep { + require.Equal(t, int(8*tc.factor), e.Value.count) + } + }) + } +} diff --git a/pkg/query/signals/options.go b/pkg/query/signals/options.go new file mode 100644 index 000000000..3d0fd8aa3 --- /dev/null +++ b/pkg/query/signals/options.go @@ -0,0 +1,55 @@ +package signals + +import "time" + +// Options configures signal storage limits and its clock. +type Options struct { + // Now supplies the clock used for evidence publication and aging. + Now func() time.Time + + // MaxBytes limits the total resident storage budget in bytes. + MaxBytes int64 + + // MaxEntryBytes limits the resident size of one evidence entry in bytes. + MaxEntryBytes int64 + + // MaxSchemas limits the number of resident schema versions. + MaxSchemas int + + // MaxRevisions limits resident exact-revision buckets per schema. + MaxRevisions int + + // MaxSchemaHandles limits owner handles attached to a stored schema. + MaxSchemaHandles int + + // Enabled names select a learning profile; nil enables the whole registry. + Enabled []string +} + +func (o Options) defaults() Options { + if o.Now == nil { + o.Now = time.Now + } + + if o.MaxBytes == 0 { + o.MaxBytes = 16 << 20 + } + + if o.MaxEntryBytes == 0 { + o.MaxEntryBytes = 256 << 10 + } + + if o.MaxSchemas == 0 { + o.MaxSchemas = 4 + } + + if o.MaxSchemaHandles == 0 { + o.MaxSchemaHandles = 64 + } + + if o.MaxRevisions == 0 { + o.MaxRevisions = 64 + } + + return o +} diff --git a/pkg/query/signals/owner.go b/pkg/query/signals/owner.go new file mode 100644 index 000000000..30d9ffb02 --- /dev/null +++ b/pkg/query/signals/owner.go @@ -0,0 +1,514 @@ +package signals + +import ( + "errors" + "fmt" + "reflect" + "strings" + "sync" + "sync/atomic" + "weak" + + "github.com/authzed/spicedb/pkg/datastore" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +type entryKey struct { + schema string + name signal.Name + revision string + key any +} + +type entry struct { + publicationMu sync.Mutex + publicationVersion uint64 // GUARDED_BY(publicationMu) + mu sync.Mutex + state signal.State // GUARDED_BY(mu) + evidence signal.Evidence // GUARDED_BY(mu) + footprint signal.Footprint + cost int64 + used uint64 +} + +type schemaState struct { + ready chan struct{} + definitions *core.StoredSchema + cost int64 + used uint64 +} + +// Owner isolates a bounded learning store for one datalayer/server. +type Owner struct { + id uint64 + mu sync.Mutex + registry *Registry + options Options + entries map[entryKey]*entry // GUARDED_BY(mu) + schemas map[string]*schemaState + enabled map[signal.Name]bool + bytes int64 + sequence uint64 +} + +var ( + epochs atomic.Uint64 + schemaHandleKey = datastore.NewDerivedCacheKey[*schemaHandles]() +) + +type schemaHandles struct { + mu sync.Mutex + owners map[uint64]weak.Pointer[Owner] // GUARDED_BY(mu) +} + +func (h *schemaHandles) bind(owner *Owner) { + h.mu.Lock() + defer h.mu.Unlock() + + for id, p := range h.owners { + if p.Value() == nil { + delete(h.owners, id) + } + } + + if _, exists := h.owners[owner.id]; exists { + return + } + + for len(h.owners) >= owner.options.MaxSchemaHandles { + oldest := ^uint64(0) + + for id := range h.owners { + oldest = min(oldest, id) + } + + delete(h.owners, oldest) + } + + h.owners[owner.id] = weak.Make(owner) +} + +// NewOwner creates a bounded signal store from a validated registry and learning profile. +func NewOwner(r *Registry, opts Options) (*Owner, error) { + if r == nil { + return nil, errors.New("nil signal registry") + } + + if !r.frozen { + if err := r.Freeze(); err != nil { + return nil, err + } + } + + opts = opts.defaults() + if opts.MaxBytes < 1 || opts.MaxEntryBytes < 1 || opts.MaxSchemas < 1 || opts.MaxRevisions < 1 || opts.MaxSchemaHandles < 1 { + return nil, errors.New("invalid signal storage limits") + } + + o := &Owner{ + id: epochs.Add(1), + registry: r, + options: opts, + entries: map[entryKey]*entry{}, + schemas: map[string]*schemaState{}, + enabled: map[signal.Name]bool{}, + } + var enable func(signal.Name) error + enable = func(n signal.Name) error { + d, ok := r.definitions[n] + if !ok { + return fmt.Errorf("unknown enabled signal %q", n) + } + + if o.enabled[n] { + return nil + } + + o.enabled[n] = true + + for _, dep := range d.Dependencies { + if err := enable(dep); err != nil { + return err + } + } + + return nil + } + + if opts.Enabled == nil { + for n := range r.definitions { + if err := enable(n); err != nil { + return nil, err + } + } + } else { + for _, n := range opts.Enabled { + if err := enable(signal.Name(n)); err != nil { + return nil, err + } + } + } + + return o, nil +} + +// Bind attaches the owner to a loaded schema and returns a revision-scoped view. +func (o *Owner) Bind(s *datastore.ReadOnlyStoredSchema, scope signal.Scope) (*View, error) { + if s == nil || s.Get().GetV1() == nil || scope.SchemaHash == "" || + s.Get().GetV1().SchemaHash != scope.SchemaHash || scope.Revision == nil || scope.Revision == datastore.NoRevision { + return nil, errors.New("invalid signal schema/revision binding") + } + + handles, err := datastore.LoadOrStoreDerived( + s, + schemaHandleKey, + func() *schemaHandles { return &schemaHandles{owners: map[uint64]weak.Pointer[Owner]{}} }, + ) + if err != nil { + return nil, err + } + + handles.bind(o) + o.mu.Lock() + o.sequence++ + + if ss, ok := o.schemas[scope.SchemaHash]; ok { + ss.used = o.sequence + ready := ss.ready + o.mu.Unlock() + + if ready != nil { + <-ready + } + + return &View{owner: o, scope: scope}, nil + } + + cost := s.EstimatedSize() + 256 + if cost > o.options.MaxBytes { + o.mu.Unlock() + return &View{owner: o, scope: scope}, nil + } + + donor := o.seedLocked() + + for len(o.schemas) >= o.options.MaxSchemas { + o.evictSchemaLocked() + } + + ss := &schemaState{definitions: s.Get(), cost: cost, used: o.sequence, ready: make(chan struct{})} + o.schemas[scope.SchemaHash] = ss + o.bytes += cost + o.trimLocked() + o.mu.Unlock() + o.transfer(scope.SchemaHash, ss, donor) + close(ss.ready) + return &View{owner: o, scope: scope}, nil +} + +// Local returns an isolated view for legacy schema readers without a shared handle. +func (o *Owner) Local(scope signal.Scope) *View { + local, _ := NewOwner(o.registry, o.options) + scope.SchemaHash = fmt.Sprintf("local-%d", epochs.Add(1)) + local.schemas[scope.SchemaHash] = &schemaState{} + return &View{owner: local, scope: scope} +} + +// ResidentBytes returns the store's estimated resident storage usage in bytes. +func (o *Owner) ResidentBytes() int64 { o.mu.Lock(); defer o.mu.Unlock(); return o.bytes } + +func (o *Owner) dropLocked(k entryKey) { + if e := o.entries[k]; e != nil { + o.bytes -= e.cost + delete(o.entries, k) + } +} + +func (o *Owner) evictSchemaLocked() { + var oldest string + seq := ^uint64(0) + + for h, s := range o.schemas { + if s.used < seq { + oldest, seq = h, s.used + } + } + + if s := o.schemas[oldest]; s != nil { + o.bytes -= s.cost + delete(o.schemas, oldest) + + for k := range o.entries { + if k.schema == oldest { + o.dropLocked(k) + } + } + } +} + +func (o *Owner) trimLocked() { + for o.bytes > o.options.MaxBytes { + if len(o.entries) == 0 { + o.evictSchemaLocked() + continue + } + + var oldest entryKey + seq := ^uint64(0) + + for k, e := range o.entries { + if e.used < seq { + oldest, seq = k, e.used + } + } + + o.dropLocked(oldest) + } +} + +func (o *Owner) binding(name signal.Name, kt, vt reflect.Type, write bool) (signal.Descriptor, error) { + d, ok := o.registry.definitions[name] + if !ok { + return d, fmt.Errorf("unknown signal %q", name) + } + + expected := d.ValueType + + if write { + expected = d.ObservationType + } + + if kt != d.KeyType || vt != expected { + return d, fmt.Errorf("wrong signal type binding for %q", name) + } + + if write && d.Source != "" { + return d, fmt.Errorf("cannot observe derived signal %q", name) + } + + return d, nil +} + +func (o *Owner) key(scope signal.Scope, d signal.Descriptor, key any) entryKey { + rev := "" + + if d.Policy == signal.RevisionBound { + if scope.Revision != nil { + rev = scope.Revision.String() + } + } + + return entryKey{schema: scope.SchemaHash, name: d.Name, revision: rev, key: key} +} + +func (o *Owner) touch(k entryKey, create bool, d signal.Descriptor) *entry { + o.mu.Lock() + defer o.mu.Unlock() + + if _, ok := o.schemas[k.schema]; !ok { + return nil + } + + if e := o.entries[k]; e != nil { + o.sequence++ + e.used = o.sequence + return e + } + + if !create { + return nil + } + + // Construct private provider state without the shared owner mutex. + o.mu.Unlock() + var state signal.State + + if d.Build != nil { + state = d.Build(o.options.Now) + } else { + state = d.New() + } + + cost := entryCost(k, state, signal.Footprint{}) + o.mu.Lock() + + if _, ok := o.schemas[k.schema]; !ok { + return nil + } + + if existing := o.entries[k]; existing != nil { + o.sequence++ + existing.used = o.sequence + return existing + } + + if cost > o.options.MaxEntryBytes { + return nil + } + + if k.revision != "" { + revs := map[string]uint64{} + + for existing, e := range o.entries { + if existing.schema == k.schema && existing.revision != "" { + revs[existing.revision] = max(revs[existing.revision], e.used) + } + } + + if _, found := revs[k.revision]; !found && len(revs) >= o.options.MaxRevisions { + old := "" + seq := ^uint64(0) + + for r, s := range revs { + if s < seq { + old, seq = r, s + } + } + + for existing := range o.entries { + if existing.schema == k.schema && existing.revision == old { + o.dropLocked(existing) + } + } + } + } + + o.sequence++ + e := &entry{ + state: state, + used: o.sequence, + evidence: signal.Evidence{ + ID: signal.EvidenceID{Epoch: epochs.Add(1)}, + SourceSchemaHash: k.schema, + SourceRevision: k.revision, + Historical: d.Policy == signal.CrossRevision, + }, + } + e.cost = cost + + if e.cost > o.options.MaxEntryBytes { + return nil + } + + // Copy string backing only when admitting a new population. Lookup hits do + // not allocate, and a short substring cannot retain an unaccounted buffer. + k.key = ownedKey(reflect.ValueOf(k.key)).Interface() + k.schema = strings.Clone(k.schema) + k.revision = strings.Clone(k.revision) + o.entries[k] = e + o.bytes += e.cost + o.trimLocked() + + if o.entries[k] != e { + return nil + } + + return e +} + +func (o *Owner) read(scope signal.Scope, d signal.Descriptor, key any) (any, signal.Availability, signal.Evidence, error) { + if !o.enabled[d.Name] { + return nil, signal.Unsupported, signal.Evidence{}, nil + } + + e := o.touch(o.key(scope, d, key), false, d) + if e == nil { + return nil, signal.Missing, signal.Evidence{}, nil + } + + e.mu.Lock() + s := e.state.Clone() + ev := e.evidence + e.mu.Unlock() + value, availability := s.Value(o.options.Now()) + if q, ok := value.(signal.Quality); ok { + ev.Samples = q.SignalSamples() + ev.EffectiveWeight = q.SignalWeight() + } + + return value, availability, ev, nil +} + +func (o *Owner) write(scope signal.Scope, d signal.Descriptor, key, observation any) error { + if !o.enabled[d.Name] || (d.Accept != nil && !d.Accept(key, observation)) { + return nil + } + + k := o.key(scope, d, key) + e := o.touch(k, true, d) + if e == nil { + return nil + } + + e.publicationMu.Lock() + defer e.publicationMu.Unlock() + now := o.options.Now() + observed := now + + if d.ObservationTime != nil { + if at := d.ObservationTime(observation); !at.IsZero() { + observed = at + } + } + + e.mu.Lock() + + if err := e.state.Add(observation); err != nil { + e.mu.Unlock() + return err + } + + e.evidence.Samples++ + e.evidence.EffectiveWeight = float64(e.evidence.Samples) + + if e.evidence.FirstObserved.IsZero() || observed.Before(e.evidence.FirstObserved) { + e.evidence.FirstObserved = observed + } + + if observed.After(e.evidence.LastObserved) { + e.evidence.LastObserved = observed + } + + e.evidence.PublishedAt = now + e.publicationVersion++ + e.evidence.ID.Version = e.publicationVersion + update := signal.UpdateID{Epoch: e.evidence.ID.Epoch, Sequence: e.evidence.ID.Version} + + if d.Footprint != nil { + e.footprint = ownedFootprint(d.Footprint(key)) + } + + cost := entryCost(k, e.state, e.footprint) + e.mu.Unlock() + o.mu.Lock() + resident := o.entries[k] == e + if resident { + if cost > o.options.MaxEntryBytes { + o.dropLocked(k) + } else { + o.bytes += cost - e.cost + e.cost = cost + o.trimLocked() + } + } + + o.mu.Unlock() + + if resident { + if d.Stamp != nil { + observation = d.Stamp(observation, update, observed) + } + + for _, n := range o.registry.order { + derived := o.registry.definitions[n] + if derived.Source == d.Name { + if err := o.write(scope, derived, key, observation); err != nil { + return err + } + } + } + } + + return nil +} diff --git a/pkg/query/signals/property_test.go b/pkg/query/signals/property_test.go new file mode 100644 index 000000000..338320c8b --- /dev/null +++ b/pkg/query/signals/property_test.go @@ -0,0 +1,60 @@ +package signals_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + "pgregory.net/rapid" + + "github.com/authzed/spicedb/internal/datastore/revisions" + "github.com/authzed/spicedb/pkg/query/signals" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +func TestRevisionEvictionPreservesPlanningSnapshots(t *testing.T) { + rapid.Check(t, func(rt *rapid.T) { + observations := rapid.SliceOfN(rapid.IntRange(-100, 100), 1, 20).Draw(rt, "observations") + revision := rapid.Uint64Range(1, 1000).Draw(rt, "revision") + key := rapid.StringMatching(`[a-z]{1,20}`).Draw(rt, "population") + r, err := signals.NewRegistry(definition("sum")) + + require.NoError(rt, err) + o, err := signals.NewOwner(r, signals.Options{MaxRevisions: 1, MaxBytes: 4096}) + + require.NoError(rt, err) + a, err := o.Bind(stored("schema"), signal.Scope{ + SchemaHash: "schema", + Revision: revisions.NewForTransactionID(revision), + }) + + require.NoError(rt, err) + total := 0 + + for _, observation := range observations { + require.NoError(rt, signal.Observe(a, sumInput, key, observation)) + total += observation + } + + plan := a.ForPlan() + before, err := signal.Lookup(plan, sumRef, key) + + require.NoError(rt, err) + require.Equal(rt, total, before.Value) + b, err := o.Bind(stored("schema"), signal.Scope{ + SchemaHash: "schema", + Revision: revisions.NewForTransactionID(revision + 1), + }) + + require.NoError(rt, err) + require.NoError(rt, signal.Observe(b, sumInput, key, 1)) + after, err := signal.Lookup(plan, sumRef, key) + + require.NoError(rt, err) + require.Equal(rt, before, after) + evicted, err := signal.Lookup(a, sumRef, key) + + require.NoError(rt, err) + require.False(rt, evicted.IsAvailable()) + require.LessOrEqual(rt, o.ResidentBytes(), int64(4096)) + }) +} diff --git a/pkg/query/signals/registry.go b/pkg/query/signals/registry.go new file mode 100644 index 000000000..703bc69d0 --- /dev/null +++ b/pkg/query/signals/registry.go @@ -0,0 +1,127 @@ +// Package signals stores and resolves named planning evidence. +package signals + +import ( + "errors" + "fmt" + "slices" + + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +// Registry holds named signal descriptors and validates their dependencies. +type Registry struct { + definitions map[signal.Name]signal.Descriptor + order []signal.Name + frozen bool +} + +// NewRegistry creates a provider catalog and validates its dependency graph. +func NewRegistry(ds ...signal.Descriptor) (*Registry, error) { + r := &Registry{definitions: map[signal.Name]signal.Descriptor{}} + + for _, d := range ds { + if err := r.Register(d); err != nil { + return nil, err + } + } + + if err := r.validate(); err != nil { + return nil, err + } + + return r, nil +} + +// Register adds a descriptor to an unfrozen catalog, rejecting invalid or duplicate names. +func (r *Registry) Register(d signal.Descriptor) error { + if r.frozen { + return errors.New("signal registry is frozen") + } + + if d.Name == "" || d.KeyType == nil || !d.KeyType.Comparable() || + d.ObservationType == nil || d.ValueType == nil || (d.New == nil && d.Build == nil) || d.Policy > signal.CrossRevision { + return fmt.Errorf("invalid signal descriptor %q", d.Name) + } + + if !supportedKey(d.KeyType) { + return fmt.Errorf("unsupported signal key type for %q", d.Name) + } + + if _, ok := r.definitions[d.Name]; ok { + return fmt.Errorf("duplicate signal %q", d.Name) + } + + d.Dependencies = slices.Clone(d.Dependencies) + r.definitions[d.Name] = d + return nil +} + +func (r *Registry) validate() error { + state := map[signal.Name]uint8{} + r.order = nil + var visit func(signal.Name) error + visit = func(n signal.Name) error { + if state[n] == 1 { + return fmt.Errorf("signal dependency cycle at %q", n) + } + + if state[n] == 2 { + return nil + } + + d, ok := r.definitions[n] + if !ok { + return fmt.Errorf("missing signal dependency %q", n) + } + + state[n] = 1 + deps := slices.Clone(d.Dependencies) + slices.Sort(deps) + + for _, dep := range deps { + if err := visit(dep); err != nil { + return err + } + } + + if d.Source != "" { + source, ok := r.definitions[d.Source] + if !ok || source.KeyType != d.KeyType || source.ObservationType != d.ObservationType { + return fmt.Errorf("incompatible source binding for %q", n) + } + } + + state[n] = 2 + r.order = append(r.order, n) + return nil + } + names := make([]signal.Name, 0, len(r.definitions)) + + for n := range r.definitions { + names = append(names, n) + } + + slices.Sort(names) + + for _, n := range names { + if err := visit(n); err != nil { + return err + } + } + + return nil +} + +// Freeze validates dependencies and prevents further registration. +func (r *Registry) Freeze() error { + if err := r.validate(); err != nil { + return err + } + + r.frozen = true + return nil +} + +// Contains reports whether a name exists in the catalog. Freeze before concurrent use. +func (r *Registry) Contains(name signal.Name) bool { _, ok := r.definitions[name]; return ok } diff --git a/pkg/query/signals/registry_test.go b/pkg/query/signals/registry_test.go new file mode 100644 index 000000000..259d7a568 --- /dev/null +++ b/pkg/query/signals/registry_test.go @@ -0,0 +1,58 @@ +package signals_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/pkg/query/signals" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +type sum struct{ n int } + +func (s *sum) Add(v int) error { s.n += v; return nil } +func (s *sum) Value(time.Time) (int, signal.Availability) { return s.n, signal.Available } +func (s *sum) Size() int64 { return 8 } +func (s *sum) Clone() signal.Accumulator[int, int] { return &sum{s.n} } +func definition(name signal.Name) signal.Descriptor { + return signal.NewSource[string, int, int](name, signal.SourceConfig[string, int, int]{ + Policy: signal.RevisionBound, + New: func() signal.Accumulator[int, int] { return &sum{} }, + }) +} + +func TestRegistryRejectsDuplicateName(t *testing.T) { + _, err := signals.NewRegistry(definition("a"), definition("a")) + require.ErrorContains(t, err, "duplicate") +} + +func TestRegistryRejectsWrongBinding(t *testing.T) { + d := definition("a") + d.ValueType = nil + _, err := signals.NewRegistry(d) + require.Error(t, err) +} + +func TestRegistryRejectsMissingDependency(t *testing.T) { + d := definition("a") + d.Dependencies = []signal.Name{"missing"} + _, err := signals.NewRegistry(d) + require.ErrorContains(t, err, "missing") +} + +func TestRegistryRejectsCycle(t *testing.T) { + a, b := definition("a"), definition("b") + a.Dependencies = []signal.Name{"b"} + b.Dependencies = []signal.Name{"a"} + _, err := signals.NewRegistry(a, b) + require.ErrorContains(t, err, "cycle") +} + +func TestFrozenRegistryRejectsMutation(t *testing.T) { + r, err := signals.NewRegistry(definition("a")) + require.NoError(t, err) + require.NoError(t, r.Freeze()) + require.Error(t, r.Register(definition("b"))) +} diff --git a/pkg/query/signals/safety_test.go b/pkg/query/signals/safety_test.go new file mode 100644 index 000000000..33d01da0e --- /dev/null +++ b/pkg/query/signals/safety_test.go @@ -0,0 +1,219 @@ +package signals_test + +import ( + "math" + "reflect" + "strings" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/internal/datastore/revisions" + "github.com/authzed/spicedb/internal/querysignalstest" + "github.com/authzed/spicedb/pkg/query/signals" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +func TestNonreflexiveKeyIsRejected(t *testing.T) { + type key struct{ Value [1]float64 } + + d := signal.NewSource[key, int, int]("sum", signal.SourceConfig[key, int, int]{ + New: func() signal.Accumulator[int, int] { return &sum{} }, + }) + r, err := signals.NewRegistry(d) + require.NoError(t, err) + o, err := signals.NewOwner(r, signals.Options{}) + require.NoError(t, err) + v := view(t, o, "a", 1) + invalid := key{Value: [1]float64{math.NaN()}} + + require.ErrorContains(t, signal.Observe(v, signal.Input[key, int]{Name: "sum"}, invalid, 1), "key") + _, err = signal.Lookup(v, signal.Reference[key, int]{Name: "sum"}, invalid) + require.ErrorContains(t, err, "key") + _, err = signal.Lookup(v.ForPlan(), signal.Reference[key, int]{Name: "sum"}, invalid) + require.ErrorContains(t, err, "key") +} + +func TestRegistryRejectsReferenceKeys(t *testing.T) { + for _, keyType := range []reflect.Type{ + reflect.TypeFor[*string](), + reflect.TypeFor[any](), + reflect.TypeFor[struct{ Value any }](), + reflect.TypeFor[[1]chan int](), + } { + t.Run(keyType.String(), func(t *testing.T) { + d := definition("sum") + d.KeyType = keyType + _, err := signals.NewRegistry(d) + + require.ErrorContains(t, err, "key") + }) + } +} + +func TestSourceRequiresConstructor(t *testing.T) { + d := signal.NewSource[string, int, int]("sum", signal.SourceConfig[string, int, int]{}) + _, err := signals.NewRegistry(d) + + require.Error(t, err) +} + +func TestSchemaTransferHasIndependentPublicationIdentity(t *testing.T) { + d := definition("sum") + d.Policy = signal.CrossRevision + d.Transfer = func(signal.SchemaChange, signal.Footprint) signal.TransferDecision { + return signal.TransferDecision{Action: signal.Retain} + } + r, err := signals.NewRegistry(d) + require.NoError(t, err) + o, err := signals.NewOwner(r, signals.Options{}) + require.NoError(t, err) + a := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(a, sumInput, "x", 1)) + b := view(t, o, "b", 2) + require.NoError(t, signal.Observe(a, sumInput, "x", 2)) + require.NoError(t, signal.Observe(b, sumInput, "x", 5)) + av, err := signal.Lookup(a, sumRef, "x") + require.NoError(t, err) + bv, err := signal.Lookup(b, sumRef, "x") + require.NoError(t, err) + require.Equal(t, 3, av.Value) + require.Equal(t, 6, bv.Value) + require.NotEqual(t, av.Evidence.ID, bv.Evidence.ID) +} + +func TestFootprintCountsTowardEntryBudget(t *testing.T) { + d := definition("sum") + d.Footprint = func(any) signal.Footprint { + return signal.Footprint{Caveats: []string{strings.Repeat("x", 4096)}} + } + r, err := signals.NewRegistry(d) + require.NoError(t, err) + o, err := signals.NewOwner(r, signals.Options{MaxEntryBytes: 1024}) + require.NoError(t, err) + v := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(v, sumInput, "x", 1)) + e, err := signal.Lookup(v, sumRef, "x") + + require.NoError(t, err) + require.False(t, e.IsAvailable()) +} + +func TestFootprintIsOwnedByEntry(t *testing.T) { + footprint := signal.Footprint{Elements: []signal.Element{{Namespace: "doc", Relation: "read"}}} + d := definition("sum") + d.Policy = signal.CrossRevision + d.Footprint = func(any) signal.Footprint { return footprint } + d.Transfer = func(_ signal.SchemaChange, f signal.Footprint) signal.TransferDecision { + require.Equal(t, "read", f.Elements[0].Relation) + return signal.TransferDecision{Action: signal.Retain} + } + r, err := signals.NewRegistry(d) + require.NoError(t, err) + o, err := signals.NewOwner(r, signals.Options{}) + require.NoError(t, err) + a := view(t, o, "a", 1) + + require.NoError(t, signal.Observe(a, sumInput, "x", 1)) + footprint.Elements[0].Relation = "edited" + b := view(t, o, "b", 2) + e, err := signal.Lookup(b, sumRef, "x") + + require.NoError(t, err) + require.Equal(t, 1, e.Value) +} + +func TestBuildOnlySource(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + started := time.Now() + d := signal.NewSource[string, int, int]("sum", signal.SourceConfig[string, int, int]{ + Build: func(now func() time.Time) signal.Accumulator[int, int] { + require.Equal(t, started, now()) + return &sum{} + }, + }) + o := querysignalstest.Owner(t, d) + v := querysignalstest.View(t, o, "a", 1) + + require.NoError(t, signal.Observe(v, sumInput, "x", 4)) + e, err := signal.Lookup(v, sumRef, "x") + + require.NoError(t, err) + require.Equal(t, 4, e.Value) + }) +} + +func TestDerivedSignalsReceiveIncrementsAcrossRevisions(t *testing.T) { + source := definition("sum") + derived := signal.NewDerived("history", sumInput, signal.SourceConfig[string, int, int]{ + Policy: signal.CrossRevision, + New: func() signal.Accumulator[int, int] { return &sum{} }, + }) + o := querysignalstest.Owner(t, source, derived) + a := querysignalstest.View(t, o, "a", 1) + b := querysignalstest.View(t, o, "a", 2) + + require.NoError(t, signal.Observe(a, sumInput, "x", 3)) + require.NoError(t, signal.Observe(a, sumInput, "x", 5)) + require.NoError(t, signal.Observe(b, sumInput, "x", 7)) + history, err := signal.Lookup(b, signal.Reference[string, int]{Name: "history"}, "x") + + require.NoError(t, err) + require.Equal(t, 15, history.Value) + require.True(t, history.Evidence.Historical) + require.Error(t, signal.Observe(b, signal.Input[string, int]{Name: "history"}, "x", 1)) +} + +func TestSelectiveSchemaTransfer(t *testing.T) { + const original = "definition user {}\ndefinition doc {\n relation viewer: user\n permission read = viewer\n}" + + for _, tc := range []struct { + name string + next string + keep bool + }{ + {"unrelated", original + "\ndefinition other {}", true}, + {"changed", "definition user {}\ndefinition doc {\n relation editor: user\n permission read = editor\n}", false}, + } { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + d := definition("sum") + d.Policy = signal.CrossRevision + d.Footprint = func(any) signal.Footprint { + return signal.Footprint{Elements: []signal.Element{{Namespace: "doc", Relation: "read"}, {Namespace: "doc", Relation: "viewer"}}} + } + d.Transfer = signal.Unchanged + o := querysignalstest.Owner(t, d) + a, err := o.Bind(querysignalstest.Schema(t, "a", original), signal.Scope{ + SchemaHash: "a", Revision: revisions.NewForTransactionID(1), + }) + + require.NoError(t, err) + require.NoError(t, signal.Observe(a, sumInput, "x", 7)) + before, err := signal.Lookup(a, sumRef, "x") + + require.NoError(t, err) + time.Sleep(time.Minute) + b, err := o.Bind(querysignalstest.Schema(t, "b", tc.next), signal.Scope{ + SchemaHash: "b", Revision: revisions.NewForTransactionID(2), + }) + + require.NoError(t, err) + after, err := signal.Lookup(b, sumRef, "x") + + require.NoError(t, err) + require.Equal(t, tc.keep, after.IsAvailable()) + + if tc.keep { + require.Equal(t, 7, after.Value) + require.Equal(t, before.Evidence.LastObserved, after.Evidence.LastObserved) + require.Equal(t, before.Evidence.PublishedAt, after.Evidence.PublishedAt) + } + }) + }) + } +} diff --git a/pkg/query/signals/signal/definition.go b/pkg/query/signals/signal/definition.go new file mode 100644 index 000000000..d33285329 --- /dev/null +++ b/pkg/query/signals/signal/definition.go @@ -0,0 +1,184 @@ +package signal + +import ( + "reflect" + "time" +) + +// Accumulator owns provider state. Value and Clone must return independent snapshots. +type Accumulator[O, V any] interface { + // Add incorporates an observation into the accumulator. + Add(O) error + + // Value returns an independent estimate snapshot at the supplied time. + Value(time.Time) (V, Availability) + + // Size returns the estimated resident size of the accumulator in bytes. + Size() int64 + + // Clone returns an independent copy of the accumulator state. + Clone() Accumulator[O, V] +} + +// State provides the type-erased accumulator contract used by the store. +type State interface { + // Add incorporates an observation into the accumulator. + Add(any) error + + // Value returns an independent estimate snapshot at the supplied time. + Value(time.Time) (any, Availability) + + // Size returns the estimated resident size of the accumulator in bytes. + Size() int64 + + // Clone returns an independent copy of the accumulator state. + Clone() State +} + +type adapter[O, V any] struct{ Accumulator[O, V] } + +// Add incorporates an observation into the accumulator. +func (a adapter[O, V]) Add(v any) error { return a.Accumulator.Add(v.(O)) } + +// Value returns an independent estimate snapshot at the supplied time. +func (a adapter[O, V]) Value(t time.Time) (any, Availability) { return a.Accumulator.Value(t) } + +// DiscountEvidence reduces existing evidence weight without renewing its age. +func (a adapter[O, V]) DiscountEvidence(f float64) bool { + if d, ok := a.Accumulator.(Discountable); ok { + d.DiscountEvidence(f) + return true + } + + return false +} + +// Clone returns an independent copy of the accumulator state. +func (a adapter[O, V]) Clone() State { return adapter[O, V]{a.Accumulator.Clone()} } + +// Descriptor defines a named provider and its lifecycle callbacks. +// It retains runtime types so the shared registry can validate named accesses and +// dependencies between providers with different generic types. +type Descriptor struct { + // Build constructs an accumulator using the store's clock, when supplied. + Build func(func() time.Time) State + + // Stamp attaches a source increment ID and publication time to an observation. + Stamp func(any, UpdateID, time.Time) any + + // Accept filters observations before publication or derived-provider delivery. + Accept func(any, any) bool + + // Name identifies a registered signal provider. + Name Name + + // Description summarizes the provider's purpose. + Description string + + // Policy selects exact-revision or cross-revision evidence storage. + Policy RevisionPolicy + + // KeyType is the runtime type of the provider's population key. + KeyType reflect.Type + + // ObservationType is the runtime type of accepted observations. + ObservationType reflect.Type + + // ValueType is the runtime type of published estimates. + ValueType reflect.Type + + // Dependencies lists providers that must be registered before this provider. + Dependencies []Name + + // Source names the provider whose accepted increments feed this provider. + Source Name + + // New constructs an empty accumulator. + New func() State + + // Footprint extracts the schema dependencies associated with a population key. + Footprint func(any) Footprint + + // Transfer decides whether evidence is compatible with a schema change. + Transfer func(SchemaChange, Footprint) TransferDecision + + // ObservationTime extracts an observation's original time for evidence aging. + ObservationTime func(any) time.Time +} + +// SourceConfig defines typed construction and lifecycle callbacks for a provider. +type SourceConfig[K comparable, O, V any] struct { + // Build constructs an accumulator using the store's clock, when supplied. + Build func(func() time.Time) Accumulator[O, V] + + // Stamp attaches a source increment ID and publication time to an observation. + Stamp func(O, UpdateID, time.Time) O + + // Accept filters observations before publication or derived-provider delivery. + Accept func(K, O) bool + + // Description summarizes the provider's purpose. + Description string + + // Policy selects exact-revision or cross-revision evidence storage. + Policy RevisionPolicy + + // New constructs an empty accumulator. + New func() Accumulator[O, V] + + // Footprint extracts the schema dependencies associated with a population key. + Footprint func(K) Footprint + + // Transfer decides whether evidence is compatible with a schema change. + Transfer func(SchemaChange, Footprint) TransferDecision + + // ObservationTime extracts an observation's original time for evidence aging. + ObservationTime func(O) time.Time +} + +// NewSource adapts a typed provider configuration into a store descriptor. +func NewSource[K comparable, O, V any](name Name, c SourceConfig[K, O, V]) Descriptor { + d := Descriptor{ + Name: name, + Description: c.Description, + Policy: c.Policy, + KeyType: reflect.TypeFor[K](), + ObservationType: reflect.TypeFor[O](), + ValueType: reflect.TypeFor[V](), + Transfer: c.Transfer, + } + + if c.New != nil { + d.New = func() State { return adapter[O, V]{c.New()} } + } + + if c.Build != nil { + d.Build = func(now func() time.Time) State { return adapter[O, V]{c.Build(now)} } + } + + if c.Stamp != nil { + d.Stamp = func(o any, id UpdateID, at time.Time) any { return c.Stamp(o.(O), id, at) } + } + + if c.Accept != nil { + d.Accept = func(k, o any) bool { return c.Accept(k.(K), o.(O)) } + } + + if c.Footprint != nil { + d.Footprint = func(k any) Footprint { return c.Footprint(k.(K)) } + } + + if c.ObservationTime != nil { + d.ObservationTime = func(o any) time.Time { return c.ObservationTime(o.(O)) } + } + + return d +} + +// NewDerived consumes accepted increments of its source, never cumulative snapshots. +func NewDerived[K comparable, O, V any](name Name, source Input[K, O], c SourceConfig[K, O, V]) Descriptor { + d := NewSource(name, c) + d.Source = source.Name + d.Dependencies = []Name{source.Name} + return d +} diff --git a/pkg/query/signals/signal/distribution.go b/pkg/query/signals/signal/distribution.go new file mode 100644 index 000000000..940b7420d --- /dev/null +++ b/pkg/query/signals/signal/distribution.go @@ -0,0 +1,93 @@ +package signal + +import ( + "math" + + "github.com/caio/go-tdigest/v4" +) + +// Distribution exposes immutable approximate numeric distribution queries. +type Distribution struct { + digest *tdigest.TDigest + parts []WeightedDistribution + min, max float64 +} + +// WeightedDistribution associates an immutable distribution with a mixture weight. +type WeightedDistribution struct { + // Distribution is the immutable approximate path-count distribution. + Distribution Distribution + + // Weight is the distribution's effective observation weight. + Weight float64 +} + +// DigestDistribution clones a t-digest into an immutable distribution with observed bounds. +func DigestDistribution(d *tdigest.TDigest, minimum, maximum float64) Distribution { + return Distribution{digest: d.Clone(), min: minimum, max: maximum} +} + +// Mixture constructs an immutable weighted mixture of distributions. +func Mixture(parts []WeightedDistribution, minimum, maximum float64) Distribution { + return Distribution{parts: append([]WeightedDistribution(nil), parts...), min: minimum, max: maximum} +} + +// CDF returns the approximate fraction of samples at or below x. +func (d Distribution) CDF(x float64) float64 { + if x < d.min { + return 0 + } + + if x >= d.max { + return 1 + } + + if d.digest != nil { + return d.digest.CDF(x) + } + + total, sum := 0.0, 0.0 + + for _, p := range d.parts { + total += p.Weight + sum += p.Weight * p.Distribution.CDF(x) + } + + if total == 0 { + return math.NaN() + } + + return sum / total +} + +// Quantile returns the approximate value at q, or NaN for an invalid quantile. +func (d Distribution) Quantile(q float64) float64 { + if q < 0 || q > 1 || math.IsNaN(q) { + return math.NaN() + } + + if q == 0 { + return d.min + } + + if q == 1 { + return d.max + } + + if d.digest != nil { + return d.digest.Quantile(q) + } + + low, high := d.min, d.max + + for range 64 { + mid := low + (high-low)/2 + if d.CDF(mid) < q { + low = mid + } else { + high = mid + } + } + + return low + (high-low)/2 +} diff --git a/pkg/query/signals/signal/distribution_test.go b/pkg/query/signals/signal/distribution_test.go new file mode 100644 index 000000000..a6a2f7818 --- /dev/null +++ b/pkg/query/signals/signal/distribution_test.go @@ -0,0 +1,62 @@ +package signal_test + +import ( + "math" + "testing" + + "github.com/caio/go-tdigest/v4" + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +func TestDistributionRetainsIndependentSnapshot(t *testing.T) { + digest, err := tdigest.New() + + require.NoError(t, err) + require.NoError(t, digest.AddWeighted(10, 100)) + require.NoError(t, digest.AddWeighted(20, 100)) + snapshot := signal.DigestDistribution(digest, 10, 20) + + require.NoError(t, digest.AddWeighted(1000, 10000)) + require.InDelta(t, 10.0, snapshot.Quantile(0), 1e-9) + require.InDelta(t, 20.0, snapshot.Quantile(1), 1e-9) + require.InDelta(t, 15.0, snapshot.Quantile(0.5), 1) + require.Zero(t, snapshot.CDF(9)) + require.InDelta(t, 1.0, snapshot.CDF(20), 1e-9) + + for _, q := range []float64{-1, 2, math.NaN()} { + require.True(t, math.IsNaN(snapshot.Quantile(q))) + } +} + +func TestWeightedMixtureOwnsParts(t *testing.T) { + low, err := tdigest.New() + + require.NoError(t, err) + require.NoError(t, low.AddWeighted(1, 100)) + high, err := tdigest.New() + + require.NoError(t, err) + require.NoError(t, high.AddWeighted(9, 100)) + parts := []signal.WeightedDistribution{ + {Distribution: signal.DigestDistribution(low, 1, 1), Weight: 1}, + {Distribution: signal.DigestDistribution(high, 9, 9), Weight: 3}, + } + mixture := signal.Mixture(parts, 1, 9) + parts[0].Weight = 1000 + + require.InDelta(t, 0.25, mixture.CDF(5), 1e-9) + require.InDelta(t, 9.0, mixture.Quantile(0.5), 1e-6) +} + +func TestExecutionFootprintRoundTrip(t *testing.T) { + footprint := signal.Footprint{ + Elements: []signal.Element{{Namespace: "doc", Relation: "view"}}, + Caveats: []string{"allowed"}, + } + key := signal.ExecutionKey{Dependencies: signal.EncodeFootprint(footprint)} + + require.Equal(t, footprint, key.Footprint()) + require.Empty(t, (signal.ExecutionKey{Dependencies: "invalid"}).Footprint()) +} diff --git a/pkg/query/signals/signal/execution.go b/pkg/query/signals/signal/execution.go new file mode 100644 index 000000000..6b281756e --- /dev/null +++ b/pkg/query/signals/signal/execution.go @@ -0,0 +1,84 @@ +package signal + +import "encoding/json" + +// Operation identifies the execution operation represented by evidence. +type Operation uint8 + +const ( + + // Check identifies a membership check. + Check Operation = iota + 1 + + // IterSubjects identifies subject enumeration. + IterSubjects + + // IterResources identifies resource enumeration. + IterResources +) + +// Terminal describes how an invocation ended. +type Terminal uint8 + +const ( + + // Complete indicates that the invocation completed normally. + Complete Terminal = iota + + // ConsumerStopped indicates that enumeration ended before exhaustion. + ConsumerStopped + + // Canceled indicates that the invocation's context was canceled. + Canceled + + // Errored indicates that the invocation ended with an error. + Errored +) + +// ExecutionKey identifies an observed execution population independently of dispatch keys. +type ExecutionKey struct { + // Identity is the full structural identity of the observed subtree. + // It must include all descendants, unlike query.CanonicalKey, which omits + // children of caveat nodes. + Identity string + + // Operation identifies the execution operation represented by evidence. + Operation Operation + + // Filter identifies the type and relation filters for this population. + Filter string + + // Mode distinguishes single and batched invocation populations. + Mode string + + // Strategy identifies the execution strategy used by this population. + Strategy string + + // Role distinguishes sender wrappers from receiver bodies. + Role string + + // Dependencies encodes the subtree's transitive schema footprint for compatibility checks. + Dependencies string +} + +// Footprint decodes the schema dependencies recorded in the execution key. +func (k ExecutionKey) Footprint() Footprint { + var f Footprint + if err := json.Unmarshal([]byte(k.Dependencies), &f); err != nil { + return Footprint{} + } + + return f +} + +// EncodeFootprint encodes schema dependencies for use in a comparable execution key. +func EncodeFootprint(f Footprint) string { b, _ := json.Marshal(f); return string(b) } + +// Quality provides sample counts and effective weight for evidence metadata. +type Quality interface { + // SignalSamples returns the number of observations represented by the estimate. + SignalSamples() uint64 + + // SignalWeight returns the effective observation weight of the estimate. + SignalWeight() float64 +} diff --git a/pkg/query/signals/signal/reference.go b/pkg/query/signals/signal/reference.go new file mode 100644 index 000000000..da900ef95 --- /dev/null +++ b/pkg/query/signals/signal/reference.go @@ -0,0 +1,65 @@ +package signal + +import "reflect" + +type ( + + // Reference binds a signal name to its key and estimate types. + Reference[K comparable, V any] struct { + // Name identifies a registered signal provider. + Name Name + } + + // Input binds a signal name to its key and observation types. + Input[K comparable, O any] struct { + // Name identifies a registered signal provider. + Name Name + } + + // ReadView provides type-checked access to published signal estimates. + ReadView interface { + // Read reads an estimate after validating its key and value types. + Read(Name, any, reflect.Type, reflect.Type) (any, Availability, Evidence, error) + } +) + +// Writer provides type-checked publication of signal observations. +type Writer interface { + // Write publishes an observation after validating its key and observation types. + Write(Name, any, any, reflect.Type, reflect.Type) error +} + +// Lookup reads a typed estimate and its availability from a signal view. +func Lookup[K comparable, V any](v ReadView, ref Reference[K, V], key K) (Estimate[V], error) { + value, status, evidence, err := v.Read(ref.Name, key, reflect.TypeFor[K](), reflect.TypeFor[V]()) + result := Estimate[V]{Availability: status, Evidence: evidence} + + if err == nil && status == Available { + result.Value = value.(V) + } + + return result, err +} + +// LookupFirst returns the first available estimate and its signal name in candidate order. +// Unavailable estimates are skipped; lookup errors stop the search and identify the failing signal. +// If no candidate is available, it returns Missing availability and an empty signal name. +func LookupFirst[K comparable, V any](v ReadView, key K, refs ...Reference[K, V]) (Estimate[V], Name, error) { + for _, ref := range refs { + estimate, err := Lookup(v, ref, key) + if err != nil { + return estimate, ref.Name, err + } + + if estimate.IsAvailable() { + return estimate, ref.Name, nil + } + } + + return Estimate[V]{Availability: Missing}, "", nil +} + +// Observe publishes a typed observation to its scoped signal writer. +func Observe[K comparable, O any](w Writer, in Input[K, O], key K, observation O) error { + return w.Write(in.Name, key, observation, reflect.TypeFor[K](), reflect.TypeFor[O]()) +} diff --git a/pkg/query/signals/signal/reference_test.go b/pkg/query/signals/signal/reference_test.go new file mode 100644 index 000000000..46be041c2 --- /dev/null +++ b/pkg/query/signals/signal/reference_test.go @@ -0,0 +1,94 @@ +package signal_test + +import ( + "errors" + "reflect" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +type lookupView struct { + values map[signal.Name]signal.Estimate[int] + errors map[signal.Name]error + reads []signal.Name +} + +// Read supplies controlled estimates and records lookup order. +func (v *lookupView) Read(name signal.Name, key any, kt, vt reflect.Type) (any, signal.Availability, signal.Evidence, error) { + v.reads = append(v.reads, name) + + if key != "population" || kt != reflect.TypeFor[string]() || vt != reflect.TypeFor[int]() { + return nil, signal.Missing, signal.Evidence{}, errors.New("wrong lookup binding") + } + + e, ok := v.values[name] + if !ok { + e.Availability = signal.Missing + } + + return e.Value, e.Availability, e.Evidence, v.errors[name] +} + +func TestLookupFirst(t *testing.T) { + first := signal.Reference[string, int]{Name: "first"} + second := signal.Reference[string, int]{Name: "second"} + provenance := signal.Evidence{ID: signal.EvidenceID{Epoch: 10, Version: 3}, Historical: true} + + t.Run("available first avoids reading fallback", func(t *testing.T) { + view := &lookupView{values: map[signal.Name]signal.Estimate[int]{ + first.Name: {Value: 7, Availability: signal.Available, Evidence: provenance}, + }} + e, name, err := signal.LookupFirst(view, "population", first, second) + require.NoError(t, err) + require.Equal(t, first.Name, name) + require.Equal(t, 7, e.Value) + require.Equal(t, provenance, e.Evidence) + require.Equal(t, []signal.Name{first.Name}, view.reads) + }) + + for _, status := range []signal.Availability{signal.Missing, signal.Expired, signal.Evicted, signal.Unsupported} { + t.Run("unavailable fallback", func(t *testing.T) { + view := &lookupView{values: map[signal.Name]signal.Estimate[int]{ + first.Name: {Availability: status}, + second.Name: {Value: 11, Availability: signal.Available, Evidence: provenance}, + }} + e, name, err := signal.LookupFirst(view, "population", first, second) + require.NoError(t, err) + require.Equal(t, second.Name, name) + require.Equal(t, 11, e.Value) + require.Equal(t, provenance, e.Evidence) + require.Equal(t, []signal.Name{first.Name, second.Name}, view.reads) + }) + } + + t.Run("no available signal", func(t *testing.T) { + view := &lookupView{} + e, name, err := signal.LookupFirst(view, "population", first, second) + require.NoError(t, err) + require.False(t, e.IsAvailable()) + require.Equal(t, signal.Missing, e.Availability) + require.Empty(t, name) + require.Equal(t, []signal.Name{first.Name, second.Name}, view.reads) + }) + + t.Run("empty candidates", func(t *testing.T) { + view := &lookupView{} + e, name, err := signal.LookupFirst[string, int](view, "population") + require.NoError(t, err) + require.False(t, e.IsAvailable()) + require.Empty(t, name) + require.Empty(t, view.reads) + }) + + t.Run("lookup error prevents fallback", func(t *testing.T) { + failure := errors.New("binding failure") + view := &lookupView{errors: map[signal.Name]error{first.Name: failure}} + _, name, err := signal.LookupFirst(view, "population", first, second) + require.ErrorIs(t, err, failure) + require.Equal(t, first.Name, name) + require.Equal(t, []signal.Name{first.Name}, view.reads) + }) +} diff --git a/pkg/query/signals/signal/schema_test.go b/pkg/query/signals/signal/schema_test.go new file mode 100644 index 000000000..467392e69 --- /dev/null +++ b/pkg/query/signals/signal/schema_test.go @@ -0,0 +1,78 @@ +package signal_test + +import ( + "maps" + "slices" + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/internal/querysignalstest" + caveattypes "github.com/authzed/spicedb/pkg/caveats/types" + "github.com/authzed/spicedb/pkg/diff" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +func TestSchemaCompatibilityFollowsDeclaredDependencies(t *testing.T) { + const original = `definition user {} + +caveat gate(flag bool) { flag } + +definition document { + relation viewer: user with gate + permission view = viewer +}` + + footprint := signal.Footprint{ + Elements: []signal.Element{ + {Namespace: "document", Relation: "viewer"}, + {Namespace: "document", Relation: "view"}, + }, + Caveats: []string{"gate"}, + } + renamedViewer := strings.Replace(original, "relation viewer:", "relation editor:", 1) + renamedViewer = strings.Replace(renamedViewer, "= viewer", "= editor", 1) + withoutCaveat := strings.Replace(original, "caveat gate(flag bool) { flag }", "", 1) + withoutCaveat = strings.Replace(withoutCaveat, " with gate", "", 1) + from := querysignalstest.Schema(t, "from", original).Get().GetV1() + + for _, tc := range []struct { + name string + next string + want signal.TransferAction + }{ + {"unchanged", original, signal.Retain}, + {"unrelated definition", original + "\ndefinition other {}", signal.Retain}, + {"definition removed", "definition user {}\ncaveat gate(flag bool) { flag }", signal.Reset}, + {"relation removed", renamedViewer, signal.Reset}, + {"unrelated relation", strings.Replace(original, "permission view", "relation editor: user\npermission view", 1), signal.Retain}, + {"changed caveat", strings.Replace(original, "{ flag }", "{ !flag }", 1), signal.Reset}, + {"removed caveat", withoutCaveat, signal.Reset}, + {"definition comment", strings.Replace(original, "definition document", "/** Resources. */\ndefinition document", 1), signal.Retain}, + {"caveat comment", strings.Replace(original, "caveat gate", "/** Context gate. */\ncaveat gate", 1), signal.Retain}, + } { + t.Run(tc.name, func(t *testing.T) { + to := querysignalstest.Schema(t, "to", tc.next).Get().GetV1() + changes, err := diff.DiffSchemas( + diff.DiffableSchema{ + ObjectDefinitions: slices.Collect(maps.Values(from.NamespaceDefinitions)), + CaveatDefinitions: slices.Collect(maps.Values(from.CaveatDefinitions)), + }, + diff.DiffableSchema{ + ObjectDefinitions: slices.Collect(maps.Values(to.NamespaceDefinitions)), + CaveatDefinitions: slices.Collect(maps.Values(to.CaveatDefinitions)), + }, + caveattypes.Default.TypeSet, + ) + + require.NoError(t, err) + decision := signal.Unchanged(signal.SchemaChange{Diff: changes}, footprint) + + require.Equal(t, tc.want, decision.Action) + }) + } + + require.Equal(t, signal.Reset, signal.Unchanged(signal.SchemaChange{}, footprint).Action) + require.Equal(t, signal.Reset, signal.Unchanged(signal.SchemaChange{Diff: &diff.SchemaDiff{}}, signal.Footprint{}).Action) +} diff --git a/pkg/query/signals/signal/types.go b/pkg/query/signals/signal/types.go new file mode 100644 index 000000000..29d84e79c --- /dev/null +++ b/pkg/query/signals/signal/types.go @@ -0,0 +1,245 @@ +// Package signal defines typed observations and estimates for planning signals. +package signal + +import ( + "time" + + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/diff" +) + +type ( + + // Name identifies a registered signal provider. + Name string + + // RevisionPolicy controls whether evidence is isolated to an exact datastore revision. + RevisionPolicy uint8 +) + +const ( + + // RevisionBound isolates evidence to the selected datastore revision. + RevisionBound RevisionPolicy = iota + + // CrossRevision allows evidence to be shared across revisions within a schema. + CrossRevision +) + +// Availability describes whether an estimate can be used. +type Availability uint8 + +const ( + + // Available indicates that an estimate is usable. + Available Availability = iota + + // Missing indicates that no usable observations exist. + Missing + + // Expired indicates that evidence has aged below the usable threshold. + Expired + + // Evicted indicates that evidence was removed by storage limits. + Evicted + + // Unsupported indicates that the requested signal or population is unsupported. + Unsupported +) + +// Scope binds evidence to a schema and selected datastore revision. +type Scope struct { + // Revision is the request-selected datastore revision. + Revision datastore.Revision + + // SchemaHash identifies the schema used by the request. + SchemaHash string +} + +type ( + + // EvidenceID identifies a published snapshot within a source epoch. + EvidenceID struct { + // Epoch identifies the source accumulator incarnation. + Epoch uint64 + + // Version is the publication sequence within the epoch. + Version uint64 + } + + // Evidence describes the provenance and age of an estimate. + Evidence struct { + // ID identifies the published evidence snapshot. + ID EvidenceID + + // Samples counts observations represented by the estimate. + Samples uint64 + + // EffectiveWeight is the weight remaining after aging and discounting. + EffectiveWeight float64 + + // FirstObserved is the earliest observation time represented by the evidence. + FirstObserved time.Time + + // LastObserved is the latest observation time represented by the evidence. + LastObserved time.Time + + // PublishedAt is the time the evidence was published. + PublishedAt time.Time + + // SourceRevision identifies the source revision when available. + SourceRevision string + + // SourceSchemaHash identifies the schema where the evidence originated. + SourceSchemaHash string + + // Historical reports whether evidence was derived across revisions. + Historical bool + } +) + +// Estimate holds a typed value, its availability, and its provenance. +type Estimate[V any] struct { + // Value is the typed estimate returned by the provider. + Value V + + // Availability describes whether an estimate can be used. + Availability Availability + + // Evidence describes the provenance and age of an estimate. + Evidence Evidence +} + +// IsAvailable reports whether the estimate contains a usable value. +func (e Estimate[V]) IsAvailable() bool { + return e.Availability == Available +} + +type ( + + // Element identifies a relation or permission in a schema definition. + Element struct { + // Namespace is the schema definition name. + Namespace string + + // Relation is the relation or permission name. + Relation string + } + + // Footprint lists the schema dependencies of an evidence population. + Footprint struct { + // Elements lists relation and permission dependencies. + Elements []Element + + // Caveats lists caveat definition dependencies. + Caveats []string + } +) + +type ( + + // SchemaChange contains the diff used to assess evidence compatibility. + SchemaChange struct { + // Diff describes changes between the source and destination schemas. + Diff *diff.SchemaDiff + } + + // TransferAction specifies how evidence crosses a schema change. + TransferAction uint8 +) + +const ( + + // Reset discards evidence during schema transfer. + Reset TransferAction = iota + + // Retain preserves compatible evidence without changing its age. + Retain + + // Discount preserves evidence with reduced weight. + Discount +) + +// TransferDecision describes a provider's schema compatibility decision. +type TransferDecision struct { + // Action selects whether to reset, retain, or discount evidence. + Action TransferAction + + // Factor is the multiplier applied when discounting evidence. + Factor float64 + + // Reason explains the transfer decision. + Reason string +} + +// Unchanged retains evidence only when its transitive schema dependencies did not change. +func Unchanged(change SchemaChange, f Footprint) TransferDecision { + if change.Diff == nil { + return TransferDecision{Action: Reset, Reason: "unknown-change"} + } + + if len(f.Elements) == 0 && len(f.Caveats) == 0 { + return TransferDecision{Action: Reset, Reason: "unknown-dependencies"} + } + + d := change.Diff + + for _, e := range f.Elements { + for _, n := range d.RemovedNamespaces { + if n == e.Namespace { + return TransferDecision{Action: Reset, Reason: "removed-definition"} + } + } + + if nd, ok := d.ChangedNamespaces[e.Namespace]; ok { + for _, delta := range nd.Deltas() { + switch string(delta.Type) { + case "namespace-comments-changed", "changed-permission-comment", "changed-relation-comment": + continue + case "added-relation", "removed-relation", "added-permission", "removed-permission", + "changed-permission-implementation", "legacy-changed-relation-implementation", + "relation-allowed-type-added", "relation-allowed-type-removed": + default: + return TransferDecision{Action: Reset, Reason: "unknown-change-kind"} + } + + if delta.RelationName == "" || delta.RelationName == e.Relation { + return TransferDecision{Action: Reset, Reason: "changed-dependency"} + } + } + } + } + + for _, name := range f.Caveats { + for _, n := range d.RemovedCaveats { + if n == name { + return TransferDecision{Action: Reset, Reason: "removed-caveat"} + } + } + + if cd, ok := d.ChangedCaveats[name]; ok { + for _, delta := range cd.Deltas() { + if string(delta.Type) != "caveat-comments-changed" { + return TransferDecision{Action: Reset, Reason: "changed-caveat"} + } + } + } + } + + return TransferDecision{Action: Retain, Factor: 1, Reason: "unchanged-dependencies"} +} + +// UpdateID identifies an accepted source increment, not a cumulative snapshot. +type UpdateID struct { + // Epoch identifies the source accumulator incarnation. + Epoch uint64 + + // Sequence identifies the accepted increment within its source epoch. + Sequence uint64 +} + +// Discountable supports schema-transfer weighting without changing evidence age. +type Discountable interface { + // DiscountEvidence reduces existing evidence weight without renewing its age. + DiscountEvidence(float64) +} diff --git a/pkg/query/signals/store_test.go b/pkg/query/signals/store_test.go new file mode 100644 index 000000000..a0ab014c6 --- /dev/null +++ b/pkg/query/signals/store_test.go @@ -0,0 +1,183 @@ +package signals_test + +import ( + "runtime" + "sync" + "testing" + "time" + "weak" + + "github.com/ccoveille/go-safecast/v2" + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/internal/datastore/revisions" + "github.com/authzed/spicedb/pkg/datastore" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/query/signals" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +func stored(hash string) *datastore.ReadOnlyStoredSchema { + return datastore.NewReadOnlyStoredSchema( + &core.StoredSchema{VersionOneof: &core.StoredSchema_V1{V1: &core.StoredSchema_V1StoredSchema{SchemaHash: hash}}}, + ) +} + +func owner(t *testing.T, opts signals.Options) *signals.Owner { + t.Helper() + r, err := signals.NewRegistry(definition("sum")) + require.NoError(t, err) + o, err := signals.NewOwner(r, opts) + require.NoError(t, err) + return o +} + +func view(t *testing.T, o *signals.Owner, hash string, rev int64) *signals.View { + t.Helper() + v, err := o.Bind( + stored(hash), + signal.Scope{SchemaHash: hash, Revision: revisions.NewForTransactionID(safecast.MustConvert[uint64](rev))}, + ) + require.NoError(t, err) + return v +} + +var ( + sumRef = signal.Reference[string, int]{Name: "sum"} + sumInput = signal.Input[string, int]{Name: "sum"} +) + +func TestExactRevisionIsolation(t *testing.T) { + o := owner(t, signals.Options{}) + a, b := view(t, o, "a", 9), view(t, o, "a", 10) + require.NoError(t, signal.Observe(a, sumInput, "x", 7)) + v, err := signal.Lookup(a, sumRef, "x") + require.NoError(t, err) + require.Equal(t, 7, v.Value) + v, err = signal.Lookup(b, sumRef, "x") + require.NoError(t, err) + require.Equal(t, signal.Missing, v.Availability) +} + +func TestOwnerAndSchemaIsolation(t *testing.T) { + a, b := owner(t, signals.Options{}), owner(t, signals.Options{}) + v := view(t, a, "a", 1) + require.NoError(t, signal.Observe(v, sumInput, "x", 7)) + + for _, other := range []*signals.View{view(t, b, "a", 1), view(t, a, "b", 1)} { + e, err := signal.Lookup(other, sumRef, "x") + require.NoError(t, err) + require.Equal(t, signal.Missing, e.Availability) + } +} + +func TestPlanReadsRemainStable(t *testing.T) { + v := view(t, owner(t, signals.Options{}), "a", 1) + require.NoError(t, signal.Observe(v, sumInput, "x", 2)) + plan := v.ForPlan() + before, err := signal.Lookup(plan, sumRef, "x") + require.NoError(t, err) + require.NoError(t, signal.Observe(v, sumInput, "x", 5)) + after, err := signal.Lookup(plan, sumRef, "x") + require.NoError(t, err) + require.Equal(t, before, after) + fresh, _ := signal.Lookup(v.ForPlan(), sumRef, "x") + require.Equal(t, 7, fresh.Value) +} + +func TestSchemaBindingRejectsHashMismatch(t *testing.T) { + o := owner(t, signals.Options{}) + _, err := o.Bind(stored("a"), signal.Scope{SchemaHash: "b", Revision: revisions.NewForTransactionID(1)}) + require.Error(t, err) +} + +func TestEvictionDoesNotMutateSnapshot(t *testing.T) { + o := owner(t, signals.Options{MaxRevisions: 1}) + a := view(t, o, "a", 1) + require.NoError(t, signal.Observe(a, sumInput, "x", 7)) + p := a.ForPlan() + before, _ := signal.Lookup(p, sumRef, "x") + b := view(t, o, "a", 2) + require.NoError(t, signal.Observe(b, sumInput, "x", 9)) + after, _ := signal.Lookup(p, sumRef, "x") + require.Equal(t, before, after) + old, _ := signal.Lookup(a, sumRef, "x") + require.Equal(t, signal.Missing, old.Availability) +} + +func TestTypedBindingAndConcurrentUpdates(t *testing.T) { + v := view(t, owner(t, signals.Options{}), "a", 1) + bad := signal.Reference[int, int]{Name: "sum"} + _, err := signal.Lookup(v, bad, 1) + require.Error(t, err) + var wg sync.WaitGroup + + for range 20 { + wg.Go(func() { + for range 10 { + require.NoError(t, signal.Observe(v, sumInput, "x", 1)) + } + }) + } + + wg.Wait() + e, err := signal.Lookup(v, sumRef, "x") + require.NoError(t, err) + require.Equal(t, 200, e.Value) +} + +func TestOversizeAdmissionIsUnavailable(t *testing.T) { + o := owner(t, signals.Options{MaxEntryBytes: 1}) + v := view(t, o, "a", 1) + require.NoError(t, signal.Observe(v, sumInput, "x", 7)) + e, err := signal.Lookup(v, sumRef, "x") + require.NoError(t, err) + require.NotEqual(t, signal.Available, e.Availability) + require.LessOrEqual(t, o.ResidentBytes(), int64(16<<20)) +} + +func TestFactoryCanInspectOwner(t *testing.T) { + var o *signals.Owner + d := definition("sum") + original := d.New + d.New = func() signal.State { _ = o.ResidentBytes(); return original() } + r, err := signals.NewRegistry(d) + require.NoError(t, err) + o, err = signals.NewOwner(r, signals.Options{}) + require.NoError(t, err) + v := view(t, o, "a", 1) + done := make(chan error, 1) + go func() { done <- signal.Observe(v, sumInput, "x", 1) }() + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("provider factory held owner mutex") + } +} + +func TestSchemaAttachmentDoesNotRetainOwner(t *testing.T) { + schema := stored("shared") + create := func() weak.Pointer[signals.Owner] { + o := owner(t, signals.Options{}) + v, err := o.Bind(schema, signal.Scope{SchemaHash: "shared", Revision: revisions.NewForTransactionID(1)}) + require.NoError(t, err) + require.NoError(t, signal.Observe(v, sumInput, "x", 1)) + return weak.Make(o) + } + discarded := create() + + for range 20 { + runtime.GC() + + if discarded.Value() == nil { + break + } + + time.Sleep(10 * time.Millisecond) + } + + require.Nil(t, discarded.Value()) + runtime.KeepAlive(schema) +} diff --git a/pkg/query/signals/transfer.go b/pkg/query/signals/transfer.go new file mode 100644 index 000000000..bb4e74030 --- /dev/null +++ b/pkg/query/signals/transfer.go @@ -0,0 +1,117 @@ +package signals + +import ( + "maps" + "slices" + + caveattypes "github.com/authzed/spicedb/pkg/caveats/types" + "github.com/authzed/spicedb/pkg/diff" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +type seedEntry struct { + key entryKey + entry *entry +} +type schemaSeed struct { + definitions *core.StoredSchema + entries []seedEntry +} + +// seedLocked chooses a single donor; it makes no claim about revision chronology. +func (o *Owner) seedLocked() schemaSeed { + var donor string + seq := uint64(0) + + for h, s := range o.schemas { + if s.used > seq && s.definitions != nil { + select { + case <-s.ready: + donor, seq = h, s.used + default: + } + } + } + + if donor == "" { + return schemaSeed{} + } + + result := schemaSeed{definitions: o.schemas[donor].definitions} + + for k, e := range o.entries { + d := o.registry.definitions[k.name] + if k.schema == donor && d.Policy == signal.CrossRevision && d.Transfer != nil { + result.entries = append(result.entries, seedEntry{k, e}) + } + } + + return result +} + +func diffable(s *core.StoredSchema) diff.DiffableSchema { + v := s.GetV1() + return diff.DiffableSchema{ + ObjectDefinitions: slices.Collect(maps.Values(v.NamespaceDefinitions)), + CaveatDefinitions: slices.Collect(maps.Values(v.CaveatDefinitions)), + } +} + +func (o *Owner) transfer(hash string, target *schemaState, donor schemaSeed) { + if donor.definitions == nil { + return + } + + changes, err := diff.DiffSchemas(diffable(donor.definitions), diffable(target.definitions), caveattypes.Default.TypeSet) + if err != nil { + return + } + + for _, seed := range donor.entries { + seed.entry.mu.Lock() + state := seed.entry.state.Clone() + ev := seed.entry.evidence + footprint := ownedFootprint(seed.entry.footprint) + seed.entry.mu.Unlock() + d := o.registry.definitions[seed.key.name] + decision := d.Transfer(signal.SchemaChange{Diff: changes}, ownedFootprint(footprint)) + + switch decision.Action { + case signal.Retain: + case signal.Discount: + if decision.Factor <= 0 || decision.Factor >= 1 { + continue + } + + discount, ok := state.(interface{ DiscountEvidence(float64) bool }) + if !ok || !discount.DiscountEvidence(decision.Factor) { + continue + } + default: + continue + } + + k := seed.key + k.schema = hash + cost := entryCost(k, state, footprint) + if cost > o.options.MaxEntryBytes { + continue + } + + // The new schema owns an independent accumulator and publication stream. + ev.ID = signal.EvidenceID{Epoch: epochs.Add(1)} + next := &entry{state: state, evidence: ev, footprint: footprint, cost: cost} + o.mu.Lock() + + if o.schemas[hash] == target && o.entries[k] == nil { + o.sequence++ + next.used = o.sequence + o.entries[k] = next + o.bytes += cost + o.trimLocked() + } + + o.mu.Unlock() + } +} diff --git a/pkg/query/signals/view.go b/pkg/query/signals/view.go new file mode 100644 index 000000000..e558266ee --- /dev/null +++ b/pkg/query/signals/view.go @@ -0,0 +1,88 @@ +package signals + +import ( + "reflect" + "sync" + + "github.com/authzed/spicedb/pkg/query/signals/signal" +) + +// View provides observation and lookup access for one schema/revision scope. +type View struct { + owner *Owner + scope signal.Scope +} + +// Read reads an estimate after validating its key and value types. +func (v *View) Read(n signal.Name, k any, kt, vt reflect.Type) (any, signal.Availability, signal.Evidence, error) { + d, err := v.owner.binding(n, kt, vt, false) + if err != nil { + return nil, signal.Missing, signal.Evidence{}, err + } + + if err := validateKey(k, d.KeyType); err != nil { + return nil, signal.Missing, signal.Evidence{}, err + } + + return v.owner.read(v.scope, d, k) +} + +// Write publishes an observation after validating its key and observation types. +func (v *View) Write(n signal.Name, k, observation any, kt, ot reflect.Type) error { + d, err := v.owner.binding(n, kt, ot, true) + if err != nil { + return err + } + + if err := validateKey(k, d.KeyType); err != nil { + return err + } + + return v.owner.write(v.scope, d, k, observation) +} + +// ForPlan returns a view that memoizes independent snapshots for one planning run. +func (v *View) ForPlan() signal.ReadView { return &planView{view: v, values: map[planKey]readResult{}} } + +type planKey struct { + name signal.Name + key any + kt, vt reflect.Type +} + +type readResult struct { + value any + availability signal.Availability + evidence signal.Evidence + err error +} + +type planView struct { + mu sync.Mutex + view *View + values map[planKey]readResult // GUARDED_BY(mu) +} + +// Read reads an estimate after validating its key and value types. +func (p *planView) Read(n signal.Name, k any, kt, vt reflect.Type) (any, signal.Availability, signal.Evidence, error) { + if err := validateKey(k, kt); err != nil { + return nil, signal.Missing, signal.Evidence{}, err + } + + p.mu.Lock() + defer p.mu.Unlock() + key := planKey{n, k, kt, vt} + r, ok := p.values[key] + if !ok { + r.value, r.availability, r.evidence, r.err = p.view.Read(n, k, kt, vt) + p.values[key] = r + } + + return r.value, r.availability, r.evidence, r.err +} + +// Enabled reports whether the owner's learning profile enables the named signal. +func (v *View) Enabled(n signal.Name) bool { return v.owner.enabled[n] } + +// Registry returns the frozen provider catalog for declaration validation. +func (v *View) Registry() *Registry { return v.owner.registry }