phase 6a - replay functionality [source cache-scoped] - #1045
Conversation
kans
changed the title
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| var scanned int | ||
| for primaries.First(); primaries.Valid(); primaries.Next() { | ||
| scanned++ | ||
| if scanned&0x3FF == 0 { | ||
| if err := ctx.Err(); err != nil { | ||
| primaries.Close() | ||
| return err | ||
| } | ||
| } | ||
| stamp, err := rawdb.ScanSourceScopeKeyRaw(primaries.Value(), scopeField) | ||
| if err != nil { | ||
| primaries.Close() | ||
| return fmt.Errorf("source cache replay: preflight %s primary %x: %w", rowKind, primaries.Key(), err) | ||
| } | ||
| if stamp != scopeKey { | ||
| continue | ||
| } | ||
| indexKey, ok := rawdb.AppendBySourceScopeKeyFromPrimary(nil, primaries.Key(), scopeKey) | ||
| if !ok { | ||
| primaries.Close() | ||
| return fmt.Errorf("source cache replay: preflight %s primary %x cannot derive source index", rowKind, primaries.Key()) | ||
| } | ||
| _, closer, err := prev.db.Get(indexKey) | ||
| if err != nil { | ||
| primaries.Close() | ||
| return fmt.Errorf("source cache replay: preflight %s primary %x missing source index: %w", rowKind, primaries.Key(), err) | ||
| } | ||
| closer.Close() | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: This preflight iterates the entire primary keyspace of the row kind in prev on every replay call, skipping only the per-primary Get for non-matching stamps. Since validateReplaySourceScope runs once per scope (via each ReplaySourceCache*), a sync that replays S scopes against a previous file of N rows is O(S·N) — a quadratic cliff for delta-query connectors with many small scopes and a large prior sync. The single-scope benchmark (source_cache_replay_bench_test.go puts all rows under one scope) scales linearly and won't detect this. Per the repo cost-contract criteria, consider a multi-scope benchmark that pins the cost curve, and/or bounding the primary-side pass (e.g. persisting a per-scope row count/digest in the manifest so the index-side scan alone suffices). The code comment already acknowledges this as a known limitation. (Confidence: high on the full-keyspace scan; medium on real-world impact — depends on connector scope granularity.)
There was a problem hiding this comment.
This will be done in follow up.
General PR Review: [WIP] phase 6a - replay functionalityBlocking Issues: 0 Criteria status: loaded Review SummaryScanned the full PR diff for security and correctness issues, plus the baton-sdk repo-local criteria (exported Go API stability, proto/wire contracts, serialized-state round-tripping, default-behavior stability, and go.mod/go.sum).
The one real correctness issue found (a Pebble value-handle leak in PutGrantRecordsIfNewer, pkg/dotc1z/engine/pebble/if_newer.go:77-80, where closer from e.db.Get stays open if marshalRecord errors) is already tracked in existing findings and is pre-existing/unchanged since the last reviewed SHA — not re-posted here. The sibling entitlement/resource IfNewer paths close on all paths. No new blocking security or correctness issues were found. The replay machinery is extensively self-verified. Security: None found. |
| if err != nil { | ||
| if errors.Is(err, pebble.ErrNotFound) { | ||
| continue | ||
| } | ||
| return err | ||
| } | ||
| if _, ok := seen[id]; ok { | ||
| continue | ||
| } | ||
| seen[id] = struct{}{} | ||
| identities = append(identities, id) | ||
| } | ||
| if len(identities) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| batch := e.db.NewRecordBatch() | ||
| defer func() { _ = batch.Close() }() | ||
| for _, id := range identities { | ||
| key := encodeGrantIdentityKey(id) | ||
| oldVal, closer, err := e.db.Get(key) | ||
| if errors.Is(err, pebble.ErrNotFound) { | ||
| continue | ||
| } | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if err := batch.StageGrantDelete(key, oldVal); err != nil { | ||
| _ = closer.Close() |
There was a problem hiding this comment.
🟡 Suggestion (low confidence): DeleteGrantRecordsBounded (and the parallel DeleteEntitlementRecords) accumulate every resolved identity into an in-memory identities slice + seen map and stage all tombstones into a single RecordBatch committed once. Memory is O(len(externalIDs)), unlike the replay path which deliberately chunks at replayBatchRows (10k). The single atomic commit is a deliberate all-or-nothing tombstone choice and the delete set is asserted to be delta-sized (see DeleteSourceCacheRows doc), so this is likely fine — but it's exactly the "cleanup/tombstone scan builds O(scope size) batch" class this PR added to docs/BUG_CATCHING.md. Worth a bound or a note if a whole-scope tombstone can ever reach here.
General PR Review: [WIP] phase 6a - replay functionalityBlocking Issues: 0 | Suggestions: 3 | Threads Resolved: 0 Review SummaryFull PR diff scanned for security and correctness. This is a HIGH-risk change per the repo triage model: it introduces new durable storage families (by_source_scope indexes, the source-cache manifest keyspace TypeSourceCache=0x0B), a new wire-stamped record field (source_scope_key), and cross-file replay that copies raw values between SDK versions — silent + durable + version-pair-dependent, so failures escape local testing and cost a fleet-wide re-sync. The concrete correctness findings below were already identified in the prior review and remain present in the current tree; no new bugs were found, and no security issues were found. Recommend the durable-format and replay paths get the escalated instrument coverage the trusted criteria calls for (a two-artifact cross-version replay harness plus a golden-artifact corpus for the new index/manifest families), which a single-shot CI review only samples. Security IssuesNone found. Correctness IssuesNone found. Suggestions
Prompt for AI agents |
General PR Review: phase 6a - replay functionality [source cache-scoped]Blocking Issues: 0 | Suggestions: 2 | Threads Resolved: 0 Review SummaryScanned the full PR diff (84 files) for security and correctness, with focused reads of the new source-cache replay engine, the Security IssuesNone found. Correctness IssuesNone found (blocking). Suggestions
Previously identified, still present at this SHA — not re-flagged inline:
Prompt for AI agents |
kans
changed the title
| func (r SyncRun) UsableAsReplaySource() bool { | ||
| return r.Type == connectorstore.SyncTypeFull && !r.Compacted | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: UsableAsReplaySource() gates on !r.Compacted, but the SQLite adapter (getFinishedSync in pkg/dotc1z/sync_runs.go:306-352) never selects or sets Compacted, so it is always false for a SQLite prior run — a compacted SQLite full sync would pass this gate. Impact is currently bounded (the pebble replay path independently re-checks via validateReplaySourceEligible at source_cache.go:129, and replay is pebble-only), but the syncer comment at syncer.go:3735 says a SQLite prior run should work as a replay source. Consider documenting that Compacted is only meaningful for engines that track it, or having the SQLite adapter conservatively derive compaction provenance. (medium confidence)
| test-extra: race-check compat-check interrupt-check fuzz-smoke differential-check bench-smoke ## Run bounded confidence checks omitted from CI. | ||
|
|
||
| .PHONY: test-nightly | ||
| test-nightly: export BATON_TEST_NIGHTLY=1 |
There was a problem hiding this comment.
🟡 Suggestion: neither new tier has an automated runner. .github/workflows/ci.yaml:39 and main.yaml:56 invoke bare go test ./... with neither BATON_TEST_EXTRA nor BATON_TEST_NIGHTLY set, and there is no schedule:/workflow_dispatch: workflow that calls make test-extra or make test-nightly. As a result the 21 newly gated tests — the full deterministic chaos corpora, scheduler soak, WAL-checkpoint race, C1Z integrity, the randomized source-cache lifecycle model, checkpoint-cut enumeration, exhaustive parallel-queue interleavings, and both differential suites — now run only when a human types a Make target, which is coverage removal rather than relocation. Consider adding a nightly scheduled job that runs make test-nightly.
| func RequireExtra(t testing.TB) { | ||
| t.Helper() | ||
| if os.Getenv(ExtraEnv) == "" && os.Getenv(NightlyEnv) == "" { | ||
| t.Skipf("set %s=1 or run the corresponding Make target", ExtraEnv) | ||
| } | ||
| } | ||
|
|
||
| // RequireNightly skips a randomized, repeated, or full-corpus test unless the | ||
| // nightly confidence suite explicitly enabled it. | ||
| func RequireNightly(t testing.TB) { | ||
| t.Helper() | ||
| if os.Getenv(NightlyEnv) == "" { | ||
| t.Skipf("set %s=1 or run make test-nightly", NightlyEnv) | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: both gates test os.Getenv(...) == "", so BATON_TEST_EXTRA=0 (or BATON_TEST_NIGHTLY=false) enables the tier. docs/TESTING.md documents the contract as =1, so a CI matrix or shell profile that sets the variable to 0 in order to opt out would silently opt in instead. Consider accepting only truthy values (an explicit == "1", or strconv.ParseBool on a non-empty value).
| // C10/C12: the replay commit seam supplies deterministic evidence that live | ||
| // batch cardinality is fixed, and lets retry be cut after one landed chunk. | ||
| func TestVerificationReplayBatchBoundAndInterruptedRetry(t *testing.T) { | ||
| testtier.RequireExtra(t) |
There was a problem hiding this comment.
🟡 Suggestion: require.Equal(t, replayBatchRows, highWater) on line 57 is the only assertion anywhere that the live replay path actually honors the production replayBatchRows = 10_000 bound — TestVerificationReplayCommittedPrefixRetryAllKinds lowers the seam to 2, and source_scope_verification_test.go:1113 only exercises validateBatchHighWater in isolation. Gating this test to the extra tier therefore removes CI's only guard against a regression that drops or changes the production batch bound on the new durable replay path. Consider splitting the cheap bound assertion into a CI-tier test and leaving only the 10k-row fixture behind RequireExtra.
| run, metaErr := previousSyncStore.SyncMeta().LatestFinishedSyncOfAnyType(ctx) | ||
| if metaErr != nil { | ||
| closeErr := previousSyncStore.Close(ctx) | ||
| if s.previousSyncC1ZPathOptional { | ||
| ctxzap.Extract(ctx).Warn("previous-sync c1z metadata unusable; syncing without source-cache replay", | ||
| zap.String("previous_sync_c1z_path", s.previousSyncC1ZPath), | ||
| zap.Error(errors.Join(metaErr, closeErr)), | ||
| ) | ||
| break | ||
| } | ||
| return nil, fmt.Errorf( | ||
| "error reading previous-sync c1z %q metadata: %w", | ||
| s.previousSyncC1ZPath, | ||
| errors.Join(metaErr, closeErr), | ||
| ) | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: this makes NewSyncer hard-fail for WithPreviousSyncC1ZPath callers on a metadata read error, where before this PR no metadata was read at all and the sync proceeded. Note the asymmetry with the branch immediately below — run == nil || !UsableAsReplaySource() only warns and degrades for the same non-optional caller. Since previousSyncReader is currently unconsumed scaffolding (the ETag read path is still t.Skipped in pebble_etag_replay_test.go), this only adds a new way for construction to fail; consider warn-and-degrade here too, reserving hard failure for open failures as the option doc describes. (confidence: medium)
| if mayExist { | ||
| oldScope, err := ScanSourceScopeKeyRaw(oldVal, field) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if oldScope != "" && oldScope != newScope { | ||
| if err := rb.deleteSourceScopeKey(key, oldScope); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: this returns the ScanSourceScopeKeyRaw(oldVal, …) error and fails the whole put, while the delete twin stageSourceScopeCleanup (records.go:452-455) deliberately degrades to deleteAllSourceScopeKeysForPrimary. Net effect: a record whose stored prior value is malformed can never be overwritten (permanent failure) but can be deleted (self-heals) — and before this PR the put path did not parse oldVal at all for grants/entitlements, so overwriting a bad row succeeded. Consider mirroring the cleanup path: on scan error fall back to deleteAllSourceScopeKeysForPrimary(key) and continue to the new-scope Set. (confidence: medium)
| func encodeGrantBySourceScopeIndexKey(scopeKey string, id grantIdentity) []byte { | ||
| key, _ := rawdb.AppendBySourceScopeKeyFromPrimary(nil, encodeGrantIdentityKey(id), scopeKey) | ||
| return key | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: the ok return from AppendBySourceScopeKeyFromPrimary is discarded, so a grant identity that fails to splice yields a nil key that flows on as a real index key. Every other splice site in this PR (setByPrincipalKey, deleteSourceScopeKey, stageSourceScopeChange) turns !ok into an error. Either propagate the bool or panic on !ok, rather than relying on downstream len(k) < 3 checks to catch it. (confidence: medium)
| closeErr := w.Close(ctx) | ||
| if closeErr != nil { | ||
| l.Error("compactPebble: error closing source store", zap.Error(closeErr), zap.String("file", sourcePath)) | ||
| } | ||
| if err := joinSourceStoreCloseError(selectErr, closeErr, sourcePath); err != nil { | ||
| return err |
There was a problem hiding this comment.
🟡 Suggestion: source stores here are read-only inputs that have already been fully consumed (SourceFile carries only Path/SyncID/Stats). Joining their Close error into the return — same at compactor_pebble.go:561-570 — changes previously log-only behavior into a hard compaction failure, so a transient temp-dir/unlink hiccup can now fail an otherwise correct compaction. Consider keeping these logged and reserving the join for destination-side close errors. (confidence: high that the behavior changed, medium that it matters in practice)
| t.Helper() | ||
| if os.Getenv(ExtraEnv) == "" && os.Getenv(NightlyEnv) == "" { | ||
| t.Skipf("set %s=1 or run the corresponding Make target", ExtraEnv) | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
🟡 Suggestion: the gate is != "", so BATON_TEST_EXTRA=0 (or =false) enables the extra tier — the opposite of what the value suggests, and the skip message tells the reader to set =1. Same shape in RequireNightly. Consider comparing against "1" (or parsing with strconv.ParseBool) so an explicit off value turns the tier off. (confidence: high, low impact)
| .PHONY: race-check | ||
| race-check: ## Run the complete Go suite with the race detector. | ||
| go test -race -tags=baton_lambda_support -count=1 -timeout=45m ./... | ||
| BATON_TEST_EXTRA=1 go test -race -tags=baton_lambda_support -count=1 -timeout=45m ./... |
There was a problem hiding this comment.
🟡 Suggestion: ci.yaml and main.yaml run bare go test ./... with neither tier variable set, so every test newly gated by testtier.RequireExtra/RequireNightly now skips in all PR and main runs — including several that previously ran unconditionally (TestParallelQueueExhaustiveInterleavings, TestTopologicalMergeDifferentialRandom{,Store}, TestTopologicalMergeChunkedDirtyFlush, TestCleanupContextDeadlineExceeded) and the whole non-short set (TestC1ZIntegrity, TestWALCheckpointRace, all TestChaosConnector*Corpus). race-check here sets only BATON_TEST_EXTRA=1, so the chaos corpora that the deleted comment said "race-check already includes" now run only under nightly. Worth confirming that's the intended net coverage, or gating the cheap deterministic ones (e.g. TestTopologicalMergeChunkedDirtyFlush) at a lower tier.
| // StaleSkipped counts index entries under the scope that did NOT | ||
| // yield a copied row: the primary was missing, or its value stamp | ||
| // named a different scope. This is the discriminator between "scope | ||
| // legitimately empty" (index prefix empty, StaleSkipped == 0) and | ||
| // "scope's rows were clobbered without index cleanup" (index says | ||
| // rows existed, none survived the stamp check). Future syncer replay | ||
| // orchestration must fail a zero-row replay when StaleSkipped > 0. | ||
| StaleSkipped int64 |
There was a problem hiding this comment.
🟡 Suggestion: StaleSkipped can never be non-zero. validateReplaySourceScope runs first and hard-errors on exactly the two conditions that increment it — an index entry whose primary is absent (preflight %s index %x has no primary) and an indexed primary whose stamp differs (index scope %q resolves to primary stamped %q) — over the same indexPrefix range the replay loop then walks on an immutable read-only prev. So the three res.StaleSkipped++ sites (lines 931, 956, 1126, 1140, 1277, 1295) are unreachable, and the documented contract "future syncer replay orchestration must fail a zero-row replay when StaleSkipped > 0" would be gating on a permanently-zero signal. Either drop the preflight's hard failure for these two cases (the replay loop already comments that stale index entries are an expected, tolerable state) or drop the field so downstream doesn't build on it.
| } | ||
|
|
||
| // A fold inherits its base's validators, but the merged winners no longer | ||
| // represent that connector snapshot. Drop only the small manifest keyspace; | ||
| // retaining the existing source-scope indexes avoids an O(base) rewrite. | ||
| if err := destEng.InvalidateSourceCacheReplayState(ctx, false); err != nil { | ||
| return "", fmt.Errorf("compactPebbleFold: invalidate source-cache replay state: %w", err) | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: bucket_plans.go has no bucket for the three by_source_scope index families or TypeSourceCache, so the fold merge copies the partial sources' scope-stamped primary rows into the grants/entitlements/resources buckets without their index entries, while dropScopeIndexes=false keeps the base's entries — including ones whose primary was just overwritten by a winner with a different (or empty) stamp. The output therefore violates the primary↔index biconditional that stageSourceScopeCleanup and validateReplaySourceScope rely on. It's contained today only because baseRec.SetCompacted(true) makes the artifact replay-ineligible; if fold output ever becomes an eligible source, this becomes silent wrong-row replay. Worth a comment recording the dependency, or adding the scope families to the fold bucket plan.
| // A connector that can cheaply revalidate upstream data — HTTP conditional | ||
| // requests (GitHub), delta queries (Microsoft Graph) — opts in by attaching | ||
| // SourceCacheCapability MODE_READ_WRITE to its Validate response. During a | ||
| // sync it looks up the previous validator for a scope via the Lookup the SDK | ||
| // provides on SyncOpAttrs, revalidates upstream, and either emits fresh rows | ||
| // tagged with SourceCacheRecord or asks the SDK to replay the previous rows | ||
| // with SourceCacheReplay. | ||
| // | ||
| // The connector owns scope computation; the SDK only keys storage by the | ||
| // connector-supplied scope key. The validator (etag, delta token) is opaque | ||
| // to the SDK. | ||
| // | ||
| // Invariant that keeps replay safe: a connector must only emit | ||
| // SourceCacheReplay for a scope whose validator it received from THIS sync's | ||
| // Lookup. The lookup need not happen in the same call that emits the | ||
| // replay: a planning call may batch-resolve many scopes and pass the | ||
| // verdicts to sibling cursors through EnqueuePageTokens page tokens — that | ||
| // satisfies the invariant, because the validator still originates from the | ||
| // consuming sync. What's forbidden is a validator that outlives a sync | ||
| // (connector-side caches, config, upstream echoes). When source cache is | ||
| // disabled or degraded (no capability, no usable previous sync, unsupported | ||
| // storage engine) the SDK installs NoopLookup, every lookup misses, and a | ||
| // well-behaved connector naturally falls back to full fetch. |
There was a problem hiding this comment.
🟡 Suggestion: this package doc describes wiring that doesn't exist yet — Lookup, SetLookup, and NoopLookup have no references anywhere outside this package and its tests, and resource.SyncOpAttrs has no source-cache field, so "the Lookup the SDK provides on SyncOpAttrs" and "the SDK installs NoopLookup" are not true at this SHA (the PR body confirms syncer orchestration is deferred). A connector author reading this new exported package would build against a surface that isn't reachable. Suggest a short "not yet wired; see phase 6b" note in the doc until the syncer consumes it.
A sync can only skip refetching a page if the previous artifact is a trustworthy source for it. This adds the storage, provenance, and eligibility machinery for that decision. Nothing is on by default: the write path is unchanged for connectors that do not opt in through sourcecache.WithScope, and no production sync consumes replay yet. Storage. Grant, resource, and entitlement records carry a source_scope_key, indexed by a by_source_scope family so mid-sync tombstones can drop a scope's rows without scanning the keyspace. Unscoped syncs must not pay for an index they never populate, so the index obligation sits behind a sourceScopeMayExist gate: probed at Open, armed only when a scoped key is actually staged, and cleared when the families are excised. Benchmarks attributed essentially all of the added ingest cost to the prior-value Get this avoids, so the gate is what keeps ordinary writes where they were. Provenance. Replay is only sound from a finished, uncompacted, full Pebble sync. SyncRunSummary gains a compacted flag that the v3 manifest header projects, and the syncer rejects SQLite and compacted inputs outright rather than degrading silently. Every compaction output is replay-ineligible by construction: k-way and overlay drop the source-scope indexes instead of playing them forward, and fold keeps whatever is cheapest for fold without merging partial inputs. Also fixes a deadlock in which the source-cache delete path re-entered closeMu through a promoted store method, and adds the instruments this work needed to stay honest: a commit-point enumeration meta-test that fails when a new batch-commit site ships without a declared failure seam, and opt-in test tiers so the long deterministic and randomized cases run in the nightly sweep instead of every pull request. Co-authored-by: Cursor <cursoragent@cursor.com>
| for i, r := range resources { | ||
| rid := r.GetId() | ||
| if err := s.Engine.DeleteResourceRecord(ctx, rid.GetResourceType(), rid.GetResource()); err != nil { | ||
| return fmt.Errorf("source cache delete resource %q: %w", ids[i], err) | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion (medium confidence): the resource branch issues one DeleteResourceRecord per id, and each of those is its own withWrite + batch.Commit(writeOpts(e.opts.durability)) — so a tombstone page of N resources costs N separate durable commits and is not atomic. The sibling branches below stage all grant/entitlement deletes into a single RecordBatch, and the scoped variants commit in bounded chunks and report committed progress on error. Here a mid-page failure leaves a partially applied page with only error returned, so a caller can't tell what landed. Consider an engine-level DeleteResourceRecords(ctx, ids) that batches like DeleteGrantRecordsBounded/DeleteEntitlementRecords.
| parentRT, parentID, parentErr := ScanResourceParentRaw(oldVal) | ||
| if parentErr != nil { | ||
| if err := rb.deleteAllResourceParentKeysForChild(childRT, childID); err != nil { | ||
| return err | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion (medium confidence): this replaces stageResourceParentDelete, which propagated the ScanResourceParentRaw error, with a silent fallback to deleteAllResourceParentKeysForChild — a full iteration of the entire by_parent index family, per deleted row. A mass delete over a corrupted resource family therefore becomes O(deleted × |by_parent|) with no error and no log line saying corruption was hit. Two things worth reconsidering: (1) emit some signal (wrapped error or a counter) so the corruption is observable rather than absorbed, and (2) note that StageResourcePut above (line 322) still hard-fails on the same malformed prior value, so overwriting a corrupt row errors while deleting it silently heals — the asymmetry is easy to trip over later.
1 participant
Summary
Implements Sync Replay Phase 6a for the Pebble-backed
dotc1zstore, and secondarily, runs the experiment of using a semi-formal verification plan, created independently of the implementation.This adds:
Verification record
This PR was developed and tested against an implementation-blind verification plan.
docs/verification/sync-replay-6a/plan.mddocs/verification/sync-replay-6a/evidence.mdThe plan defines C01–C43, their oracles, coverage levels, failure models, and closure requirements. The evidence record states exactly which criteria are verified, sampled, incomplete, excluded, or deferred.
Verification-driven fixes
The verification work found and corrected defects involving:
Scoped tombstone operations now commit in bounded chunks. If a later chunk fails, they report only committed progress, preserve primary/index agreement, mark the store dirty, and converge on retry.
Change orders
The committed plan records four post-freeze change orders:
Deliberately deferred
This PR does not implement:
Executable exclusions identify API-boundary cells that Phase 6a cannot represent. An exclusion is not counted as a behavioral pass.
Evidence
Passing gates: