fix(sync): handle dangling entitlements incrementally

pkg/sync/expand/drop_stats_test.go

@@ -78,4 +78,25 @@ func TestTopologicalSourceMissingRecordedOnDropStats(t *testing.T) {
 	defer stats.mu.Unlock()
 	require.Equal(t, int64(1), stats.sourceMissing, "the dropped source edge must land on the aggregate")
 	require.Contains(t, stats.seen, source.GetId())
+	require.Contains(t, graph.DanglingEntitlementIDs, source.GetId(),
+		"the topological entitlement load must record the missing source")
+}
+
+func TestTopologicalDestinationMissingRecordsDanglingID(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+
+	group := makeResource("group", "org")
+	source := makeEntitlement("ent:source", group)
+	dest := makeEntitlement("ent:dest", group)
+	store.AddEntitlement(source)
+
+	graph := NewEntitlementGraph(ctx)
+	graph.AddEntitlementID(source.GetId())
+	graph.AddEntitlementID(dest.GetId())
+	require.NoError(t, graph.AddEdge(ctx, source.GetId(), dest.GetId(), false, nil))
+
+	require.NoError(t, NewExpander(store, graph).RunTopologicalMergeStreaming(ctx))
+	require.Contains(t, graph.DanglingEntitlementIDs, dest.GetId(),
+		"the production topological path must record the missing destination")
 }

pkg/sync/expand/expander.go

@@ -433,6 +433,8 @@ func (e *Expander) runAction(ctx context.Context, action *EntitlementGraphAction
 			// the per-sync report. The batch shares one source, but every
 			// destination edge drops — count them all.
 			e.dropStats.RecordSourceMissingEdges(action.SourceEntitlementID, len(dests))
+			// The edges are deleted below, so a later seed could not reach them.
+			e.graph.NoteUnrecoverableDangling()
 			l.Debug("runAction: source entitlement not found, dropping batch edges",
 				zap.String("source_entitlement_id", action.SourceEntitlementID))
 			for _, d := range dests {
@@ -469,6 +471,8 @@ func (e *Expander) runAction(ctx context.Context, action *EntitlementGraphAction
 				// A single missing descendant drops only its own edge; the rest
 				// of the batch still expands. Debug + aggregate, as above.
 				e.dropStats.RecordDestinationMissing(d.EntitlementID)
+				// The edge is deleted below, so a later seed could not reach it.
+				e.graph.NoteUnrecoverableDangling()
 				l.Debug("runAction: descendant entitlement not found, dropping edge",
 					zap.String("descendant_entitlement_id", d.EntitlementID))
 				_ = e.graph.DeleteEdge(ctx, action.SourceEntitlementID, d.EntitlementID)

pkg/sync/expand/graph.go

@@ -91,6 +91,24 @@ type EntitlementGraph struct {
 	HasNoCycles           bool                      `json:"has_no_cycles"`
 	ExpansionPlan         *EntitlementGraphPlan     `json:"plan,omitempty"`
 	ExpansionMetrics      *EntitlementGraphMetrics  `json:"metrics,omitempty"`
+	// DanglingEntitlementIDs are edge endpoints this expansion skipped because
+	// they had no entitlement row in the store. If a later sync supplies one of
+	// those rows it adds no edge and no grant — the only two things incremental
+	// expansion seeds from — so nothing would seed the walk and the fast path
+	// would miss grants full expansion writes. Persisted so the next compaction
+	// can precheck these ids: any that now resolve seed expansion, while those
+	// still missing remain recorded without seeding a walk.
+	//
+	// Scoped deliberately. Most dangling references are permanent (connector
+	// magic-id bugs, disabled-by-default resource types — see DroppedEdgeStats),
+	// so a graph-wide "something was dangling" flag would re-arm on every full
+	// expansion and disable the fast path forever for those tenants.
+	DanglingEntitlementIDs map[string]struct{} `json:"dangling_entitlement_ids,omitempty"`
+	// DanglingOverflow means the dangling set hit its cap, or an evaluator
+	// dropped the edge outright so seeding cannot recover it. Either way the id
+	// list is no longer a complete description of what was skipped, so the next
+	// compaction must decline rather than seed.
+	DanglingOverflow bool `json:"dangling_overflow,omitempty"`
 }
 
 type EntitlementGraphMetrics struct {
@@ -112,6 +130,43 @@ func NewEntitlementGraph(_ context.Context) *EntitlementGraph {
 	}
 }
 
+// maxDanglingEntitlementIDs caps the persisted dangling set so a pathologically
+// broken connector cannot bloat the sidecar. Past the cap the graph records
+// overflow and the next compaction declines instead of seeding. A var so tests
+// can lower it.
+var maxDanglingEntitlementIDs = 4096
+
+// NoteDanglingReference records that an expansion skipped entitlementID because
+// it has no row in the store. Nil-safe: the legacy expander reaches it through
+// a graph that may not be installed yet.
+func (g *EntitlementGraph) NoteDanglingReference(entitlementID string) {
+	if g == nil || entitlementID == "" {
+		return
+	}
+	if g.DanglingEntitlementIDs == nil {
+		g.DanglingEntitlementIDs = make(map[string]struct{})
+	}
+	if _, ok := g.DanglingEntitlementIDs[entitlementID]; ok {
+		return
+	}
+	if len(g.DanglingEntitlementIDs) >= maxDanglingEntitlementIDs {
+		g.DanglingOverflow = true
+		return
+	}
+	g.DanglingEntitlementIDs[entitlementID] = struct{}{}
+}
+
+// NoteUnrecoverableDangling records a skipped endpoint whose edge was dropped
+// from the graph, so seeding the id on a later run could not reach it. Only the
+// source-batched evaluator deletes edges, and it never runs on Pebble, but the
+// graph must still refuse to be reused if it ever does.
+func (g *EntitlementGraph) NoteUnrecoverableDangling() {
+	if g == nil {
+		return
+	}
+	g.DanglingOverflow = true
+}
+
 // isNodeExpanded - is every outgoing edge from this node Expanded?
 func (g *EntitlementGraph) isNodeExpanded(nodeID int) bool {
 	for _, edgeID := range g.SourcesToDestinations[nodeID] {

pkg/sync/expand/graph_blob.go

@@ -19,7 +19,11 @@ type graphBlobEnvelope struct {
 	Graph         *EntitlementGraph               `json:"graph"`
 }
 
-const graphBlobFormatVersion uint32 = 2
+// Version 3 adds EntitlementGraph.DanglingEntitlementIDs and DanglingOverflow.
+// A v2 blob predates that bookkeeping, so what it skipped is unknowable and it
+// must not be trusted as an incremental base: the version check below rejects
+// it and the reader falls back to full expansion.
+const graphBlobFormatVersion uint32 = 3
 
 // MarshalGraphBlob serializes a legacy, unbound graph blob for compatibility
 // tests. Transient state is stripped first (a reload rebuilds it).

pkg/sync/expand/graph_blob_test.go

@@ -246,3 +246,61 @@ func BenchmarkMarshalGraphBlob(b *testing.B) {
 func entName(i int) string {
 	return fmt.Sprintf("group:g%06d:member", i)
 }
+
+// TestUnmarshalGraphBlob_RejectsPreDanglingVersion: a v2 blob predates
+// dangling bookkeeping, so what it skipped is unknowable and it
+// must not be reused as an incremental base. Rejection is silent — (nil, nil)
+// — so the reader falls back to full expansion rather than failing.
+func TestUnmarshalGraphBlob_RejectsPreDanglingVersion(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("ent-a")
+
+	v2Blob, err := json.Marshal(map[string]any{
+		"format_version": 2,
+		"sync_id":        "sync-1",
+		"graph":          g,
+	})
+	require.NoError(t, err)
+
+	got, digest, err := UnmarshalGraphBlobWithGrantDigest(v2Blob, "sync-1")
+	require.NoError(t, err)
+	require.Nil(t, got)
+	require.Nil(t, digest)
+}
+
+// TestGraphBlob_RoundTripsDanglingState: the recorded ids are what the next run
+// seeds from, so losing them in serialization would silently reintroduce the
+// divergence this whole mechanism exists to prevent.
+func TestGraphBlob_RoundTripsDanglingState(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("ent-a")
+	g.NoteDanglingReference("missing:one")
+	g.NoteDanglingReference("missing:two")
+
+	data, err := MarshalGraphBlob("sync-1", g)
+	require.NoError(t, err)
+	got, err := UnmarshalGraphBlob(data, "sync-1")
+	require.NoError(t, err)
+	require.NotNil(t, got)
+	require.Contains(t, got.DanglingEntitlementIDs, "missing:one")
+	require.Contains(t, got.DanglingEntitlementIDs, "missing:two")
+	require.False(t, got.DanglingOverflow)
+}
+
+// TestGraphBlob_RoundTripsDanglingOverflow: overflow is the flag that makes the
+// compactor decline outright, so it must survive too.
+func TestGraphBlob_RoundTripsDanglingOverflow(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("ent-a")
+	g.NoteUnrecoverableDangling()
+
+	data, err := MarshalGraphBlob("sync-1", g)
+	require.NoError(t, err)
+	got, err := UnmarshalGraphBlob(data, "sync-1")
+	require.NoError(t, err)
+	require.NotNil(t, got)
+	require.True(t, got.DanglingOverflow)
+}

pkg/sync/expand/incremental.go

@@ -45,6 +45,13 @@ func (g *EntitlementGraph) Clone() (*EntitlementGraph, error) {
 		Loaded:                g.Loaded,
 		Depth:                 g.Depth,
 		HasNoCycles:           g.HasNoCycles,
+		DanglingOverflow:      g.DanglingOverflow,
+	}
+	if g.DanglingEntitlementIDs != nil {
+		out.DanglingEntitlementIDs = make(map[string]struct{}, len(g.DanglingEntitlementIDs))
+		for id := range g.DanglingEntitlementIDs {
+			out.DanglingEntitlementIDs[id] = struct{}{}
+		}
 	}
 	for id, node := range g.Nodes {
 		node.EntitlementIDs = append([]string(nil), node.EntitlementIDs...)
@@ -110,6 +117,9 @@ func (g *EntitlementGraph) reinitMaps() {
 	if g.Edges == nil {
 		g.Edges = map[int]Edge{}
 	}
+	if g.DanglingEntitlementIDs == nil {
+		g.DanglingEntitlementIDs = map[string]struct{}{}
+	}
 }
 
 // ErrIncrementalFallback means a new edge closed a cycle; the caller should
@@ -127,6 +137,14 @@ var ErrIncrementalRevocationDecline = errors.New("incremental expansion: revocat
 // that normal full expansion is the safer bounded-cost path.
 var ErrIncrementalDenseChangeDecline = errors.New("incremental expansion: dense affected graph, fall back to full expansion")
 
+// ErrIncrementalDanglingReferenceDecline means the base graph can no longer
+// describe what it skipped — the dangling set overflowed its cap, or an
+// evaluator dropped the edges outright — so prechecking the recorded ids cannot
+// make the fast path agree with full expansion. Ordinary dangling references do
+// NOT reach this: still-missing ids remain recorded without seeding a walk, and
+// ids that now resolve seed their affected closure.
+var ErrIncrementalDanglingReferenceDecline = errors.New("incremental expansion: base graph dangling set is not seedable, fall back to full expansion")
+
 const (
 	incrementalDenseGraphMinNodes = 1000
 	incrementalMaxAffectedPercent = 10
@@ -184,8 +202,35 @@ func NewIncrementalExpander(store ExpanderStore, graph *EntitlementGraph) *Incre
 //
 // The walk reads current membership from the store, so changed members
 // (already merged in) propagate without being passed in. Returns
-// ErrIncrementalFallback if a new edge closes a cycle.
+// ErrIncrementalFallback if a new edge closes a cycle. This method mutates the
+// graph while applying changes and may do so before returning an error; callers
+// that intend to retry must discard or restore the graph first.
 func (ie *IncrementalExpander) ExpandChanges(ctx context.Context, newEdges []NewEdge, changedEntitlementIDs []string) (*IncrementalResult, error) {
+	if ie.graph.DanglingOverflow {
+		return nil, ErrIncrementalDanglingReferenceDecline
+	}
+	resolvedDanglingIDs, err := ie.precheckDanglingEntitlements(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if len(resolvedDanglingIDs) > 0 {
+		// Keep the exported method from appending into or sorting the caller's
+		// backing array while combining the internally discovered seeds.
+		changedEntitlementIDs = append([]string(nil), changedEntitlementIDs...)
+		seen := make(map[string]struct{}, len(changedEntitlementIDs)+len(resolvedDanglingIDs))
+		for _, id := range changedEntitlementIDs {
+			seen[id] = struct{}{}
+		}
+		for _, id := range resolvedDanglingIDs {
+			if _, ok := seen[id]; ok {
+				continue
+			}
+			seen[id] = struct{}{}
+			changedEntitlementIDs = append(changedEntitlementIDs, id)
+		}
+		sort.Strings(changedEntitlementIDs)
+	}
+
 	if len(newEdges) == 0 && len(changedEntitlementIDs) == 0 {
 		return &IncrementalResult{}, nil
 	}
@@ -258,6 +303,45 @@ func (ie *IncrementalExpander) ExpandChanges(ctx context.Context, newEdges []New
 	return result, nil
 }
 
+// precheckDanglingEntitlements separates recorded endpoints that are still
+// missing from those that now resolve. Still-missing ids stay persisted but do
+// not seed the walk, avoiding a useless recomputation of their entire forward
+// closure on every run. Resolved ids become normal changed-entitlement seeds.
+//
+// Build the replacement set locally and publish it only after every lookup
+// succeeds, so a transient store error does not partially mutate the graph.
+func (ie *IncrementalExpander) precheckDanglingEntitlements(ctx context.Context) ([]string, error) {
+	if len(ie.graph.DanglingEntitlementIDs) == 0 {
+		return nil, nil
+	}
+
+	ids := make([]string, 0, len(ie.graph.DanglingEntitlementIDs))
+	for id := range ie.graph.DanglingEntitlementIDs {
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+
+	stillMissing := make(map[string]struct{}, len(ids))
+	resolved := make([]string, 0, len(ids))
+	for _, id := range ids {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		entitlement, err := ie.getEntitlement(ctx, id)
+		if err != nil {
+			return nil, err
+		}
+		if entitlement == nil {
+			stillMissing[id] = struct{}{}
+			continue
+		}
+		resolved = append(resolved, id)
+	}
+
+	ie.graph.DanglingEntitlementIDs = stillMissing
+	return resolved, nil
+}
+
 func topologicalAffectedNodeOrder(g *EntitlementGraph, affected map[int]struct{}) ([]int, error) {
 	inDegree := make(map[int]int, len(affected))
 	for nodeID := range affected {
@@ -376,7 +460,9 @@ func (ie *IncrementalExpander) recomputeDestination(ctx context.Context, nodeID
 	}
 	if destEnt == nil {
 		// Dangling ref: skip-with-warn, matching the full evaluator (don't
-		// error into a fallback).
+		// error into a fallback). Record it on the persisted graph so a later
+		// incremental run can precheck it and seed the walk if it resolves.
+		ie.graph.NoteDanglingReference(destEntitlementID)
 		ctxzap.Extract(ctx).Warn("incremental expansion: destination entitlement not in store; skipping",
 			zap.String("entitlement_id", destEntitlementID))
 		return 0, nil
@@ -406,6 +492,7 @@ func (ie *IncrementalExpander) recomputeDestination(ctx context.Context, nodeID
 				return 0, err
 			}
 			if sourceEnt == nil {
+				ie.graph.NoteDanglingReference(sourceEntitlementID)
 				ctxzap.Extract(ctx).Warn("incremental expansion: source entitlement not in store; skipping",
 					zap.String("entitlement_id", sourceEntitlementID))
 				continue