pebble

func (*Adapter) CheckpointSync ¶

func (a *Adapter) CheckpointSync(ctx context.Context, syncToken string) error

CheckpointSync persists a step token to the open sync's record.

func (*Adapter) Cleanup ¶

func (a *Adapter) Cleanup(ctx context.Context) error

Cleanup on the bare Adapter is a no-op. The real Pebble retention policy lives on *registeredStore (see register.go) — it needs access to caller-supplied options (SyncLimit, SkipCleanup) that the engine itself doesn't track, plus the dirty-flag plumbing on the wrapper.

Callers that open through dotc1z.NewStore(..., WithEngine(EnginePebble)) get the real Cleanup; callers that build a bare Adapter (unit tests, embedding) silently get retention=disabled. The method is kept on the Adapter only to satisfy the connectorstore.Writer interface contract regardless of how the adapter was constructed.

func (*Adapter) Close ¶

func (a *Adapter) Close(ctx context.Context) error

Close shuts down the engine. After Close, all methods return errors.

func (*Adapter) CurrentDBSizeBytes ¶

func (a *Adapter) CurrentDBSizeBytes() (int64, error)

CurrentDBSizeBytes returns the current uncompressed Pebble working-set size on disk. Implements connectorstore.DBSizeProvider for progress logging parity with the SQLite-backed *dotc1z.C1File.

func (*Adapter) CurrentSyncStep ¶

func (a *Adapter) CurrentSyncStep(ctx context.Context) (string, error)

CurrentSyncStep returns the current sync's step string, or "".

func (*Adapter) DeleteGrant ¶

func (a *Adapter) DeleteGrant(ctx context.Context, grantID string) error

DeleteGrant removes a grant by id.

func (*Adapter) EndSync ¶

func (a *Adapter) EndSync(ctx context.Context) error

EndSync stamps the open sync_run's ended_at and detaches it. After EndSync, the adapter has no current sync; SetCurrentSync or StartNewSync are required for further writes.

func (*Adapter) FileOps ¶

func (a *Adapter) FileOps() dotc1z.FileOps

FileOps returns the FileOps sub-store backed by the Pebble adapter. Implements dotc1z.C1ZStore.FileOps(). Both methods are fully implemented: CloneSync materializes one sync's data into a fresh c1z (used by `baton clone`), and GenerateSyncDiff computes the additions-only set difference between two ended syncs and emits a new SyncTypePartial sync (used by the local differ CLI).

func (*Adapter) FileOpsWithEncoding ¶

func (a *Adapter) FileOpsWithEncoding(encoding dotc1z.PayloadEncoding) dotc1z.FileOps

FileOpsWithEncoding returns a FileOps that emits new c1z files using the given payload encoding. Used by registeredStore to propagate its configured encoding (e.g. PayloadEncodingTar for callers that want uncompressed envelopes) through to CloneSync's output file.

func (*Adapter) GetAsset ¶

func (a *Adapter) GetAsset(ctx context.Context, req *v2.AssetServiceGetAssetRequest) (string, io.Reader, error)

GetAsset returns the (content_type, data-reader) for the given asset. The returned reader is backed by a bytes.Reader over the fully-materialized blob.

func (*Adapter) GetEntitlement ¶

func (a *Adapter) GetEntitlement(ctx context.Context, req *reader_v2.EntitlementsReaderServiceGetEntitlementRequest) (*reader_v2.EntitlementsReaderServiceGetEntitlementResponse, error)

GetEntitlement fetches a single entitlement by ID. Implements reader_v2.EntitlementsReaderServiceServer.

func (*Adapter) GetGrant ¶

func (a *Adapter) GetGrant(ctx context.Context, req *reader_v2.GrantsReaderServiceGetGrantRequest) (*reader_v2.GrantsReaderServiceGetGrantResponse, error)

GetGrant fetches a single grant by ID.

func (*Adapter) GetLatestFinishedSync ¶

func (a *Adapter) GetLatestFinishedSync(ctx context.Context, req *reader_v2.SyncsReaderServiceGetLatestFinishedSyncRequest) (*reader_v2.SyncsReaderServiceGetLatestFinishedSyncResponse, error)

GetLatestFinishedSync returns the most-recently-ended sync_run, optionally filtered by sync_type. Implements reader_v2.SyncsReaderServiceServer. Delegates to Engine.LatestFinishedSyncRecord.

Future work: if this becomes hot, a "latest by type" sidecar key keyed on (typeFinishedSyncByType | type) → sync_id would let Engine.LatestFinishedSyncRecord short-circuit to O(1).

func (*Adapter) GetResource ¶

func (a *Adapter) GetResource(ctx context.Context, req *reader_v2.ResourcesReaderServiceGetResourceRequest) (*reader_v2.ResourcesReaderServiceGetResourceResponse, error)

GetResource fetches a single resource by (resource_type_id, resource_id). Implements reader_v2.ResourcesReaderServiceServer.

func (*Adapter) GetResourceType ¶

func (a *Adapter) GetResourceType(ctx context.Context, req *reader_v2.ResourceTypesReaderServiceGetResourceTypeRequest) (*reader_v2.ResourceTypesReaderServiceGetResourceTypeResponse, error)

GetResourceType fetches a single resource_type by ID. Implements reader_v2.ResourceTypesReaderServiceServer.

func (*Adapter) GetSync ¶

func (a *Adapter) GetSync(ctx context.Context, req *reader_v2.SyncsReaderServiceGetSyncRequest) (*reader_v2.SyncsReaderServiceGetSyncResponse, error)

GetSync fetches a single sync_run record by ID. Implements reader_v2.SyncsReaderServiceServer.

func (*Adapter) GrantStats ¶

func (a *Adapter) GrantStats(ctx context.Context, syncType connectorstore.SyncType, syncID string) (map[string]int64, error)

GrantStats returns just the grant count, partitioned by entitlement resource_type_id. Used by progresslog to show per-RT progress. Like Stats, this is an exact count via iteration.

syncType is validated against the sync_run's actual type to match SQLite's C1File.GrantStats: SyncTypeAny accepts any sync; otherwise we require the sync_run.Type to match exactly. A type mismatch returns (nil, nil) — the syncer treats that as "nothing to show for this view," same as the SQLite contract.

func (*Adapter) Grants ¶

func (a *Adapter) Grants() dotc1z.GrantStore

Grants returns the GrantStore implementation backed by the Pebble adapter. Implements dotc1z.C1ZStore.Grants(); used by the expander, the c1-side fileClientWrapper, and the differ.

Caveat: the *Page methods that filter on needs_expansion currently return no candidates because grant expansion (Stack 6 per RFC v4) is deferred. The methods are CORRECT (they don't break the syncer's expansion phase — they make it a no-op) but they don't actually find expandable grants. Once Stack 6 lands, the engine will start flagging needs_expansion at PutGrants and the index walk here will return real rows.

func (*Adapter) LatestFinishedSyncID ¶

func (a *Adapter) LatestFinishedSyncID(ctx context.Context, syncType connectorstore.SyncType) (string, error)

LatestFinishedSyncID returns the most-recently-finished sync ID of the given type. Implements connectorstore.LatestFinishedSyncIDFetcher. Delegates to Engine.LatestFinishedSyncRecord; see that method for the predicate + tiebreaker contract.

func (*Adapter) ListEntitlements ¶

func (a *Adapter) ListEntitlements(ctx context.Context, req *v2.EntitlementsServiceListEntitlementsRequest) (*v2.EntitlementsServiceListEntitlementsResponse, error)

ListEntitlements returns up to page_size entitlements, optionally filtered by Resource (resource_type_id, resource_id). Pagination matches SQLite (see ListGrants).

func (*Adapter) ListEntitlementsByIds ¶

func (a *Adapter) ListEntitlementsByIds(
	ctx context.Context,
	req *reader_v2.EntitlementsReaderServiceListEntitlementsByIdsRequest,
) (*reader_v2.EntitlementsReaderServiceListEntitlementsByIdsResponse, error)

ListEntitlementsByIds returns entitlements for the requested external_ids. Missing rows are silently omitted.

func (*Adapter) ListGrants ¶

func (a *Adapter) ListGrants(ctx context.Context, req *v2.GrantsServiceListGrantsRequest) (*v2.GrantsServiceListGrantsResponse, error)

ListGrants returns up to page_size grants on the active sync. Pagination matches the SQLite engine's semantics:

• page_size == 0 || page_size > MaxPageSize → DefaultPageSize (10000)
• page_token is opaque base64; pass nextPageToken back verbatim
• filter by req.Resource — the entitlement-side resource of each grant — when set; uses the by_entitlement_resource index. This matches SQLite's `listGrantsGeneric` which filters on grants.resource_id / resource_type_id (the entitlement's resource columns). Callers who want to filter by principal should use ListGrantsForPrincipal instead.

func (*Adapter) ListGrantsForEntitlement ¶

func (a *Adapter) ListGrantsForEntitlement(
	ctx context.Context,
	req *reader_v2.GrantsReaderServiceListGrantsForEntitlementRequest,
) (*reader_v2.GrantsReaderServiceListGrantsForEntitlementResponse, error)

ListGrantsForEntitlement paginates grants on a specific entitlement, optionally narrowed by principal_id or principal_resource_type_ids. Implements reader_v2.GrantsReaderServiceServer.

func (*Adapter) ListGrantsForEntitlements ¶

func (a *Adapter) ListGrantsForEntitlements(
	ctx context.Context,
	req *reader_v2.GrantsReaderServiceListGrantsForEntitlementsRequest,
) (*reader_v2.GrantsReaderServiceListGrantsForEntitlementsResponse, error)

ListGrantsForEntitlements is the batched counterpart to ListGrantsForEntitlement (RFC §A4). It walks K entitlements in one RPC and returns a flat grant list ordered by visitation; callers group by Grant.Entitlement.Id.

Cursor encoding (Option C):

varint(entitlement_index) || varint(intra_cursor_len) ||
intra_cursor_bytes || crc32(list_checksum)

list_checksum is crc32 over the sorted entitlement IDs in the request — if the caller changes the entitlement list between pages we detect the mismatch and restart from the beginning rather than silently mis-paginate.

func (*Adapter) ListGrantsForPrincipal ¶

func (a *Adapter) ListGrantsForPrincipal(
	ctx context.Context,
	req *reader_v2.GrantsReaderServiceListGrantsForEntitlementRequest,
) (*reader_v2.GrantsReaderServiceListGrantsForEntitlementResponse, error)

ListGrantsForPrincipal is the Go-level convenience method that matches C1File.ListGrantsForPrincipal. It is NOT a gRPC RPC — the explorer / cel-search consumers reach C1File directly today. Adapter exposes the same shape for callers that take a typed store (refactor to a shared interface is tracked separately).

Semantically equivalent to ListGrantsForEntitlement(req) where the request carries a principal filter — the underlying PaginateGrantsByPrincipal index walk is what makes this O(K).

func (*Adapter) ListGrantsForResourceType ¶

func (a *Adapter) ListGrantsForResourceType(
	ctx context.Context,
	req *reader_v2.GrantsReaderServiceListGrantsForResourceTypeRequest,
) (*reader_v2.GrantsReaderServiceListGrantsForResourceTypeResponse, error)

ListGrantsForResourceType paginates grants whose principal is of the given resource_type_id, via idxGrantByPrincipalResourceType. The cursor is the index key.

Implements reader_v2.GrantsReaderServiceServer.

func (*Adapter) ListResourceTypes ¶

func (a *Adapter) ListResourceTypes(ctx context.Context, req *v2.ResourceTypesServiceListResourceTypesRequest) (*v2.ResourceTypesServiceListResourceTypesResponse, error)

ListResourceTypes returns up to page_size resource_types. Pagination matches SQLite (see ListGrants).

func (*Adapter) ListResources ¶

func (a *Adapter) ListResources(ctx context.Context, req *v2.ResourcesServiceListResourcesRequest) (*v2.ResourcesServiceListResourcesResponse, error)

ListResources returns up to page_size resources, optionally filtered by parent resource id and/or resource_type_id. Pagination matches SQLite (see ListGrants).

Note on the resource_type_id filter: when parent is also set, the by_parent index is used and we post-filter by resource_type_id (the index doesn't carry resource_type_id in the lookup prefix). When only resource_type_id is set, we still iterate the full primary range and post-filter — adding a by_resource_type index is a future-work item if this path becomes hot.

func (*Adapter) ListResourcesByIds ¶

func (a *Adapter) ListResourcesByIds(
	ctx context.Context,
	req *reader_v2.ResourcesReaderServiceListResourcesByIdsRequest,
) (*reader_v2.ResourcesReaderServiceListResourcesByIdsResponse, error)

ListResourcesByIds returns all resources matching the supplied (resource_type_id, resource_id) pairs. Missing rows are silently omitted. Callers detect partial misses by length comparison.

func (*Adapter) ListStaticEntitlements ¶

func (a *Adapter) ListStaticEntitlements(
	_ context.Context,
	_ *v2.EntitlementsServiceListStaticEntitlementsRequest,
) (*v2.EntitlementsServiceListStaticEntitlementsResponse, error)

ListStaticEntitlements is the always-empty counterpart to ListEntitlements that some connectors expose for static (compile- time-known) entitlements. C1File returns an empty list; the Pebble adapter mirrors that contract.

func (*Adapter) ListSyncs ¶

func (a *Adapter) ListSyncs(ctx context.Context, req *reader_v2.SyncsReaderServiceListSyncsRequest) (*reader_v2.SyncsReaderServiceListSyncsResponse, error)

ListSyncs paginates sync_run records across the engine. Order is the natural sync_id (KSUID) order — KSUIDs sort by timestamp, so this is also chronological. Implements reader_v2.SyncsReaderServiceServer.

func (*Adapter) Metadata ¶

func (a *Adapter) Metadata() connectorstore.StoreMetadata

Metadata describes the storage backing this adapter. The Pebble adapter always reports the v3 format; PayloadEncoding is set by the writer at envelope time and is not directly visible on the Adapter itself — registeredStore (in pkg/dotc1z/engine/pebble/ register.go) overrides this method to fill PayloadEncoding from its configured value.

Strings are inlined rather than referencing dotc1z constants because this subpackage is imported by dotc1z, so the reverse import would cycle. The values match dotc1z.EnginePebble.String() and dotc1z.C1ZFormatV3.String() — see connectorstore.StoreMetadata docs for the canonical value list.

func (*Adapter) OutputFilepath ¶

func (a *Adapter) OutputFilepath() (string, error)

OutputFilepath returns the on-disk path the adapter writes to. The SQLite equivalent returns the .c1z file path; for Pebble we don't have a single file, so this returns the engine's directory. Used by tooling that wants to copy or inspect the storage.

func (*Adapter) PutAsset ¶

func (a *Adapter) PutAsset(ctx context.Context, assetRef *v2.AssetRef, contentType string, data []byte) error

PutAsset writes a single asset row. assetRef carries the (resource_type, resource_id) pair we use as the external_id — joined with a "/" separator since the engine's AssetRecord PK is (sync_id, external_id).

func (*Adapter) PutEntitlements ¶

func (a *Adapter) PutEntitlements(ctx context.Context, entitlements ...*v2.Entitlement) error

PutEntitlements writes a batch of entitlements in a single Pebble batch.

func (*Adapter) PutGrants ¶

func (a *Adapter) PutGrants(ctx context.Context, grants ...*v2.Grant) error

PutGrants writes a batch of grants in a single Pebble batch. v2 is translated to v3 first; the engine then commits the whole batch with one fsync (or NoSync during a fresh sync — see MarkFreshSync).

The translation uses per-shard arenas (grantTranslateArena) so the 3 × N proto-struct allocations from V2GrantToV3's builder pattern collapse to 3 slice allocations per shard. For large fresh-sync writes this substantially reduces GC scan pressure during the engine's parallel build phase.

The translation itself runs in parallel across translateShards workers when the input is large enough — protobuf Get/Set methods on the underlying v2.Grant and v3 arena structs are thread-safe for read+arena-private-write access patterns. Each worker writes to a disjoint range of the records slice and uses its own arena, so no shared mutable state across workers.

func (*Adapter) PutResourceTypes ¶

func (a *Adapter) PutResourceTypes(ctx context.Context, rts ...*v2.ResourceType) error

PutResourceTypes writes a batch of resource types in a single Pebble batch.

func (*Adapter) PutResources ¶

func (a *Adapter) PutResources(ctx context.Context, resources ...*v2.Resource) error

PutResources writes a batch of resources in a single Pebble batch.

func (*Adapter) ResumeSync ¶

func (a *Adapter) ResumeSync(ctx context.Context, syncType connectorstore.SyncType, syncID string) (string, error)

ResumeSync attaches to an existing sync_run by id. Returns the caller-provided id if it matches an existing record.

func (*Adapter) SetCurrentSync ¶

func (a *Adapter) SetCurrentSync(ctx context.Context, syncID string) error

SetCurrentSync rebinds the engine's current sync without creating a new SyncRunRecord. Used by callers that previously called StartNewSync/ResumeSync.

func (*Adapter) StartNewSync ¶

func (a *Adapter) StartNewSync(ctx context.Context, syncType connectorstore.SyncType, parentSyncID string) (string, error)

StartNewSync creates a new sync_run record. Returns the new sync_id.

func (*Adapter) StartOrResumeSync ¶

func (a *Adapter) StartOrResumeSync(ctx context.Context, syncType connectorstore.SyncType, syncID string) (string, bool, error)

StartOrResumeSync resumes if syncID-like state exists for the given type, else starts a new sync. Returns (id, started_new, err).

func (*Adapter) Stats ¶

func (a *Adapter) Stats(ctx context.Context, syncType connectorstore.SyncType, syncID string) (map[string]int64, error)

Stats returns a map of {table_name: row_count} for the given sync. Used by pkg/sync/progresslog to show the "syncing X resources, Y entitlements, Z grants" progress lines. The SQLite engine answers this with a fast COUNT(*) per table; for Pebble we iterate each per-record-type bucket and count keys in the [sync_lo, sync_hi) range. Counts are exact (not estimates).

Mirrors connectorstore semantics: syncType == SyncTypeAny accepts any sync; otherwise we filter by the sync_run's type.

func (*Adapter) StreamEntitlements ¶

func (a *Adapter) StreamEntitlements(
	ctx context.Context,
	syncID string,
) iter.Seq2[*v2.Entitlement, error]

StreamEntitlements yields all entitlements for syncID. Implements connectorstore.StreamingReader.

func (*Adapter) StreamGrants ¶

func (a *Adapter) StreamGrants(
	ctx context.Context,
	syncID string,
	opts connectorstore.StreamGrantsOptions,
) iter.Seq2[*v2.Grant, error]

StreamGrants yields grants for syncID, optionally narrowed by opts. Implements connectorstore.StreamingReader.

func (*Adapter) StreamResources ¶

func (a *Adapter) StreamResources(
	ctx context.Context,
	syncID string,
	opts connectorstore.StreamResourcesOptions,
) iter.Seq2[*v2.Resource, error]

StreamResources yields resources for syncID, optionally narrowed by resource_type. Implements connectorstore.StreamingReader.

func (*Adapter) SyncMeta ¶

func (a *Adapter) SyncMeta() dotc1z.SyncMeta

SyncMeta returns the SyncMeta sub-store backed by the Pebble adapter. Implements dotc1z.C1ZStore.SyncMeta(). The sync-run records live in Pebble under typeSyncRun; methods here walk IterateAllSyncRuns + GetSyncRunRecord + PutSyncRunRecord on the engine.

func (*Engine) CheckpointTo ¶

func (e *Engine) CheckpointTo(ctx context.Context, destDir string) error

CheckpointTo writes a self-contained Pebble directory snapshot to destDir. destDir must not exist yet. Pebble creates it and hard-links SSTs where possible.

The source engine stays writable after CheckpointTo returns; this is the building block dotc1z's higher-level Save wraps with the v3 envelope format.

func (*Engine) Close ¶

func (e *Engine) Close() error

Close shuts down the engine. After Close, all methods return ErrEngineClosing. If the engine was Quiesce'd first, the in-flight writes have already drained; otherwise Close blocks until they complete.

func (*Engine) CompactSyncRanges ¶

func (e *Engine) CompactSyncRanges(ctx context.Context, syncID string) error

CompactSyncRanges runs pebble.Compact over each sync-scoped range for syncID. Called after DeleteSyncData to reclaim disk space from the tombstones DeleteRange leaves behind. Without this, the next checkpoint would still include the deleted bytes.

Compaction is bounded per-range, not engine-wide, so the cost scales with the data being pruned rather than the engine's total size. Errors are best-effort: a Compact failure on one range doesn't block the others — pebble retries compaction in the background and the next Cleanup call will re-attempt.

ctx is honored between ranges; if the caller cancels mid-pass we surface ctx.Err() verbatim so the orchestrator (registeredStore .Cleanup) can flip the sync to ErrSyncNotComplete.

func (*Engine) CurrentDBSizeBytes ¶

func (e *Engine) CurrentDBSizeBytes() (int64, error)

CurrentDBSizeBytes returns the total size of regular files in the Pebble database directory. This is the Pebble equivalent of C1File's SQLite DBSizeProvider capability: it reports the uncompressed working set on disk, including WAL/log, MANIFEST, OPTIONS, and SST files currently present.

func (*Engine) DB ¶

func (e *Engine) DB() *pebble.DB

DB returns the underlying *pebble.DB. Exported for the synccompactor/pebble package; callers must not Close it directly (use Engine.Close) and must respect the engine's lifecycle. Returns nil after Close.

func (*Engine) DBDir ¶

func (e *Engine) DBDir() string

DBDir returns the on-disk path the engine writes to. Exported so the Adapter can implement OutputFilepath / CurrentDBSizeBytes.

func (*Engine) DeleteAssetRecord ¶

func (e *Engine) DeleteAssetRecord(ctx context.Context, syncID, externalID string) error

func (*Engine) DeleteEntitlementRecord ¶

func (e *Engine) DeleteEntitlementRecord(ctx context.Context, syncID, externalID string) error

func (*Engine) DeleteGrantRecord ¶

func (e *Engine) DeleteGrantRecord(ctx context.Context, syncID, externalID string) error

DeleteGrantRecord removes a grant and its index entries.

func (*Engine) DeleteResourceRecord ¶

func (e *Engine) DeleteResourceRecord(ctx context.Context, syncID, resourceTypeID, resourceID string) error

func (*Engine) DeleteResourceTypeRecord ¶

func (e *Engine) DeleteResourceTypeRecord(ctx context.Context, syncID, externalID string) error

func (*Engine) DeleteSyncData ¶

func (e *Engine) DeleteSyncData(ctx context.Context, syncID string) error

DeleteSyncData removes every key scoped to syncID from the engine. This is the Pebble counterpart to the SQLite Cleanup path's per-table DELETE WHERE sync_id = ? loop.

Refuses to operate on a sync that is currently being written. "Currently being written" means the engine is in the fresh-sync fast path (between MarkFreshSync and EndFreshSync) for this sync_id. The mere fact that e.currentSync still points at a previously-ended sync is not enough to lock it out — EndFreshSync intentionally leaves currentSync set so Reader-side empty-syncID reads keep working until SetCurrentSync rebinds.

Each DeleteRange call writes a single tombstone to the memtable (~µs), so we deliberately do NOT check ctx between ranges. A partial DeleteSyncData would leave some keyspaces tombstoned and others not — a "half-deleted sync" the next Cleanup pass would have to re-tombstone. Callers that need ctx-aware cancellation should check ctx around the DeleteSyncData call, not inside it.

After this returns the keys are tombstoned but the physical bytes remain in the LSM until compaction. Pair with CompactSyncRanges (or one broader Compact call) to actually reclaim disk before CheckpointTo.

func (*Engine) DeleteSyncRunRecord ¶

func (e *Engine) DeleteSyncRunRecord(ctx context.Context, syncID string) error

func (*Engine) EndFreshSync ¶

func (e *Engine) EndFreshSync(ctx context.Context) error

EndFreshSync clears the fresh-sync flag and flushes the memtable + fsyncs the WAL so the data written during the sync is on disk before the caller returns. Called by Adapter.EndSync.

func (*Engine) Flush ¶

func (e *Engine) Flush(ctx context.Context) error

Flush forces the engine's memtable to disk. Exported so the cleanup orchestration can ensure tombstones are durable before the next checkpoint reads the LSM.

This is a thin wrapper over pebble.DB.Flush + a WAL fsync; the EndFreshSync path uses the same combination at sync end. ctx is checked before the blocking calls so a cancelled deadline doesn't trigger a full memtable flush.

func (*Engine) GetAssetRecord ¶

func (e *Engine) GetAssetRecord(ctx context.Context, syncID, externalID string) (*v3.AssetRecord, error)

func (*Engine) GetEntitlementRecord ¶

func (e *Engine) GetEntitlementRecord(ctx context.Context, syncID, externalID string) (*v3.EntitlementRecord, error)

func (*Engine) GetGrantRecord ¶

func (e *Engine) GetGrantRecord(ctx context.Context, syncID, externalID string) (*v3.GrantRecord, error)

GetGrantRecord fetches a grant record by sync_id + external_id. syncID may be empty to use the engine's currently-set sync.

func (*Engine) GetResourceRecord ¶

func (e *Engine) GetResourceRecord(ctx context.Context, syncID, resourceTypeID, resourceID string) (*v3.ResourceRecord, error)

func (*Engine) GetResourceTypeRecord ¶

func (e *Engine) GetResourceTypeRecord(ctx context.Context, syncID, externalID string) (*v3.ResourceTypeRecord, error)

func (*Engine) GetSyncRunRecord ¶

func (e *Engine) GetSyncRunRecord(ctx context.Context, syncID string) (*v3.SyncRunRecord, error)

func (*Engine) IsFreshSync ¶

func (e *Engine) IsFreshSync() bool

IsFreshSync reports whether the engine is in the fresh-sync write path (set by MarkFreshSync).

func (*Engine) IterateAllSyncRuns ¶

func (e *Engine) IterateAllSyncRuns(ctx context.Context, yield func(*v3.SyncRunRecord) bool) error

IterateAllSyncRuns iterates every sync_run record in the engine. Used by callers that want "what syncs do I have available?".

func (*Engine) IterateAssetsBySync ¶

func (e *Engine) IterateAssetsBySync(ctx context.Context, syncID string, yield func(*v3.AssetRecord) bool) error

func (*Engine) IterateEntitlementsByResource ¶

func (e *Engine) IterateEntitlementsByResource(ctx context.Context, syncID, resourceTypeID, resourceID string, yield func(*v3.EntitlementRecord) bool) error

func (*Engine) IterateEntitlementsBySync ¶

func (e *Engine) IterateEntitlementsBySync(ctx context.Context, syncID string, yield func(*v3.EntitlementRecord) bool) error

func (*Engine) IterateGrantsByEntitlement ¶

func (e *Engine) IterateGrantsByEntitlement(ctx context.Context, syncID, entitlementID string, yield func(*v3.GrantRecord) bool) error

IterateGrantsByEntitlement iterates the by_entitlement index for the given entitlement_id, yielding each grant in encoded principal- key order. yield returns false to stop.

func (*Engine) IterateGrantsByNeedsExpansion ¶

func (e *Engine) IterateGrantsByNeedsExpansion(ctx context.Context, syncID string, yield func(*v3.GrantRecord) bool) error

IterateGrantsByNeedsExpansion iterates the needs_expansion index for a sync, yielding each grant whose NeedsExpansion flag is currently set. yield returns false to stop.

Pebble-equivalent of the SQLite partial index `WHERE needs_expansion = 1`. Backs PendingExpansionPage on the grant store.

func (*Engine) IterateGrantsByPrincipal ¶

func (e *Engine) IterateGrantsByPrincipal(ctx context.Context, syncID, principalRT, principalID string, yield func(*v3.GrantRecord) bool) error

IterateGrantsByPrincipal iterates the by_principal index.

func (*Engine) IterateGrantsByPrincipalResourceType ¶

func (e *Engine) IterateGrantsByPrincipalResourceType(ctx context.Context, syncID, principalRT string, yield func(*v3.GrantRecord) bool) error

IterateGrantsByPrincipalResourceType iterates the by-principal-RT index for a sync. Yields each grant whose principal carries the given resource_type, in encoded external_id order. Stops when yield returns false.

func (*Engine) IterateGrantsBySync ¶

func (e *Engine) IterateGrantsBySync(ctx context.Context, syncID string, yield func(*v3.GrantRecord) bool) error

IterateGrantsBySync iterates all grants in a sync in primary-key order. yield returns false to stop iteration.

func (*Engine) IterateResourceTypesBySync ¶

func (e *Engine) IterateResourceTypesBySync(ctx context.Context, syncID string, yield func(*v3.ResourceTypeRecord) bool) error

func (*Engine) IterateResourcesByParent ¶

func (e *Engine) IterateResourcesByParent(ctx context.Context, syncID, parentRT, parentID string, yield func(*v3.ResourceRecord) bool) error

func (*Engine) IterateResourcesBySync ¶

func (e *Engine) IterateResourcesBySync(ctx context.Context, syncID string, yield func(*v3.ResourceRecord) bool) error

func (*Engine) LatestFinishedSyncRecord ¶

func (e *Engine) LatestFinishedSyncRecord(ctx context.Context, typeOK func(v3.SyncType) bool) (*v3.SyncRunRecord, error)

LatestFinishedSyncRecord returns the SyncRunRecord with the latest ended_at that matches typeOK, or (nil, nil) when no finished sync matches. typeOK may be nil to accept any sync type.

This is the single source of truth for "pick the latest finished sync" on the Pebble engine; all three external entry points (connectorstore.LatestFinishedSyncIDFetcher, dotc1z.SyncMeta.LatestFullSync / LatestFinishedSyncOfAnyType, and reader_v2.SyncsReaderService.GetLatestFinishedSync) call here so the tiebreaker and predicate semantics stay consistent.

Ties on ended_at are broken by sync_id (KSUIDs sort by time, so the lexicographically greater id is the later sync). Matches the SQLite-side `ORDER BY ended_at DESC, sync_id DESC` in pkg/dotc1z/sync_runs.go:getFinishedSync (commit 1627b047) which closes a Windows coarse-time-resolution race where adjacent EndSync calls can produce identical ended_at timestamps.

O(N) in the count of sync_runs (one record per sync; typically a few hundred over a c1z's lifetime, not millions).

func (*Engine) LatestUnfinishedSyncRecord ¶

func (e *Engine) LatestUnfinishedSyncRecord(ctx context.Context, typeOK func(v3.SyncType) bool) (*v3.SyncRunRecord, error)

LatestUnfinishedSyncRecord returns the most-recently-started sync_run that has NOT ended and was started within the last week, matching typeOK (nil = any type), or (nil, nil) when none match.

Mirrors SQLite's getLatestUnfinishedSync (pkg/dotc1z/sync_runs.go): it selects ended_at-null records started after the unfinishedSyncMaxAge cutoff, ordered by started_at (sync_id tiebreaker for coarse clock resolution). It is the read-path fallback after LatestFinishedSyncRecord, so a c1z whose only sync was interrupted before EndSync still resolves to that in-progress sync instead of returning no sync at all — matching the SQLite resolveSyncIDForRead cascade.

O(N) in the count of sync_runs (one record per sync).

func (*Engine) MarkFreshSync ¶

func (e *Engine) MarkFreshSync(syncID string) error

MarkFreshSync sets currentSync AND flags the sync as freshly started (no prior records under this sync_id). The engine then takes the perf-fast write path: pebble.NoSync per commit and skip read-before-write index cleanup. The host crash semantics match SQLite's PRAGMA synchronous=NORMAL — the connector is the source of truth during the sync; a crash forces re-sync rather than silent data loss.

Callers should call EndFreshSync (via Flush) at sync end to harden the data with a single fsync.

func (*Engine) PaginateEntitlementsByResource ¶

func (e *Engine) PaginateEntitlementsByResource(
	ctx context.Context, syncID, resourceTypeID, resourceID, cursor string, limit int,
) ([]*v3.EntitlementRecord, string, error)

PaginateEntitlementsByResource uses the by_resource index.

func (*Engine) PaginateEntitlementsBySync ¶

func (e *Engine) PaginateEntitlementsBySync(
	ctx context.Context, syncID, cursor string, limit int,
) ([]*v3.EntitlementRecord, string, error)

PaginateEntitlementsBySync returns a page of entitlements in primary key order.

func (*Engine) PaginateGrantsByEntitlement ¶

func (e *Engine) PaginateGrantsByEntitlement(
	ctx context.Context, syncID, entitlementID, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsByEntitlement uses the by_entitlement index. The index keys carry the principal-rt/principal-id/external-id tail; each match triggers a secondary primary-key Get to materialize the full grant. Cursor is the index key, not the primary key.

func (*Engine) PaginateGrantsByEntitlementResource ¶

func (e *Engine) PaginateGrantsByEntitlementResource(
	ctx context.Context, syncID, entRT, entRID, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsByEntitlementResource walks the by_entitlement_resource index — all grants in `syncID` whose entitlement's resource is (entRT, entRID). Cursor is the index key.

Drives Adapter.ListGrants and ListWithAnnotationsForResourcePage when req.Resource is set, matching SQLite's `listGrantsGeneric` filter on grants.resource_id / resource_type_id (the entitlement- side resource columns). The pre-existing Pebble path used PaginateGrantsByPrincipal here, which returned empty for the common "grants on this group" semantic.

func (*Engine) PaginateGrantsByNeedsExpansion ¶

func (e *Engine) PaginateGrantsByNeedsExpansion(
	ctx context.Context, syncID, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsByNeedsExpansion returns a page of grants whose NeedsExpansion flag is set. Backs the GrantStore PendingExpansionPage path; the SQLite equivalent is a query guarded by the partial index `WHERE needs_expansion = 1`.

Cursor is the needs_expansion index key.

func (*Engine) PaginateGrantsByPrincipal ¶

func (e *Engine) PaginateGrantsByPrincipal(
	ctx context.Context, syncID, principalRT, principalID, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsByPrincipal uses the by_principal index. Same shape as PaginateGrantsByEntitlement.

func (*Engine) PaginateGrantsByPrincipalResourceType ¶

func (e *Engine) PaginateGrantsByPrincipalResourceType(
	ctx context.Context, syncID, principalRT, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsByPrincipalResourceType walks the by-principal-RT index. Cursor is the index key. Drives the new fast path for the Adapter's ListGrantsForResourceType.

func (*Engine) PaginateGrantsBySync ¶

func (e *Engine) PaginateGrantsBySync(
	ctx context.Context, syncID, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsBySync returns up to `limit` grants from the primary-key range, starting strictly after `cursor`. Returns the next cursor (empty if no more) plus the materialized records.

Uses a per-call grantReadArena to pre-allocate the GrantRecord + nested EntitlementRef + PrincipalRef slots. The vtproto-generated UnmarshalVT detects the pre-set pointers and reuses the arena slots instead of heap-allocating a fresh nested struct per record, collapsing 2N nested allocs into 2 slice allocs per page. The arena lifetime ends with this function — callers receive pointers into the arena's backing arrays, so retention is intentional.

func (*Engine) PaginateResourceTypesBySync ¶

func (e *Engine) PaginateResourceTypesBySync(
	ctx context.Context, syncID, cursor string, limit int,
) ([]*v3.ResourceTypeRecord, string, error)

PaginateResourceTypesBySync returns a page of resource_types.

func (*Engine) PaginateResourcesByParent ¶

func (e *Engine) PaginateResourcesByParent(
	ctx context.Context, syncID, parentRT, parentID, cursor string, limit int,
) ([]*v3.ResourceRecord, string, error)

PaginateResourcesByParent uses the by_parent index.

func (*Engine) PaginateResourcesBySync ¶

func (e *Engine) PaginateResourcesBySync(
	ctx context.Context, syncID, cursor string, limit int,
) ([]*v3.ResourceRecord, string, error)

PaginateResourcesBySync returns a page of resources in primary key order.

func (*Engine) PersistSyncStats ¶

func (e *Engine) PersistSyncStats(ctx context.Context, syncID string) error

PersistSyncStats computes and writes the stats sidecar for syncID. Exposed for the EndFreshSync caller and the on-Open migration backfill.

func (*Engine) PutAssetRecord ¶

func (e *Engine) PutAssetRecord(ctx context.Context, r *v3.AssetRecord) error

PutAssetRecord — assets have no secondary indexes.

func (*Engine) PutEntitlementRecord ¶

func (e *Engine) PutEntitlementRecord(ctx context.Context, r *v3.EntitlementRecord) error

PutEntitlementRecord writes an entitlement + its by_resource index.

func (*Engine) PutEntitlementRecords ¶

func (e *Engine) PutEntitlementRecords(ctx context.Context, records ...*v3.EntitlementRecord) error

PutEntitlementRecords writes N entitlements in two pebble.Batches — primary keys in one, by_resource index keys in the other. Mirrors the PutGrantRecords pattern (RFC §3a Tier-B/C):

• within-call dedup pre-pass keyed by (sync, external_id) drops earlier occurrences;
• the first PutEntitlementRecords call of a fresh sync skips the read-before-write Get (keyspace provably empty);
• subsequent calls fall back to read-before-write so cross-call duplicates can clean up the prior call's index entries.

func (*Engine) PutEntitlementRecordsIfNewer ¶

func (e *Engine) PutEntitlementRecordsIfNewer(ctx context.Context, records ...*v3.EntitlementRecord) error

PutEntitlementRecordsIfNewer writes entitlements only when newer.

func (*Engine) PutGrantRecord ¶

func (e *Engine) PutGrantRecord(ctx context.Context, r *v3.GrantRecord) error

PutGrantRecord writes a grant record + its by_entitlement and by_principal index entries, atomically in a single pebble.Batch.

This is the engine's canonical write path; other record types follow the same shape (read the previous primary if any → delete its index entries → write the new primary → write the new index entries → commit).

func (*Engine) PutGrantRecords ¶

func (e *Engine) PutGrantRecords(ctx context.Context, records ...*v3.GrantRecord) error

PutGrantRecords writes N grants in two pebble.Batches — one for primary-key writes, one for index-key writes — and commits them sequentially. This is the bulk path the adapter's PutGrants uses.

Mutation safety. Connectors can legitimately emit the same external_id twice within a single sync (paginated sources, deduplication bugs in upstream APIs, etc.). To prevent orphan index entries from earlier duplicates, we pre-scan records to find the latest occurrence of each (sync_id, external_id) tuple and process only those — earlier duplicates are dropped before any batch byte is written. db.Get doesn't see in-batch writes either way, so this dedup pass is the load-bearing safety net that neither the old read-before-write path nor a pure skip-Get path provides.

Read-before-write index cleanup. On a NON-fresh sync the engine must Get the prior primary value so it can delete the index keys the previous sync wrote (entitlement/principal can change between syncs). On a FRESH sync the keyspace under this sync is empty by construction (MarkFreshSync ranges-delete it) so the Get is guaranteed to return ErrNotFound and we skip it — saves an LSM lookup per record on the bulk path.

Two batches. The primary keys arrive in roughly sorted order from the adapter (V2→V3 translator preserves the connector's record order, and most connectors emit grants clustered by entitlement which tends to cluster external_ids). The index keys are sorted on a totally different field (entitlement_id / principal), so interleaving them in one batch makes Pebble's flushable-batch promotion path quote sort. Splitting them lets each batch's natural order survive.

Fresh-sync still uses pebble.NoSync — EndFreshSync does one Flush+fsync at sync end to harden the data.

func (*Engine) PutGrantRecordsIfNewer ¶

func (e *Engine) PutGrantRecordsIfNewer(ctx context.Context, records ...*v3.GrantRecord) error

PutGrantRecordsIfNewer writes records that are strictly newer than the stored copy. Records without a discovered_at are treated as "always write" (caller is asserting freshness explicitly).

func (*Engine) PutResourceRecord ¶

func (e *Engine) PutResourceRecord(ctx context.Context, r *v3.ResourceRecord) error

PutResourceRecord writes a resource record + its by_parent index entry (if the resource has a parent). Read-before-write index cleanup follows the canonical pattern from grants.go.

func (*Engine) PutResourceRecords ¶

func (e *Engine) PutResourceRecords(ctx context.Context, records ...*v3.ResourceRecord) error

PutResourceRecords writes N resources in two pebble.Batches — primary keys in one, by_parent index keys in the other. Mirrors the PutGrantRecords pattern (RFC §3a Tier-B/C):

• within-call dedup pre-pass keyed by (sync, rt, res_id) drops earlier occurrences of repeated resources;
• the first PutResourceRecords call of a fresh sync skips the read-before-write Get (keyspace provably empty);
• subsequent calls fall back to read-before-write so cross-call duplicates can clean up index entries the prior call wrote.

func (*Engine) PutResourceRecordsIfNewer ¶

func (e *Engine) PutResourceRecordsIfNewer(ctx context.Context, records ...*v3.ResourceRecord) error

PutResourceRecordsIfNewer writes resources only when the incoming discovered_at is strictly newer than the stored copy.

func (*Engine) PutResourceTypeRecord ¶

func (e *Engine) PutResourceTypeRecord(ctx context.Context, r *v3.ResourceTypeRecord) error

PutResourceTypeRecord writes a resource_type record. No secondary indexes, so this is a single-key write.

func (*Engine) PutResourceTypeRecords ¶

func (e *Engine) PutResourceTypeRecords(ctx context.Context, records ...*v3.ResourceTypeRecord) error

PutResourceTypeRecords writes N resource_types in one batch. Fresh-sync uses pebble.NoSync (one fsync at EndFreshSync).

func (*Engine) PutResourceTypeRecordsIfNewer ¶

func (e *Engine) PutResourceTypeRecordsIfNewer(ctx context.Context, records ...*v3.ResourceTypeRecord) error

PutResourceTypeRecordsIfNewer writes resource_types only when newer.

func (*Engine) PutSyncRunRecord ¶

func (e *Engine) PutSyncRunRecord(ctx context.Context, r *v3.SyncRunRecord) error

PutSyncRunRecord — one row per sync. sync_id is the only PK component; the record reads/writes use only the sync_id.

Unlike other writer methods this does NOT consult resolveSyncBytes fallback — a SyncRunRecord must always carry its own sync_id (since it's *defining* what's a valid sync, not operating under one).

func (*Engine) Quiesce ¶

func (e *Engine) Quiesce(ctx context.Context) error

Quiesce is the strict write-barrier from RFC v4 §3.7. After Quiesce returns, all subsequent Writer method calls return ErrEngineQuiesced. In-flight writes drain via the engine's WaitGroup. Concurrent callers serialize on closeMu; the second caller observes a fully-quiesced engine.

Idempotent: calling Quiesce twice is a no-op the second time.

There is no Unquiesce. Callers that want to keep writing open a new engine instance via Open.

func (*Engine) Save ¶

func (e *Engine) Save(ctx context.Context, dest string) error

func (*Engine) SetCurrentSync ¶

func (e *Engine) SetCurrentSync(syncID string) error

SetCurrentSync sets the engine's tracked current sync_id from a string KSUID. Subsequent Put*/List* calls with an empty syncID use this value. Clears the freshSync flag — a bare SetCurrentSync is conservative (treats the sync as resumable, so writes keep fsync + read-before-write).