sync: require the page ledger and delete the token-checkpoint sync path

pkg/dotc1z/engine_registry.go

@@ -103,7 +103,8 @@ var defaultEngineRegistry = &engineRegistry{
 }
 
 // RegisterEngine registers a storage engine driver with the process-global
-// dotc1z engine registry.
+// dotc1z engine registry. sync.(*syncer).setStore requires Pebble and
+// PageLedgerStore.
 func RegisterEngine(driver EngineDriver) error {
 	return defaultEngineRegistry.register(driver)
 }

pkg/sync/chaos_cancellation_test.go

@@ -196,7 +196,6 @@ func TestChaosExternalCancelStopsQuietlyAndCheckpoints(t *testing.T) {
 			accountingAfterCancel.Store(true)
 		}
 	}
-	sc.testHooks.checkpointHook = func(string) { t.Error("Pebble wrote a checkpoint token") }
 
 	done := make(chan error, 1)
 	go func() { done <- harness.Syncer.Sync(firstCtx) }()
@@ -234,7 +233,7 @@ func TestChaosExternalCancelStopsQuietlyAndCheckpoints(t *testing.T) {
 	require.Zero(t, run.Runtime().ActiveOperations(),
 		"external cancel must release the blocked connector call")
 
-	require.Nil(t, findEntry(capturedEntries(), zapcore.ErrorLevel, "cancelling context due to error in action"),
+	require.Nil(t, findEntry(capturedEntries(), zapcore.WarnLevel, "cancelling context due to error in action"),
 		"workers observing shutdown must not log action-failure errors (RFC 0009 §4.2)")
 	require.True(t, accountingAfterCancel.Load(),
 		"run accounting must persist with a detached context after external cancellation")
@@ -310,10 +309,10 @@ func TestParallelBatchSecondGenuineFailureStillLogs(t *testing.T) {
 	core, capturedEntries := newCaptureCore()
 	ctx = ctxzap.ToContext(ctx, zap.New(core))
 
-	st := newEmptySchedulerState(t)
-	first := st.pushAction(ctx, Action{Op: SyncGrantsOp, ResourceID: "group-1"})
-	second := st.pushAction(ctx, Action{Op: SyncGrantsOp, ResourceID: "group-2"})
-	s := &syncer{run: st, stats: newRunStats(), graph: newExpansionGraph(), cfg: syncConfig{workerCount: 2}}
+	s, _ := newLedgerSchedulerFixture(t, 2)
+	s.run.pushAction(ctx, Action{Op: SyncGrantsOp, ResourceID: "group-1"})
+	s.run.pushAction(ctx, Action{Op: SyncGrantsOp, ResourceID: "group-2"})
+	seedLedgerTestRun(t, s, nil)
 
 	// Bounded rendezvous, not a WaitGroup: if the pool ever stops dispatching
 	// these two actions concurrently, the premise is broken and this must fail
@@ -337,7 +336,10 @@ func TestParallelBatchSecondGenuineFailureStillLogs(t *testing.T) {
 		return fmt.Errorf("injected genuine failure for %s", action.ResourceID)
 	}
 
-	_, err := s.syncParallel(ctx, newTestRetryer(ctx), []*Action{first, second}, f)
+	_, err := s.syncParallel(ctx, newTestRetryer(ctx), s.run.peekMatchingActions(ctx, SyncGrantsOp), f)
+	mu.Lock()
+	require.Equal(t, 2, arrived)
+	mu.Unlock()
 	require.Error(t, err)
 	require.False(t, premiseFailed.Load(),
 		"premise: both actions must be in flight before either fails, so the second failure is a genuine discovery")
@@ -367,21 +369,24 @@ func TestParallelBatchFailureDuringStopStaysQuiet(t *testing.T) {
 	preBatchCtx, stop := context.WithCancelCause(ctx)
 	defer stop(nil)
 
-	st := newEmptySchedulerState(t)
-	action := st.pushAction(ctx, Action{Op: SyncGrantsOp, ResourceID: "group-1"})
-	s := &syncer{run: st, stats: newRunStats(), graph: newExpansionGraph(), cfg: syncConfig{workerCount: 1}}
+	s, _ := newLedgerSchedulerFixture(t, 1)
+	s.run.pushAction(ctx, Action{Op: SyncGrantsOp, ResourceID: "group-1"})
+	seedLedgerTestRun(t, s, nil)
 
+	var ran atomic.Bool
 	f := func(ctx context.Context, action *Action) error {
+		ran.Store(true)
 		stop(context.DeadlineExceeded)
 		return fmt.Errorf("genuine failure racing the run-duration deadline")
 	}
 
-	_, err := s.syncParallel(preBatchCtx, newTestRetryer(ctx), []*Action{action}, f)
+	_, err := s.syncParallel(preBatchCtx, newTestRetryer(ctx), s.run.peekMatchingActions(ctx, SyncGrantsOp), f)
+	require.True(t, ran.Load())
 	require.Error(t, err, "the failure must still reach the caller")
 
 	for _, entry := range capturedEntries() {
 		require.False(t,
-			entry.level == zapcore.ErrorLevel && contains(entry.message, "cancelling context due to error in action"),
+			entry.level == zapcore.WarnLevel && contains(entry.message, "cancelling context due to error in action"),
 			"a failure observed while the run is stopping must not log as a discovery")
 	}
 }

pkg/sync/chaos_lifecycle_test.go

@@ -59,9 +59,6 @@ func runDataPolicyLifecycleCase(t *testing.T, transport chaosTransport, corpusCa
 	)
 	initialConcrete, ok := initialHarness.Syncer.(*syncer)
 	require.True(t, ok)
-	// Persist every completed page so the drop case proves counters restore
-	// from the checkpoint instead of merely being reconstructed by replay.
-	initialConcrete.checkpointInterval = 0
 	require.ErrorIs(t, initialHarness.Syncer.Sync(ctx), chaosconnector.ErrInterruptRequested)
 	require.NoError(t, initialHarness.Close(t.Context()))
 	require.NoError(t, initialRun.Runtime().VerifyRequired())

pkg/sync/checkpoint_cut_test.go

@@ -343,7 +343,6 @@ func runCutAttempt(
 		return nil
 	})
 	var commits atomic.Int64
-	sc.testHooks.checkpointHook = func(string) { t.Error("Pebble wrote a checkpoint token") }
 	sc.testHooks.ledgerCommitted = func(c1zstore.LedgerRow) {
 		n := commits.Add(1)
 		if spec.commit > 0 && n == int64(spec.commit) {

pkg/sync/external_principal_batch_delete_test.go

@@ -9,7 +9,6 @@ import (
 
 	v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2"
 	"github.com/conductorone/baton-sdk/pkg/annotations"
-	"github.com/conductorone/baton-sdk/pkg/connectorstore"
 	"github.com/conductorone/baton-sdk/pkg/dotc1z/c1zstore"
 )
 
@@ -28,14 +27,8 @@ type batchDeleteRouteStore struct {
 	putGrantIDs    []string
 }
 
-func (s *batchDeleteRouteStore) Metadata() connectorstore.StoreMetadata {
-	return connectorstore.StoreMetadata{}
-}
-
 func (s *batchDeleteRouteStore) Grants() c1zstore.GrantStore { return &s.grants }
 
-// SyncMeta answers the capability resolution setStore performs at attach; see
-// legacyPaginatedCheckpointStore.SyncMeta.
 func (s *batchDeleteRouteStore) SyncMeta() c1zstore.SyncMeta { return nil }
 
 func (s *batchDeleteRouteStore) PutGrants(_ context.Context, grants ...*v2.Grant) error {
@@ -113,7 +106,7 @@ func newExternalMatchSyncer(store c1zstore.Store) *syncer {
 	run := newRunState()
 	run.setFact(factHasExternalResourceGrants)
 	s := &syncer{run: run, stats: newRunStats(), graph: newExpansionGraph()}
-	s.setStore(store)
+	s.setStore(withTripwireLedgerCaps(store))
 	return s
 }
 

pkg/sync/external_principal_cleanup_scale_test.go

@@ -10,7 +10,6 @@ import (
 
 	v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2"
 	"github.com/conductorone/baton-sdk/pkg/annotations"
-	"github.com/conductorone/baton-sdk/pkg/connectorstore"
 	"github.com/conductorone/baton-sdk/pkg/dotc1z/c1zstore"
 )
 
@@ -44,12 +43,6 @@ type cleanupScaleStore struct {
 	grantDeletes       int
 }
 
-// SyncMeta answers the capability resolution setStore performs at attach; see
-// legacyPaginatedCheckpointStore.SyncMeta.
-func (s *cleanupScaleStore) Metadata() connectorstore.StoreMetadata {
-	return connectorstore.StoreMetadata{}
-}
-
 func (s *cleanupScaleStore) SyncMeta() c1zstore.SyncMeta { return nil }
 
 func (s *cleanupScaleStore) ListResources(
@@ -129,7 +122,7 @@ func TestChaosConnectorExternalPrincipalCleanupUsesOnePassPerKeyspace(t *testing
 			}
 
 			syncer := &syncer{}
-			syncer.setStore(store)
+			syncer.setStore(withTripwireLedgerCaps(store))
 			require.NoError(t, syncer.deleteStaleExternalPrincipals(t.Context(), nil))
 			require.Equal(t, 1, store.resourceScans)
 			require.Equal(t, 1, store.entitlementScans)

pkg/sync/external_principal_index_verification_test.go

@@ -402,7 +402,7 @@ func TestVerificationExternalPrincipalMatchDeleteCutResumesToGolden(t *testing.T
 	require.NoError(t, err)
 	cutState, _, _ := decodeTestRun(t, currentToken)
 	cutSyncer := &syncer{run: cutState, stats: newRunStats(), graph: newExpansionGraph()}
-	cutSyncer.setStore(cutWrapper)
+	cutSyncer.setStore(withLedgerCaps(t, cutWrapper, cutStore))
 	err = cutSyncer.processGrantsWithExternalPrincipals(ctx, principals)
 	require.ErrorIs(t, err, errVerificationDeleteCut)
 	require.Equal(t, []int{9}, cutWrapper.putBatchSizes, "test premise: cut happened after the bulk put")
@@ -421,7 +421,7 @@ func TestVerificationExternalPrincipalMatchDeleteCutResumesToGolden(t *testing.T
 
 	resumedWrapper := &interruptingExternalMatchStore{Store: resumedStore}
 	resumedSyncer := &syncer{run: resumedState, stats: newRunStats(), graph: newExpansionGraph()}
-	resumedSyncer.setStore(resumedWrapper)
+	resumedSyncer.setStore(withLedgerCaps(t, resumedWrapper, resumedStore))
 	require.NoError(t, resumedSyncer.processGrantsWithExternalPrincipals(ctx, principals))
 	require.Equal(t, []int{33}, resumedWrapper.putBatchSizes,
 		"resume scans two remaining carriers plus nine expanded grants, each matching three principals")
@@ -468,7 +468,7 @@ func TestVerificationExternalPrincipalMatchPutCutResumesToGolden(t *testing.T) {
 	cutState, _, _ := decodeTestRun(t, currentToken)
 	cutSyncer := &syncer{run: cutState, stats: newRunStats(), graph: newExpansionGraph()}
 	cutSyncer.testHooks.externalMatchGrantPutChunk = &putChunk
-	cutSyncer.setStore(cutWrapper)
+	cutSyncer.setStore(withLedgerCaps(t, cutWrapper, cutStore))
 	err = cutSyncer.processGrantsWithExternalPrincipals(ctx, principals)
 	require.ErrorIs(t, err, errVerificationPutCut)
 	require.Equal(t, []int{putChunk, putChunk}, cutWrapper.putBatchSizes,
@@ -492,7 +492,7 @@ func TestVerificationExternalPrincipalMatchPutCutResumesToGolden(t *testing.T) {
 
 	resumedWrapper := &interruptingExternalMatchStore{Store: resumedStore}
 	resumedSyncer := &syncer{run: resumedState, stats: newRunStats(), graph: newExpansionGraph()}
-	resumedSyncer.setStore(resumedWrapper)
+	resumedSyncer.setStore(withLedgerCaps(t, resumedWrapper, resumedStore))
 	require.NoError(t, resumedSyncer.processGrantsWithExternalPrincipals(ctx, principals))
 	require.Equal(t, []int{21}, resumedWrapper.putBatchSizes,
 		"resume scans three carriers plus four expanded grants, each matching three principals")

pkg/sync/hooks.go

@@ -29,10 +29,6 @@ type syncTestHooks struct {
 	// halt sweep uses it to prove crash/resume equivalence at every
 	// ordering-sensitive point.
 	ingestHaltHook func(stage string) error
-	// checkpointHook, when non-nil, observes every durably written
-	// checkpoint token. The cut harness uses it to count checkpoints
-	// and to simulate a crash immediately after a chosen one.
-	checkpointHook func(token string)
 	// queueAudit, when non-nil, records every parallelActionQueue
 	// event (seed/dequeue/commit/abort/done) for post-hoc verification
 	// of the queue contract.

pkg/sync/ingest_filter.go

@@ -348,14 +348,6 @@ func (s *syncer) filterFreshGrantsWithStats(ctx context.Context, grants []*v2.Gr
 	return out, nil
 }
 
-// filterFreshGrantResource reports whether a grant-discovered resource
-// (InsertResourceGrants) references a scheduled resource type. Uplift skips
-// resources whose type is absent from the sync's resource types, so storing
-// them is dead data even though the resource row itself could be written.
-func (s *syncer) filterFreshGrantResource(ctx context.Context, resource *v2.Resource) (bool, error) {
-	return s.filterFreshGrantResourceWithStats(ctx, resource, &s.ingestFilterStats)
-}
-
 func (s *syncer) filterFreshGrantResourceWithStats(ctx context.Context, resource *v2.Resource, stats *ingestFilterStats) (bool, error) {
 	if s.cfg.syncType != connectorstore.SyncTypeFull || resource.GetId() == nil {
 		return true, nil

pkg/sync/ingest_filter_test.go

@@ -645,7 +645,7 @@ func TestFreshIngestFilterProbeFailureFailsSyncNotDrops(t *testing.T) {
 	t.Run("local store failure", func(t *testing.T) {
 		failing := &resourceTypeProbeStore{Store: store, err: status.Error(codes.Internal, "probe boom")}
 		s := &syncer{cfg: syncConfig{syncType: connectorstore.SyncTypeFull}}
-		s.setStore(failing)
+		s.setStore(withLedgerCaps(t, failing, store))
 
 		_, err := s.filterFreshEntitlements(ctx, []*v2.Entitlement{entitlement})
 		require.ErrorContains(t, err, "probe boom")
@@ -679,7 +679,7 @@ func TestFreshIngestFilterCachesProbeVerdicts(t *testing.T) {
 
 	counting := &resourceTypeProbeStore{Store: store, calls: map[string]int{}}
 	s := &syncer{cfg: syncConfig{syncType: connectorstore.SyncTypeFull}}
-	s.setStore(counting)
+	s.setStore(withLedgerCaps(t, counting, store))
 
 	group := ingestFilterResource("group", "g1")
 	disabled := ingestFilterResource("iam_policy", "p1")

pkg/sync/ingest_invariants.go

@@ -631,10 +631,8 @@ func (s *syncer) runIngestionInvariants(ctx context.Context) error {
 			s.ingestFilterStats.blockReplay(ingestQualityReasonRetainedInvalid)
 		},
 	}
-	if s.ledgered {
-		policy.scheduledChild = func(ctx context.Context, child, parentType, parent string) (bool, error) {
-			return s.caps.pageLedger.HasScheduledWork(ctx, "resource:"+childScheduleKey(child, parentType, parent))
-		}
+	policy.scheduledChild = func(ctx context.Context, child, parentType, parent string) (bool, error) {
+		return s.caps.pageLedger.HasScheduledWork(ctx, "resource:"+childScheduleKey(child, parentType, parent))
 	}
 	// Leaving the predicate nil is how the policy says "no scheduler
 	// evidence, skip I10". The nil check has to be here: a bound method

pkg/sync/initial_actions.go

@@ -60,32 +60,22 @@ func (s *syncer) initialActions(targetedResources []*v2.Resource) ([]Action, []s
 
 func (s *syncer) initializeAction(ctx context.Context, action *Action, targetedResources []*v2.Resource) error {
 	children, facts := s.initialActions(targetedResources)
-	if s.ledgered {
-		invocation, ok := ctx.Value(ledgerInvocationKey{}).(*ledgerInvocation)
-		if !ok {
-			return errors.New("ledger Init requires an open page")
-		}
-		if err := s.recordFirstReportOptions(invocation.page); err != nil {
+	invocation, ok := ctx.Value(ledgerInvocationKey{}).(*ledgerInvocation)
+	if !ok {
+		return errors.New("ledger Init requires an open page")
+	}
+	if err := s.recordFirstReportOptions(invocation.page); err != nil {
+		return err
+	}
+	if s.stats.ingestQuality() != nil {
+		if err := invocation.page.setFact(ledgerFactIngestKnown); err != nil {
 			return err
 		}
-		if s.stats.ingestQuality() != nil {
-			if err := invocation.page.setFact(ledgerFactIngestKnown); err != nil {
-				return err
-			}
-		}
-		for _, fact := range facts {
-			if err := invocation.page.setFact(fact); err != nil {
-				return err
-			}
-		}
-		return s.nextPageOrFinishAction(ctx, action, "", children...)
 	}
-	s.finishAction(ctx, action)
 	for _, fact := range facts {
-		s.run.setFact(fact)
-	}
-	for _, child := range children {
-		s.run.pushAction(ctx, child)
+		if err := invocation.page.setFact(fact); err != nil {
+			return err
+		}
 	}
-	return s.Checkpoint(ctx, true)
+	return s.nextPageOrFinishAction(ctx, action, "", children...)
 }

pkg/sync/ledger_attachment_test.go

@@ -65,14 +65,13 @@ func TestLedgerPublicRegisteredPathAttachment(t *testing.T) {
 				afterWrites := len(f.audit.events)
 				f.audit.mu.Unlock()
 				require.Equal(t, writes, afterWrites)
-				valid := engine == "pebble" && capability || engine != "pebble" && !capability
-				if !valid {
+				if engine != "pebble" || !capability {
 					require.Error(t, err)
 					require.Equal(t, 1, closeStore.closes)
 					return
 				}
 				require.NoError(t, err)
-				require.Equal(t, engine == "pebble", s.ledgered)
+				require.Nil(t, s.storeAttachErr)
 				require.Zero(t, closeStore.closes)
 				require.NoError(t, s.Close(ctx))
 				require.Equal(t, 1, closeStore.closes)
@@ -94,7 +93,7 @@ func TestLedgerNilStoreFallsBackToPath(t *testing.T) {
 				require.NoError(t, err)
 				s := created.(*syncer)
 				require.NoError(t, s.loadStore(t.Context()))
-				require.Equal(t, engine == c1zstore.EnginePebble, s.ledgered)
+				require.Nil(t, s.storeAttachErr)
 				require.NoError(t, s.Close(t.Context()))
 			})
 		}

pkg/sync/ledger_external_model_test.go

@@ -35,7 +35,7 @@ func (p *externalStreamingProbe) ListGrantsForEntitlement(
 }
 
 func TestLedgerExternalImportsPagesDirectly(t *testing.T) {
-	s, f, source := externalPageFixture(t, true)
+	s, f, source := externalPageFixture(t)
 	fresh := externalMatchPrincipal(t, "fresh", nil)
 	for _, id := range []string{"one", "two", "three"} {
 		require.NoError(t, source.store.PutGrants(t.Context(), gt.NewGrant(fresh, "member", v2.ResourceId_builder{ResourceType: "user", Resource: id}.Build())))