sync

package
v0.18.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 15, 2026 License: Apache-2.0 Imports: 44 Imported by: 0

Documentation

Index

Constants

View Source 
const StateTokenVersion = 1

If you make a breaking change to the state token, you must increment this version.

Variables

View Source 
var ErrNoSyncIDFound = fmt.Errorf("no syncID found after starting or resuming sync")
View Source 
var ErrSyncNotComplete = fmt.Errorf("sync exited without finishing")
View Source 
var ErrTooManyWarnings = fmt.Errorf("too many warnings, exiting sync")

Functions

func BuildCompactedToken added in v0.18.3 

func BuildCompactedToken(baseToken string, in CompactionTokenInput) (string, error)

BuildCompactedToken rewrites a compacted output's sync token with a compaction provenance section. baseToken is the base input's token ("" for rebuild outputs, which start empty); its resume state, skip flags, and timing stats are preserved. Chained compactions accumulate: the original StatsSyncID, the uncapped partial count, and the partials aggregate carry forward from an existing compaction section.

func GetExpandableAnnotation added in v0.2.92 

func GetExpandableAnnotation(annos annotations.Annotations) (*v2.GrantExpandable, error)

func GetExternalResourceMatchAllAnnotation added in v0.2.84 

func GetExternalResourceMatchAllAnnotation(annos annotations.Annotations) (*v2.ExternalResourceMatchAll, error)

func GetExternalResourceMatchAnnotation added in v0.2.84 

func GetExternalResourceMatchAnnotation(annos annotations.Annotations) (*v2.ExternalResourceMatch, error)

func GetExternalResourceMatchIDAnnotation added in v0.2.90 

func GetExternalResourceMatchIDAnnotation(annos annotations.Annotations) (*v2.ExternalResourceMatchID, error)

func IsSyncPreservable added in v0.6.18 

func IsSyncPreservable(err error) bool

IsSyncPreservable returns true if the error returned by Sync() means that the sync artifact is useful. This either means that there was no error, or that the error is recoverable (we can resume the sync and possibly succeed next time).

func NeedsExpansion added in v0.12.2 

func NeedsExpansion(stateStr string) (bool, error)

func NormalizeWorkerCount added in v0.9.1 

func NormalizeWorkerCount(count int) int

NormalizeWorkerCount maps raw worker-count inputs (CLI / config sentinels) to the syncer's internal worker count: -1 selects min(max(GOMAXPROCS, 1), 4); any other value uses max(count, 0).

func PrepareExpansionReplayToken added in v0.12.6 

func PrepareExpansionReplayToken(stateStr string) (string, error)

PrepareExpansionReplayToken rewrites a finished sync's state token so the sync can be re-run through grant expansion, preserving the token's other recorded state rather than discarding it. It marks the sync as needing expansion and, when the action stack is empty, pushes an InitOp so the resumed syncer drives its work from the top. A finished sync's token has an empty action stack, so without the InitOp a resume would find nothing to do and exit before expanding; clearing the whole token would also drop the skip flags and exclusion-group bookkeeping the token carries.

Types

type Action 

type Action struct {
	ID                   string   `json:"id,omitempty"`
	Op                   ActionOp `json:"operation,omitempty"`
	PageToken            string   `json:"page_token,omitempty"`
	ResourceTypeID       string   `json:"resource_type_id,omitempty"`
	ResourceID           string   `json:"resource_id,omitempty"`
	ParentResourceTypeID string   `json:"parent_resource_type_id,omitempty"`
	ParentResourceID     string   `json:"parent_resource_id,omitempty"`
}

Action stores the current operation, page token, and optional fields for which resource is being worked with.

type ActionOp 

type ActionOp uint8

ActionOp represents a sync operation. 

const (
	UnknownOp ActionOp = iota
	InitOp
	SyncResourceTypesOp
	SyncResourcesOp
	SyncEntitlementsOp
	ListResourcesForEntitlementsOp
	SyncGrantsOp
	SyncExternalResourcesOp
	SyncAssetsOp
	SyncGrantExpansionOp
	SyncTargetedResourceOp
	SyncStaticEntitlementsOp
)

Do not change the order of these constants, and only append new ones at the end. Otherwise resuming a sync started by an older version of baton-sdk will cause very strange behavior.

func (ActionOp) MarshalJSON 

func (s ActionOp) MarshalJSON() ([]byte, error)

MarshalJSON marshals the ActionOp into a json string.

func (ActionOp) String 

func (s ActionOp) String() string

String() returns the string representation for an ActionOp. This is used for marshalling the op.

func (*ActionOp) UnmarshalJSON 

func (s *ActionOp) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals the input byte slice and updates this action op.

type CompactionAggregate added in v0.18.3 

type CompactionAggregate struct {
	Executions         int64                         `json:"executions"`
	StepDurationsMs    map[string]int64              `json:"step_durations_ms,omitempty"`
	ConnectorCallStats map[string]*ConnectorCallStat `json:"connector_call_stats,omitempty"`
}

CompactionAggregate sums timing stats across the partial syncs merged into a compacted output. The values describe a population of Executions distinct sync runs — never one execution — and accumulate across chained folds.

type CompactionRecordCounts added in v0.18.3 

type CompactionRecordCounts struct {
	Output   int64 `json:"output"`
	Added    int64 `json:"added,omitempty"`
	Replaced int64 `json:"replaced,omitempty"`
	Carried  int64 `json:"carried,omitempty"`
}

CompactionRecordCounts describes one record type's provenance in a compacted output. Output is the record count in the compacted artifact. Added counts records admitted with no incumbent, Replaced counts records that overrode a strictly-older incumbent, and Carried counts incumbents that survived untouched. Added/Replaced/Carried are only populated by the fold strategy — rebuild merges lose per-source attribution in their run-file paths, so rebuild outputs carry Output only.

type CompactionTokenInput added in v0.18.3 

type CompactionTokenInput struct {
	Mode           string
	BaseSyncID     string
	PartialSyncIDs []string
	// PartialTokens are the partials' marshalled sync tokens, used for the
	// timing aggregate. Entries may be empty or unparseable (e.g. converted
	// sqlite inputs carry no token); those contribute no stats.
	PartialTokens []string
	RecordCounts  map[string]CompactionRecordCounts
}

CompactionTokenInput carries one compaction run's provenance into BuildCompactedToken.

type CompactionTokenStats added in v0.18.3 

type CompactionTokenStats struct {
	// Mode is the compaction strategy that produced this artifact
	// (fold / overlay / kway).
	Mode string `json:"mode"`
	// StatsSyncID is the sync whose execution the token's top-level
	// step_durations_ms / connector_call_stats describe. It is carried
	// unchanged across chained folds so the attribution stays truthful.
	StatsSyncID string `json:"stats_sync_id,omitempty"`
	// BaseSyncID is the immediate base input of this compaction run (the id
	// the artifact carried before the rename).
	BaseSyncID string `json:"base_sync_id,omitempty"`
	// PartialSyncIDs lists merged partial sync ids, capped at
	// maxCompactionPartialIDs. PartialCount is the uncapped total and
	// accumulates across chained folds.
	PartialSyncIDs []string `json:"partial_sync_ids,omitempty"`
	PartialCount   int64    `json:"partial_count"`
	// RecordCounts is keyed by record type (resource_types, resources,
	// entitlements, grants).
	RecordCounts map[string]*CompactionRecordCounts `json:"record_counts,omitempty"`
	// PartialsAggregate sums the merged partials' timing stats; nil when no
	// partial carried stats.
	PartialsAggregate *CompactionAggregate `json:"partials_aggregate,omitempty"`
}

CompactionTokenStats is the compaction provenance section of a compacted sync token. Its presence re-attributes the token's top-level timing stats: they describe StatsSyncID's collection run, not the compacted artifact.

func CompactionStatsFromToken added in v0.18.3 

func CompactionStatsFromToken(token string) (*CompactionTokenStats, error)

CompactionStatsFromToken returns the compaction provenance section of a marshalled sync token, or nil when the token is empty or carries none.

type ConnectorCallStat added in v0.18.3 

type ConnectorCallStat struct {
	Count   int64 `json:"count"`
	TotalMs int64 `json:"total_ms"`
	MaxMs   int64 `json:"max_ms"`
}

ConnectorCallStat contains cumulative latency statistics for one connector method.

type Progress added in v0.0.25 

type Progress struct {
	Action               string
	ResourceTypeID       string
	ResourceID           string
	ParentResourceTypeID string
	ParentResourceID     string
	Count                uint32
}

func NewProgress added in v0.0.25 

func NewProgress(a *Action, c uint32) *Progress

type SessionStoreStat added in v0.18.3 

type SessionStoreStat struct {
	Count    int64 `json:"count"`
	Errors   int64 `json:"errors,omitempty"`
	Timeouts int64 `json:"timeouts,omitempty"`
	TotalMs  int64 `json:"total_ms"`
	MaxMs    int64 `json:"max_ms"`
}

SessionStoreStat contains cumulative latency and outcome counters for one session-store operation. Timeouts is the deadline-exceeded subset of Errors; MaxMs pinned at a fixed value with Timeouts ≈ Count is the signature of a backend whose every request times out.

type State 

type State interface {
	PushAction(ctx context.Context, action Action)
	FinishAction(ctx context.Context, action *Action)
	NextPage(ctx context.Context, actionID string, pageToken string) error
	EntitlementGraph(ctx context.Context) *expand.EntitlementGraph
	ClearEntitlementGraph(ctx context.Context)
	Current() *Action
	GetAction(id string) *Action
	PeekMatchingActions(ctx context.Context, op ActionOp) []*Action
	Marshal() (string, error)
	Unmarshal(input string) error
	NeedsExpansion() bool
	SetNeedsExpansion()
	HasExternalResourcesGrants() bool
	SetHasExternalResourcesGrants()
	ShouldFetchRelatedResources() bool
	SetShouldFetchRelatedResources()
	ShouldSkipEntitlementsAndGrants() bool
	SetShouldSkipEntitlementsAndGrants()
	ShouldSkipGrants() bool
	SetShouldSkipGrants()
	GetCompletedActionsCount() uint64
	AddStepDuration(bucket string, duration time.Duration)
	StepDurations() map[string]int64
	RecordConnectorCall(method string, duration time.Duration)
	ConnectorCallStats() map[string]ConnectorCallStat
	RecordSessionOp(op string, duration time.Duration, opErr error, timedOut bool)
	MergeSessionStat(op string, add SessionStoreStat)
	SessionStoreStats() map[string]SessionStoreStat
	// CheckAndSetExclusionGroupResourceType records that exclusionGroupID is
	// used on resourceTypeID. If the group was already recorded against a
	// different resource type, the prior value is returned with conflict=true
	// and the map is not modified. Otherwise the map is updated (if needed)
	// and ("", false) is returned. The existing return value is meaningful
	// only when conflict is true.
	CheckAndSetExclusionGroupResourceType(exclusionGroupID, resourceTypeID string) (existing string, conflict bool)
	// CheckAndSetExclusionGroupDefault records that entitlementID is the
	// default entitlement for exclusionGroupID. If the group already has a
	// recorded default that is not entitlementID, the prior value is returned
	// with conflict=true and the map is not modified. Otherwise the map is
	// updated (if needed) and ("", false) is returned. The existing return
	// value is meaningful only when conflict is true.
	CheckAndSetExclusionGroupDefault(exclusionGroupID, entitlementID string) (existing string, conflict bool)
	// IncrementExclusionGroupCount increments the running count of entitlements
	// observed for exclusionGroupID and returns the new count. The count is
	// global per exclusion group id (across all resources) and survives resume.
	IncrementExclusionGroupCount(exclusionGroupID string) uint32
	// ClearExclusionGroupTracking drops the exclusion group bookkeeping maps so
	// they don't bloat the token of a completed sync, mirroring
	// ClearEntitlementGraph.
	ClearExclusionGroupTracking(ctx context.Context)
}

type SyncOpt 

type SyncOpt func(s *syncer)

func WithC1ZPath added in v0.1.0 

func WithC1ZPath(path string) SyncOpt

WithC1ZPath sets the path to the c1z file. Either this or WithConnectorStore must be provided to create a new syncer.

func WithConnectorStore added in v0.1.0 

func WithConnectorStore(store c1zstore.Store) SyncOpt

WithConnectorStore sets the connector store to use. This is the preferred option. Either this or WithC1ZPath must be provided to create a new syncer.

func WithDontExpandGrants added in v0.3.48 

func WithDontExpandGrants() SyncOpt

WithDontExpandGrants sets whether to skip expanding grants. This is used for speeding up service mode connectors and reducing their c1z upload size. C1 will process the uploaded c1z and expand grants itself.

func WithExternalResourceC1ZPath added in v0.2.84 

func WithExternalResourceC1ZPath(path string) SyncOpt

func WithExternalResourceEntitlementIdFilter added in v0.2.84 

func WithExternalResourceEntitlementIdFilter(entitlementId string) SyncOpt

func WithMetricsHandler added in v0.12.5 

func WithMetricsHandler(h metrics.Handler) SyncOpt

WithMetricsHandler attaches a metrics.Handler that the syncer forwards to progresslog.NewProgressCounts so the grant-expansion OTel instruments (baton.sync.expand.actions_remaining / actions_burned / decompressed_bytes / decompressed_bytes_delta) actually reach the configured exporter instead of the default no-op handler.

Callers should pre-tag the handler with the dimensions they want to slice by (e.g. tenant_id, connector_id) via Handler.WithTags before passing it in — baton-sdk has no view of those identifiers.

func WithOnlyExpandGrants added in v0.3.8 

func WithOnlyExpandGrants() SyncOpt

WithOnlyExpandGrants sets whether to skip syncing resources and only expand grants.

func WithOptionalPreviousSyncC1ZPath added in v0.15.0 

func WithOptionalPreviousSyncC1ZPath(path string) SyncOpt

WithOptionalPreviousSyncC1ZPath is WithPreviousSyncC1ZPath with best-effort semantics: if the file is missing, corrupt, or written by an incompatible SDK, NewSyncer logs and proceeds WITHOUT replay instead of failing. Intended for cache-style replay sources the caller maintains automatically (the service-mode previous-sync spare) — a bad cache file must never fail a sync. Callers that name a specific file deliberately should use WithPreviousSyncC1ZPath, which surfaces open failures.

func WithPreviousSyncC1ZPath added in v0.15.0 

func WithPreviousSyncC1ZPath(path string) SyncOpt

WithPreviousSyncC1ZPath points ETag-replay at a separate c1z holding the previous sync, instead of reading a previous sync from inside the live store.

This is required for the single-sync v3 (Pebble) engine: a Pebble c1z holds exactly one sync by contract, so there is no in-file "previous sync" to replay from (StartNewSync replaces the prior sync). Supplying the prior run's c1z here lets the syncer recover unchanged resources' ETags and carry their grants forward across runs. When unset, replay falls back to reading a previous sync from the live store (the SQLite multi-sync behavior), so existing callers are unaffected.

The file is opened read-only and engine-agnostically (the magic byte selects SQLite or Pebble), so the previous-sync c1z may use either engine.

func WithProgressHandler added in v0.0.25 

func WithProgressHandler(f func(s *Progress)) SyncOpt

WithProgressHandler sets a `progressHandler` for `NewSyncer` Options. The progress handler is called for sync action, such as listing resources, entitlements, grants, etc. If running in parallel mode, this function must be thread-safe.

func WithRunDuration 

func WithRunDuration(d time.Duration) SyncOpt

WithRunDuration sets a `time.Duration` for `NewSyncer` Options. `d` represents a duration. The elapsed time between two instants as an int64 nanosecond count.

func WithSessionStore added in v0.5.0 

func WithSessionStore(sessionStore sessions.SetSessionStore) SyncOpt

func WithSkipEntitlementsAndGrants added in v0.3.40 

func WithSkipEntitlementsAndGrants(skip bool) SyncOpt

WithSkipEntitlementsAndGrants sets whether to skip syncing entitlements and grants for resources. If true, only resources will be synced.

func WithSkipFullSync added in v0.2.15 

func WithSkipFullSync() SyncOpt

WithSkipFullSync skips syncing entirely.

func WithSkipGrants added in v0.5.1 

func WithSkipGrants(skip bool) SyncOpt

WithSkipGrants sets whether to skip syncing grants for resources. Entitlements will still be synced.

func WithStorageEngine added in v0.13.5 

func WithStorageEngine(engine c1zstore.Engine) SyncOpt

WithStorageEngine selects the dotc1z storage engine when opening the c1z file via WithC1ZPath. Empty uses the baton-sdk default.

func WithSyncID added in v0.3.8 

func WithSyncID(syncID string) SyncOpt

func WithSyncIdentity added in v0.12.5 

func WithSyncIdentity(id uotel.SyncIdentity) SyncOpt

WithSyncIdentity stamps connector identity onto sync and dotc1z spans (and the c1z size metric) so a single connector's work is filterable in APM. Attribute keys match the pprof.Do labels set by the platform sync activity, so spans and CPU profiles line up. Sync injects this into the run context via uotel.WithSyncIdentity, which is how it reaches dotc1z spans too.

func WithSyncResourceTypes added in v0.5.0 

func WithSyncResourceTypes(resourceTypeIDs []string) SyncOpt

WithSyncResourceTypes sets the resource types to sync. If empty (the default), all resource types will be synced.

func WithTargetedSyncResources added in v0.6.6 

func WithTargetedSyncResources(resources []*v2.Resource) SyncOpt

func WithTmpDir added in v0.1.8 

func WithTmpDir(path string) SyncOpt

func WithTransitionHandler 

func WithTransitionHandler(f func(s Action)) SyncOpt

WithTransitionHandler sets a `transitionHandler` for `NewSyncer` Options.

func WithWorkerCount added in v0.8.0 

func WithWorkerCount(count int) SyncOpt

WithWorkerCount sets the number of workers to use. If <=1, 1 worker is used (default). If > 1, parallel sync is used. If -1, the number of workers is set to the number of CPU cores or 4, whichever is lower. If < -1, 1 worker is used. (Nothing should do this, but there's no way to return an error in this option.)

type Syncer 

type Syncer interface {
	Sync(context.Context) error
	Close(context.Context) error
}

func NewSyncer 

func NewSyncer(ctx context.Context, c types.ConnectorClient, opts ...SyncOpt) (Syncer, error)

NewSyncer returns a new syncer object.

Source Files

View all Source files

Directories

Path Synopsis
scc
Package scc provides an iterative FW–BW SCC condensation for directed graphs, adapted for Baton’s entitlement graph.
 Package scc provides an iterative FW–BW SCC condensation for directed graphs, adapted for Baton’s entitlement graph.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.