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 pkg/dotc1z's Pebble store wrapper (pebble_store.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) CleanupCandidates ¶ added in v0.13.0

func (a *Adapter) CleanupCandidates(ctx context.Context) ([]c1zstore.SyncRun, error)

CleanupCandidates walks every sync_run record and projects it into the engine-neutral c1zstore.SyncRun shape, sorted oldest-first. c1zstore.SelectSyncsToDelete depends on this ordering so "drop the oldest overflow" trims the right end. Used by pkg/dotc1z's Pebble store to drive the retention policy at Cleanup.

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) CurrentSyncID ¶ added in v0.13.0

func (a *Adapter) CurrentSyncID() string

CurrentSyncID returns the adapter's current sync id, or "" when no sync is active. Used by pkg/dotc1z's Pebble store to drive the retention policy at Cleanup.

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() c1zstore.FileOps

FileOps returns the FileOps sub-store backed by the Pebble adapter. Implements c1zstore.Store.FileOps(). CloneSync materializes the single sync's data into a fresh c1z (used by `baton clone`); GenerateSyncDiff is unsupported (single-sync contract — see ErrDiffUnsupported).

func (*Adapter) FileOpsWithEncoding ¶

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

FileOpsWithEncoding returns a FileOps that emits new c1z files using the given payload encoding. Used by the registered store in pkg/dotc1z 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() c1zstore.GrantStore

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

needs_expansion is populated at PutGrants time: V2GrantToV3 extracts the GrantExpandable annotation and sets NeedsExpansion, which keys the idxGrantByNeedsExpansion index. The *Page methods walk that index and return real expandable rows (see TestExpansionAnnotationRoundtrip and the live PendingExpansion path).

func (*Adapter) GrantsForEntitlementPrincipalSorted ¶ added in v0.15.3

func (a *Adapter) GrantsForEntitlementPrincipalSorted() bool

GrantsForEntitlementPrincipalSorted reports that ListGrantsForEntitlement yields grants in non-decreasing principal (resource_type, resource) order. The by_entitlement index is keyed entitlement_id|principal_rt|principal_id|external_id, and pagination walks it in key order, so consumers can group by principal without buffering and sorting the whole entitlement.

func (*Adapter) InitCurrentSync ¶ added in v0.13.6

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

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) ListGrantPrincipalKeysForEntitlement ¶ added in v0.12.3

func (a *Adapter) ListGrantPrincipalKeysForEntitlement(
	ctx context.Context,
	entitlement *v2.Entitlement,
	pageToken string,
	pageSize uint32,
) ([]string, string, error)

ListGrantPrincipalKeysForEntitlement returns the compact principal keys used by grant expansion prefetch. It avoids materializing full grant records when the caller only needs to know which principals already have a descendant entitlement grant.

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) ListSyncRuns ¶ added in v0.15.1

func (a *Adapter) ListSyncRuns(ctx context.Context, pageToken string, pageSize uint32) ([]*c1zstore.SyncRun, string, error)

ListSyncRuns returns every sync-run record projected into the engine-neutral c1zstore.SyncRun shape, oldest-first (parent before child for a well-formed chain), in a single page. A v3 Pebble c1z holds exactly one sync, so pagination is trivial: the whole set is returned on the first call and the next-page token is always empty. pageToken and pageSize are accepted for parity with the SQLite ListSyncRuns and are not used. It backs the c1z sanitizer's source-side sync-graph-metadata read (linked_sync_id, supports_diff), which the gRPC reader surface does not carry.

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 — pkg/dotc1z's Pebble store wrapper (pebble_store.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) PebbleEngine ¶ added in v0.13.0

func (a *Adapter) PebbleEngine() *Engine

PebbleEngine returns the underlying *Engine. Nil-safe so AsEngine can probe arbitrary writers.

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 under a freshly-minted sync_id. Returns the new sync_id.

func (*Adapter) StartNewSyncWithID ¶ added in v0.15.0

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

StartNewSyncWithID is StartNewSync but adopts the caller-supplied syncID instead of minting one. It exists for conversion/compaction (e.g. ToPebble) that must preserve the source sync's identity so the produced file's sync_id matches the snapshot it was derived from. syncID must be non-empty.

func (*Adapter) StartOrResumeSync ¶

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

StartOrResumeSync resumes an existing sync if one is resumable, else starts a new sync. Returns (id, started_new, err).

Resume precedence mirrors SQLite's StartOrResumeSync→ResumeSync cascade (pkg/dotc1z/sync_runs.go):

1. A caller-supplied syncID that names an existing sync_run.
2. With an empty syncID, the latest in-progress (not-yet-ended) sync of the requested type started within the last week — the same predicate as Engine.LatestUnfinishedSyncRecord. This is the path the syncer drives across activity windows: it calls StartOrResumeSync(ctx, syncType, "") with no id, expecting an interrupted sync to resume where it checkpointed. Without this, every window started a brand-new sync (wiping prior data via ResetForNewSync and resetting the FSM to InitOp), so any sync exceeding one window never finished.

Only when nothing is resumable do we start a new sync.

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() c1zstore.SyncMeta

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

func (*Adapter) UnsafePutUniqueGrants ¶ added in v0.12.2

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

UnsafePutUniqueGrants writes grants on the trusted-import path: records are encoded in parallel and written unconditionally, with no read-before-write and no dedup pass. Do not use it for live connector output. The destination sync must be fresh, and the caller MUST guarantee each external_id appears at most once across the whole sync (not just within this batch). Live connector writes should use PutGrants.

func (*BulkGrantShard) AddGrants ¶ added in v0.13.7

func (s *BulkGrantShard) AddGrants(ctx context.Context, grants ...*v2.Grant) error

AddGrants translates and appends grants to this shard, in ascending external-id order (see BulkGrantShard). Translation, the v3 marshal, and key encoding run in the caller's goroutine; all writers are shard-private, so concurrent shards never contend.

func (*BulkGrantShard) Close ¶ added in v0.13.7

func (s *BulkGrantShard) Close()

Close finalizes the shard's primary SST and flushes its index tail chunks into the background sorters. Must be called once per shard before Finish; AddGrants is invalid afterwards.

func (*BulkSyncImport) Abort ¶ added in v0.13.7

func (b *BulkSyncImport) Abort()

Abort closes and removes all staged files without ingesting. Safe to call after Finish (no-op) or on a partially-failed import.

func (*BulkSyncImport) AddEntitlements ¶ added in v0.13.7

func (b *BulkSyncImport) AddEntitlements(ctx context.Context, ents ...*v2.Entitlement) error

AddEntitlements translates and appends entitlements, which must arrive sorted by external id. by_resource index keys are derived and spilled with the same resource guard as writeEntitlementIndexes.

func (*BulkSyncImport) AddResourceTypes ¶ added in v0.13.7

func (b *BulkSyncImport) AddResourceTypes(ctx context.Context, rts ...*v2.ResourceType) error

AddResourceTypes translates and appends resource types, which must arrive sorted by external id.

func (*BulkSyncImport) AddResources ¶ added in v0.13.7

func (b *BulkSyncImport) AddResources(ctx context.Context, resources ...*v2.Resource) error

AddResources translates and appends resources, which must arrive sorted by (resource_type_id, resource_id). by_parent index keys are derived and spilled with the same parent guard as writeResourceIndexes.

func (*BulkSyncImport) ComputedStats ¶ added in v0.13.7

func (b *BulkSyncImport) ComputedStats() *v3.SyncStatsRecord

ComputedStats returns a SyncStatsRecord built from the counts the import accumulated while streaming (primary record counts, resources by resource type, entitlements by resource type, grants by entitlement resource type). Valid after all grant shards are Closed. Assets do not flow through the importer — the caller sets that count itself before stashing the record via Engine.StashComputedSyncStats.

func (*BulkSyncImport) Finish ¶ added in v0.13.7

func (b *BulkSyncImport) Finish(ctx context.Context) error

Finish finalizes the import: every spill sorter drains its background sorts, each index family k-way merges its sorted runs (combined across all shards) into one SST in parallel, the ordered primary SSTs (including each shard's grant SST) are collected, and everything is ingested in one pebble Ingest call (the key ranges are pairwise disjoint, so a single call is legal and costs one manifest rotation). The staging directory is removed on the way out regardless of outcome.

func (*BulkSyncImport) NewGrantShard ¶ added in v0.13.7

func (b *BulkSyncImport) NewGrantShard() (*BulkGrantShard, error)

NewGrantShard registers a new grant import shard. Safe to call concurrently.

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; writes are only blocked while the checkpoint is cut. This is the building block dotc1z's higher-level Save wraps with the v3 envelope format.

Read-only engines cannot call pebble.DB.Checkpoint (it copies OPTIONS via d.optionsFileNum, which Pebble never populates on read-only open). Those engines clone the on-disk tree with vfs.Clone instead.

The explicit Flush is what makes the snapshot WAL-independent: every committed write lands in SSTs before the checkpoint is cut. We deliberately do NOT pass pebble.WithFlushedWAL() — it would be redundant after the flush, and it appends a WAL record, guaranteeing the checkpoint carries a WAL file.

CheckpointTo takes the engine write barrier for the whole Flush→Checkpoint→truncate window. That prevents a write from committing between the Flush and Checkpoint — such a write would otherwise exist only in the WAL, which truncateCheckpointWALs discards.

func (*Engine) Close ¶

func (e *Engine) Close() error

Close shuts down the engine. After Close, all methods return ErrEngineClosing. Close blocks until all in-flight writes complete.

func (*Engine) CompactAllRanges ¶ added in v0.15.0

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

CompactAllRanges runs pebble.Compact over every sync-scoped range to reclaim disk from the tombstones ResetForNewSync (or any bulk delete) leaves behind. The file holds one sync, so "all ranges" is the whole data keyspace. Without this, the next checkpoint would still include the deleted bytes.

Errors are best-effort: a Compact failure on one range doesn't block the others — pebble retries compaction in the background. ctx is honored between ranges and surfaced verbatim on cancellation.

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, externalID string) error

func (*Engine) DeleteEntitlementRecord ¶

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

func (*Engine) DeleteGrantRecord ¶

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

DeleteGrantRecord removes a grant and its index entries.

func (*Engine) DeleteResourceRecord ¶

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

func (*Engine) DeleteResourceTypeRecord ¶

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

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, externalID string) (*v3.AssetRecord, error)

func (*Engine) GetEntitlementRecord ¶

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

func (*Engine) GetGrantRecord ¶

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

GetGrantRecord fetches a grant record by external_id.

func (*Engine) GetResourceRecord ¶

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

func (*Engine) GetResourceTypeRecord ¶

func (e *Engine) GetResourceTypeRecord(ctx context.Context, 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) IterateAssets ¶ added in v0.15.0

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

func (*Engine) IterateEntitlements ¶ added in v0.15.0

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

func (*Engine) IterateEntitlementsByResource ¶

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

func (*Engine) IterateGrants ¶ added in v0.15.0

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

IterateGrants iterates all grants in primary-key order. yield returns false to stop iteration.

func (*Engine) IterateGrantsByEntitlement ¶

func (e *Engine) IterateGrantsByEntitlement(ctx context.Context, 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, yield func(*v3.GrantRecord) bool) error

IterateGrantsByNeedsExpansion iterates the needs_expansion index, 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, 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, principalRT string, yield func(*v3.GrantRecord) bool) error

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

func (*Engine) IterateResourceTypes ¶ added in v0.15.0

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

func (*Engine) IterateResources ¶ added in v0.15.0

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

func (*Engine) IterateResourcesByParent ¶

func (e *Engine) IterateResourcesByParent(ctx context.Context, parentRT, parentID 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) PaginateEntitlements ¶ added in v0.15.0

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

PaginateEntitlementsBySync returns a page of entitlements in primary key order.

func (*Engine) PaginateEntitlementsByResource ¶

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

PaginateEntitlementsByResource uses the by_resource index.

func (*Engine) PaginateGrantPrincipalKeysByEntitlement ¶ added in v0.12.3

func (e *Engine) PaginateGrantPrincipalKeysByEntitlement(
	ctx context.Context, entitlementID, cursor string, limit int,
) ([]string, string, error)

PaginateGrantPrincipalKeysByEntitlement scans the existing by_entitlement index and returns only principal identity keys for each matching grant. The key format is principal_resource_type + "\x00" + principal_resource_id, matching pkg/sync/expand's descendantGrantKey.

func (*Engine) PaginateGrants ¶ added in v0.15.0

func (e *Engine) PaginateGrants(
	ctx context.Context, 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) PaginateGrantsByEntitlement ¶

func (e *Engine) PaginateGrantsByEntitlement(
	ctx context.Context, 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) PaginateGrantsByEntitlementPrincipal ¶ added in v0.12.3

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

PaginateGrantsByEntitlementPrincipal uses the by_entitlement index narrowed to the entitlement_id + principal tuple. This is the hot path for grant expansion, where callers repeatedly ask whether a single principal already has a grant on a descendant entitlement.

func (*Engine) PaginateGrantsByEntitlementResource ¶

func (e *Engine) PaginateGrantsByEntitlementResource(
	ctx context.Context, 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, 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, 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, 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) PaginateResourceTypes ¶ added in v0.15.0

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

PaginateResourceTypesBySync returns a page of resource_types.

func (*Engine) PaginateResources ¶ added in v0.15.0

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

PaginateResourcesBySync returns a page of resources in primary key order.

func (*Engine) PaginateResourcesByParent ¶

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

PaginateResourcesByParent uses the by_parent index.

func (*Engine) PersistComputedSyncStats ¶ added in v0.13.2

func (e *Engine) PersistComputedSyncStats(ctx context.Context, syncID string, rec *v3.SyncStatsRecord) error

PersistComputedSyncStats writes a caller-computed stats record — e.g. one accumulated while the synccompactor wrote merge winners — without re-scanning the keyspaces. SyncId and WrittenAt are set here so callers only fill counts. Durability matches PersistSyncStats (pebble.Sync via writeSyncStats).

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. If a caller-computed record was stashed for this sync (StashComputedSyncStats), it is persisted instead of re-scanning the keyspaces.

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 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) PutExpandedGrantRecords ¶ added in v0.15.3

func (e *Engine) PutExpandedGrantRecords(ctx context.Context, records []*v3.GrantRecord) error

PutExpandedGrantRecords is the grant-expander write path — the engine side of GrantStore.StoreExpandedGrants, and its only caller.

Two properties distinguish it from PutGrantRecords:

• Single read-before-write. The expander must preserve each grant's existing Expansion / NeedsExpansion / DiscoveredAt side-state, which requires reading the prior primary value. That same read also yields the bytes needed to delete the prior value's stale index entries. The old path did BOTH a GetGrantRecord in the adapter (to preserve side-state) AND a db.Get here (to clean indexes) — two point lookups per grant. This path issues one and uses it for both.

• NoSync commit. Expanded grants are fully regenerable from the sync (the expander recomputes them from the entitlement graph), so a per-batch fsync buys nothing. Writes commit with pebble.NoSync and are hardened by the single Flush at sync end (EndFreshSync) or Close — the same bargain the fresh-sync fast path strikes, extended to resumed syncs where IsFreshSync() is false.

records arrive as freshly translated v3 GrantRecords with NO preservation or discovered_at stamping applied; this method performs the merge so the read it already issues does double duty.

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 external_id 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 is empty by construction (StartNewSync excises 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 (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 writes the file's single sync-run record at the fixed sync-run key. The sync_id lives in the record value (the only place a sync_id is stored — never in a key), and must be non-empty since this record is *defining* what counts as a valid sync.

func (*Engine) ResetForNewSync ¶ added in v0.15.0

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

ResetForNewSync wipes every sync-scoped keyspace (records, indexes, the sync-run record, and the stats sidecar) so a freshly started sync begins on an empty keyspace.

Why this exists: a v3 Pebble c1z holds exactly one sync, and the keys carry no sync_id. StartNewSync calls this before binding a new sync so the replacement sync never inherits orphan records from a prior sync — records the new sync doesn't happen to overwrite would otherwise linger under identical keys.

The wipe uses pebble.DB.Excise rather than DeleteRange tombstones: excise drops fully-covered SSTs from the manifest outright (O(metadata), immediate disk reclaim) instead of leaving range tombstones whose dead bytes survive until compaction. With tombstones, a replacement sync that finishes before background compaction catches up would hard-link the prior sync's dead SSTs into the CheckpointTo envelope at save — the same bloat the old multi-sync Cleanup path ran CompactSyncRanges to avoid. Excise also keeps the MarkFreshSync "empty by construction" fast path honest physically, not just logically.

Two spans cover everything sync-scoped:

• [v3|typeResourceType, v3|typeEngineMeta): every record type, the sync-run record, all secondary indexes, assets, and the reserved counter/session bytes — one contiguous span, so almost every SST is fully covered and dropped whole.
• the stats sidecar range inside engine-meta.

Engine-global metadata — the keyspace-version stamp and index-migration markers — lives elsewhere in engine-meta and is intentionally preserved.

Refuses while a fresh sync is in progress (between MarkFreshSync and EndFreshSync): wiping mid-sync would corrupt the in-flight sync.

func (*Engine) Save ¶

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

func (*Engine) SessionClear ¶ added in v0.16.0

func (e *Engine) SessionClear(ctx context.Context, opt ...sessions.SessionStoreOption) error

func (*Engine) SessionDelete ¶ added in v0.16.0

func (e *Engine) SessionDelete(ctx context.Context, key string, opt ...sessions.SessionStoreOption) error

func (*Engine) SessionGet ¶ added in v0.16.0

func (e *Engine) SessionGet(ctx context.Context, key string, opt ...sessions.SessionStoreOption) ([]byte, bool, error)

func (*Engine) SessionGetAll ¶ added in v0.16.0

func (e *Engine) SessionGetAll(ctx context.Context, pageToken string, opt ...sessions.SessionStoreOption) (map[string][]byte, string, error)

func (*Engine) SessionGetMany ¶ added in v0.16.0

func (e *Engine) SessionGetMany(ctx context.Context, keys []string, opt ...sessions.SessionStoreOption) (map[string][]byte, []string, error)

func (*Engine) SessionSet ¶ added in v0.16.0

func (e *Engine) SessionSet(ctx context.Context, key string, value []byte, opt ...sessions.SessionStoreOption) error

func (*Engine) SessionSetMany ¶ added in v0.16.0

func (e *Engine) SessionSetMany(ctx context.Context, values map[string][]byte, opt ...sessions.SessionStoreOption) 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).

func (*Engine) StartBulkSyncImport ¶ added in v0.13.7

func (e *Engine) StartBulkSyncImport(ctx context.Context, syncID string, tmpDir string) (*BulkSyncImport, error)

StartBulkSyncImport opens a bulk import targeting syncID, which must be the engine's current FRESH sync (see BulkSyncImport contract). Working files are staged in a fresh directory under tmpDir ("" = system temp dir) and removed by Finish/Abort.

func (*Engine) StashComputedSyncStats ¶ added in v0.13.7

func (e *Engine) StashComputedSyncStats(syncID string, rec *v3.SyncStatsRecord)

StashComputedSyncStats registers a caller-computed stats record to be persisted by the next PersistSyncStats call for syncID instead of the O(N) keyspace re-scan. Intended for trusted bulk writers (e.g. the BulkSyncImport conversion path) that counted every record they wrote; the caller owns the record's correctness. The stash is consumed by exactly one PersistSyncStats call.

func (*Engine) UnsafePutUniqueGrantRecords ¶ added in v0.12.2

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

UnsafePutUniqueGrantRecords is the trusted-import write path: it writes records unconditionally, with NO read-before-write and NO dedup pass. Do not use it for live connector output. The engine must currently be in fresh-sync mode, and the caller must guarantee each external_id appears at most once across the whole sync (not just within this batch). Primary + index key/value encoding — including the proto marshal — runs in parallel across GOMAXPROCS workers; a single goroutine then Sets the pre-encoded bytes into two batches and commits them (NoSync during a fresh sync).

Unlike PutGrantRecords this skips the per-record db.Get that PutGrantRecords performs on every batch after the first of a fresh sync. That read-before- write only exists to clean up stale index entries when an external_id is rewritten within a sync — impossible when the caller guarantees global uniqueness for the imported sync.