feat(query): add reusable planning signals

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
+}

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,
+			},
+		},
+	})
+}

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.

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()
+}