From 35f6c8e2b762fe68754975f73845265a9edd4f94 Mon Sep 17 00:00:00 2001 From: Barak Michener Date: Fri, 22 May 2026 12:01:49 -0700 Subject: [PATCH 1/4] refactor: remove in-progress keys and hash based on the plan --- internal/dispatch/caching/caching.go | 26 +- .../dispatch/caching/cachingdispatch_test.go | 6 +- internal/dispatch/dispatch.go | 21 +- internal/dispatch/dispatch_optimizer.go | 20 +- internal/dispatch/executor.go | 333 ++++++------------ internal/dispatch/executor_test.go | 81 +---- internal/dispatch/keys/computed.go | 39 +- internal/dispatch/keys/computed_test.go | 12 +- internal/dispatch/keys/hasher.go | 24 ++ .../singleflight/singleflight_test.go | 6 +- pkg/proto/dispatch/v1/dispatch.pb.go | 21 +- pkg/proto/dispatch/v1/dispatch_vtproto.pb.go | 61 ---- proto/internal/dispatch/v1/dispatch.proto | 22 +- 13 files changed, 225 insertions(+), 447 deletions(-) diff --git a/internal/dispatch/caching/caching.go b/internal/dispatch/caching/caching.go index 474a79a818..0240080be7 100644 --- a/internal/dispatch/caching/caching.go +++ b/internal/dispatch/caching/caching.go @@ -410,26 +410,28 @@ func (cd *Dispatcher) DispatchLookupSubjects(req *v1.DispatchLookupSubjectsReque return nil } -// LookupPlanCheck probes the cache for a Plan-Check answer using a lightweight -// descriptor. This is the cache-hit fast path that avoids serializing the -// iterator subtree on the sender side: the executor calls this first, and only -// builds a full DispatchQueryPlanRequest (which is what forces plan -// serialization) when this misses. +// LookupPlanCheck probes the cache for a Plan-Check answer using the +// serialized iterator bytes the caller has already produced. This is the +// cache-hit fast path: the executor serializes once at dispatch time, calls +// LookupPlanCheck with the bytes, and only on miss reuses those same bytes +// for the full DispatchQueryPlanRequest. Cache hits skip the request/stream +// machinery (~200 allocs in our memory-engine measurements) but still pay +// the serialize cost. // -// On a hit we increment both queryPlanTotal and queryPlanFromCache so the cache -// hit ratio across LookupPlanCheck + DispatchQueryPlan stays sensible; on a -// miss we *do not* increment queryPlanTotal here — the caller will issue the -// full DispatchQueryPlan next, which counts the dispatch itself. +// On a hit we increment both queryPlanTotal and queryPlanFromCache so the +// cache hit ratio across LookupPlanCheck + DispatchQueryPlan stays sensible; +// on a miss we do not increment queryPlanTotal here — the caller will issue +// the full DispatchQueryPlan next, which counts the dispatch itself. // -// We also forward to the delegate on miss so a chain of caching dispatchers -// (rare today, but allowed by the layering) all get probed. +// We forward to the delegate on miss so a chain of caching dispatchers (rare +// today, but allowed by the layering) all get probed. func (cd *Dispatcher) LookupPlanCheck(ctx context.Context, lookup dispatch.PlanCheckLookup) (*v1.ResultPath, bool, error) { if lookup.PlanContext == nil { return cd.d.LookupPlanCheck(ctx, lookup) } cacheKey := keys.PlanCheckLookupKey( lookup.PlanContext.Revision, - lookup.CanonicalKey, + lookup.Plan, lookup.Resource, lookup.Subject, lookup.PlanContext.CaveatContext, diff --git a/internal/dispatch/caching/cachingdispatch_test.go b/internal/dispatch/caching/cachingdispatch_test.go index bfd517f00f..8b61873af9 100644 --- a/internal/dispatch/caching/cachingdispatch_test.go +++ b/internal/dispatch/caching/cachingdispatch_test.go @@ -307,10 +307,10 @@ func TestDispatchQueryPlanRecordsCachingMetrics(t *testing.T) { Operation: v1.PlanOperation_PLAN_OPERATION_CHECK, Resource: tuple.ONRStringToCore("document", "doc1", "viewer"), Subject: tuple.ONRStringToCore("user", "alice", "..."), + Plan: []byte("document#viewer"), PlanContext: &v1.PlanContext{ - Revision: "1", - SchemaHash: []byte(datalayer.NoSchemaHashForTesting), - InProgressKeys: []string{"document#viewer"}, + Revision: "1", + SchemaHash: []byte(datalayer.NoSchemaHashForTesting), }, } diff --git a/internal/dispatch/dispatch.go b/internal/dispatch/dispatch.go index d0b9b266e3..80841d155d 100644 --- a/internal/dispatch/dispatch.go +++ b/internal/dispatch/dispatch.go @@ -84,17 +84,18 @@ type LookupSubjects interface { // PlanStream is an alias for the stream to which plan results will be written. type PlanStream = Stream[*v1.DispatchQueryPlanResponse] -// PlanCheckLookup is the small, cheap-to-construct descriptor a caller passes -// to LookupPlanCheck to consult dispatcher caches before paying to serialize a -// full iterator subtree into a DispatchQueryPlanRequest. The fields mirror the -// inputs the Plan-Check cache key hashes over (see keys.PlanCheckLookupKey): -// callers that produce identical descriptors for the same underlying problem -// will hit the same cache entry as the corresponding DispatchQueryPlan call. +// PlanCheckLookup is the descriptor a caller passes to LookupPlanCheck to +// consult dispatcher caches without going through DispatchQueryPlan's full +// request/stream machinery. The serialized iterator bytes are the dispatch +// identity: two lookups with the same Plan bytes (and same resource/subject/ +// plan context) target the same cache slot, identical to the slot the +// corresponding DispatchQueryPlan request would compute (see +// keys.PlanCheckLookupKey). type PlanCheckLookup struct { - CanonicalKey string - Resource *core.ObjectAndRelation - Subject *core.ObjectAndRelation - PlanContext *v1.PlanContext + Plan []byte + Resource *core.ObjectAndRelation + Subject *core.ObjectAndRelation + PlanContext *v1.PlanContext } // Plan interface describes the methods required to dispatch plan-based query planner requests. diff --git a/internal/dispatch/dispatch_optimizer.go b/internal/dispatch/dispatch_optimizer.go index 665865404d..2f4b1f69a9 100644 --- a/internal/dispatch/dispatch_optimizer.go +++ b/internal/dispatch/dispatch_optimizer.go @@ -53,12 +53,20 @@ func wrapAliasWithDispatch(outline query.Outline) query.Outline { } } -// containsUnmatchedRecursiveSentinelOutline is the outline-time mirror of -// containsUnmatchedRecursiveSentinel (executor.go): it walks an outline -// subtree and reports whether any RecursiveSentinel has no RecursiveIterator -// with the same (definition, relation) key in the same subtree. An unmatched -// sentinel means dispatching this subtree would sever the sentinel from the -// collection context its iterator establishes. +// recursiveKey uniquely identifies a recursion pair by definition and relation +// name. Used by containsUnmatchedRecursiveSentinelOutline to pair up sentinels +// with their managing RecursiveIterators at optimization time. +type recursiveKey struct { + definitionName string + relationName string +} + +// containsUnmatchedRecursiveSentinelOutline walks an outline subtree and +// reports whether any RecursiveSentinel has no RecursiveIterator with the +// same (definition, relation) key in the same subtree. An unmatched sentinel +// means dispatching this subtree would sever the sentinel from the collection +// context its iterator establishes — the optimizer uses this to keep such +// aliases unwrapped (and therefore non-dispatching) at planning time. func containsUnmatchedRecursiveSentinelOutline(root query.Outline) bool { var iteratorKeys map[recursiveKey]struct{} var sentinelKeys map[recursiveKey]struct{} diff --git a/internal/dispatch/executor.go b/internal/dispatch/executor.go index 3abc73d624..aa01f63d53 100644 --- a/internal/dispatch/executor.go +++ b/internal/dispatch/executor.go @@ -31,53 +31,6 @@ var _ query.Executor = &DispatchExecutor{} // the executor was constructed with chunkSize == 0. const defaultDispatchChunkSize = 100 -// recursiveKey uniquely identifies a recursion pair by definition and relation name. -type recursiveKey struct { - definitionName string - relationName string -} - -// containsUnmatchedRecursiveSentinel walks the entire iterator tree and checks whether -// there is a RecursiveSentinelIterator whose (definitionName, relationName) pair has no -// matching RecursiveIterator anywhere in the tree. A sentinel is "managed" (safe to -// dispatch) only when a RecursiveIterator with the same key exists to handle its -// collection context. This handles the double-recursion case where a RecursiveIterator -// for one recursion pair contains a sentinel for a different pair in its subtree. -func containsUnmatchedRecursiveSentinel(it query.Iterator) bool { - // Collect all keys from RecursiveIterator and RecursiveSentinelIterator nodes. - var iteratorKeys map[recursiveKey]struct{} - var sentinelKeys map[recursiveKey]struct{} - - var walk func(query.Iterator) - walk = func(sub query.Iterator) { - if ri, ok := sub.(*query.RecursiveIterator); ok { - if iteratorKeys == nil { - iteratorKeys = make(map[recursiveKey]struct{}) - } - iteratorKeys[recursiveKey{ri.DefinitionName(), ri.RelationName()}] = struct{}{} - } - if si, ok := sub.(*query.RecursiveSentinelIterator); ok { - if sentinelKeys == nil { - sentinelKeys = make(map[recursiveKey]struct{}) - } - sentinelKeys[recursiveKey{si.DefinitionName(), si.RelationName()}] = struct{}{} - } - for _, child := range sub.Subiterators() { - walk(child) - } - } - walk(it) - - // Every sentinel must have a matching iterator; an unmatched sentinel means - // its collection context won't be established and dispatch would break. - for key := range sentinelKeys { - if _, managed := iteratorKeys[key]; !managed { - return true - } - } - return false -} - // NewDispatchExecutor creates a new DispatchExecutor that dispatches alias // iterator operations through the given Dispatcher chain. dispatchChunkSize // caps the number of inputs per CheckMany RPC; if zero, defaults to 100. @@ -92,66 +45,30 @@ func NewDispatchExecutor(dispatcher Dispatcher, planContext *v1.PlanContext, dis } } -// wrappedAlias extracts the AliasIterator wrapped by a DispatchIterator. The -// dispatch-wrap optimizer (internal/dispatch/dispatch_optimizer.go) always -// builds DispatchIterator(Alias(...)), so the assertion is normally safe; we -// still return ok=false for any other shape so a future caller handing us an -// odd tree doesn't panic the executor. -func wrappedAlias(it query.Iterator) (*query.AliasIterator, bool) { - disp, ok := it.(*DispatchIterator) - if !ok { - return nil, false - } - subs := disp.Subiterators() - if len(subs) != 1 { - return nil, false - } - alias, ok := subs[0].(*query.AliasIterator) - if !ok { - return nil, false - } - return alias, true -} - -// shouldDispatch reports whether the iterator is a DispatchIterator-wrapped -// alias that the executor should ship over RPC, vs. running locally. Two -// reasons to refuse: -// - the wrap contains a sentinel whose matching RecursiveIterator lives -// outside its own subtree (sentinel-based recursion guard, ancestor case). -// - the alias's canonical key is already in the active dispatch chain -// (in-progress guard, cross-relation cycle case). Without this, a standalone -// plan that re-expands the same key in a sibling subtree produces an -// infinite dispatch loop. +// shouldDispatch reports whether the iterator is a DispatchIterator the +// executor should ship over RPC vs run locally. The wrap is the authoritative +// dispatch boundary — the dispatch-wrap optimizer +// (internal/dispatch/dispatch_optimizer.go) decides which alias positions get +// a wrap and excludes unsafe sentinel positions at planning time; this +// predicate just checks for the marker. // -// The wrap itself is the authoritative dispatch boundary — bare AliasIterators -// produced by the planner are intentionally not dispatched. The optimizer -// decides which alias positions get a wrap; this predicate just consults it. -func (e *DispatchExecutor) shouldDispatch(it query.Iterator) (string, bool) { - alias, ok := wrappedAlias(it) - if !ok { - return "", false - } - if containsUnmatchedRecursiveSentinel(it) { - return "", false - } - key := alias.DefinitionName() + "#" + alias.Relation() - for _, k := range e.planContext.GetInProgressKeys() { - if k == key { - return "", false - } - } - return key, true +// No runtime cycle-break: the compiler emits RecursiveIterator + +// RecursiveSentinelIterator for every cyclic schema, so the compiled tree is +// finite by construction and a dispatch chain can't loop on the same key. +func (e *DispatchExecutor) shouldDispatch(it query.Iterator) bool { + _, ok := it.(*DispatchIterator) + return ok } func (e *DispatchExecutor) Check(ctx *query.Context, it query.Iterator, resource query.Object, subject query.ObjectAndRelation) (*query.Path, error) { - if _, ok := e.shouldDispatch(it); ok { + if e.shouldDispatch(it) { return e.dispatchCheck(ctx, it, resource, subject) } return it.CheckImpl(ctx, resource, subject) } func (e *DispatchExecutor) CheckManySubjects(ctx *query.Context, it query.Iterator, resource query.Object, subjects []query.ObjectAndRelation) ([]*query.Path, error) { - if _, ok := e.shouldDispatch(it); !ok { + if !e.shouldDispatch(it) { out := make([]*query.Path, len(subjects)) for i, s := range subjects { p, err := it.CheckImpl(ctx, resource, s) @@ -166,7 +83,7 @@ func (e *DispatchExecutor) CheckManySubjects(ctx *query.Context, it query.Iterat } func (e *DispatchExecutor) CheckManyResources(ctx *query.Context, it query.Iterator, resources []query.Object, subject query.ObjectAndRelation) ([]*query.Path, error) { - if _, ok := e.shouldDispatch(it); !ok { + if !e.shouldDispatch(it) { out := make([]*query.Path, len(resources)) for i, r := range resources { p, err := it.CheckImpl(ctx, r, subject) @@ -181,7 +98,7 @@ func (e *DispatchExecutor) CheckManyResources(ctx *query.Context, it query.Itera } func (e *DispatchExecutor) IterSubjects(ctx *query.Context, it query.Iterator, resource query.Object, filterSubjectType query.ObjectType) (query.PathSeq, error) { - if _, ok := e.shouldDispatch(it); ok { + if e.shouldDispatch(it) { return e.dispatchIterSubjects(ctx, it, resource, filterSubjectType) } pathSeq, err := it.IterSubjectsImpl(ctx, resource, filterSubjectType) @@ -192,7 +109,7 @@ func (e *DispatchExecutor) IterSubjects(ctx *query.Context, it query.Iterator, r } func (e *DispatchExecutor) IterResources(ctx *query.Context, it query.Iterator, subject query.ObjectAndRelation, filterResourceType query.ObjectType) (query.PathSeq, error) { - if _, ok := e.shouldDispatch(it); ok { + if e.shouldDispatch(it) { return e.dispatchIterResources(ctx, it, subject, filterResourceType) } pathSeq, err := it.IterResourcesImpl(ctx, subject, filterResourceType) @@ -206,14 +123,15 @@ func (e *DispatchExecutor) dispatchCheck(ctx *query.Context, it query.Iterator, subCtx, cancel := context.WithCancel(ctx) defer cancel() - // Probe the dispatcher chain for a cached Plan-Check answer before paying - // to serialize the iterator. The lookup descriptor is cheap to build — - // just the wrapped alias's (def, rel) plus the resource/subject ONRs and - // parent plan context — and the cache key it produces is identical to the - // one DispatchQueryPlan would compute (see keys.PlanCheckLookupKey), so a - // hit here is the same hit the slow path would have found. - alias, _ := wrappedAlias(it) - canonicalKey := alias.DefinitionName() + "#" + alias.Relation() + // The serialized iterator bytes are the dispatch identity — we serialize + // once and reuse the bytes for both the cache lookup and (on miss) the + // dispatched request. Cache hits skip the full DispatchQueryPlan machinery + // (request alloc, stream setup, delegate hop) but still pay this serialize + // cost — fine, because dispatching mandates the serialize anyway. + plan, err := serializePlan(it) + if err != nil { + return nil, err + } resourceONR := &core.ObjectAndRelation{ Namespace: resource.ObjectType, ObjectId: resource.ObjectID, @@ -224,20 +142,17 @@ func (e *DispatchExecutor) dispatchCheck(ctx *query.Context, it query.Iterator, Relation: subject.Relation, } if path, ok, err := e.dispatcher.LookupPlanCheck(subCtx, PlanCheckLookup{ - CanonicalKey: canonicalKey, - Resource: resourceONR, - Subject: subjectONR, - PlanContext: e.planContext, + Plan: plan, + Resource: resourceONR, + Subject: subjectONR, + PlanContext: e.planContext, }); err != nil { return nil, err } else if ok { return resultPathToQueryPath(path), nil } - req, err := e.buildRequest(ctx, v1.PlanOperation_PLAN_OPERATION_CHECK, it, resource, subject) - if err != nil { - return nil, err - } + req := e.buildRequest(ctx, v1.PlanOperation_PLAN_OPERATION_CHECK, plan, resourceONR, subjectONR) stream := NewCollectingDispatchStream[*v1.DispatchQueryPlanResponse](subCtx) if err := e.dispatcher.DispatchQueryPlan(req, stream); err != nil { return nil, err @@ -256,18 +171,25 @@ func (e *DispatchExecutor) dispatchIterSubjects(ctx *query.Context, it query.Ite subCtx, cancel := context.WithCancel(ctx) defer cancel() + plan, err := serializePlan(it) + if err != nil { + return nil, err + } + // For LookupSubjects the proto Subject field carries the *filter* subject // type+relation rather than a real subject ONR — the receiver needs the // filter to apply the same reachability/optimizer decisions the sender // would have applied. The actual receiver-side iteration uses NoObjectFilter // since the sender filters the returned paths itself. - req, err := e.buildRequest(ctx, v1.PlanOperation_PLAN_OPERATION_LOOKUP_SUBJECTS, it, resource, query.ObjectAndRelation{ - ObjectType: filterSubjectType.Type, - Relation: filterSubjectType.Subrelation, - }) - if err != nil { - return nil, err + resourceONR := &core.ObjectAndRelation{ + Namespace: resource.ObjectType, + ObjectId: resource.ObjectID, + } + subjectONR := &core.ObjectAndRelation{ + Namespace: filterSubjectType.Type, + Relation: filterSubjectType.Subrelation, } + req := e.buildRequest(ctx, v1.PlanOperation_PLAN_OPERATION_LOOKUP_SUBJECTS, plan, resourceONR, subjectONR) stream := NewCollectingDispatchStream[*v1.DispatchQueryPlanResponse](subCtx) if err := e.dispatcher.DispatchQueryPlan(req, stream); err != nil { return nil, err @@ -281,13 +203,21 @@ func (e *DispatchExecutor) dispatchIterResources(ctx *query.Context, it query.It subCtx, cancel := context.WithCancel(ctx) defer cancel() - req, err := e.buildRequest(ctx, v1.PlanOperation_PLAN_OPERATION_LOOKUP_RESOURCES, it, query.Object{ - ObjectType: subject.ObjectType, - ObjectID: subject.ObjectID, - }, subject) + plan, err := serializePlan(it) if err != nil { return nil, err } + + resourceONR := &core.ObjectAndRelation{ + Namespace: subject.ObjectType, + ObjectId: subject.ObjectID, + } + subjectONR := &core.ObjectAndRelation{ + Namespace: subject.ObjectType, + ObjectId: subject.ObjectID, + Relation: subject.Relation, + } + req := e.buildRequest(ctx, v1.PlanOperation_PLAN_OPERATION_LOOKUP_RESOURCES, plan, resourceONR, subjectONR) stream := NewCollectingDispatchStream[*v1.DispatchQueryPlanResponse](subCtx) if err := e.dispatcher.DispatchQueryPlan(req, stream); err != nil { return nil, err @@ -301,13 +231,22 @@ func (e *DispatchExecutor) dispatchCheckManySubjects(ctx *query.Context, it quer subCtx, cancel := context.WithCancel(ctx) defer cancel() + plan, err := serializePlan(it) + if err != nil { + return nil, err + } + resourceONR := &core.ObjectAndRelation{ + Namespace: resource.ObjectType, + ObjectId: resource.ObjectID, + } + out := make([]*query.Path, len(subjects)) idx := 0 for chunk := range slices.Chunk(subjects, int(e.dispatchChunkSize)) { - req, err := e.buildManyRequest(ctx, v1.PlanOperation_PLAN_OPERATION_CHECK_MANY_SUBJECTS, it, resource, query.ObjectAndRelation{}, nil, chunk) - if err != nil { - return nil, err - } + // CHECK_MANY_SUBJECTS: real resource ONR, empty subject ONR (real subjects + // ride on `many`). Empty placeholder rather than nil so the receiver's + // dereferences of req.Subject.* don't NPE. + req := e.buildManyRequest(ctx, v1.PlanOperation_PLAN_OPERATION_CHECK_MANY_SUBJECTS, plan, resourceONR, &core.ObjectAndRelation{}, nil, chunk) stream := NewCollectingDispatchStream[*v1.DispatchQueryPlanResponse](subCtx) if err := e.dispatcher.DispatchQueryPlan(req, stream); err != nil { return nil, err @@ -332,13 +271,23 @@ func (e *DispatchExecutor) dispatchCheckManyResources(ctx *query.Context, it que subCtx, cancel := context.WithCancel(ctx) defer cancel() + plan, err := serializePlan(it) + if err != nil { + return nil, err + } + subjectONR := &core.ObjectAndRelation{ + Namespace: subject.ObjectType, + ObjectId: subject.ObjectID, + Relation: subject.Relation, + } + out := make([]*query.Path, len(resources)) idx := 0 for chunk := range slices.Chunk(resources, int(e.dispatchChunkSize)) { - req, err := e.buildManyRequest(ctx, v1.PlanOperation_PLAN_OPERATION_CHECK_MANY_RESOURCES, it, query.Object{}, subject, chunk, nil) - if err != nil { - return nil, err - } + // CHECK_MANY_RESOURCES: empty resource ONR (real resources on `many`), + // real subject ONR. Empty placeholder rather than nil so the receiver's + // dereferences of req.Resource.* don't NPE. + req := e.buildManyRequest(ctx, v1.PlanOperation_PLAN_OPERATION_CHECK_MANY_RESOURCES, plan, &core.ObjectAndRelation{}, subjectONR, chunk, nil) stream := NewCollectingDispatchStream[*v1.DispatchQueryPlanResponse](subCtx) if err := e.dispatcher.DispatchQueryPlan(req, stream); err != nil { return nil, err @@ -359,40 +308,18 @@ func (e *DispatchExecutor) dispatchCheckManyResources(ctx *query.Context, it que return out, nil } -// buildManyRequest constructs a DispatchQueryPlanRequest for CheckMany operations. -// Exactly one of (manyResources, manySubjects) is non-nil; the corresponding -// singular field (resource for CHECK_MANY_RESOURCES, subject for CHECK_MANY_SUBJECTS) -// must be left zero — the receiver inspects `many` for those operations. -func (e *DispatchExecutor) buildManyRequest(ctx *query.Context, op v1.PlanOperation, it query.Iterator, resource query.Object, subject query.ObjectAndRelation, manyResources []query.Object, manySubjects []query.ObjectAndRelation) (*v1.DispatchQueryPlanRequest, error) { - // it is a *DispatchIterator wrapping the alias the executor decided to - // ship — see shouldDispatch. Serialize only the inner alias subtree: the - // receiver runs it locally (its executor won't re-dispatch a bare alias), - // and we save a small handful of header bytes per request. - alias, ok := wrappedAlias(it) - if !ok { - return nil, fmt.Errorf("DispatchQueryPlan: buildManyRequest expected DispatchIterator(Alias), got %T", it) - } - key := alias.DefinitionName() + "#" + alias.Relation() - plan, err := serializePlan(alias) - if err != nil { - return nil, err - } - // The current dispatch's canonical key rides on PlanContext.InProgressKeys - // (appended by planContextForDispatch). Receiver-side cache key computation - // reads it back from the last entry there; there is no top-level - // canonical_key field on the request anymore. +// buildManyRequest constructs a DispatchQueryPlanRequest for CheckMany +// operations. Exactly one of (manyResources, manySubjects) is non-nil; the +// corresponding singular field (resource for CHECK_MANY_RESOURCES, subject +// for CHECK_MANY_SUBJECTS) must be left zero — the receiver inspects `many` +// for those operations. Callers serialize the iterator into `plan` before +// calling so the bytes can also be used for cache lookups when relevant. +func (e *DispatchExecutor) buildManyRequest(ctx *query.Context, op v1.PlanOperation, plan []byte, resource *core.ObjectAndRelation, subject *core.ObjectAndRelation, manyResources []query.Object, manySubjects []query.ObjectAndRelation) *v1.DispatchQueryPlanRequest { req := &v1.DispatchQueryPlanRequest{ - Operation: op, - Resource: &core.ObjectAndRelation{ - Namespace: resource.ObjectType, - ObjectId: resource.ObjectID, - }, - Subject: &core.ObjectAndRelation{ - Namespace: subject.ObjectType, - ObjectId: subject.ObjectID, - Relation: subject.Relation, - }, - PlanContext: planContextForDispatch(e.planContext, key, ctx.TopLevelOperation), + Operation: op, + Resource: resource, + Subject: subject, + PlanContext: planContextForDispatch(e.planContext, ctx.TopLevelOperation), Plan: plan, } if len(manyResources) > 0 { @@ -414,12 +341,13 @@ func (e *DispatchExecutor) buildManyRequest(ctx *query.Context, op v1.PlanOperat } } } - return req, nil + return req } -// serializePlan emits the iterator subtree using the pkg/query wire format so -// the receiver can deserialize and execute it directly without rebuilding from -// the (definition, relation) canonical key. +// serializePlan emits the iterator subtree using the pkg/query wire format. +// The bytes are simultaneously: (a) the dispatch identity used to compute +// cache/singleflight/cluster keys, and (b) the payload the receiver +// deserializes and executes. One serialize per dispatch, used for both. func serializePlan(it query.Iterator) ([]byte, error) { var buf bytes.Buffer if err := it.Serialize(&buf); err != nil { @@ -428,50 +356,30 @@ func serializePlan(it query.Iterator) ([]byte, error) { return buf.Bytes(), nil } -func (e *DispatchExecutor) buildRequest(ctx *query.Context, op v1.PlanOperation, it query.Iterator, resource query.Object, subject query.ObjectAndRelation) (*v1.DispatchQueryPlanRequest, error) { - // dispatch{Check,IterSubjects,IterResources} only enter buildRequest after - // shouldDispatch accepted `it` as a DispatchIterator(Alias). Unwrap the - // wrap and serialize the inner alias — the receiver runs it locally without - // re-dispatching, and we keep the wire shape identical to what shipped - // before the wrap became authoritative. The dispatch boundary's "def#rel" - // is appended to PlanContext.InProgressKeys via planContextForDispatch; - // receiver-side cache key computation pulls it back out of that slice. - alias, ok := wrappedAlias(it) - if !ok { - return nil, fmt.Errorf("DispatchQueryPlan: buildRequest expected DispatchIterator(Alias), got %T", it) - } - key := alias.DefinitionName() + "#" + alias.Relation() - plan, err := serializePlan(alias) - if err != nil { - return nil, err - } +// buildRequest constructs a DispatchQueryPlanRequest for the single-target +// dispatch operations (CHECK, LOOKUP_RESOURCES, LOOKUP_SUBJECTS). Callers +// serialize the iterator into `plan` before calling — for CHECK the same +// bytes drive the cache lookup, so we don't re-serialize inside. +func (e *DispatchExecutor) buildRequest(ctx *query.Context, op v1.PlanOperation, plan []byte, resource *core.ObjectAndRelation, subject *core.ObjectAndRelation) *v1.DispatchQueryPlanRequest { return &v1.DispatchQueryPlanRequest{ - Operation: op, - Resource: &core.ObjectAndRelation{ - Namespace: resource.ObjectType, - ObjectId: resource.ObjectID, - }, - Subject: &core.ObjectAndRelation{ - Namespace: subject.ObjectType, - ObjectId: subject.ObjectID, - Relation: subject.Relation, - }, - PlanContext: planContextForDispatch(e.planContext, key, ctx.TopLevelOperation), + Operation: op, + Resource: resource, + Subject: subject, + PlanContext: planContextForDispatch(e.planContext, ctx.TopLevelOperation), Plan: plan, - }, nil + } } -// planContextForDispatch returns a shallow copy of pc with `key` appended to -// the in-progress dispatch keys and `topLevelOp` recorded as the user-facing -// operation. Two receiver-side guards depend on these: -// - InProgressKeys breaks cross-relation cycles produced by standalone plan -// rebuilds at each receiver hop. -// - TopLevelOperation propagates the original API operation across hops so -// iterators that consult TopLevelOperation see consistent user intent. -func planContextForDispatch(pc *v1.PlanContext, key string, topLevelOp query.Operation) *v1.PlanContext { +// planContextForDispatch returns a shallow copy of pc with `topLevelOp` +// recorded as the user-facing operation. The receiver-side guard +// TopLevelOperation propagates the original API operation across hops so +// iterators that consult TopLevelOperation (e.g. AliasIterator's self-edge +// logic) see consistent user intent. No InProgressKeys append — the plan +// bytes are the dispatch identity now (see keys.PlanCheckLookupKey), and +// cycle-breaks are guaranteed by the compiler's RecursiveIterator wrapping. +func planContextForDispatch(pc *v1.PlanContext, topLevelOp query.Operation) *v1.PlanContext { if pc == nil { return &v1.PlanContext{ - InProgressKeys: []string{key}, TopLevelOperation: queryOpToPlanOperation(topLevelOp), } } @@ -483,19 +391,12 @@ func planContextForDispatch(pc *v1.PlanContext, key string, topLevelOp query.Ope if tlo == v1.PlanOperation_PLAN_OPERATION_CHECK { tlo = queryOpToPlanOperation(topLevelOp) } - - // Duplicate and append the in progress keys - inProgress := make([]string, len(pc.InProgressKeys)+1) - copy(inProgress, pc.InProgressKeys) - inProgress[len(pc.InProgressKeys)] = key - return &v1.PlanContext{ Revision: pc.Revision, CaveatContext: pc.CaveatContext, MaxRecursionDepth: pc.MaxRecursionDepth, OptionalDatastoreLimit: pc.OptionalDatastoreLimit, SchemaHash: pc.SchemaHash, - InProgressKeys: inProgress, TopLevelOperation: tlo, } } diff --git a/internal/dispatch/executor_test.go b/internal/dispatch/executor_test.go index 59d4ebd3ea..4759349522 100644 --- a/internal/dispatch/executor_test.go +++ b/internal/dispatch/executor_test.go @@ -248,60 +248,11 @@ func TestDispatchExecutor_IterResources_NonAliasLocal(t *testing.T) { require.Empty(t, receiver.planCalls) } -func TestDispatchExecutor_Check_AliasWithRecursiveSentinelDoesNotDispatch(t *testing.T) { - receiver := &testDispatcher{} - - sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() - - // AliasIterator wrapping a RecursiveSentinel — should NOT dispatch - sentinel := query.NewRecursiveSentinelIterator("document", "viewer", false) - alias := query.NewAliasIterator("", "viewer", sentinel) - - path, err := sender.Check(ctx, NewDispatchIterator(alias), query.Object{ObjectType: "document", ObjectID: "doc1"}, query.ObjectAndRelation{ObjectType: "user", ObjectID: "alice", Relation: "..."}) - require.NoError(t, err) - // RecursiveSentinel.CheckImpl returns nil - require.Nil(t, path) - - // Verify NO dispatch was called - require.Empty(t, receiver.planCalls) -} - -func TestDispatchExecutor_IterSubjects_AliasWithRecursiveSentinelDoesNotDispatch(t *testing.T) { - receiver := &testDispatcher{} - - sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() - - sentinel := query.NewRecursiveSentinelIterator("document", "viewer", false) - alias := query.NewAliasIterator("", "viewer", sentinel) - - pathSeq, err := sender.IterSubjects(ctx, NewDispatchIterator(alias), query.Object{ObjectType: "document", ObjectID: "doc1"}, query.NoObjectFilter()) - require.NoError(t, err) - - paths, err := query.CollectAll(pathSeq) - require.NoError(t, err) - require.Empty(t, paths) - require.Empty(t, receiver.planCalls) -} - -func TestDispatchExecutor_IterResources_AliasWithRecursiveSentinelDoesNotDispatch(t *testing.T) { - receiver := &testDispatcher{} - - sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() - - sentinel := query.NewRecursiveSentinelIterator("document", "viewer", false) - alias := query.NewAliasIterator("", "viewer", sentinel) - - pathSeq, err := sender.IterResources(ctx, NewDispatchIterator(alias), query.ObjectAndRelation{ObjectType: "user", ObjectID: "alice", Relation: "..."}, query.NoObjectFilter()) - require.NoError(t, err) - - paths, err := query.CollectAll(pathSeq) - require.NoError(t, err) - require.Empty(t, paths) - require.Empty(t, receiver.planCalls) -} +// Sentinel-skip behavior is now enforced by the dispatch-wrap optimizer +// (containsUnmatchedRecursiveSentinelOutline in dispatch_optimizer.go) at +// planning time — the executor trusts the wrap and dispatches whenever one +// is present. The optimizer's outline-level check has its own coverage in +// dispatch_optimizer_test.go. func TestDispatchExecutor_Check_AliasWithSentinelInsideRecursiveDispatches(t *testing.T) { receiver := &testDispatcher{ @@ -399,28 +350,6 @@ func TestDispatchExecutor_IterResources_AliasWithSentinelInsideRecursiveDispatch require.Len(t, receiver.planCalls, 1) } -func TestDispatchExecutor_Check_DoubleRecursionUnmatchedSentinelDoesNotDispatch(t *testing.T) { - receiver := &testDispatcher{} - - sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() - - // Double recursion: RecursiveIterator for "folder#viewer" contains a - // RecursiveSentinelIterator for "document#reader" in its subtree. - // The document#reader sentinel is unmatched (no RecursiveIterator for it), - // so dispatch should be blocked. - documentSentinel := query.NewRecursiveSentinelIterator("document", "reader", false) - folderRecursive := query.NewRecursiveIterator(documentSentinel, "folder", "viewer") - alias := query.NewAliasIterator("", "viewer", folderRecursive) - - path, err := sender.Check(ctx, NewDispatchIterator(alias), query.Object{ObjectType: "folder", ObjectID: "f1"}, query.ObjectAndRelation{ObjectType: "user", ObjectID: "alice", Relation: "..."}) - require.NoError(t, err) - require.Nil(t, path) - - // Verify NO dispatch was called — unmatched sentinel blocks dispatch - require.Empty(t, receiver.planCalls) -} - func TestDispatchExecutor_Check_DoubleRecursionBothMatchedDispatches(t *testing.T) { receiver := &testDispatcher{ planResponses: []*v1.DispatchQueryPlanResponse{{ diff --git a/internal/dispatch/keys/computed.go b/internal/dispatch/keys/computed.go index 18d8353a26..63a147e0b8 100644 --- a/internal/dispatch/keys/computed.go +++ b/internal/dispatch/keys/computed.go @@ -132,36 +132,25 @@ func lookupSubjectsRequestToKey(req *v1.DispatchLookupSubjectsRequest) DispatchC func planCheckRequestToKey(req *v1.DispatchQueryPlanRequest) DispatchCacheKey { return PlanCheckLookupKey( req.PlanContext.Revision, - currentDispatchKey(req.PlanContext), + req.Plan, req.Resource, req.Subject, req.PlanContext.CaveatContext, ) } -// currentDispatchKey returns the canonical "def#rel" key for the dispatch -// currently being processed. By contract — see planContextForDispatch in -// internal/dispatch/executor.go — the sender appends the current dispatch's -// key to PlanContext.InProgressKeys before sending, so the last entry is -// always the current dispatch's key. Returns "" if InProgressKeys is empty, -// which is invalid for a Plan-dispatched request but is tolerated to keep -// this helper safe to call on partial test fixtures. -func currentDispatchKey(pc *v1.PlanContext) string { - if pc == nil || len(pc.InProgressKeys) == 0 { - return "" - } - return pc.InProgressKeys[len(pc.InProgressKeys)-1] -} - -// PlanCheckLookupKey computes the same Plan-Check cache key as -// planCheckRequestToKey, but takes the inputs directly so callers can probe the -// cache without first having to construct a DispatchQueryPlanRequest (which -// forces iterator serialization). The hashed component set must stay identical -// to planCheckRequestToKey or the cache-hit fast path and the DispatchQueryPlan -// slow path will end up on different cache entries. -func PlanCheckLookupKey(revision string, canonicalKey string, resource, subject *core.ObjectAndRelation, caveatContext *structpb.Struct) DispatchCacheKey { +// PlanCheckLookupKey computes the Plan-Check cache key from the serialized +// iterator bytes plus the request inputs. Same shape as planCheckRequestToKey +// so the cache-hit fast path (LookupPlanCheck) and the DispatchQueryPlan slow +// path produce identical keys for identical inputs. +// +// The plan bytes are the dispatch identity — content-addressed; two dispatches +// with byte-identical plans collide into the same cache slot, and any +// structural difference (advisor reorderings, optimizer variations) lands in +// a separate slot. +func PlanCheckLookupKey(revision string, plan []byte, resource, subject *core.ObjectAndRelation, caveatContext *structpb.Struct) DispatchCacheKey { return dispatchCacheKeyHash(planCheckPrefix, revision, - hashableString(canonicalKey), + hashableBytes(plan), hashableOnr{resource}, hashableOnr{subject}, hashableContext{Struct: caveatContext}, @@ -171,7 +160,7 @@ func PlanCheckLookupKey(revision string, canonicalKey string, resource, subject // planLookupResourcesRequestToKey converts a plan lookup resources request into a cache key func planLookupResourcesRequestToKey(req *v1.DispatchQueryPlanRequest) DispatchCacheKey { return dispatchCacheKeyHash(planLookupResourcesPrefix, req.PlanContext.Revision, - hashableString(currentDispatchKey(req.PlanContext)), + hashableBytes(req.Plan), hashableOnr{req.Subject}, hashableContext{Struct: req.PlanContext.CaveatContext}, ) @@ -180,7 +169,7 @@ func planLookupResourcesRequestToKey(req *v1.DispatchQueryPlanRequest) DispatchC // planLookupSubjectsRequestToKey converts a plan lookup subjects request into a cache key func planLookupSubjectsRequestToKey(req *v1.DispatchQueryPlanRequest) DispatchCacheKey { return dispatchCacheKeyHash(planLookupSubjectsPrefix, req.PlanContext.Revision, - hashableString(currentDispatchKey(req.PlanContext)), + hashableBytes(req.Plan), hashableOnr{req.Resource}, hashableContext{Struct: req.PlanContext.CaveatContext}, ) diff --git a/internal/dispatch/keys/computed_test.go b/internal/dispatch/keys/computed_test.go index aa420c5aad..7ed08a9236 100644 --- a/internal/dispatch/keys/computed_test.go +++ b/internal/dispatch/keys/computed_test.go @@ -556,9 +556,9 @@ var generatorFuncs = map[string]generatorFunc{ return planCheckRequestToKey(&v1.DispatchQueryPlanRequest{ Resource: ONR(resourceRelation.Namespace, resourceIds[0], resourceRelation.Relation), Subject: ONR(subjectRelation.Namespace, subjectIds[0], subjectRelation.Relation), + Plan: []byte(resourceRelation.Relation), PlanContext: &v1.PlanContext{ - Revision: metadata.AtRevision, - InProgressKeys: []string{resourceRelation.Relation}, + Revision: metadata.AtRevision, }, }), []string{ resourceRelation.Relation, @@ -580,9 +580,9 @@ var generatorFuncs = map[string]generatorFunc{ ) (DispatchCacheKey, []string) { return planLookupResourcesRequestToKey(&v1.DispatchQueryPlanRequest{ Subject: ONR(subjectRelation.Namespace, subjectIds[0], subjectRelation.Relation), + Plan: []byte(resourceRelation.Relation), PlanContext: &v1.PlanContext{ - Revision: metadata.AtRevision, - InProgressKeys: []string{resourceRelation.Relation}, + Revision: metadata.AtRevision, }, }), []string{ resourceRelation.Relation, @@ -602,9 +602,9 @@ var generatorFuncs = map[string]generatorFunc{ ) (DispatchCacheKey, []string) { return planLookupSubjectsRequestToKey(&v1.DispatchQueryPlanRequest{ Resource: ONR(resourceRelation.Namespace, resourceIds[0], resourceRelation.Relation), + Plan: []byte(resourceRelation.Relation), PlanContext: &v1.PlanContext{ - Revision: metadata.AtRevision, - InProgressKeys: []string{resourceRelation.Relation}, + Revision: metadata.AtRevision, }, }), []string{ resourceRelation.Relation, diff --git a/internal/dispatch/keys/hasher.go b/internal/dispatch/keys/hasher.go index a07e57307f..5eb7cae5ae 100644 --- a/internal/dispatch/keys/hasher.go +++ b/internal/dispatch/keys/hasher.go @@ -21,6 +21,7 @@ type hashableValue interface { type hasherInterface interface { WriteString(value string) + Write(p []byte) } type hashableRelationReference struct { @@ -70,6 +71,18 @@ func (hs hashableString) AppendToHash(hasher hasherInterface) { hasher.WriteString(string(hs)) } +// hashableBytes feeds raw bytes into the cache-key hash. Used for Plan-Check +// keys, where the serialized iterator subtree is the dispatch identity — +// hashing the bytes directly avoids needing any parallel "canonical key" +// string alongside. +type hashableBytes []byte + +func (hb hashableBytes) AppendToHash(hasher hasherInterface) { + if len(hb) > 0 { + hasher.Write(hb) + } +} + type hashableLimit uint32 func (hl hashableLimit) AppendToHash(hasher hasherInterface) { @@ -177,6 +190,17 @@ func (h *dispatchCacheKeyHasher) WriteString(value string) { h.mustWriteString(value) } +// Write feeds raw bytes into the underlying xxhash digest. Used by +// hashableBytes so callers (e.g., Plan-Check keys hashing the serialized +// iterator) don't have to round-trip through string conversion. +func (h *dispatchCacheKeyHasher) Write(p []byte) { + // xxhash.Digest.Write never returns an error; mirror mustWriteString. + _, err := h.stableHasher.Write(p) + if err != nil { + panic(fmt.Errorf("got an error from writing to the stable hasher: %w", err)) + } +} + func (h *dispatchCacheKeyHasher) mustWriteString(value string) { // NOTE: xxhash doesn't seem to ever return an error for WriteString, but we check it just // to be on the safe side. diff --git a/internal/dispatch/singleflight/singleflight_test.go b/internal/dispatch/singleflight/singleflight_test.go index eb9cf6e9ce..717ca363cc 100644 --- a/internal/dispatch/singleflight/singleflight_test.go +++ b/internal/dispatch/singleflight/singleflight_test.go @@ -364,10 +364,10 @@ func TestDispatchQueryPlanCheckCoalescesConcurrentCalls(t *testing.T) { Operation: v1.PlanOperation_PLAN_OPERATION_CHECK, Resource: tuple.ONRStringToCore("document", "doc1", "viewer"), Subject: tuple.ONRStringToCore("user", "alice", "..."), + Plan: []byte("document#viewer"), PlanContext: &v1.PlanContext{ - Revision: "1234", - SchemaHash: []byte(datalayer.NoSchemaHashForTesting), - InProgressKeys: []string{"document#viewer"}, + Revision: "1234", + SchemaHash: []byte(datalayer.NoSchemaHashForTesting), }, } diff --git a/pkg/proto/dispatch/v1/dispatch.pb.go b/pkg/proto/dispatch/v1/dispatch.pb.go index 7f79ecafe7..bc96a0e9b4 100644 --- a/pkg/proto/dispatch/v1/dispatch.pb.go +++ b/pkg/proto/dispatch/v1/dispatch.pb.go @@ -1740,13 +1740,6 @@ type PlanContext struct { MaxRecursionDepth int32 `protobuf:"varint,3,opt,name=max_recursion_depth,json=maxRecursionDepth,proto3" json:"max_recursion_depth,omitempty"` OptionalDatastoreLimit uint64 `protobuf:"varint,4,opt,name=optional_datastore_limit,json=optionalDatastoreLimit,proto3" json:"optional_datastore_limit,omitempty"` SchemaHash []byte `protobuf:"bytes,5,opt,name=schema_hash,json=schemaHash,proto3" json:"schema_hash,omitempty"` - // in_progress_keys are the dispatch canonical keys (\"def#rel\") that ancestors - // of this dispatch are currently computing. The receiver-side DispatchExecutor - // refuses to dispatch any alias whose key is in this set, falling back to local - // execution. This is what breaks dispatch loops that arise when the standalone - // plan for a key contains another structurally complete instance of itself - // (cross-relation recursion the sentinel machinery can't see). - InProgressKeys []string `protobuf:"bytes,6,rep,name=in_progress_keys,json=inProgressKeys,proto3" json:"in_progress_keys,omitempty"` // top_level_operation carries the user-facing operation (Check / LookupSubjects // / LookupResources) that started the dispatch chain. The receiver pre-seals // its qctx with this value so iterators that consult TopLevelOperation (e.g. @@ -1824,13 +1817,6 @@ func (x *PlanContext) GetSchemaHash() []byte { return nil } -func (x *PlanContext) GetInProgressKeys() []string { - if x != nil { - return x.InProgressKeys - } - return nil -} - func (x *PlanContext) GetTopLevelOperation() PlanOperation { if x != nil { return x.TopLevelOperation @@ -2313,16 +2299,15 @@ const file_dispatch_v1_dispatch_proto_rawDesc = "" + "\aUNKNOWN\x10\x00\x12\f\n" + "\bRELATION\x10\x01\x12\x0e\n" + "\n" + - "PERMISSION\x10\x02\"\xea\x02\n" + + "PERMISSION\x10\x02\"\xd8\x02\n" + "\vPlanContext\x12\x1a\n" + "\brevision\x18\x01 \x01(\tR\brevision\x12>\n" + "\x0ecaveat_context\x18\x02 \x01(\v2\x17.google.protobuf.StructR\rcaveatContext\x12.\n" + "\x13max_recursion_depth\x18\x03 \x01(\x05R\x11maxRecursionDepth\x128\n" + "\x18optional_datastore_limit\x18\x04 \x01(\x04R\x16optionalDatastoreLimit\x12\x1f\n" + "\vschema_hash\x18\x05 \x01(\fR\n" + - "schemaHash\x12(\n" + - "\x10in_progress_keys\x18\x06 \x03(\tR\x0einProgressKeys\x12J\n" + - "\x13top_level_operation\x18\a \x01(\x0e2\x1a.dispatch.v1.PlanOperationR\x11topLevelOperation\"\xd8\x02\n" + + "schemaHash\x12J\n" + + "\x13top_level_operation\x18\a \x01(\x0e2\x1a.dispatch.v1.PlanOperationR\x11topLevelOperationJ\x04\b\x06\x10\aR\x10in_progress_keys\"\xd8\x02\n" + "\x18DispatchQueryPlanRequest\x128\n" + "\toperation\x18\x01 \x01(\x0e2\x1a.dispatch.v1.PlanOperationR\toperation\x126\n" + "\bresource\x18\x03 \x01(\v2\x1a.core.v1.ObjectAndRelationR\bresource\x124\n" + diff --git a/pkg/proto/dispatch/v1/dispatch_vtproto.pb.go b/pkg/proto/dispatch/v1/dispatch_vtproto.pb.go index dbf0524433..d52418332e 100644 --- a/pkg/proto/dispatch/v1/dispatch_vtproto.pb.go +++ b/pkg/proto/dispatch/v1/dispatch_vtproto.pb.go @@ -659,11 +659,6 @@ func (m *PlanContext) CloneVT() *PlanContext { copy(tmpBytes, rhs) r.SchemaHash = tmpBytes } - if rhs := m.InProgressKeys; rhs != nil { - tmpContainer := make([]string, len(rhs)) - copy(tmpContainer, rhs) - r.InProgressKeys = tmpContainer - } if len(m.unknownFields) > 0 { r.unknownFields = make([]byte, len(m.unknownFields)) copy(r.unknownFields, m.unknownFields) @@ -1710,15 +1705,6 @@ func (this *PlanContext) EqualVT(that *PlanContext) bool { if string(this.SchemaHash) != string(that.SchemaHash) { return false } - if len(this.InProgressKeys) != len(that.InProgressKeys) { - return false - } - for i, vx := range this.InProgressKeys { - vy := that.InProgressKeys[i] - if vx != vy { - return false - } - } if this.TopLevelOperation != that.TopLevelOperation { return false } @@ -3621,15 +3607,6 @@ func (m *PlanContext) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i-- dAtA[i] = 0x38 } - if len(m.InProgressKeys) > 0 { - for iNdEx := len(m.InProgressKeys) - 1; iNdEx >= 0; iNdEx-- { - i -= len(m.InProgressKeys[iNdEx]) - copy(dAtA[i:], m.InProgressKeys[iNdEx]) - i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.InProgressKeys[iNdEx]))) - i-- - dAtA[i] = 0x32 - } - } if len(m.SchemaHash) > 0 { i -= len(m.SchemaHash) copy(dAtA[i:], m.SchemaHash) @@ -4745,12 +4722,6 @@ func (m *PlanContext) SizeVT() (n int) { if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } - if len(m.InProgressKeys) > 0 { - for _, s := range m.InProgressKeys { - l = len(s) - n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) - } - } if m.TopLevelOperation != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.TopLevelOperation)) } @@ -9104,38 +9075,6 @@ func (m *PlanContext) UnmarshalVT(dAtA []byte) error { m.SchemaHash = []byte{} } iNdEx = postIndex - case 6: - if wireType != 2 { - return fmt.Errorf("proto: wrong wireType = %d for field InProgressKeys", wireType) - } - var stringLen uint64 - for shift := uint(0); ; shift += 7 { - if shift >= 64 { - return protohelpers.ErrIntOverflow - } - if iNdEx >= l { - return io.ErrUnexpectedEOF - } - b := dAtA[iNdEx] - iNdEx++ - stringLen |= uint64(b&0x7F) << shift - if b < 0x80 { - break - } - } - intStringLen := int(stringLen) - if intStringLen < 0 { - return protohelpers.ErrInvalidLength - } - postIndex := iNdEx + intStringLen - if postIndex < 0 { - return protohelpers.ErrInvalidLength - } - if postIndex > l { - return io.ErrUnexpectedEOF - } - m.InProgressKeys = append(m.InProgressKeys, string(dAtA[iNdEx:postIndex])) - iNdEx = postIndex case 7: if wireType != 0 { return fmt.Errorf("proto: wrong wireType = %d for field TopLevelOperation", wireType) diff --git a/proto/internal/dispatch/v1/dispatch.proto b/proto/internal/dispatch/v1/dispatch.proto index 1d80a5e127..c0f9306b33 100644 --- a/proto/internal/dispatch/v1/dispatch.proto +++ b/proto/internal/dispatch/v1/dispatch.proto @@ -239,13 +239,14 @@ message PlanContext { int32 max_recursion_depth = 3; uint64 optional_datastore_limit = 4; bytes schema_hash = 5; - // in_progress_keys are the dispatch canonical keys (\"def#rel\") that ancestors - // of this dispatch are currently computing. The receiver-side DispatchExecutor - // refuses to dispatch any alias whose key is in this set, falling back to local - // execution. This is what breaks dispatch loops that arise when the standalone - // plan for a key contains another structurally complete instance of itself - // (cross-relation recursion the sentinel machinery can't see). - repeated string in_progress_keys = 6; + // Tag 6 was `repeated string in_progress_keys`, the receiver-side + // cross-relation cycle guard. Removed once the serialize-over-wire plan + // change eliminated receiver-side outline rebuilds: the compiler emits + // RecursiveIterator + RecursiveSentinelIterator for every cyclic schema + // (`pkg/query/build_tree.go`), so a finite compiled plan can't contain + // nested same-key dispatches, and the runtime guard is unreachable. + reserved 6; + reserved "in_progress_keys"; // top_level_operation carries the user-facing operation (Check / LookupSubjects // / LookupResources) that started the dispatch chain. The receiver pre-seals // its qctx with this value so iterators that consult TopLevelOperation (e.g. @@ -258,10 +259,9 @@ message PlanContext { message DispatchQueryPlanRequest { PlanOperation operation = 1; - // Tag 2 was `string canonical_key`, removed once `plan` carried the - // iterator subtree on the wire and the sender-side caching/routing layers - // were taught to read the current dispatch's "def#rel" key from the last - // entry of PlanContext.in_progress_keys instead. + // Tag 2 was `string canonical_key`. The receiver no longer needs a key — + // the plan bytes are the dispatch identity (sender-side caching/routing + // layers hash `plan` directly). reserved 2; reserved "canonical_key"; // For CHECK_MANY_RESOURCES, resource is empty and `many` carries the resources. From cebf2b19514cc8f8058baaaabe13d3f2ad1d741d Mon Sep 17 00:00:00 2001 From: Barak Michener Date: Fri, 22 May 2026 12:31:25 -0700 Subject: [PATCH 2/4] refactor: re-advise after dispatch --- internal/dispatch/dispatch_optimizer.go | 60 ++++++++++++-- internal/dispatch/dispatch_optimizer_test.go | 85 ++++++++++++++++---- internal/dispatch/graph/graph.go | 8 ++ pkg/query/advisor_count.go | 18 +++-- pkg/query/queryplan_metadata.go | 33 ++++++++ 5 files changed, 174 insertions(+), 30 deletions(-) diff --git a/internal/dispatch/dispatch_optimizer.go b/internal/dispatch/dispatch_optimizer.go index 2f4b1f69a9..27f44eefa0 100644 --- a/internal/dispatch/dispatch_optimizer.go +++ b/internal/dispatch/dispatch_optimizer.go @@ -30,14 +30,19 @@ func init() { }) } -// wrapAliasWithDispatch is a bottom-up OutlineMutation that wraps each -// AliasIteratorType node in a DispatchIteratorType node. The wrap is the -// static portion of the old DispatchExecutor.shouldDispatchAlias check: -// - the "is this an alias" test becomes "is this outline node an alias?" -// - the sentinel-recursion guard runs over the outline subtree +// wrapAliasWithDispatch is a bottom-up OutlineMutation that wraps an +// AliasIteratorType node in a DispatchIteratorType when dispatching the alias +// is worth the per-dispatch overhead. Three exclusions: +// - non-aliases are passed through untouched; +// - aliases whose subtree contains an unmatched RecursiveSentinel can't be +// dispatched without severing the sentinel's collection context; +// - aliases whose body is "leaf-like" (a single Datastore read, or a Union +// whose direct children are all Datastores) are too cheap to amortize the +// dispatch wrap — see aliasBodyIsLeafLike for the rationale. // -// The remaining runtime guard (in-progress dispatch-key cycle break) cannot -// be made static and stays in the executor. +// After Phase 1's executor migration, the wrap is the authoritative dispatch +// boundary: every wrap → a dispatch, no wrap → local execution. So skipping +// a wrap here means skipping the dispatch entirely. func wrapAliasWithDispatch(outline query.Outline) query.Outline { if outline.Type != query.AliasIteratorType { return outline @@ -45,6 +50,9 @@ func wrapAliasWithDispatch(outline query.Outline) query.Outline { if containsUnmatchedRecursiveSentinelOutline(outline) { return outline } + if aliasBodyIsLeafLike(outline) { + return outline + } // New node has ID==0; ApplyOptimizations runs FillMissingNodeIDs after the // transform, which assigns it a fresh ID and records its canonical key. return query.Outline{ @@ -53,6 +61,44 @@ func wrapAliasWithDispatch(outline query.Outline) query.Outline { } } +// aliasBodyIsLeafLike reports whether the alias's single body is too cheap to +// be worth wrapping for dispatch: +// - a bare DatastoreIterator (one relationship read), or +// - a Union whose direct children are all DatastoreIterators (a fan-in of +// reads to compose multiple direct relations like `view = viewer + editor`). +// +// "Direct children" only — we intentionally don't recurse into nested Unions. +// Anything richer than this (an Arrow, an Intersection, a Recursive, a nested +// Alias-now-wrapped-as-Dispatch) is heavy enough that dispatch + caching pays +// for itself, so the wrap stays. +// +// Optimizers run bottom-up via MutateOutline, so the body we inspect here is +// the post-mutation outline. If a descendant alias was itself a candidate for +// wrapping it has already become DispatchIteratorType — which is not Datastore, +// so this check correctly classifies the parent as "not leaf-like" and keeps +// its wrap. +func aliasBodyIsLeafLike(outline query.Outline) bool { + if len(outline.SubOutlines) != 1 { + return false + } + body := outline.SubOutlines[0] + switch body.Type { + case query.DatastoreIteratorType: + return true + case query.UnionIteratorType: + if len(body.SubOutlines) == 0 { + return false + } + for _, child := range body.SubOutlines { + if child.Type != query.DatastoreIteratorType { + return false + } + } + return true + } + return false +} + // recursiveKey uniquely identifies a recursion pair by definition and relation // name. Used by containsUnmatchedRecursiveSentinelOutline to pair up sentinels // with their managing RecursiveIterators at optimization time. diff --git a/internal/dispatch/dispatch_optimizer_test.go b/internal/dispatch/dispatch_optimizer_test.go index 01c5e9fbe2..ac21dd520c 100644 --- a/internal/dispatch/dispatch_optimizer_test.go +++ b/internal/dispatch/dispatch_optimizer_test.go @@ -50,6 +50,15 @@ func dsOutline() query.Outline { } } +// arrowOutline builds an Arrow node — useful for constructing alias bodies +// that are NOT "leaf-like" so the dispatch-wrap optimizer keeps the wrap. +func arrowOutline(left, right query.Outline) query.Outline { + return query.Outline{ + Type: query.ArrowIteratorType, + SubOutlines: []query.Outline{left, right}, + } +} + // runDispatchWrap canonicalizes outline then runs only the dispatch-wrap // optimization on it, returning the resulting root outline. func runDispatchWrap(t *testing.T, outline query.Outline) query.Outline { @@ -62,17 +71,16 @@ func runDispatchWrap(t *testing.T, outline query.Outline) query.Outline { } func TestDispatchWrapOptimizer(t *testing.T) { - t.Run("wraps a single alias", func(t *testing.T) { - input := aliasOutline("document", "viewer", dsOutline()) + t.Run("wraps alias whose body is an Arrow", func(t *testing.T) { + input := aliasOutline("document", "viewer", + arrowOutline(dsOutline(), dsOutline()), + ) got := runDispatchWrap(t, input) require.Equal(t, DispatchIteratorType, got.Type) require.Len(t, got.SubOutlines, 1) require.Equal(t, query.AliasIteratorType, got.SubOutlines[0].Type) require.Equal(t, "viewer", got.SubOutlines[0].Args.RelationName) - // child unchanged - require.Len(t, got.SubOutlines[0].SubOutlines, 1) - require.Equal(t, query.DatastoreIteratorType, got.SubOutlines[0].SubOutlines[0].Type) }) t.Run("leaves non-alias outlines alone", func(t *testing.T) { @@ -81,8 +89,47 @@ func TestDispatchWrapOptimizer(t *testing.T) { require.Equal(t, query.DatastoreIteratorType, got.Type) }) - t.Run("wraps nested aliases at every level", func(t *testing.T) { + t.Run("skips wrap when alias body is a single Datastore", func(t *testing.T) { + // Alias("view") → Datastore — too cheap to amortize a dispatch. + input := aliasOutline("document", "view", dsOutline()) + got := runDispatchWrap(t, input) + require.Equal(t, query.AliasIteratorType, got.Type, + "leaf-like alias (body = Datastore) must NOT be wrapped") + require.Equal(t, "view", got.Args.RelationName) + }) + + t.Run("skips wrap when alias body is a Union of direct Datastores", func(t *testing.T) { + // Alias("view") → Union[DS, DS] — still just data reads; the dispatch + // overhead would dwarf the work. + input := aliasOutline("document", "view", query.Outline{ + Type: query.UnionIteratorType, + SubOutlines: []query.Outline{dsOutline(), dsOutline()}, + }) + got := runDispatchWrap(t, input) + require.Equal(t, query.AliasIteratorType, got.Type, + "union-of-datastores body must NOT be wrapped") + }) + + t.Run("wraps when Union has a non-Datastore direct child", func(t *testing.T) { + // Alias("viewer") → Union[DS, Arrow(...)] — the Arrow makes it heavy + // enough that the wrap pays back. + input := aliasOutline("document", "viewer", query.Outline{ + Type: query.UnionIteratorType, + SubOutlines: []query.Outline{ + dsOutline(), + arrowOutline(dsOutline(), dsOutline()), + }, + }) + got := runDispatchWrap(t, input) + require.Equal(t, DispatchIteratorType, got.Type, + "mixed-children union body must be wrapped") + }) + + t.Run("nested aliases: outer wraps, leaf-like inner stays unwrapped", func(t *testing.T) { // Alias("perm")(Union[ Alias("view")(DS), DS ]) + // - Inner Alias("view")(DS) is leaf-like → stays unwrapped. + // - Outer Union has a non-Datastore direct child (the inner Alias), + // so the outer Alias is NOT leaf-like → wraps. inner := aliasOutline("document", "view", dsOutline()) union := query.Outline{ Type: query.UnionIteratorType, @@ -92,25 +139,24 @@ func TestDispatchWrapOptimizer(t *testing.T) { got := runDispatchWrap(t, input) - // Outer alias is wrapped - require.Equal(t, DispatchIteratorType, got.Type) + require.Equal(t, DispatchIteratorType, got.Type, "outer alias should be wrapped") outerAlias := got.SubOutlines[0] require.Equal(t, query.AliasIteratorType, outerAlias.Type) require.Equal(t, "perm", outerAlias.Args.RelationName) - // Inner alias inside the union is also wrapped gotUnion := outerAlias.SubOutlines[0] require.Equal(t, query.UnionIteratorType, gotUnion.Type) require.Len(t, gotUnion.SubOutlines, 2) - require.Equal(t, DispatchIteratorType, gotUnion.SubOutlines[0].Type) - require.Equal(t, query.AliasIteratorType, gotUnion.SubOutlines[0].SubOutlines[0].Type) - require.Equal(t, "view", gotUnion.SubOutlines[0].SubOutlines[0].Args.RelationName) + // Inner alias is leaf-like and stays unwrapped. + require.Equal(t, query.AliasIteratorType, gotUnion.SubOutlines[0].Type, + "inner leaf-like alias must stay unwrapped") + require.Equal(t, "view", gotUnion.SubOutlines[0].Args.RelationName) }) t.Run("wraps alias whose subtree has a matched sentinel/iterator pair", func(t *testing.T) { // Alias("view")(Recursive("folder","viewer")(Sentinel("folder","viewer"))) - // The recursive iterator inside the alias matches the sentinel's key, so - // the dispatch boundary is safe. + // The recursive iterator inside the alias matches the sentinel's key, + // so the dispatch boundary is safe; body is also not leaf-like. input := aliasOutline("document", "view", recursiveOutline("folder", "viewer", sentinelOutline("folder", "viewer")), ) @@ -122,7 +168,8 @@ func TestDispatchWrapOptimizer(t *testing.T) { // Alias("view")(Union[ Sentinel("folder","viewer"), DS ]) // No RecursiveIterator with key folder#viewer exists in the alias's // subtree, so dispatching the alias would sever the sentinel from its - // collection context. + // collection context. (The sentinel guard runs before the leaf-like + // check.) input := aliasOutline("document", "view", query.Outline{ Type: query.UnionIteratorType, @@ -138,7 +185,9 @@ func TestDispatchWrapOptimizer(t *testing.T) { }) t.Run("assigns IDs and canonical keys to newly-wrapped nodes", func(t *testing.T) { - input := aliasOutline("document", "viewer", dsOutline()) + input := aliasOutline("document", "viewer", + arrowOutline(dsOutline(), dsOutline()), + ) co, err := query.CanonicalizeOutline(input) require.NoError(t, err) res, err := ApplyDispatchWrap(co, queryopt.RequestParams{}) @@ -151,7 +200,9 @@ func TestDispatchWrapOptimizer(t *testing.T) { }) t.Run("optimized outline compiles to DispatchIterator above AliasIterator", func(t *testing.T) { - input := aliasOutline("document", "viewer", dsOutline()) + input := aliasOutline("document", "viewer", + arrowOutline(dsOutline(), dsOutline()), + ) co, err := query.CanonicalizeOutline(input) require.NoError(t, err) res, err := ApplyDispatchWrap(co, queryopt.RequestParams{}) diff --git a/internal/dispatch/graph/graph.go b/internal/dispatch/graph/graph.go index 7e4ad34ce7..6809fc6806 100644 --- a/internal/dispatch/graph/graph.go +++ b/internal/dispatch/graph/graph.go @@ -532,6 +532,14 @@ func (ld *localDispatcher) DispatchQueryPlan( return err } + // Re-advise against the receiver's locally accumulated stats. The sender + // baked its hint choices (arrow direction) into the wire bytes against its + // own stats; this call gives the receiver a chance to refine those choices + // using whatever counts it has observed. Structurally inert — the iterator + // tree shape (and therefore its canonical key, and therefore the cache key + // the upstream caching layer computed from req.Plan) does not change. + ld.queryPlanMetadata.ReAdviseIterator(it) + // Use DispatchExecutor so nested alias boundaries in the subtree re-dispatch // through the full dispatch chain. dl := datalayer.MustFromContext(ctx) diff --git a/pkg/query/advisor_count.go b/pkg/query/advisor_count.go index 3c8f015c78..bf0c846a9a 100644 --- a/pkg/query/advisor_count.go +++ b/pkg/query/advisor_count.go @@ -23,12 +23,19 @@ func (a *CountAdvisor) GetHints(outline Outline, keySource CanonicalKeySource) ( if outline.Type != ArrowIteratorType || len(outline.SubOutlines) != 2 { return nil, nil } - leftKey := keySource.GetCanonicalKey(outline.SubOutlines[0].ID) rightKey := keySource.GetCanonicalKey(outline.SubOutlines[1].ID) + return []Hint{ArrowDirectionHint(chooseArrowDirection(leftKey, rightKey, a.stats))}, nil +} - leftStats := a.stats[leftKey] - rightStats := a.stats[rightKey] +// chooseArrowDirection picks the arrow execution direction by comparing +// observed left/right fan-outs. Shared between the outline-time advisor +// (compile-time hint emission) and the iterator-time re-advise on dispatch +// receipt — both need the same comparison logic against the same stats map, +// just sourced from different points in the pipeline. +func chooseArrowDirection(leftKey, rightKey CanonicalKey, stats map[CanonicalKey]CountStats) arrowDirection { + leftStats := stats[leftKey] + rightStats := stats[rightKey] leftFanout := defaultArrowFanout rightFanout := defaultArrowFanout @@ -36,15 +43,14 @@ func (a *CountAdvisor) GetHints(outline Outline, keySource CanonicalKeySource) ( if leftStats.IterSubjectsCalls != 0 { leftFanout = float64(leftStats.IterSubjectsResults) / float64(leftStats.IterSubjectsCalls) } - if rightStats.IterResourcesCalls != 0 { rightFanout = float64(rightStats.IterResourcesResults) / float64(rightStats.IterResourcesCalls) } if rightFanout < leftFanout { - return []Hint{ArrowDirectionHint(rightToLeft)}, nil + return rightToLeft } - return []Hint{ArrowDirectionHint(leftToRight)}, nil + return leftToRight } // GetMutations is a stub — no structural mutations from count data yet. diff --git a/pkg/query/queryplan_metadata.go b/pkg/query/queryplan_metadata.go index 9a1eaea2e2..14b0169665 100644 --- a/pkg/query/queryplan_metadata.go +++ b/pkg/query/queryplan_metadata.go @@ -60,3 +60,36 @@ func (m *QueryPlanMetadata) ApplyAdvisor(co CanonicalOutline) (CanonicalOutline, } return ApplyAdvisor(co, NewCountAdvisor(stats)) } + +// ReAdviseIterator walks a compiled iterator tree and refreshes per-iterator +// optimization choices against the locally accumulated stats. Structurally +// inert — only hint-level fields (currently ArrowIterator.direction) change, +// so the iterator's CanonicalKey is preserved and the cache key the receiver +// computed from req.Plan bytes remains valid. +// +// Used by the dispatch receiver: the sender baked its hints into the wire +// bytes against its stats; ReAdviseIterator gives the receiver one shot to +// adjust those choices against its own stats before execution. +func (m *QueryPlanMetadata) ReAdviseIterator(it Iterator) { + stats := m.GetStats() + if len(stats) == 0 { + return + } + reAdviseIterator(it, stats) +} + +// reAdviseIterator is the recursive worker for ReAdviseIterator. Walks the +// iterator tree depth-first; for each ArrowIterator pulls its left/right +// sub-iterators' canonical keys and updates the direction field directly via +// chooseArrowDirection. No new iterators allocated, no decompile/recompile. +func reAdviseIterator(it Iterator, stats map[CanonicalKey]CountStats) { + if arrow, ok := it.(*ArrowIterator); ok { + subs := arrow.Subiterators() + if len(subs) == 2 { + arrow.direction = chooseArrowDirection(subs[0].CanonicalKey(), subs[1].CanonicalKey(), stats) + } + } + for _, sub := range it.Subiterators() { + reAdviseIterator(sub, stats) + } +} From e421abab3bf2ed76f1a40eef524773f73a72ecff Mon Sep 17 00:00:00 2001 From: Barak Michener Date: Fri, 22 May 2026 13:26:23 -0700 Subject: [PATCH 3/4] refactor: dispatch taking into account operation --- internal/dispatch/executor.go | 38 ++++++++++++-------- internal/dispatch/executor_test.go | 48 ++++++++++++++----------- internal/dispatch/graph/graph.go | 24 +++++++++++++ internal/dispatch/keys/computed_test.go | 6 ++-- internal/dispatch/keys/hasher.go | 5 +-- 5 files changed, 82 insertions(+), 39 deletions(-) diff --git a/internal/dispatch/executor.go b/internal/dispatch/executor.go index aa01f63d53..b7895292c0 100644 --- a/internal/dispatch/executor.go +++ b/internal/dispatch/executor.go @@ -45,30 +45,40 @@ func NewDispatchExecutor(dispatcher Dispatcher, planContext *v1.PlanContext, dis } } -// shouldDispatch reports whether the iterator is a DispatchIterator the -// executor should ship over RPC vs run locally. The wrap is the authoritative -// dispatch boundary — the dispatch-wrap optimizer -// (internal/dispatch/dispatch_optimizer.go) decides which alias positions get -// a wrap and excludes unsafe sentinel positions at planning time; this -// predicate just checks for the marker. +// shouldDispatch reports whether to ship this iterator over RPC vs run it +// locally. Two conditions must hold: +// +// 1. The iterator is a DispatchIterator — the optimizer's authoritative +// dispatch boundary (see internal/dispatch/dispatch_optimizer.go). +// +// 2. currentOp matches ctx.TopLevelOperation — the request type at this +// wrap matches the user-facing request type that started the work. The +// receiver-side executor only knows how to drive one operation per +// dispatch (the request's op == top-level op by construction at the +// edge), so dispatching a wrap reached via a *different* sub-operation +// (e.g. Arrow(RTL) drives IterResources on its right subtree while the +// user request is Check) would launch an RPC whose results we then +// re-process locally — pure overhead. Stay local; let CheckImpl drive. // // No runtime cycle-break: the compiler emits RecursiveIterator + // RecursiveSentinelIterator for every cyclic schema, so the compiled tree is // finite by construction and a dispatch chain can't loop on the same key. -func (e *DispatchExecutor) shouldDispatch(it query.Iterator) bool { - _, ok := it.(*DispatchIterator) - return ok +func (e *DispatchExecutor) shouldDispatch(ctx *query.Context, it query.Iterator, currentOp query.Operation) bool { + if _, ok := it.(*DispatchIterator); !ok { + return false + } + return currentOp == ctx.TopLevelOperation } func (e *DispatchExecutor) Check(ctx *query.Context, it query.Iterator, resource query.Object, subject query.ObjectAndRelation) (*query.Path, error) { - if e.shouldDispatch(it) { + if e.shouldDispatch(ctx, it, query.OperationCheck) { return e.dispatchCheck(ctx, it, resource, subject) } return it.CheckImpl(ctx, resource, subject) } func (e *DispatchExecutor) CheckManySubjects(ctx *query.Context, it query.Iterator, resource query.Object, subjects []query.ObjectAndRelation) ([]*query.Path, error) { - if !e.shouldDispatch(it) { + if !e.shouldDispatch(ctx, it, query.OperationCheck) { out := make([]*query.Path, len(subjects)) for i, s := range subjects { p, err := it.CheckImpl(ctx, resource, s) @@ -83,7 +93,7 @@ func (e *DispatchExecutor) CheckManySubjects(ctx *query.Context, it query.Iterat } func (e *DispatchExecutor) CheckManyResources(ctx *query.Context, it query.Iterator, resources []query.Object, subject query.ObjectAndRelation) ([]*query.Path, error) { - if !e.shouldDispatch(it) { + if !e.shouldDispatch(ctx, it, query.OperationCheck) { out := make([]*query.Path, len(resources)) for i, r := range resources { p, err := it.CheckImpl(ctx, r, subject) @@ -98,7 +108,7 @@ func (e *DispatchExecutor) CheckManyResources(ctx *query.Context, it query.Itera } func (e *DispatchExecutor) IterSubjects(ctx *query.Context, it query.Iterator, resource query.Object, filterSubjectType query.ObjectType) (query.PathSeq, error) { - if e.shouldDispatch(it) { + if e.shouldDispatch(ctx, it, query.OperationIterSubjects) { return e.dispatchIterSubjects(ctx, it, resource, filterSubjectType) } pathSeq, err := it.IterSubjectsImpl(ctx, resource, filterSubjectType) @@ -109,7 +119,7 @@ func (e *DispatchExecutor) IterSubjects(ctx *query.Context, it query.Iterator, r } func (e *DispatchExecutor) IterResources(ctx *query.Context, it query.Iterator, subject query.ObjectAndRelation, filterResourceType query.ObjectType) (query.PathSeq, error) { - if e.shouldDispatch(it) { + if e.shouldDispatch(ctx, it, query.OperationIterResources) { return e.dispatchIterResources(ctx, it, subject, filterResourceType) } pathSeq, err := it.IterResourcesImpl(ctx, subject, filterResourceType) diff --git a/internal/dispatch/executor_test.go b/internal/dispatch/executor_test.go index 4759349522..176b62670e 100644 --- a/internal/dispatch/executor_test.go +++ b/internal/dispatch/executor_test.go @@ -63,8 +63,15 @@ func (d *testDispatcher) ReadyState() ReadyState { return ReadyState{IsReady: tr var _ Dispatcher = &testDispatcher{} -func newTestContext() *query.Context { - return query.NewLocalContext(context.Background()) +// newTestContext returns a query.Context pre-sealed with topLevelOp so that +// the DispatchExecutor's shouldDispatch sees a matching TopLevelOperation +// and actually fires (in production this seal happens in ctx.Check / +// ctx.IterX / on dispatch receipt via MarkAsOperation; the executor unit +// tests bypass those entry points and drive the executor directly). +func newTestContext(topLevelOp query.Operation) *query.Context { + ctx := query.NewLocalContext(context.Background()) + ctx.TopLevelOperation = topLevelOp + return ctx } func TestDispatchExecutor_Check_AliasDispatches(t *testing.T) { @@ -82,7 +89,7 @@ func TestDispatchExecutor_Check_AliasDispatches(t *testing.T) { } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) // AliasIterator wrapping a FixedIterator — should trigger dispatch inner := query.NewFixedIterator() @@ -105,7 +112,7 @@ func TestDispatchExecutor_Check_NonAliasLocal(t *testing.T) { receiver := &testDispatcher{} sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) // FixedIterator (not an alias) — should delegate locally, no dispatch fixed := query.NewFixedIterator(query.Path{ @@ -131,7 +138,7 @@ func TestDispatchExecutor_Check_AliasNoResult(t *testing.T) { } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) @@ -152,7 +159,7 @@ func TestDispatchExecutor_IterSubjects_AliasDispatches(t *testing.T) { } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterSubjects) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) @@ -173,7 +180,7 @@ func TestDispatchExecutor_IterSubjects_NonAliasLocal(t *testing.T) { receiver := &testDispatcher{} sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterSubjects) fixed := query.NewFixedIterator( query.Path{ @@ -208,7 +215,7 @@ func TestDispatchExecutor_IterResources_AliasDispatches(t *testing.T) { } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterResources) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) @@ -229,7 +236,7 @@ func TestDispatchExecutor_IterResources_NonAliasLocal(t *testing.T) { receiver := &testDispatcher{} sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterResources) fixed := query.NewFixedIterator( query.Path{ @@ -269,7 +276,7 @@ func TestDispatchExecutor_Check_AliasWithSentinelInsideRecursiveDispatches(t *te } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) // AliasIterator wrapping a RecursiveIterator that contains a RecursiveSentinel. // The sentinel is managed by the RecursiveIterator's context, so dispatch is allowed. @@ -301,7 +308,7 @@ func TestDispatchExecutor_IterSubjects_AliasWithSentinelInsideRecursiveDispatche } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterSubjects) sentinel := query.NewRecursiveSentinelIterator("document", "viewer", false) recursive := query.NewRecursiveIterator(sentinel, "document", "viewer") @@ -333,7 +340,7 @@ func TestDispatchExecutor_IterResources_AliasWithSentinelInsideRecursiveDispatch } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterResources) sentinel := query.NewRecursiveSentinelIterator("document", "viewer", false) recursive := query.NewRecursiveIterator(sentinel, "document", "viewer") @@ -365,7 +372,7 @@ func TestDispatchExecutor_Check_DoubleRecursionBothMatchedDispatches(t *testing. } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) // Double recursion with both pairs matched: // RecursiveIterator(folder#viewer) contains RecursiveIterator(document#reader) @@ -400,7 +407,7 @@ func TestDispatchExecutor_StreamBatching(t *testing.T) { } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationIterResources) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) @@ -426,7 +433,7 @@ func TestDispatchExecutor_PlanContextForwarded(t *testing.T) { OptionalDatastoreLimit: 100, } sender := NewDispatchExecutor(receiver, pc, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) _, _ = sender.Check(ctx, NewDispatchIterator(alias), query.Object{ObjectType: "document", ObjectID: "doc1"}, query.ObjectAndRelation{ObjectType: "user", ObjectID: "alice", Relation: "..."}) @@ -447,6 +454,7 @@ func TestDispatchExecutor_ContextCancellation(t *testing.T) { sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) qctx := query.NewLocalContext(ctx) + qctx.TopLevelOperation = query.OperationCheck alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) _, err := sender.Check(qctx, NewDispatchIterator(alias), query.Object{ObjectType: "document", ObjectID: "doc1"}, query.ObjectAndRelation{ObjectType: "user", ObjectID: "alice", Relation: "..."}) @@ -573,7 +581,7 @@ func TestPaginationLimitFromPlanContext_Zero(t *testing.T) { func TestDispatchExecutor_CheckManyResources_NonAliasLocal(t *testing.T) { receiver := &testDispatcher{} sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) // FixedIterator (not an alias) — must delegate locally, no dispatch. fixed := query.NewFixedIterator(query.Path{ @@ -606,7 +614,7 @@ func TestDispatchExecutor_CheckManyResources_AliasDispatchesAndPairsByONR(t *tes }}, } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) resources := []query.Object{ @@ -640,7 +648,7 @@ func TestDispatchExecutor_CheckManySubjects_AliasDispatches(t *testing.T) { }}, } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, 100) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) subjects := []query.ObjectAndRelation{ @@ -726,7 +734,7 @@ func TestDispatchExecutor_CheckManyResources_ChunksByDispatchChunkSize(t *testin }, } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, chunkSize) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) @@ -773,7 +781,7 @@ func TestDispatchExecutor_CheckManySubjects_ChunksByDispatchChunkSize(t *testing }, } sender := NewDispatchExecutor(receiver, &v1.PlanContext{Revision: "rev1"}, chunkSize) - ctx := newTestContext() + ctx := newTestContext(query.OperationCheck) alias := query.NewAliasIterator("", "viewer", query.NewFixedIterator()) subjects := make([]query.ObjectAndRelation, total) diff --git a/internal/dispatch/graph/graph.go b/internal/dispatch/graph/graph.go index 6809fc6806..b694f381dc 100644 --- a/internal/dispatch/graph/graph.go +++ b/internal/dispatch/graph/graph.go @@ -632,6 +632,21 @@ func (ld *localDispatcher) DispatchQueryPlan( case v1.PlanOperation_PLAN_OPERATION_CHECK_MANY_RESOURCES: // Iterate CheckImpl directly: dispatch boundary already crossed for this // alias. qctx still uses DispatchExecutor, so nested aliases re-dispatch. + // + // shortCircuit: when the user-facing op is CHECK, the whole chain only + // needs a single yes/no — once any entry yields an uncaveated match the + // remaining entries can't change the outcome, so we stop the loop and + // the sender's CheckMany result map fills the unvisited slots with nil + // (which the BatchedArrow caller treats as "no match" and skips over). + // CAVEAT: this is *unsafe* for IntersectionArrow-rooted callers, whose + // AND semantics require knowing whether each entry actually matched — + // a short-circuit nil and a true nil are indistinguishable on the wire, + // so an early stop can produce false negatives for `relation.all(...)`. + // Today we accept that risk because Arrow is by far the common case; + // fixing IntersectionArrow correctness here will need a per-call hint + // (e.g. a new CHECK_ANY_* op, or a flag on the request) so the receiver + // only short-circuits when the caller opts in. + shortCircuit := req.PlanContext.GetTopLevelOperation() == v1.PlanOperation_PLAN_OPERATION_CHECK for _, res := range req.Many { resource := query.Object{ObjectType: res.Namespace, ObjectID: res.ObjectId} path, err := it.CheckImpl(qctx, resource, subject) @@ -644,11 +659,17 @@ func (ld *localDispatcher) DispatchQueryPlan( }); err != nil { return err } + if shortCircuit && path.Caveat == nil { + return nil + } } } return nil case v1.PlanOperation_PLAN_OPERATION_CHECK_MANY_SUBJECTS: + // See CHECK_MANY_RESOURCES above for the shortCircuit rationale and the + // IntersectionArrow caveat — the two cases are symmetric. + shortCircuit := req.PlanContext.GetTopLevelOperation() == v1.PlanOperation_PLAN_OPERATION_CHECK for _, sub := range req.Many { subject := query.ObjectAndRelation{ ObjectType: sub.Namespace, @@ -665,6 +686,9 @@ func (ld *localDispatcher) DispatchQueryPlan( }); err != nil { return err } + if shortCircuit && path.Caveat == nil { + return nil + } } } return nil diff --git a/internal/dispatch/keys/computed_test.go b/internal/dispatch/keys/computed_test.go index 7ed08a9236..f037b3ed2e 100644 --- a/internal/dispatch/keys/computed_test.go +++ b/internal/dispatch/keys/computed_test.go @@ -556,7 +556,7 @@ var generatorFuncs = map[string]generatorFunc{ return planCheckRequestToKey(&v1.DispatchQueryPlanRequest{ Resource: ONR(resourceRelation.Namespace, resourceIds[0], resourceRelation.Relation), Subject: ONR(subjectRelation.Namespace, subjectIds[0], subjectRelation.Relation), - Plan: []byte(resourceRelation.Relation), + Plan: []byte(resourceRelation.Relation), PlanContext: &v1.PlanContext{ Revision: metadata.AtRevision, }, @@ -580,7 +580,7 @@ var generatorFuncs = map[string]generatorFunc{ ) (DispatchCacheKey, []string) { return planLookupResourcesRequestToKey(&v1.DispatchQueryPlanRequest{ Subject: ONR(subjectRelation.Namespace, subjectIds[0], subjectRelation.Relation), - Plan: []byte(resourceRelation.Relation), + Plan: []byte(resourceRelation.Relation), PlanContext: &v1.PlanContext{ Revision: metadata.AtRevision, }, @@ -602,7 +602,7 @@ var generatorFuncs = map[string]generatorFunc{ ) (DispatchCacheKey, []string) { return planLookupSubjectsRequestToKey(&v1.DispatchQueryPlanRequest{ Resource: ONR(resourceRelation.Namespace, resourceIds[0], resourceRelation.Relation), - Plan: []byte(resourceRelation.Relation), + Plan: []byte(resourceRelation.Relation), PlanContext: &v1.PlanContext{ Revision: metadata.AtRevision, }, diff --git a/internal/dispatch/keys/hasher.go b/internal/dispatch/keys/hasher.go index 5eb7cae5ae..a28689cb76 100644 --- a/internal/dispatch/keys/hasher.go +++ b/internal/dispatch/keys/hasher.go @@ -12,6 +12,7 @@ import ( "github.com/authzed/spicedb/pkg/caveats" core "github.com/authzed/spicedb/pkg/proto/core/v1" v1 "github.com/authzed/spicedb/pkg/proto/dispatch/v1" + "github.com/authzed/spicedb/pkg/spiceerrors" "github.com/authzed/spicedb/pkg/tuple" ) @@ -197,7 +198,7 @@ func (h *dispatchCacheKeyHasher) Write(p []byte) { // xxhash.Digest.Write never returns an error; mirror mustWriteString. _, err := h.stableHasher.Write(p) if err != nil { - panic(fmt.Errorf("got an error from writing to the stable hasher: %w", err)) + spiceerrors.MustPanicf("got an error from writing to the stable hasher: %w", err) } } @@ -206,7 +207,7 @@ func (h *dispatchCacheKeyHasher) mustWriteString(value string) { // to be on the safe side. _, err := h.stableHasher.WriteString(value) if err != nil { - panic(fmt.Errorf("got an error from writing to the stable hasher: %w", err)) + spiceerrors.MustPanicf("got an error from writing to the stable hasher: %w", err) } } From 0933a2b2de213999a6b08c152b835e28667d985f Mon Sep 17 00:00:00 2001 From: Tanner Stirrat Date: Fri, 5 Jun 2026 09:22:44 -0600 Subject: [PATCH 4/4] chore: picking things up --- internal/dispatch/keys/hasher.go | 5 ++--- pkg/query/recursive.go | 8 +++++++- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/internal/dispatch/keys/hasher.go b/internal/dispatch/keys/hasher.go index a28689cb76..d7b673758b 100644 --- a/internal/dispatch/keys/hasher.go +++ b/internal/dispatch/keys/hasher.go @@ -2,7 +2,6 @@ package keys import ( "encoding/hex" - "fmt" "slices" "strconv" @@ -198,7 +197,7 @@ func (h *dispatchCacheKeyHasher) Write(p []byte) { // xxhash.Digest.Write never returns an error; mirror mustWriteString. _, err := h.stableHasher.Write(p) if err != nil { - spiceerrors.MustPanicf("got an error from writing to the stable hasher: %w", err) + spiceerrors.MustPanicf("got an error from writing to the stable hasher: %s", err) } } @@ -207,7 +206,7 @@ func (h *dispatchCacheKeyHasher) mustWriteString(value string) { // to be on the safe side. _, err := h.stableHasher.WriteString(value) if err != nil { - spiceerrors.MustPanicf("got an error from writing to the stable hasher: %w", err) + spiceerrors.MustPanicf("got an error from writing to the stable hasher: %s", err) } } diff --git a/pkg/query/recursive.go b/pkg/query/recursive.go index 0e898c723a..9c69140f2c 100644 --- a/pkg/query/recursive.go +++ b/pkg/query/recursive.go @@ -6,6 +6,8 @@ import ( "io" "time" + "github.com/ccoveille/go-safecast/v2" + "github.com/authzed/spicedb/internal/caveats" core "github.com/authzed/spicedb/pkg/proto/core/v1" "github.com/authzed/spicedb/pkg/tuple" @@ -759,7 +761,11 @@ func (r *RecursiveIterator) Serialize(w io.Writer) error { return err } if nonDefault { - if _, err := buf.Write([]byte{byte(r.checkStrategy)}); err != nil { + writeableCheckStrategy, err := safecast.Convert[byte](r.checkStrategy) + if err != nil { + return err + } + if _, err := buf.Write([]byte{writeableCheckStrategy}); err != nil { return err } }