pebble: move expansion merge and rollback methods onto Expansion

pkg/dotc1z/engine/pebble/adapter_reader.go

@@ -447,7 +447,7 @@ func (e *Engine) ListSyncs(ctx context.Context, req *reader_v2.SyncsReaderServic
 	if err != nil {
 		return nil, err
 	}
-	iter, err := e.NewIter(&pebble.IterOptions{
+	iter, err := e.expansion.NewIter(&pebble.IterOptions{
 		LowerBound: lower,
 		UpperBound: upper,
 	})

pkg/dotc1z/engine/pebble/bulk_import_dedupe_test.go

@@ -36,7 +36,7 @@ func TestBulkImportAbortAfterEngineClose(t *testing.T) {
 	store, b := startBulkImport(t, ctx)
 	require.NoError(t, b.AddResourceTypes(ctx, v2.ResourceType_builder{Id: "app"}.Build()))
 
-	require.NoError(t, store.PebbleEngine().Close())
+	require.NoError(t, store.Expansion().PebbleEngine().Close())
 	require.NotPanics(t, b.Abort, "Abort after engine Close must tear down staging dirs, not panic")
 
 	// And the other retained-handle shape: Finish after Abort/Close is

pkg/dotc1z/engine/pebble/cleanup_test.go

@@ -66,14 +66,14 @@ func TestResetForNewSyncReclaimsDiskImmediately(t *testing.T) {
 
 	dataLo := []byte{versionV3, typeResourceType}
 	dataHi := []byte{versionV3, typeEngineMeta}
-	before, err := e.EstimateDiskUsage(dataLo, dataHi)
+	before, err := e.Expansion().EstimateDiskUsage(dataLo, dataHi)
 	require.NoErrorf(t, err, "EstimateDiskUsage (before)")
 	require.NotZero(t, before, "sanity: expected non-zero on-disk usage for the finished sync's data span")
 
 	// Replacement sync: StartNewSync excises the prior sync's data.
 	_, err = a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoErrorf(t, err, "StartNewSync (replacement)")
-	after, err := e.EstimateDiskUsage(dataLo, dataHi)
+	after, err := e.Expansion().EstimateDiskUsage(dataLo, dataHi)
 	require.NoErrorf(t, err, "EstimateDiskUsage (after)")
 	// The excise drops fully-covered SSTs from the manifest, so the
 	// data span's estimated usage collapses to (near) zero immediately

pkg/dotc1z/engine/pebble/endsync_concurrent_writers_test.go

@@ -40,7 +40,7 @@ func TestEndSyncConcurrentWritersNeverMissByPrincipal(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 
 	const workers = 8
 	type accepted struct {

pkg/dotc1z/engine/pebble/engine.go

@@ -142,9 +142,10 @@ type Engine struct {
 	entIDLookup         map[string][]entitlementIdentity
 	entIDLookupBuiltGen uint64
 
-	ledger   Ledger
-	syncRuns SyncRuns
-	digests  Digests
+	ledger    Ledger
+	syncRuns  SyncRuns
+	digests   Digests
+	expansion Expansion
 
 	// migratedOnOpen reports that this Open ran the in-place id-index
 	// migration. The store layer uses it to mark a writable store dirty so
@@ -206,6 +207,7 @@ func Open(ctx context.Context, dir string, opts ...Option) (*Engine, error) {
 	e.ledger.e = e
 	e.syncRuns.e = e
 	e.digests.e = e
+	e.expansion.e = e
 	if s, ok := pebbleOpts.Experimental.CompactionScheduler.(*pausableCompactionScheduler); ok {
 		e.compactionScheduler = s
 	}

pkg/dotc1z/engine/pebble/expand_write_path_stats.go

@@ -1,17 +1,23 @@
 package pebble
 
+type Expansion struct {
+	e *Engine
+}
+
+func (e *Engine) Expansion() *Expansion { return &e.expansion }
+
 type ExpandWritePathStats struct {
 	ExpandedCalls    int64
 	ExpandedRows     int64
 	SynthesizedCalls int64
 	SynthesizedRows  int64
 }
 
-func (e *Engine) ExpandWritePathStats() ExpandWritePathStats {
+func (x *Expansion) ExpandWritePathStats() ExpandWritePathStats {
 	return ExpandWritePathStats{
-		ExpandedCalls:    e.expandedWriteCalls.Load(),
-		ExpandedRows:     e.expandedWriteRows.Load(),
-		SynthesizedCalls: e.synthesizedWriteCalls.Load(),
-		SynthesizedRows:  e.synthesizedWriteRows.Load(),
+		ExpandedCalls:    x.e.expandedWriteCalls.Load(),
+		ExpandedRows:     x.e.expandedWriteRows.Load(),
+		SynthesizedCalls: x.e.synthesizedWriteCalls.Load(),
+		SynthesizedRows:  x.e.synthesizedWriteRows.Load(),
 	}
 }

pkg/dotc1z/engine/pebble/fast_path_proof_verification_test.go

@@ -94,7 +94,7 @@ func TestVerificationBindCurrentSyncDisarmsAllFastPathProofs(t *testing.T) {
 	a := newAdapter(t)
 	syncID, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 	for _, kind := range []sourcecache.RowKind{
 		sourcecache.RowKindResources,
 		sourcecache.RowKindEntitlements,
@@ -118,7 +118,7 @@ func TestVerificationBulkImportFinishDisarmsAllFastPathProofs(t *testing.T) {
 	a := newAdapter(t)
 	syncID, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 	bulk, err := e.StartBulkSyncImport(ctx, syncID, t.TempDir(), 1)
 	require.NoError(t, err)
 	defer bulk.Abort()
@@ -154,20 +154,20 @@ func TestVerificationReplayClearCommittedPrefixDisarmsFastPath(t *testing.T) {
 			prev := newAdapter(t)
 			_, err := prev.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 			require.NoError(t, err)
-			putFastPathProofRow(t, prev.PebbleEngine(), kind, "scope-a", "source", 0)
-			sealReplaySource(ctx, t, prev.PebbleEngine(), kind, "scope-a")
+			putFastPathProofRow(t, prev.Expansion().PebbleEngine(), kind, "scope-a", "source", 0)
+			sealReplaySource(ctx, t, prev.Expansion().PebbleEngine(), kind, "scope-a")
 
 			current := newAdapter(t)
 			_, err = current.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 			require.NoError(t, err)
 			for row := range 3 {
-				putFastPathProofRow(t, current.PebbleEngine(), kind, "scope-a", "destination", row)
+				putFastPathProofRow(t, current.Expansion().PebbleEngine(), kind, "scope-a", "destination", row)
 			}
-			armFastPathProof(current.PebbleEngine(), kind)
-			require.True(t, fastPathProofArmed(current.PebbleEngine(), kind), "fault premise was not armed")
-			current.PebbleEngine().test.sourceCacheReplayBatchRows = 2
+			armFastPathProof(current.Expansion().PebbleEngine(), kind)
+			require.True(t, fastPathProofArmed(current.Expansion().PebbleEngine(), kind), "fault premise was not armed")
+			current.Expansion().PebbleEngine().test.sourceCacheReplayBatchRows = 2
 			commitCalls := 0
-			current.PebbleEngine().test.sourceCacheReplayClearCommitHook = func(
+			current.Expansion().PebbleEngine().test.sourceCacheReplayClearCommitHook = func(
 				gotKind string,
 				_ int,
 				_ bool,
@@ -183,31 +183,31 @@ func TestVerificationReplayClearCommittedPrefixDisarmsFastPath(t *testing.T) {
 			var replayErr error
 			switch kind {
 			case sourcecache.RowKindResources:
-				_, replayErr = current.PebbleEngine().ReplaySourceCacheResources(
+				_, replayErr = current.Expansion().PebbleEngine().ReplaySourceCacheResources(
 					ctx,
-					prev.PebbleEngine(),
+					prev.Expansion().PebbleEngine(),
 					"scope-a",
 				)
 			case sourcecache.RowKindEntitlements:
-				_, replayErr = current.PebbleEngine().ReplaySourceCacheEntitlements(
+				_, replayErr = current.Expansion().PebbleEngine().ReplaySourceCacheEntitlements(
 					ctx,
-					prev.PebbleEngine(),
+					prev.Expansion().PebbleEngine(),
 					"scope-a",
 				)
 			case sourcecache.RowKindGrants:
-				_, replayErr = current.PebbleEngine().ReplaySourceCacheGrants(
+				_, replayErr = current.Expansion().PebbleEngine().ReplaySourceCacheGrants(
 					ctx,
-					prev.PebbleEngine(),
+					prev.Expansion().PebbleEngine(),
 					"scope-a",
 				)
 			}
 			require.ErrorIs(t, replayErr, injected)
 			require.Equal(t, 2, commitCalls, "fault did not occur after one committed clear batch")
-			armedAfterFailure := fastPathProofArmed(current.PebbleEngine(), kind)
-			current.PebbleEngine().test.sourceCacheReplayClearCommitHook = nil
+			armedAfterFailure := fastPathProofArmed(current.Expansion().PebbleEngine(), kind)
+			current.Expansion().PebbleEngine().test.sourceCacheReplayClearCommitHook = nil
 
-			putFastPathProofRow(t, current.PebbleEngine(), kind, "scope-b", "destination", 2)
-			require.NoError(t, auditSourceScopeBiconditional(current.PebbleEngine()),
+			putFastPathProofRow(t, current.Expansion().PebbleEngine(), kind, "scope-b", "destination", 2)
+			require.NoError(t, auditSourceScopeBiconditional(current.Expansion().PebbleEngine()),
 				"colliding write used stale empty-keyspace proof after committed clear")
 			require.False(t, armedAfterFailure, "committed clear did not consume %s fast-path proof", kind)
 		})
@@ -219,15 +219,15 @@ func TestVerificationReplayWriteBarrierDrainsBeforeClose(t *testing.T) {
 	prev := newAdapter(t)
 	_, err := prev.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	putFastPathProofRow(t, prev.PebbleEngine(), sourcecache.RowKindResources, "scope-a", "source", 0)
-	sealReplaySource(ctx, t, prev.PebbleEngine(), sourcecache.RowKindResources, "scope-a")
+	putFastPathProofRow(t, prev.Expansion().PebbleEngine(), sourcecache.RowKindResources, "scope-a", "source", 0)
+	sealReplaySource(ctx, t, prev.Expansion().PebbleEngine(), sourcecache.RowKindResources, "scope-a")
 
 	current := newAdapter(t)
 	_, err = current.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
 	enteredCommit := make(chan struct{})
 	releaseCommit := make(chan struct{})
-	current.PebbleEngine().test.sourceCacheReplayCommitHook = func(kind string, rows int, final bool) error {
+	current.Expansion().PebbleEngine().test.sourceCacheReplayCommitHook = func(kind string, rows int, final bool) error {
 		if kind != string(sourcecache.RowKindResources) || rows != 1 || !final {
 			return fmt.Errorf("unexpected replay commit kind=%q rows=%d final=%t", kind, rows, final)
 		}
@@ -238,9 +238,9 @@ func TestVerificationReplayWriteBarrierDrainsBeforeClose(t *testing.T) {
 
 	replayDone := make(chan error, 1)
 	go func() {
-		_, replayErr := current.PebbleEngine().ReplaySourceCacheResources(
+		_, replayErr := current.Expansion().PebbleEngine().ReplaySourceCacheResources(
 			context.Background(),
-			prev.PebbleEngine(),
+			prev.Expansion().PebbleEngine(),
 			"scope-a",
 		)
 		replayDone <- replayErr
@@ -253,7 +253,7 @@ func TestVerificationReplayWriteBarrierDrainsBeforeClose(t *testing.T) {
 
 	closeDone := make(chan error, 1)
 	go func() {
-		closeDone <- current.PebbleEngine().Close()
+		closeDone <- current.Expansion().PebbleEngine().Close()
 	}()
 	select {
 	case closeErr := <-closeDone:

pkg/dotc1z/engine/pebble/ingest_facts_security_test.go

@@ -49,7 +49,7 @@ func TestSecurity_IngestScansRejectNonCanonicalKeys(t *testing.T) {
 				a := newAdapter(t)
 				_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 				require.NoError(t, err)
-				e := a.PebbleEngine()
+				e := a.Expansion().PebbleEngine()
 				require.NoError(t, e.db.UnsafeForTesting().Set(tc.key, nil, nil))
 
 				err = scans[scan](ctx, e)
@@ -68,7 +68,7 @@ func TestIngestScansVisitEachGroupOnce(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 
 	resource := v2.Resource_builder{Id: v2.ResourceId_builder{ResourceType: "app", Resource: "github"}.Build()}.Build()
 	grant := func(entID, principalID string) *v2.Grant {

pkg/dotc1z/engine/pebble/ingest_repair_orphan_index_test.go

@@ -48,7 +48,7 @@ func TestDanglingPrincipalSweepHealsAllOrphanIndexEntries(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 
 	// "ghost" has index entries but no grant rows and no resource row:
 	// the all-orphan shape. Without the heal this would visit as
@@ -87,7 +87,7 @@ func TestEnsureGrantIndexesClearsDeferredMarker(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 
 	// The deferred regime arms the marker.
 	require.NoError(t, a.Grants().StoreExpandedGrants(ctx, mkV2Grant("", "ent-x", "user", "alice")))
@@ -119,7 +119,7 @@ func TestHealOrphanPrincipalIndexEntriesKeepsLiveRows(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 
 	require.NoError(t, a.PutGrants(ctx, mkV2Grant("", "ent-live", "user", "victim")))
 	plantOrphanPrincipalIndexEntry(t, e, "victim", "ghost-ent")
@@ -152,7 +152,7 @@ func TestIngestScansFailLoudOnMalformedKeys(t *testing.T) {
 		a := newAdapter(t)
 		_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 		require.NoError(t, err)
-		e := a.PebbleEngine()
+		e := a.Expansion().PebbleEngine()
 		// A grant-family key whose tail has too few tuple components.
 		bad := append(encodeGrantPrefix(), 0x00, 'x')
 		require.NoError(t, e.db.UnsafeForTesting().Set(bad, nil, nil))
@@ -169,7 +169,7 @@ func TestIngestScansFailLoudOnMalformedKeys(t *testing.T) {
 		a := newAdapter(t)
 		_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 		require.NoError(t, err)
-		e := a.PebbleEngine()
+		e := a.Expansion().PebbleEngine()
 		bad := append(encodeEntitlementPrefix(), 0x00, 'x')
 		require.NoError(t, e.db.UnsafeForTesting().Set(bad, nil, nil))
 
@@ -182,7 +182,7 @@ func TestIngestScansFailLoudOnMalformedKeys(t *testing.T) {
 		a := newAdapter(t)
 		_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 		require.NoError(t, err)
-		e := a.PebbleEngine()
+		e := a.Expansion().PebbleEngine()
 		bad := []byte{versionV3, typeIndex, idxGrantByPrincipal, 0x00, 'x'}
 		require.NoError(t, e.db.UnsafeForTesting().Set(bad, nil, nil))
 
@@ -201,7 +201,7 @@ func TestIngestScanSurfaceAfterCloseReturnsClosing(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 	require.NoError(t, e.Close())
 
 	visit2 := func(string, string) error { return nil }
@@ -227,7 +227,7 @@ func TestDanglingPrincipalSweepKeepsMixedPopulations(t *testing.T) {
 	a := newAdapter(t)
 	_, err := a.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
 	require.NoError(t, err)
-	e := a.PebbleEngine()
+	e := a.Expansion().PebbleEngine()
 
 	// A real grant to a principal with no resource row (a genuine
 	// dangling reference) PLUS a planted orphan entry under the same

pkg/dotc1z/engine/pebble/merge_accessor.go

@@ -31,11 +31,11 @@ type engineAccessor interface {
 
 // PebbleEngine returns the engine. Nil-safe so AsEngine can probe
 // arbitrary writers.
-func (e *Engine) PebbleEngine() *Engine {
-	if e == nil {
+func (x *Expansion) PebbleEngine() *Engine {
+	if x == nil {
 		return nil
 	}
-	return e
+	return x.e
 }
 
 // AsEngine recovers the underlying *Engine from a connectorstore.Writer
@@ -57,11 +57,11 @@ func AsEngine(w connectorstore.Writer) (*Engine, bool) {
 // counters, labeled with the caller's phase name. Diagnostic only: used to
 // attribute wall time lost to background compactions triggered by large
 // ingests (e.g. the synthesized-grant layer segments) during benchmark runs.
-func (e *Engine) LogCompactionMetrics(ctx context.Context, phase string) {
-	if e == nil || e.db == nil {
+func (x *Expansion) LogCompactionMetrics(ctx context.Context, phase string) {
+	if x == nil || x.e.db == nil {
 		return
 	}
-	m := e.db.Metrics()
+	m := x.e.db.Metrics()
 	ctxzap.Extract(ctx).Info("pebble compaction metrics",
 		zap.String("phase", phase),
 		zap.Int64("compact_count", m.Compact.Count),