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 its raw public id, resolved through the bare-id lookup edge. Callers holding the full grant should prefer DeleteGrantByRefs, which needs no id-string resolution.

func (*Adapter) DeleteGrantByRefs ¶ added in v0.18.0

func (a *Adapter) DeleteGrantByRefs(ctx context.Context, grant *v2.Grant) error

DeleteGrantByRefs removes a grant addressed by the structured refs of the supplied v2 grant — the exact delete path, no lossy id string involved. Incomplete refs are an error, never a fallback to bare-id resolution: this is a sync-internal surface, and string resolution is reserved for interactive/CLI edges (see lookup.go). A grant whose refs cannot derive an identity could not have been stored in the first place, so there is nothing a string could correctly address here.

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) GetEntitlementGrantDigest ¶ added in v0.18.3

func (a *Adapter) GetEntitlementGrantDigest(ctx context.Context, ent *v2.Entitlement) (connectorstore.GrantDigest, bool, error)

GetEntitlementGrantDigest implements connectorstore.EntitlementGrantDigestReader. It returns the stored grant-digest root (content hash + grant count) for the entitlement under the reader's active sync. found is false when no digest exists — either no active sync, an unknown bare id, or no digest was built for it (e.g. WithGrantDigestIndex(false), a file that predates the digest, or a post-seal mutation invalidated it). A nil error with found=false means "no digest".

func (*Adapter) GetEntitlementGrantDigestNodes ¶ added in v0.18.3

func (a *Adapter) GetEntitlementGrantDigestNodes(ctx context.Context, ent *v2.Entitlement, level int) ([]connectorstore.GrantDigestNode, bool, error)

GetEntitlementGrantDigestNodes implements connectorstore.EntitlementGrantDigestReader. It lists the entitlement's grant-digest rollup nodes at the requested level (2^level buckets; level 0 = the root). For 0 <= level <= the digest's native level it folds the stored leaves — one scan of the digest keyspace. For a finer level it scans the grant index directly (O(grants)) instead of erroring; the level is clamped to the bucket-hash resolution (digestMaxWidthBits).

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 whether ListGrantsForEntitlement yields grants in non-decreasing principal (resource_type, resource) order. True under the structured key layout: the primary grant key ends in the tuple-encoded (principal_rt, principal_id) components and the tuple codec is order-preserving, so a per-entitlement primary prefix scan IS a principal-ordered scan. This gates the topological-merge expansion path; streamingPrincipalGroupStream additionally fails loudly if the order invariant is ever violated, and the differential/parity suites compare the sorted and unsorted evaluators against SQLite.

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 primary grant entitlement-resource prefixes. 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 entitlement IDs in REQUEST ORDER — the cursor resumes by positional index, so a reorder is as fatal as a drop/add; any change to the list (including order) restarts from the beginning rather than silently mis-paginating.

func (*Adapter) ListGrantsForPrincipal ¶

func (a *Adapter) ListGrantsForPrincipal(
	ctx context.Context,
	req *reader_v2.GrantsReaderServiceListGrantsForPrincipalRequest,
) (*reader_v2.GrantsReaderServiceListGrantsForPrincipalResponse, error)

ListGrantsForPrincipal returns all grants where the given principal_id is the principal, via the O(K) PaginateGrantsByPrincipal index walk. The optional Entitlement field narrows results to a single entitlement — since entitlement + principal is the full primary grant key, that case is an O(1) point lookup. Implements reader_v2.GrantsReaderServiceServer.

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 the by_principal index prefix. The cursor is the by_principal 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 c1zstore.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) ScanEntitlementGrantBucket ¶ added in v0.18.3

func (a *Adapter) ScanEntitlementGrantBucket(ctx context.Context, ent *v2.Entitlement, bucket connectorstore.GrantDigestBucket, yield func(*v2.Grant) bool) error

ScanEntitlementGrantBucket implements connectorstore.EntitlementGrantDigestReader. It yields every grant in the given digest bucket of the entitlement, translated to v2.Grant. Bucket Level 0 scans the whole entitlement; a finer Level is clamped to the bucket-hash resolution. Yields nothing when there is no active sync or a bare entitlement id resolves to nothing.

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) AddGrantsWithDiscoveredAt ¶ added in v0.17.0

func (s *BulkGrantShard) AddGrantsWithDiscoveredAt(ctx context.Context, grants []*v2.Grant, discoveredAt []*timestamppb.Timestamp) error

AddGrantsWithDiscoveredAt is AddGrants with an optional per-record discovered_at, aligned by index with grants (nil slice or nil entries fall back to now). Conversions use it to preserve the source record's original discovery time: compaction merges pick winners by newest discovered_at, so re-stamping conversion wall-clock time would make converted records override genuinely newer data.

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. Rows are keyed by structural identity and re-sorted through a spill sorter, so arrival order does not matter (the converter's external-id scan order does NOT match identity-tuple order). Identity embeds the raw external id, so duplicates are impossible for well-formed sources and fail the merge.

func (*BulkSyncImport) AddEntitlementsWithDiscoveredAt ¶ added in v0.17.0

func (b *BulkSyncImport) AddEntitlementsWithDiscoveredAt(ctx context.Context, ents []*v2.Entitlement, discoveredAt []*timestamppb.Timestamp) error

AddEntitlementsWithDiscoveredAt is AddEntitlements with an optional per-record discovered_at aligned by index with ents; nil entries fall back to now. See AddGrantsWithDiscoveredAt for why conversions must preserve source discovery times.

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) AddResourceTypesWithDiscoveredAt ¶ added in v0.17.0

func (b *BulkSyncImport) AddResourceTypesWithDiscoveredAt(ctx context.Context, rts []*v2.ResourceType, discoveredAt []*timestamppb.Timestamp) error

AddResourceTypesWithDiscoveredAt is AddResourceTypes with an optional per-record discovered_at aligned by index with rts; nil entries fall back to now. See AddGrantsWithDiscoveredAt for why conversions must preserve source discovery times.

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) AddResourcesWithDiscoveredAt ¶ added in v0.17.0

func (b *BulkSyncImport) AddResourcesWithDiscoveredAt(ctx context.Context, resources []*v2.Resource, discoveredAt []*timestamppb.Timestamp) error

AddResourcesWithDiscoveredAt is AddResources with an optional per-record discovered_at aligned by index with resources; nil entries fall back to now. See AddGrantsWithDiscoveredAt for why conversions must preserve source discovery times.

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; call it after Finish so the counts reflect any duplicate-identity grant rows that folded at merge time. 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) AbortSynthesizedGrantLayer ¶ added in v0.18.0

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

AbortSynthesizedGrantLayer discards an in-flight layer session: already ingested segments remain in the DB (their rows are idempotent overwrites on retry), staged chunks are dropped. Safe to call with no open session.

Deliberately does NOT take the engine write barrier — Close calls it after setting the closing flag (withWrite would refuse), and it must stay callable as a cleanup path when a writer holding writeMu panicked. The synthLayerMu take keeps the pointer handoff race-free against Begin/Add/Finish/Close.

func (*Engine) AddSynthesizedGrantLayerContributions ¶ added in v0.18.0

func (e *Engine) AddSynthesizedGrantLayerContributions(ctx context.Context, records []synthesizedGrantRecord) error

AddSynthesizedGrantLayerContributions encodes records into the open layer session. Rows become readable as their segment is ingested; callers must not rely on visibility before FinishSynthesizedGrantLayer returns.

func (*Engine) BeginSynthesizedGrantLayer ¶ added in v0.18.0

func (e *Engine) BeginSynthesizedGrantLayer(ctx context.Context) (bool, error)

BeginSynthesizedGrantLayer opens a layer-scoped layer session. The ingested SSTs carry primary rows only; the by_principal index is always rebuilt at EndSync, so no inline index maintenance is skipped by taking this path. The boolean is part of the store-level contract (non-Pebble stores report false and callers fall back to StoreNewExpandedGrantContributions).

func (*Engine) BuildDeferredGrantIndexes ¶ added in v0.18.0

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

BuildDeferredGrantIndexes rebuilds the remaining scattered expansion index family, by_principal, from entitlement-first primary grant keys. The expansion write path can skip by_principal inline because expansion reads by entitlement only; this method rewrites the whole by_principal range as one sorted SST at EndSync.

The same scan also rebuilds the primary grant keyspace itself: every raw (key, value) is teed into rolling SSTs which replace the whole grant range via IngestAndExcise. Expansion's layer ingests are SSTs whose spans interleave, so they stack in the LSM's upper levels; consolidating them via compaction rewrites the data once per level hop (a whale measured ~60s / 431 compactions), while this tee rewrites each byte exactly once on top of a scan that is already being paid for the index build. After the excise the grant range is one flat sorted run in the bottom level: compaction debt ~0 without running the compactor at all. Requires no concurrent grant writers, which EndSync guarantees.

The scan also accumulates the grant portion of the sync-stats sidecar (total count + per-entitlement-resource-type grouping) and stashes it so the subsequent PersistSyncStats call can skip its own full grant scan. The stats accumulation is best-effort: a value that fails the shallow field scan just disables the stash and PersistSyncStats falls back to scanning.

The whole build runs under the engine write barrier (withWrite): the IngestAndExcise calls destroy everything in their key ranges, so a grant row committed by a concurrent writer after the scan's iterator snapshot would be silently erased. EndSync's callers are expected to have quiesced writers already — holding writeMu for the duration converts that convention into an enforced invariant (a straggler write blocks until the build finishes instead of racing the excise), and writeWG participation means Close waits the build out instead of tearing down e.db under it.

func (*Engine) BuildGrantDigests ¶ added in v0.18.3

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

BuildGrantDigests is the full-file standalone digest build: grants written through PutGrantRecords / UnsafePutUniqueGrantRecords / the bulk importer maintain by_principal inline and never arm the deferred-index marker, so a sync without expansion-path writes reaches EndSync with deferredIdxPending false — and possibly millions of grants. This runs its own primary-grant scan feeding the same spill-sorter → merge+fold → ingest machinery the fused pass uses (buildGrantDigestsFromSpill). When the deferred pass DOES run, the fused build supersedes this (same output, one shared scan).

Callers: Adapter.endSyncFinalize calls RepairMissingGrantDigests rather than this directly — RepairMissingGrantDigests falls back to this exact function whenever no digest exists yet at all (grantDigestsPresent false), which is always true for a brand-new sync (ResetForNewSync excises the digest keyspace at StartNewSync), so the common case is unchanged. This function stays exported and self-sufficient for direct callers (tests, explicit repair) that want an unconditional from-scratch rebuild regardless of what already exists.

Failure semantics match the fused pass: any build error (except context cancellation, which stays fatal) downgrades to a loud drop of the digest state — absent digests are safe, half-built ones are not.

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. It also takes checkpointMu exclusively for the same window: the synth-layer session's background worker ingests SSTs outside writeMu (see ingestSynthLayerSegment), and a flushable ingest landing mid-window would be a WAL-only record the truncate 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.

Refuses with ErrEngineSealed after EndSync (via checkWritable): manual compactions go through the same CompactionScheduler as automatic ones, so on a sealed (paused) engine db.Compact would block forever waiting for a grant — and, because we hold writeWG, deadlock Engine.Close too. Bind a sync (SetCurrentSync) first.

KNOWN LIMITATION: the gate only refuses calls made after the seal. A CompactAllRanges already inside its loop when EndSync pauses the scheduler blocks in db.Compact indefinitely (and holds writeWG, so a later Close hangs too). Do not run this concurrently with EndSync; no in-tree caller does.

func (*Engine) ComputeEntitlementBucketDigest ¶ added in v0.18.3

func (e *Engine) ComputeEntitlementBucketDigest(ctx context.Context, id entitlementIdentity, bucket DigestBucket) ([]byte, int64, error)

ComputeEntitlementBucketDigest folds the grant hash index over a single bucket of an entitlement (the zero bucket = the whole entitlement) — the authoritative on-demand counterpart of the stored digest nodes, for verifying or subdividing a digest that EXISTS. Only meaningful while the entitlement's digest is built (its root is stored): the hash index lives and dies with the digest nodes, so against a never-built or invalidated entitlement this folds an absent index range and returns {0, 0} — "zero grants", not "unknown". Never use it as a fallback for a missing root; see GetEntitlementDigestRoot and computeBucketDigest's precondition.

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. Callers that write the entitlement keyspace through this handle (ingests, excises) must call InvalidateBareIDLookups afterwards.

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

DeleteEntitlementRecord deletes by raw public id. A missing id is a no-op; an ambiguous id is an error (a lossy string must never guess a delete).

func (*Engine) DeleteGrantByIdentityRefs ¶ added in v0.18.0

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

DeleteGrantByIdentityRefs removes a grant addressed by its structural refs — the exact delete path for callers that hold the full grant. No lossy id string is involved. Deleting a non-existent grant is a no-op.

func (*Engine) DeleteGrantRecord ¶

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

DeleteGrantRecord removes a grant and its index entries by raw public id. A missing/unresolvable id is a no-op; an ambiguous id is an error (a lossy string must never guess a delete).

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) DirtyEntitlementBuckets ¶ added in v0.18.3

func (e *Engine) DirtyEntitlementBuckets(ctx context.Context, other *Engine, id entitlementIdentity) ([]DigestBucket, error)

DirtyEntitlementBuckets compares this engine's entitlement against other's and returns the buckets whose grants differ — see dirtyPartitionBuckets for the comparison contract (zero bucket = whole entitlement; nil = identical).

func (*Engine) DropAllGrantDigestState ¶ added in v0.18.3

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

DropAllGrantDigestState removes every stored digest node AND the whole by_entitlement_principal_hash index, and clears the engine's digests-present flag. For callers that are about to mutate grants through paths that bypass the engine's index maintenance entirely (the synccompactor's fold merge): after this, digests read as "missing — recalculate", never as stale-but-present, and subsequent engine-path writes skip per-entitlement invalidation tombstones. Two range tombstones: the digest keyspace and the hash-index keyspace are NOT adjacent (typeCounter/typeSession sit between them), so this must never be collapsed into one span.

func (*Engine) DropAllGrantDigests ¶ added in v0.18.3

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

DropAllGrantDigests removes every stored digest node. Called when a seal-time build fails partway: a partially built digest that LOOKS present would violate the present-means-exact contract, whereas absent digests just make readers re-read the grants until the next successful seal recalculates them.

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.

Uses withWrite (not a bare writeMu) so the flush participates in the closing check and writeWG: Close tears e.db down after writeWG.Wait, and a bare-mutex EndFreshSync racing Close would flush a nil db.

func (*Engine) ExpandWritePathStats ¶ added in v0.18.0

func (e *Engine) ExpandWritePathStats() ExpandWritePathStats

func (*Engine) FinishSynthesizedGrantLayer ¶ added in v0.18.0

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

FinishSynthesizedGrantLayer flushes the session's tail segment, waits for the background worker to merge and ingest every queued segment, and closes the session. No-op if no session is open or the session saw no rows.

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) GetEntitlementDigestRoot ¶ added in v0.18.3

func (e *Engine) GetEntitlementDigestRoot(ctx context.Context, id entitlementIdentity) (DigestRoot, bool, error)

GetEntitlementDigestRoot returns the stored grant-digest root for an entitlement. ok is false when no digest has been built for it (or it was invalidated) — which means the caller must re-read the entitlement's grants (or treat the whole entitlement as dirty), NOT fall back to ComputeEntitlementBucketDigest: the hash index that fold reads is only ever written and dropped alongside the digest nodes, so with no root it is absent too and the fold would report "zero grants" for an entitlement that may have millions — the false-clean trap dirtyPartitionBuckets' doc comment describes.

func (*Engine) GetEntitlementRecord ¶

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

GetEntitlementRecord fetches an entitlement by its raw public id via the bare-id lookup (exact string-match, exactly-one rule — see lookup.go).

func (*Engine) GetGrantDigestGlobalRoot ¶ added in v0.18.3

func (e *Engine) GetGrantDigestGlobalRoot(ctx context.Context) (DigestRoot, bool, error)

GetGrantDigestGlobalRoot returns the whole-file grant digest root — the XOR fold of every entitlement's grant-digest root's content-hash plus the total grant count, across the sync's single sync. ok is false when no digest has been built for this file (present-means- exact: absence means "recalculate", never "zero grants"). Written only alongside a full grant-digest build (the seal-time build, or a compaction that runs BuildGrantDigests) and dropped by the same invalidation paths that drop any per-entitlement root — see stageGrantDigestInvalidation and the Drop* functions below.

func (*Engine) GetGrantRecord ¶

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

GetGrantRecord fetches a grant record by its raw public id via the bare-id lookup (candidate-split probing, exactly-one rule — lookup.go).

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) GrantDigestIndexEnabled ¶ added in v0.18.3

func (e *Engine) GrantDigestIndexEnabled() bool

GrantDigestIndexEnabled reports whether the seal-time deferred pass builds the by_entitlement_principal_hash index and grant digests. See WithGrantDigestIndex.

func (*Engine) InvalidateBareIDLookups ¶ added in v0.18.0

func (e *Engine) InvalidateBareIDLookups()

InvalidateBareIDLookups invalidates the lazily built bare-id lookup state (see lookup.go). Engine write paths call this internally; it is exported for callers that mutate the keyspace through DB() directly.

func (*Engine) InvalidateGrantDigestPartitions ¶ added in v0.18.3

func (e *Engine) InvalidateGrantDigestPartitions(ctx context.Context, partitions []string) error

InvalidateGrantDigestPartitions drops the digest + hash-index state for exactly the given entitlement partitions, plus the whole-file global root — which folds over every partition and goes stale the moment any one of them does. Every other partition's state is left completely untouched.

For callers that mutate the primary grant keyspace directly, bypassing the engine's own per-record invalidation path (stageGrantDigestInvalidation) — the fold compactor's raw merge — to report exactly what they changed. RepairMissingGrantDigests recomputes exactly what this drops (and only that), so the pair gives compaction per-entitlement-scoped digest maintenance without an O(file) rebuild. No-op when partitions is empty or no digest has ever been built for this file.

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) IsSealed ¶ added in v0.18.0

func (e *Engine) IsSealed() bool

IsSealed reports whether the engine is in the post-EndSync sealed state.

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 primary grant keyspace for the given raw entitlement id, yielding each grant in encoded principal-key order. The id resolves through the bare-id lookup; an id matching no entitlement iterates nothing. yield returns false to stop.

func (*Engine) IterateGrantsByEntitlementBucket ¶ added in v0.18.3

func (e *Engine) IterateGrantsByEntitlementBucket(ctx context.Context, id entitlementIdentity, bucket DigestBucket, yield func(*v3.GrantRecord) bool) error

IterateGrantsByEntitlementBucket yields the grants in one principal-hash bucket of an entitlement (the zero bucket = the whole entitlement). This is the dirty-bucket loader: after a digest comparison flags a bucket, the caller materializes only those grants. The primary key is reconstructed from each index key by byte splice (no decode); the point Get per entry is the cost of MATERIALIZING a changed grant, not of finding it. Orphan index entries are skipped.

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 index narrowed to a principal resource type. Yields each grant whose principal carries the given resource_type. 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, c1zstore.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) LogCompactionMetrics ¶ added in v0.18.0

func (e *Engine) LogCompactionMetrics(ctx context.Context, phase string)

LogCompactionMetrics logs a snapshot of the LSM's compaction and ingest counters, labeled with the caller's phase name. Diagnostic only: used to attribute wall time lost to background compactions triggered by large ingests (e.g. the synthesized-grant layer segments) during benchmark runs.

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) MigratedOnOpen ¶ added in v0.18.0

func (e *Engine) MigratedOnOpen() bool

MigratedOnOpen reports whether this Open ran the in-place id-index migration (see migratedOnOpen).

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 entitlement primary key prefix.

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

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

PaginateGrantPrincipalKeysByEntitlement scans the primary grant keyspace under the entitlement identity prefix 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, entID entitlementIdentity, cursor string, limit int,
) ([]*v3.GrantRecord, string, error)

PaginateGrantsByEntitlement scans the primary grant keyspace under the entitlement identity prefix. Cursor is the primary key. Callers resolve the identity from structured refs (or the bare-id lookup) — the engine never parses id strings here.

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

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

PaginateGrantsByEntitlementPrincipal uses the structured primary key for a point lookup by entitlement + principal. 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 primary grant keyspace for all grants whose entitlement's resource is (entRT, entRID). Cursor is the primary 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 by structured primary key. The identity key is a pure function of the record (it contains the raw external id), so overwrites are idempotent and no read-before-write or index cleanup is needed; a within-call dedup pre-pass keeps last-wins semantics for same-identity duplicates in one batch.

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 remaining secondary 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) PutSynthesizedGrantContributions ¶ added in v0.18.0

func (e *Engine) PutSynthesizedGrantContributions(ctx context.Context, records []synthesizedGrantRecord) error

PutSynthesizedGrantContributions batch-writes one destination's synthesized contributions. It is the fallback for stores/engines that cannot run a layer-scoped layer session (see BeginSynthesizedGrantLayer); the layer path is preferred because it publishes sorted SSTs instead of out-of-order batch commits.

func (*Engine) PutSynthesizedGrantRecords ¶ added in v0.18.0

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

PutSynthesizedGrantRecords writes expander-synthesized grants that the caller guarantees are brand-new by structured grant identity. It skips the read-before-write Get in PutExpandedGrantRecords because there is no prior value whose Expansion/NeedsExpansion/DiscoveredAt or index entries must be preserved/cleaned.

func (*Engine) RepairMissingGrantDigests ¶ added in v0.18.3

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

RepairMissingGrantDigests rebuilds the by_entitlement_principal_hash rows and digest nodes for exactly the entitlements missing a digest root — trusting every entitlement that already has one — then recomputes the whole-file global root if anything changed. The whole operation holds the engine's write lock for its entire duration, like BuildGrantDigests/BuildDeferredGrantIndexes, so no concurrent grant write can interleave with a partition mid-repair and leave it silently wrong.

When digests were never built for this file at all (or were dropped wholesale — grantDigestsPresent false), a full BuildGrantDigests is strictly cheaper than a point-Get per entitlement that would all miss anyway, so this delegates to it instead.

Failure policy: a repair error for one partition is logged and that partition is left missing (safe — present-means-exact — and self-healing on the next repair or seal) rather than failing the whole call; only context cancellation propagates. Unlike a fused full rebuild, a partial failure here does not require dropping already-repaired partitions — each one commits its own correct, complete state independently.

Orphan grants (an entitlement identity with grants but no entitlement RECORD) are covered: the missing-partition scan discovers grant-bearing partitions from the grant primary keyspace itself, not from entitlement records, so an invalidated orphan is rebuilt like any other partition and the recomputed global root always includes it. That keeps the global root a pure function of the stored grant set — identical whether it was produced by a seal (whose fold covers orphans too) or by a fold + targeted repair — which the manifest's header-only "did the grants change?" comparison depends on. Without it, a touched orphan would go missing forever: the recomputed root's presence would satisfy the repair fast path (repairMissingGrantDigestsAttempt) on every later call.

Called from Adapter.endSyncFinalize in place of a bare BuildGrantDigests whenever the deferred (by_principal-rebuilding) pass didn't run: for a brand-new sync grantDigestsPresent is always false (ResetForNewSync excises the digest keyspace at StartNewSync), so this reduces to exactly BuildGrantDigests's existing full scan — no behavior change for the common single-EndSync-per-sync case. It only diverges — into the cheap fast path or the targeted repair — when EndSync runs a SECOND time on an already-digested, rebound sync (SetCurrentSync, not StartNewSync) without a fresh reset in between: the shape grant-expansion's own follow-up sync produces, and exactly the case that made a full rebuild here wasteful (see RFC 0003 / compactPebbleFold).

Like BuildGrantDigests, a failure that isn't context cancellation is logged and downgraded to a full state drop rather than propagated — digest problems must never fail EndSync. This only covers failures in the outer scan/orchestration (e.g. the entitlement-keyspace scan itself erroring): a failure repairing one INDIVIDUAL partition is already handled without dropping anything (see repairMissingGrantDigestsLocked).

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.