[EXPERIMENTAL] c1z sync replay#1002
Conversation
|
|
Co-authored-by: Cursor <cursoragent@cursor.com>
kans
force-pushed
the
kans/c1z-sync-replay
branch
from
2c28a48 to
ae0ccd4
Compare
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| if err := batch.Set(priKey, val, nil); err != nil { | ||
| closer.Close() | ||
| return err | ||
| } | ||
| closer.Close() | ||
| if err := batch.Set(iter.Key(), nil, nil); err != nil { | ||
| return err | ||
| } | ||
| res.Rows++ | ||
| rowsInBatch++ | ||
| if rowsInBatch >= replayBatchRows { | ||
| if err := batch.Commit(opts); err != nil { | ||
| return err | ||
| } | ||
| _ = batch.Close() | ||
| batch = e.db.NewBatch() | ||
| rowsInBatch = 0 | ||
| } | ||
| } | ||
| if err := iter.Error(); err != nil { | ||
| return err | ||
| } | ||
| return batch.Commit(opts) |
There was a problem hiding this comment.
🟠 Bug: ReplaySourceCacheEntitlements mutates the entitlement primary keyspace (batch.Set(priKey, val, ...)) but never calls e.noteEntitlementKeyspaceWrite(). Every other entitlement writer bumps that generation counter (entitlements.go:98, :159, if_newer.go:219, cleanup.go:101, bulk_import.go:690, id_index_migration.go:104); lookup.go:74 documents "call after ANY mutation of the entitlement primary keyspace." Within a source-cache sync, once the lazy entIDLookup map has been built, replayed entitlements won't appear in it, so the deleted_ids tombstone path (DeleteEntitlementRecord → resolveEntitlementIdentityByExternalID, plus grant candidate resolution via entitlementIdentitiesForExternalID) silently misses them and the delete no-ops — a row upstream reported deleted survives. Fix: call e.noteEntitlementKeyspaceWrite() after a successful replay (inside withWrite, after the final commit). (Confidence: medium.)
|
|
||
| if v.GetString(field.PreviousSyncC1ZField.GetName()) != "" { | ||
| previousSyncC1ZPath := v.GetString(field.PreviousSyncC1ZField.GetName()) | ||
| _, err := os.Open(previousSyncC1ZPath) |
There was a problem hiding this comment.
🟡 Suggestion: os.Open returns a *os.File that is assigned to _ and never closed, leaking a file descriptor on every invocation. Since this is only an existence check, use os.Stat instead (which also avoids opening the file). Confidence: high.
| _, err := os.Open(previousSyncC1ZPath) | |
| if _, err := os.Stat(previousSyncC1ZPath); err != nil { | |
| return fmt.Errorf("the specified previous sync c1z file does not exist: %s", previousSyncC1ZPath) | |
| } |
| _, err := os.Open(previousSyncC1ZPath) | ||
| if err != nil { | ||
| return fmt.Errorf("the specified previous sync c1z file does not exist: %s", previousSyncC1ZPath) | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion (supersedes my prior single-line comment above): os.Open returns a *os.File assigned to _ and never closed, leaking a file descriptor. Since this is only an existence check, use os.Stat. Confidence: high.
| _, err := os.Open(previousSyncC1ZPath) | |
| if err != nil { | |
| return fmt.Errorf("the specified previous sync c1z file does not exist: %s", previousSyncC1ZPath) | |
| } | |
| if _, err := os.Stat(previousSyncC1ZPath); err != nil { | |
| return fmt.Errorf("the specified previous sync c1z file does not exist: %s", previousSyncC1ZPath) | |
| } |
| // | ||
| // Complexity: O(scope size) tuple-walks per call regardless of tombstone | ||
| // count — callers batch a page's tombstones into one call. | ||
| func (e *Engine) DeleteGrantsByPrincipalsInScope(ctx context.Context, scopeHash string, principalIDs map[string]struct{}) (int64, error) { |
There was a problem hiding this comment.
🟡 Suggestion: unlike the ReplaySourceCache* functions (which cap batch growth at replayBatchRows with intermediate commits), the three scoped-delete functions (DeleteGrantsByPrincipalsInScope here, plus DeleteGrantsByExternalIDsInScope at line 207 and DeleteResourcesByIDsInScope at line 294) accumulate every matched deletion into a single unbounded pebble.Batch committed once at the end. A delta round tombstoning a large fraction of a big scope could grow the batch to a large in-memory size. Consider the same periodic-commit bound used by replay. Confidence: medium.
General PR Review: [EXPERIMENTAL] c1z sync replayBlocking Issues: 0 | Suggestions: 4 | Threads Resolved: 0 Review SummaryFull PR diff scanned for security and correctness, with extra focus on the SDK compatibility surfaces called out in the repo-local criteria (proto wire fields, serialized pagination/replay state, exported interfaces, defaults, and Security IssuesNone found. Correctness IssuesNone found. Suggestions
Prompt for AI agents``` SuggestionsIn `pkg/sync/syncer.go`:
In `pkg/sync/source_cache.go`:
In `pkg/sourcecache/continuation.go`:
In `pkg/cli/commands.go`:
|
kans
force-pushed
the
kans/c1z-sync-replay
branch
from
5fd87b4 to
b00db4e
Compare
| if hasSpawn { | ||
| for _, tok := range spawn.GetPageTokens() { | ||
| if tok == "" { | ||
| continue | ||
| } | ||
| spawned = append(spawned, Action{ | ||
| Op: SyncGrantsOp, | ||
| ResourceTypeID: action.ResourceTypeID, | ||
| ResourceID: action.ResourceID, // empty on type-scoped actions | ||
| PageToken: tok, | ||
| Spawned: true, | ||
| }) | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: spawn.GetPageTokens() is iterated with no cap on count, and each spawned cursor's own response can spawn again with no depth ceiling, so a buggy connector could push an unbounded number of actions onto the (checkpointed) state stack. Note SourceCacheLookupAsk.queries is bounded with max_items: 4096 in the proto, but SpawnCursors.page_tokens has no validate.rules bound — consider a per-response cap and/or a per-sync total-spawned ceiling that errors loudly, mirroring the bounce cap. (confidence: medium)
| page.deletedIDs = append(replay.GetDeletedIds(), scope.GetDeletedIds()...) | ||
| page.deletedPrincipalIDs = append(replay.GetDeletedPrincipalIds(), scope.GetDeletedPrincipalIds()...) |
There was a problem hiding this comment.
🟡 Suggestion: append(replay.GetDeletedIds(), scope.GetDeletedIds()...) appends into the slice returned by the proto getter; if that backing array has spare capacity the append mutates the proto message in place (and page.deletedIDs aliases it). Freshly-unmarshaled repeated fields usually have len == cap so this rarely bites, but it's safer to copy explicitly, e.g. append(append([]string{}, replay.GetDeletedIds()...), scope.GetDeletedIds()...). (confidence: low)
| func AnswersFromProto(msg *v2.SourceCacheLookupAnswers) []Answer { | ||
| out := make([]Answer, 0, len(msg.GetAnswers())) | ||
| for _, a := range msg.GetAnswers() { | ||
| out = append(out, Answer{ | ||
| Query: Query{RowKind: RowKind(a.GetRowKind()), ScopeHash: a.GetScopeHash()}, | ||
| Found: a.GetFound(), | ||
| ETag: a.GetEtag(), | ||
| }) | ||
| } | ||
| return out | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: AnswersFromProto trusts RowKind, ScopeHash, and an unbounded ETag without validation, unlike the sibling QueriesFromProto (lines 192-204) which validates row kind and scope hash. These parent-supplied answers are the point where lookup verdicts enter the connector; validating row kind / scope hash and bounding answer count + etag length here would add defense-in-depth against a malformed or oversized answer set. (confidence: medium)
|
|
||
| if v.GetString(field.PreviousSyncC1ZField.GetName()) != "" { | ||
| previousSyncC1ZPath := v.GetString(field.PreviousSyncC1ZField.GetName()) | ||
| _, err := os.Open(previousSyncC1ZPath) |
There was a problem hiding this comment.
🟡 Suggestion: this existence check uses os.Open but discards the returned *os.File and never closes it, leaking one FD for the process lifetime (the pre-existing external-resource-c1z check at line 467 has the same shape). Since only existence is being checked, prefer os.Stat(previousSyncC1ZPath). (confidence: high)
1 participant
POCs:
ConductorOne/baton-okta#182
ConductorOne/baton-github#177
https://github.com/ConductorOne/baton-microsoft-entra/pull/164
Source-cache replay: skip refetching data upstream says is unchanged (Pebble-only)
Connectors talking to providers with cheap change detection — HTTP conditional requests (GitHub), delta queries (Microsoft Graph) — can now tell the SDK to replay a scope's rows from the previous sync's c1z instead of re-fetching, re-parsing, and re-writing them. On a warm sync, unchanged pages cost one conditional request (a
304is free against GitHub's rate limit) and a bulk row copy, instead of full fetch + parse + write.Wire contract (
annotation_source_cache.proto)SourceCacheCapability{MODE_READ_WRITE}on the Validate response opts a connector in.SyncOpAttrs.SourceCache) for the previous sync's validator (etag / delta token), revalidates upstream, and either:304/ unchanged →SourceCacheReplay{scope_hash, etag}with an empty page: the SDK bulk-copies the previous sync's rows for that scope;SourceCacheScope{scope_hash, etag}: rows are stamped with the scope and the validator is persisted.overlay: truereplays the base then upserts changed rows on top; every page can carry tombstones (deleted_idsfor canonical IDs,deleted_principal_idsfor bare object IDs applied scope-relatively) so multi-page delta rounds never buffer; validators rotate per round.BatonSourceCacheServiceon the existing control-plane listener (shared conn with the session service).Lookup continuation (ask/answer): lookups over single-shot transports
Production runs connectors in environments that cannot call back to the syncer mid-request (gRPC-over-Lambda: one invoke, one response, no ingress to the worker). The lookup for those topologies is a continuation protocol on the existing tunnel — the response is the request:
SourceCacheLookupOfferto list requests when it can answer (capability declared + warm previous sync). The connector's lookup defers: it records the scopes it needs and the RPC answers withSourceCacheLookupAskinstead of rows. The syncer resolves the queries against its local previous c1z (the same file replay copies from — lookup and replay cannot disagree) and re-invokes the same request withSourceCacheLookupAnswers. The connector code is unchanged across topologies: direct lookup in-process, loopback gRPC in subprocess, deferred ask on Lambda.sourcecache.LookupManyis the topology-uniform batch API: a loop of point-reads on direct lookups, ONE ask for the whole batch when deferring. First-class planner shape: batch-resolve a page's scopes, embed verdicts inSpawnCursorstokens, spawned cursors never ask (Entra: ~5k scopes resolved in ~90 bounces instead of ~5k; Okta/GitHub equivalents tested).ErrLookupDeferred(must propagate,%w-wrapping fine — matched viaerrors.Is) is caught by connectorbuilder before failure metrics; an ask carrying rows/tokens/scope/replay/SpawnCursorsis a hard error; an ask without an offer is a hard error; a bounce cap (4 per REQUEST — page-token advances reset it, so multi-page planners asking once per page are fine) stops non-progressing connectors.Design + prod topology analysis:
docs/tasks/source-cache-lambda-lookup.md.Storage (Pebble engine only; SQLite is untouched and silently degrades)
source_scope_hashon v3 Resource/Entitlement/Grant records with partialby_source_scopeindex families whose tails are identity tuples — replay derives every primary key from the index key and copies raw values with no proto unmarshal on the hot path.typeSourceCachekeyspace holds one(row_kind, scope_hash) → etagmanifest entry per scope (zero-row scopes included). Empty for every connector that doesn't opt in.DeleteGrantsByPrincipalsInScope/DeleteResourcesByIDsInScope(one index scan per tombstone-carrying page, zero value reads),DeleteGrantsByExternalIDsInScope, and a bounded canonical-ID delete that never falls back to the O(all-grants) scan.Output stability (hard requirement)
A warm sync must reproduce what a cold sync would produce — same grants, same IDs. Replayed rows are verbatim copies; expander-written
Sources(classified by self-source, mirroringRollbackExpansion) are stripped at copy time so expansion recomputes provenance from current state, while connector-setSourcessurvive byte-for-byte. Replay also re-arms grant expansion vianeeds_expansionreporting — a sync where every expandable page replays still expands (pinned by test).SpawnCursors: connector-spawned sibling cursors (annotation_type_scoped_grants.proto)A ListGrants response may enqueue additional independent cursors; each token becomes its own action — scheduled by the worker pool, rate-limited, and checkpointed like any other pagination. Honored on both call shapes:
Spawned cursors are ORDINARY pages enqueued eagerly — a spawned page may replay, fetch cold (boundary shifted), chain via
NextPageToken(how fresh tail pages get discovered), or itself spawn. Progress accounting counts a resource once, when its origin action's chain ends — actions carry aSpawnedmarker (checkpoint-compatible) so fan-out never trips the "more grant resources than resources" anomaly warning.Type-scoped grants (
annotation_type_scoped_grants.proto)Grants-phase analogue of StaticEntitlements, for providers whose natural enumeration unit is a collection rather than a resource (e.g. one Graph delta stream covering 50 groups). A resource type annotated
TypeScopedGrantsis excluded from the per-resource ListGrants fan-out; its syncer implementsGrantsForResourceType. Source-cache scopes/replay/tombstones work unchanged on these pages; scope-stamped and unstamped fresh pages coexist in one cursor's stream; scoped pages are authoritative upsert streams (per-sync resource dedupe bypassed — required for overlay correctness). Progress is count-only for these types.Plumbing
--previous-sync-c1zflag for local/one-shot syncs; service mode reuses the existing spare-c1z path (WithKeepPreviousSyncC1Z). Both flags are hidden unless the connector declares replay capability (connectorrunner.DeclaresPreviousSyncCapability).field.WithConnectorDefault: connectors can declare first-class defaults for shared SDK fields (e.g. a per-connector--workersdefault); explicit user values win.SyncOpAttrsgainsSourceCache(never nil — no-op when disabled); session store is likewise nil-safe.Testing
errors.Ismatching through wrapping, answers served to unchanged connector code, swallowed deferral caught in-process, direct lookup wins over offer) and syncer-level (warm replay via ask/answer across a chained multi-sync run, phase-2SpawnCursorswith verdict-carrying tokens and zero sibling asks, multi-page planner exceeding the per-request cap with per-page asks, cap/ask-with-rows/ask-with-spawn/ask-without-offer all fail loudly,TypeScopedGrantsmarker survives re-invokes), at workers 0 and 4.make lint+make testgreen.Real-provider validation (baton-github conditional requests, baton-entra delta, baton-okta timestamp validators) is in flight on connector branches; this PR is the SDK surface they build against. The production (Temporal + Lambda) rollout additionally needs a c1-side change to materialize the previous completed c1z into the sync activity — tracked in the design doc; no c1 protocol code or infra is required beyond the SDK bump.