distribution

package
v0.0.0-...-0febee4 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: AGPL-3.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// DefaultCatalogDeltaRetention bounds the durable reconnect window. A
	// watcher that falls behind this many catalog versions receives a snapshot
	// reset instead of an incomplete delta sequence.
	DefaultCatalogDeltaRetention uint64 = 1024
	DefaultCatalogDeltaBatchSize        = 128
	MaxCatalogDeltaBatchSize            = 1024
	CatalogWatchProtocolVersion  uint32 = 1
)
View Source 
const (
	MigrationFamilyUser uint32 = iota + 1
	MigrationFamilyTxnIntent
	MigrationFamilyTxnCommit
	MigrationFamilyTxnRollback
	MigrationFamilyTxnSuccess
	MigrationFamilyTxnMeta
	MigrationFamilyTxnLock
	MigrationFamilyListMeta
	MigrationFamilyListItem
	MigrationFamilyListMetaDelta
	MigrationFamilyListClaim
	MigrationFamilyRedisLegacy
	MigrationFamilyHash
	MigrationFamilySet
	MigrationFamilyZSet
	MigrationFamilyStreamMeta
	MigrationFamilyStreamEntry
	MigrationFamilyDynamoTableMeta
	MigrationFamilyDynamoTableGeneration
	MigrationFamilyDynamoItem
	MigrationFamilyDynamoGSI
	MigrationFamilySQSQueueMeta
	MigrationFamilySQSQueueGeneration
	MigrationFamilySQSQueueSequence
	MigrationFamilySQSQueueTombstone
	MigrationFamilySQSMessageData
	MigrationFamilySQSMessageVisibility
	MigrationFamilySQSMessageDedup
	MigrationFamilySQSMessageGroup
	MigrationFamilySQSMessageByAge
	MigrationFamilySQSPartitionedMessageData
	MigrationFamilySQSPartitionedMessageVisibility
	MigrationFamilySQSPartitionedMessageDedup
	MigrationFamilySQSPartitionedMessageGroup
	MigrationFamilySQSPartitionedMessageByAge
	MigrationFamilyS3BucketMeta
	MigrationFamilyS3BucketGeneration
	MigrationFamilyS3ObjectManifest
	MigrationFamilyS3UploadMeta
	MigrationFamilyS3UploadPart
	MigrationFamilyS3Blob
	MigrationFamilyS3GCUpload
	MigrationFamilyLegacyListMetaDelta
	MigrationFamilyS3ChunkRef
	MigrationFamilyFilesystemChunk
	MigrationFamilyFilesystemUsage
)
View Source 
const DefaultRouteHistoryDepth = 32

DefaultRouteHistoryDepth is the size of Engine's versioned-snapshot ring used by the Composed-1 M2 plumbing. 32 is conservative against current single-leader catalog churn (operator-frequency, not data-plane) per the design doc §9 Q2; raise if a future control plane generates more than ~tens of versions per second.

Variables

View Source 
var (
	ErrCatalogStoreRequired        = errors.New("catalog store is required")
	ErrCatalogVersionMismatch      = errors.New("catalog version mismatch")
	ErrCatalogVersionOverflow      = errors.New("catalog version overflow")
	ErrCatalogRouteIDOverflow      = errors.New("catalog route id overflow")
	ErrCatalogRouteIDRequired      = errors.New("catalog route id is required")
	ErrCatalogGroupIDRequired      = errors.New("catalog group id is required")
	ErrCatalogDuplicateRouteID     = errors.New("catalog route id must be unique")
	ErrCatalogInvalidRouteRange    = errors.New("catalog route range is invalid")
	ErrCatalogInvalidVersionRecord = errors.New("catalog version record is invalid")
	ErrCatalogInvalidNextRouteID   = errors.New("catalog next route id record is invalid")
	ErrCatalogInvalidRouteRecord   = errors.New("catalog route record is invalid")
	ErrCatalogInvalidRouteState    = errors.New("catalog route state is invalid")
	ErrCatalogInvalidRouteKey      = errors.New("catalog route key is invalid")
	ErrCatalogRouteKeyIDMismatch   = errors.New("catalog route key and record route id mismatch")
	ErrCatalogRouteV2WriteDisabled = errors.New("catalog route descriptor v2 writes are disabled")
)
View Source 
var (
	ErrCatalogInvalidDeltaRecord   = errors.New("catalog delta record is invalid")
	ErrCatalogInvalidDeltaMutation = errors.New("catalog delta mutation is invalid")
	ErrCatalogDeltaVersionGap      = errors.New("catalog delta version is not contiguous")
	ErrCatalogDeltaBaseMismatch    = errors.New("catalog delta base version does not match durable catalog")
	ErrCatalogDeltaVersionFuture   = errors.New("catalog delta cursor is ahead of catalog version")
	ErrCatalogDeltaLimitInvalid    = errors.New("catalog delta limit must be positive")
)
View Source 
var (
	ErrEngineSnapshotVersionStale = errors.New("engine snapshot version is stale")
	ErrEngineSnapshotDuplicateID  = errors.New("engine snapshot has duplicate route id")
	ErrEngineSnapshotRouteOverlap = errors.New("engine snapshot has overlapping routes")
	ErrEngineSnapshotRouteOrder   = errors.New("engine snapshot has invalid route order")
	ErrEngineDeltaVersionGap      = errors.New("engine catalog delta version is not contiguous")
)
View Source 
var (
	ErrCatalogWatchClientRequired = errors.New("catalog watch client is required")
	ErrCatalogWatchEventInvalid   = errors.New("catalog watch event is invalid")
)
View Source 
var (
	ErrMigrationReservedRange      = errors.New("migration range intersects reserved control prefix")
	ErrMigrationInvalidRoute       = errors.New("migration route is invalid")
	ErrMigrationDataMoveRequired   = errors.New("migration data move is not implemented")
	ErrMigrationSourceRouteChanged = errors.New("migration source route does not match split job")
)
View Source 
var (
	ErrCatalogSplitJobIDRequired          = errors.New("catalog split job id is required")
	ErrCatalogSplitJobSourceRouteRequired = errors.New("catalog split job source route id is required")
	ErrCatalogSplitJobTargetGroupRequired = errors.New("catalog split job target group id is required")
	ErrCatalogInvalidSplitJobRecord       = errors.New("catalog split job record is invalid")
	ErrCatalogInvalidSplitJobPhase        = errors.New("catalog split job phase is invalid")
	ErrCatalogInvalidSplitJobBarrierState = errors.New("catalog split job barrier state is invalid")
	ErrCatalogInvalidSplitJobExportPhase  = errors.New("catalog split job export phase is invalid")
	ErrCatalogInvalidSplitJobKey          = errors.New("catalog split job key is invalid")
	ErrCatalogInvalidNextSplitJobID       = errors.New("catalog next split job id record is invalid")
	ErrCatalogSplitJobIDOverflow          = errors.New("catalog split job id overflow")
	ErrCatalogSplitJobKeyIDMismatch       = errors.New("catalog split job key and record job id mismatch")
	ErrCatalogSplitJobConflict            = errors.New("catalog split job conflict")
	ErrCatalogSplitJobTerminalRequired    = errors.New("catalog split job terminal state is required")
	ErrSplitJobOverlap                    = errors.New("split job overlaps requested route")
)
View Source 
var ErrEngineRequired = errors.New("engine is required")

Functions

func CatalogDeltaFloorKey 

func CatalogDeltaFloorKey() []byte

CatalogDeltaFloorKey stores the oldest retained delta version.

func CatalogDeltaKey 

func CatalogDeltaKey(version uint64) []byte

CatalogDeltaKey returns the ordered durable key for a catalog delta.

func CatalogDeltaVersionFromKey 

func CatalogDeltaVersionFromKey(key []byte) (uint64, bool)

CatalogDeltaVersionFromKey parses the version from a durable delta key.

func CatalogNextRouteIDKey 

func CatalogNextRouteIDKey() []byte

CatalogNextRouteIDKey returns the reserved key used for next route id storage.

func CatalogNextSplitJobIDKey 

func CatalogNextSplitJobIDKey() []byte

CatalogNextSplitJobIDKey returns the reserved key used for split-job ID allocation.

func CatalogRouteIDFromKey 

func CatalogRouteIDFromKey(key []byte) (uint64, bool)

CatalogRouteIDFromKey parses the route ID from a catalog route key.

func CatalogRouteKey 

func CatalogRouteKey(routeID uint64) []byte

CatalogRouteKey returns the reserved key used for a route descriptor.

func CatalogSplitJobHistoryKey 

func CatalogSplitJobHistoryKey(terminalAtMs int64, jobID uint64) []byte

CatalogSplitJobHistoryKey returns the reserved history key for a terminal split job.

func CatalogSplitJobHistoryKeyParts 

func CatalogSplitJobHistoryKeyParts(key []byte) (int64, uint64, bool)

CatalogSplitJobHistoryKeyParts parses terminal time and job ID from a history key.

func CatalogSplitJobIDFromKey 

func CatalogSplitJobIDFromKey(key []byte) (uint64, bool)

CatalogSplitJobIDFromKey parses the job ID from a live split-job key.

func CatalogSplitJobKey 

func CatalogSplitJobKey(jobID uint64) []byte

CatalogSplitJobKey returns the reserved live-job key for a split job.

func CatalogVersionKey 

func CatalogVersionKey() []byte

CatalogVersionKey returns the reserved key used for catalog version storage.

func CloneBytes 

func CloneBytes(b []byte) []byte

CloneBytes returns a copied byte slice.

func DecodeCatalogNextRouteID 

func DecodeCatalogNextRouteID(raw []byte) (uint64, error)

DecodeCatalogNextRouteID deserializes a next route id record.

func DecodeCatalogNextSplitJobID 

func DecodeCatalogNextSplitJobID(raw []byte) (uint64, error)

DecodeCatalogNextSplitJobID deserializes a next split-job ID record.

func DecodeCatalogVersion 

func DecodeCatalogVersion(raw []byte) (uint64, error)

DecodeCatalogVersion deserializes a catalog version record.

func EncodeCatalogDelta 

func EncodeCatalogDelta(delta CatalogDelta) ([]byte, error)

EncodeCatalogDelta serializes a validated catalog transition.

func EncodeCatalogNextRouteID 

func EncodeCatalogNextRouteID(nextRouteID uint64) []byte

EncodeCatalogNextRouteID serializes a next route id record.

func EncodeCatalogNextSplitJobID 

func EncodeCatalogNextSplitJobID(nextJobID uint64) []byte

EncodeCatalogNextSplitJobID serializes a next split-job ID record.

func EncodeCatalogVersion 

func EncodeCatalogVersion(version uint64) []byte

EncodeCatalogVersion serializes a catalog version record.

func EncodeRouteDescriptor 

func EncodeRouteDescriptor(route RouteDescriptor) ([]byte, error)

EncodeRouteDescriptor serializes a route descriptor record.

func EncodeRouteDescriptorForCatalogWrite 

func EncodeRouteDescriptorForCatalogWrite(route RouteDescriptor, allowV2 bool) ([]byte, error)

func EncodeRouteDescriptorForCatalogWriteWithSplitAtHLCOffset 

func EncodeRouteDescriptorForCatalogWriteWithSplitAtHLCOffset(route RouteDescriptor, allowV2 bool) ([]byte, uint64, error)

EncodeRouteDescriptorForCatalogWriteWithSplitAtHLCOffset serializes a route descriptor and returns the byte offset of its SplitAtHLC field.

func EncodeSplitJob 

func EncodeSplitJob(job SplitJob) ([]byte, error)

EncodeSplitJob serializes a SplitJob record as version byte + protobuf body.

func IsCatalogDeltaKey 

func IsCatalogDeltaKey(key []byte) bool

IsCatalogDeltaKey reports whether key belongs to the delta log keyspace.

func IsCatalogRouteKey 

func IsCatalogRouteKey(key []byte) bool

IsCatalogRouteKey reports whether key belongs to the route catalog keyspace.

func IsCatalogSplitJobHistoryKey 

func IsCatalogSplitJobHistoryKey(key []byte) bool

IsCatalogSplitJobHistoryKey reports whether key belongs to the split-job history namespace.

func IsCatalogSplitJobKey 

func IsCatalogSplitJobKey(key []byte) bool

IsCatalogSplitJobKey reports whether key belongs to the live split-job namespace.

func IsMigrationKnownInternalKey 

func IsMigrationKnownInternalKey(key []byte) bool

IsMigrationKnownInternalKey reports whether a raw key belongs to a concrete internal family owned by an explicit export bracket, by the txn-lock drain, or by a peer-local store that the migrator must not export at all.

func MigrationKnownInternalPrefixes 

func MigrationKnownInternalPrefixes() [][]byte

MigrationKnownInternalPrefixes returns the concrete internal data/control prefixes that the user bracket must exclude. It intentionally does not include broad umbrellas such as !txn|, !ddb|, !sqs|, !s3|, or !stream|.

func NextRouteIDFloor 

func NextRouteIDFloor(routes []RouteDescriptor) (uint64, error)

NextRouteIDFloor returns the minimum valid next route ID for routes. It is shared by catalog persistence and split planning to keep route-ID allocation rules consistent.

func NextSplitJobIDFloor 

func NextSplitJobIDFloor(jobs []SplitJob) (uint64, error)

NextSplitJobIDFloor returns the minimum valid next split-job ID for jobs.

func RunCatalogWatcher 

func RunCatalogWatcher(ctx context.Context, catalog *CatalogStore, engine *Engine, logger *slog.Logger, opts ...CatalogWatcherOption) error

RunCatalogWatcher runs CatalogWatcher with optional logger override.

func ValidateMigrationRouteRange 

func ValidateMigrationRouteRange(routeStart, routeEnd []byte) error

ValidateMigrationRouteRange rejects route intervals that intersect reserved distribution/migration control namespaces.

Types

type CatalogChangeSet 

type CatalogChangeSet struct {
	Reset  *CatalogSnapshot
	Deltas []CatalogDelta
}

CatalogChangeSet contains either a snapshot reset or a contiguous delta batch. Reset is non-nil only when the requested cursor predates retention.

type CatalogDelta 

type CatalogDelta struct {
	PreviousVersion uint64
	Version         uint64
	Mutations       []CatalogRouteMutation
}

CatalogDelta is the durable transition from PreviousVersion to Version.

func DecodeCatalogDelta 

func DecodeCatalogDelta(raw []byte) (CatalogDelta, error)

DecodeCatalogDelta deserializes and validates a catalog transition.

type CatalogMigrationStoreResolver 

type CatalogMigrationStoreResolver func(groupID uint64) (store.MVCCStore, error)

CatalogMigrationStoreResolver resolves the local MVCC store for a data group.

type CatalogMutationOp 

type CatalogMutationOp byte

CatalogMutationOp describes one route-level catalog change. 

const (
	CatalogMutationUpsert CatalogMutationOp = 1
	CatalogMutationDelete CatalogMutationOp = 2
)

type CatalogRouteMutation 

type CatalogRouteMutation struct {
	Op      CatalogMutationOp
	RouteID uint64
	Route   RouteDescriptor
}

CatalogRouteMutation changes one durable route descriptor.

type CatalogSnapshot 

type CatalogSnapshot struct {
	Version uint64
	Routes  []RouteDescriptor
	ReadTS  uint64
}

CatalogSnapshot is a point-in-time snapshot of the route catalog.

func EnsureCatalogSnapshot 

func EnsureCatalogSnapshot(ctx context.Context, catalog *CatalogStore, engine *Engine) (CatalogSnapshot, error)

EnsureCatalogSnapshot makes engine and durable catalog consistent at startup.

If the durable catalog is empty (version 0, no routes), the current in-memory engine routes are persisted as the initial catalog snapshot with generated non-zero RouteIDs. Then the resolved snapshot is applied back to engine.

type CatalogSnapshotObserver 

type CatalogSnapshotObserver func(CatalogSnapshot)

CatalogSnapshotObserver receives each snapshot successfully applied by the watcher. Implementations run synchronously on the watcher goroutine and must not block.

type CatalogStore 

type CatalogStore struct {
	// contains filtered or unexported fields
}

CatalogStore provides persistence helpers for route catalog state.

func NewCatalogStore 

func NewCatalogStore(st store.MVCCStore, opts ...CatalogStoreOption) *CatalogStore

NewCatalogStore creates a route catalog persistence helper.

func (*CatalogStore) AllowsRouteDescriptorV2Writes 

func (s *CatalogStore) AllowsRouteDescriptorV2Writes() bool

func (*CatalogStore) BuildDeltaMutationsAt 

func (s *CatalogStore) BuildDeltaMutationsAt(ctx context.Context, readTS uint64, delta CatalogDelta) ([]*store.KVPairMutation, error)

BuildDeltaMutationsAt returns the durable delta/floor/retention mutations that must be committed atomically with the catalog version transition.

func (*CatalogStore) ChangesSince 

func (s *CatalogStore) ChangesSince(ctx context.Context, afterVersion uint64, limit int) (CatalogChangeSet, error)

ChangesSince reads a consistent reconnect batch after afterVersion.

func (*CatalogStore) CreateSplitJob 

func (s *CatalogStore) CreateSplitJob(ctx context.Context, job SplitJob) error

CreateSplitJob creates a live split job and advances next_job_id when needed.

func (*CatalogStore) DeleteSplitJob 

func (s *CatalogStore) DeleteSplitJob(ctx context.Context, jobID uint64) error

DeleteSplitJob deletes a live split job.

func (*CatalogStore) DeltaFloor 

func (s *CatalogStore) DeltaFloor(ctx context.Context) (uint64, error)

DeltaFloor returns the oldest retained durable delta version. Zero means the catalog predates delta logging and a watcher must request a snapshot reset.

func (*CatalogStore) LatestCommitTS 

func (s *CatalogStore) LatestCommitTS() uint64

LatestCommitTS returns the local catalog store watermark without reading catalog data. Callers use it as the pre-Phase-D legacy timestamp input before selecting the transaction snapshot through the dedicated TSO.

func (*CatalogStore) ListSplitJobs 

func (s *CatalogStore) ListSplitJobs(ctx context.Context) ([]SplitJob, error)

ListSplitJobs returns live and history split jobs at the latest timestamp.

func (*CatalogStore) ListSplitJobsAt 

func (s *CatalogStore) ListSplitJobsAt(ctx context.Context, ts uint64) ([]SplitJob, error)

ListSplitJobsAt returns live and history split jobs at a snapshot timestamp.

func (*CatalogStore) MoveSplitJobToHistory 

func (s *CatalogStore) MoveSplitJobToHistory(ctx context.Context, expected SplitJob, job SplitJob) error

MoveSplitJobToHistory moves a terminal split job from live state to history.

func (*CatalogStore) NextRouteID 

func (s *CatalogStore) NextRouteID(ctx context.Context) (uint64, error)

NextRouteID reads the next route id counter from catalog metadata.

func (*CatalogStore) NextRouteIDAt 

func (s *CatalogStore) NextRouteIDAt(ctx context.Context, ts uint64) (uint64, error)

NextRouteIDAt reads the next route id counter at a given snapshot timestamp.

func (*CatalogStore) NextSplitJobID 

func (s *CatalogStore) NextSplitJobID(ctx context.Context) (uint64, error)

NextSplitJobID reads the next split-job ID counter from catalog metadata.

func (*CatalogStore) NextSplitJobIDAt 

func (s *CatalogStore) NextSplitJobIDAt(ctx context.Context, ts uint64) (uint64, error)

NextSplitJobIDAt reads the next split-job ID counter at a given snapshot timestamp.

func (*CatalogStore) Save 

func (s *CatalogStore) Save(ctx context.Context, expectedVersion uint64, routes []RouteDescriptor) (CatalogSnapshot, error)

Save updates the route catalog using optimistic version checks and bumps the catalog version by exactly one on success.

func (*CatalogStore) SaveSplitJob 

func (s *CatalogStore) SaveSplitJob(ctx context.Context, expected SplitJob, job SplitJob) error

SaveSplitJob updates a live split job if it still matches expected.

func (*CatalogStore) Snapshot 

func (s *CatalogStore) Snapshot(ctx context.Context) (CatalogSnapshot, error)

Snapshot reads a consistent route catalog snapshot at the store's latest known commit timestamp.

func (*CatalogStore) SnapshotAt 

func (s *CatalogStore) SnapshotAt(ctx context.Context, ts uint64) (CatalogSnapshot, error)

SnapshotAt reads a consistent route catalog snapshot at a specific MVCC timestamp.

func (*CatalogStore) SplitJob 

func (s *CatalogStore) SplitJob(ctx context.Context, jobID uint64) (SplitJob, bool, error)

SplitJob reads a live or history split job by ID at the latest timestamp.

func (*CatalogStore) SplitJobAt 

func (s *CatalogStore) SplitJobAt(ctx context.Context, jobID uint64, ts uint64) (SplitJob, bool, error)

SplitJobAt reads a live or history split job by ID at a snapshot timestamp.

func (*CatalogStore) Version 

func (s *CatalogStore) Version(ctx context.Context) (uint64, error)

Version reads only the durable catalog version at the latest commit timestamp.

type CatalogStoreOption 

type CatalogStoreOption func(*CatalogStore)

func WithCatalogMigrationStoreResolver 

func WithCatalogMigrationStoreResolver(resolver CatalogMigrationStoreResolver) CatalogStoreOption

WithCatalogMigrationStoreResolver makes split-job finalization retire target-local migration metadata from the target data group store.

func WithCatalogRouteDescriptorV2Writes 

func WithCatalogRouteDescriptorV2Writes(enabled bool) CatalogStoreOption

type CatalogWatchClientProvider 

type CatalogWatchClientProvider func(context.Context) (pb.DistributionClient, error)

CatalogWatchClientProvider resolves the Distribution endpoint used for one capability negotiation and stream attempt. Production resolves the current default-group leader again after every disconnect.

type CatalogWatcher 

type CatalogWatcher struct {
	// contains filtered or unexported fields
}

CatalogWatcher periodically refreshes Engine from durable catalog snapshots.

func NewCatalogWatcher 

func NewCatalogWatcher(catalog *CatalogStore, engine *Engine, opts ...CatalogWatcherOption) *CatalogWatcher

NewCatalogWatcher creates a watcher that polls the durable route catalog and applies newer snapshots to the in-memory engine.

func (*CatalogWatcher) Run 

func (w *CatalogWatcher) Run(ctx context.Context) error

Run starts polling and only returns when ctx is canceled or initialization requirements are not met. Snapshot read/apply failures are retried.

func (*CatalogWatcher) SyncOnce 

func (w *CatalogWatcher) SyncOnce(ctx context.Context) error

SyncOnce applies durable catalog deltas and notifies observers when the engine reaches a catalog snapshot that has not been observed yet.

type CatalogWatcherOption 

type CatalogWatcherOption func(*CatalogWatcher)

CatalogWatcherOption customizes CatalogWatcher behavior.

func WithCatalogWatcherBatchSize 

func WithCatalogWatcherBatchSize(batchSize int) CatalogWatcherOption

WithCatalogWatcherBatchSize sets the maximum number of deltas applied by one synchronization pass.

func WithCatalogWatcherInterval 

func WithCatalogWatcherInterval(interval time.Duration) CatalogWatcherOption

WithCatalogWatcherInterval sets the catalog polling interval.

func WithCatalogWatcherLogger 

func WithCatalogWatcherLogger(logger *slog.Logger) CatalogWatcherOption

WithCatalogWatcherLogger sets the logger for watcher background retries.

func WithCatalogWatcherSnapshotObserver 

func WithCatalogWatcherSnapshotObserver(observer CatalogSnapshotObserver) CatalogWatcherOption

WithCatalogWatcherSnapshotObserver installs a post-apply snapshot callback.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine holds in-memory metadata of routes and provides timestamp generation.

func NewEngine 

func NewEngine() *Engine

NewEngine creates an Engine with no hotspot splitting.

func NewEngineWithDefaultRoute 

func NewEngineWithDefaultRoute() *Engine

NewEngineWithDefaultRoute creates an Engine and registers a default route covering the full keyspace with a default group ID. The default route is also recorded in the M2 history ring as the version-0 snapshot so transactions that observed catalogVersion = 0 can resolve their read-set owner through SnapshotAt(0).

func (*Engine) AppliedCatalogSnapshot 

func (e *Engine) AppliedCatalogSnapshot() CatalogSnapshot

AppliedCatalogSnapshot returns the routes the engine is currently serving, paired with the catalog version they were applied at. It is the engine's own view rather than the persisted catalog's, which is what a caller mid-catch-up needs: the store may already be several delta batches ahead.

ReadTS is left zero because these routes come from applied deltas, not from a point-in-time catalog read.

func (*Engine) ApplyDelta 

func (e *Engine) ApplyDelta(delta CatalogDelta) error

ApplyDelta atomically publishes one contiguous catalog transition. Readers observe either the complete previous route table or the complete next table.

func (*Engine) ApplySnapshot 

func (e *Engine) ApplySnapshot(snapshot CatalogSnapshot) error

ApplySnapshot atomically replaces all in-memory routes with the provided catalog snapshot when the snapshot version is newer.

func (*Engine) Current 

func (e *Engine) Current() (RouteHistorySnapshot, bool)

Current returns the route catalog snapshot at the engine's current catalogVersion. Returns (zero, false) when the history ring has not been initialised (bare-struct Engine). Used by the M3 Composed-1 cross-version-read fence (design doc §4.4) — the gate compares the txn's observed-version owner against the current owner so a route shift between BeginTxn and Commit is caught before it can produce a G1c anomaly across a cross-group MoveRange / SplitRange.

func (*Engine) GetIntersectingRoutes 

func (e *Engine) GetIntersectingRoutes(start, end []byte) []Route

GetIntersectingRoutes returns all routes whose key ranges intersect with [start, end). A route [rStart, rEnd) intersects with [start, end) if: - rStart < end (or end is nil, meaning unbounded scan) - start < rEnd (or rEnd is nil, meaning unbounded route)

func (*Engine) GetIntersectingRoutesWithVersion 

func (e *Engine) GetIntersectingRoutesWithVersion(start, end []byte) ([]Route, uint64)

GetIntersectingRoutesWithVersion returns intersecting routes and the catalog version from the same locked snapshot.

func (*Engine) GetRoute 

func (e *Engine) GetRoute(key []byte) (Route, bool)

GetRoute finds a route for the given key using right half-open intervals.

func (*Engine) GetRouteWithVersion 

func (e *Engine) GetRouteWithVersion(key []byte) (Route, uint64, bool)

GetRouteWithVersion finds a route and returns the catalog version from the same locked snapshot. Callers can use the version as a read-routing fence.

func (*Engine) HistoryDepth 

func (e *Engine) HistoryDepth() int

HistoryDepth returns the configured ring depth for diagnostics.

func (*Engine) NextTimestamp 

func (e *Engine) NextTimestamp() uint64

NextTimestamp returns a monotonic increasing timestamp.

func (*Engine) ResolveRoutesWithVersion 

func (e *Engine) ResolveRoutesWithVersion(queries ...RouteQuery) ([][]Route, uint64)

ResolveRoutesWithVersion answers every query from a single locked catalog snapshot and returns the one version that produced all of them. Callers that need more than one route candidate must use this instead of repeating GetRouteWithVersion / GetIntersectingRoutesWithVersion: separate calls can straddle a catalog update and pair a stale route with a newer version, which turns that version into a fence the route actually read was never checked against.

func (*Engine) SetHistoryDepthForTest 

func (e *Engine) SetHistoryDepthForTest(depth int)

SetHistoryDepthForTest overrides the FIFO ring depth from outside the package. Test-only. Callers should set the depth before sharing the Engine with concurrent SnapshotAt/Current readers to avoid interleaving surprises around the eviction watermark, but the write itself is lock-protected (e.mu.Lock below) so it is safe to call from any goroutine that does not also expect a consistent SnapshotAt view across the depth change.

Exists so tests in the kv package can drive eviction-trigger scenarios without adding a constructor option just for tests (claude review on PR #894). Production code must use DefaultRouteHistoryDepth (32) or a future operator-exposed config knob.

Fails fast on depth <= 0 (coderabbit minor on PR #895): recordHistorySnapshotLocked's eviction path indexes historyOrder[0], so a zero/negative depth would surface as a confusing index-out-of-range deep in the apply path instead of at the misconfigured test seam. When shrinking depth below the current ring size, evict the excess oldest entries immediately rather than letting the next record see len(historyOrder) > historyDepth (gemini medium on PR #895 — without this trim, the next recordHistorySnapshotLocked's `make([]uint64, len-1, historyDepth)` would panic on len-1 > historyDepth).

func (*Engine) SnapshotAt 

func (e *Engine) SnapshotAt(v uint64) (RouteHistorySnapshot, bool)

SnapshotAt returns the route catalog snapshot recorded at version v. Returns (zero, false) when v is not in the ring — either because v is in the future (> catalogVersion), or because the FIFO ring has evicted v (it was older than the historyDepth-most-recent versions). The M3 Composed-1 gate (design doc §4.3) treats the not-found case as a hard error and triggers a coordinator retry, so retention depth is a liveness knob, not a safety knob.

func (*Engine) Stats 

func (e *Engine) Stats() []Route

Stats returns a snapshot of current ranges and their load counters.

func (*Engine) UpdateRoute 

func (e *Engine) UpdateRoute(start, end []byte, group uint64)

UpdateRoute registers or updates a route for the given key range. Routes are stored sorted by Start.

func (*Engine) Version 

func (e *Engine) Version() uint64

Version returns current route catalog version applied to the engine.

type GRPCCatalogWatcher 

type GRPCCatalogWatcher struct {
	// contains filtered or unexported fields
}

GRPCCatalogWatcher mirrors a remote Distribution catalog into Engine.

func NewGRPCCatalogWatcher 

func NewGRPCCatalogWatcher(client pb.DistributionClient, engine *Engine, opts ...GRPCCatalogWatcherOption) *GRPCCatalogWatcher

NewGRPCCatalogWatcher creates a capability-negotiated remote mirror.

func NewResolvingGRPCCatalogWatcher 

func NewResolvingGRPCCatalogWatcher(
	provider CatalogWatchClientProvider,
	engine *Engine,
	opts ...GRPCCatalogWatcherOption,
) *GRPCCatalogWatcher

NewResolvingGRPCCatalogWatcher creates a mirror that re-resolves its remote endpoint before every reconnect attempt.

func (*GRPCCatalogWatcher) Run 

func (w *GRPCCatalogWatcher) Run(ctx context.Context) error

Run reconnects from the last atomically published Engine version. Servers without capability negotiation use ListRoutes snapshot polling.

type GRPCCatalogWatcherOption 

type GRPCCatalogWatcherOption func(*GRPCCatalogWatcher)

GRPCCatalogWatcherOption customizes a remote catalog mirror.

func WithGRPCCatalogWatcherLogger 

func WithGRPCCatalogWatcherLogger(logger *slog.Logger) GRPCCatalogWatcherOption

WithGRPCCatalogWatcherLogger sets the reconnect logger.

func WithGRPCCatalogWatcherRetryInterval 

func WithGRPCCatalogWatcherRetryInterval(interval time.Duration) GRPCCatalogWatcherOption

WithGRPCCatalogWatcherRetryInterval sets reconnect and legacy polling delay.

type MigrationBracket 

type MigrationBracket struct {
	BracketID             uint64
	Family                uint32
	Start                 []byte
	End                   []byte
	ExcludePrefixes       [][]byte
	ExcludeKnownInternal  bool
	DrainOnly             bool
	RequiresRouteKeyCheck bool
	RequiresDecodedS3     bool
}

MigrationBracket is a raw MVCC export or drain slice used by the migrator.

func PlanExportBrackets 

func PlanExportBrackets(routeStart, routeEnd []byte) ([]MigrationBracket, error)

PlanExportBrackets returns the data-copy bracket plan. Intent locks are deliberately absent because the source drains them route-faithfully before cutover and the target must not materialize in-flight intents as data.

func PlanMigrationBrackets 

func PlanMigrationBrackets(routeStart, routeEnd []byte) ([]MigrationBracket, error)

PlanMigrationBrackets returns the full M2 migration plan, including the drain-only transaction lock bracket. Data export callers should use PlanExportBrackets, which omits drain-only control state.

func (MigrationBracket) ContainsRawKey 

func (b MigrationBracket) ContainsRawKey(rawKey []byte) bool

ContainsRawKey reports whether rawKey is inside the bracket's raw scan interval after applying bracket-local exclusions. Route ownership still requires the caller's RouteKeyFilter for every bracket.

func (MigrationBracket) ContainsRoutedKey 

func (b MigrationBracket) ContainsRoutedKey(rawKey, routeStart, routeEnd []byte, routeKey func([]byte) []byte) bool

ContainsRoutedKey applies both the bracket's raw family interval and its route ownership predicate. S3 bucket-level auxiliary rows do not encode an object route key, so they are matched by bucket route-prefix intersection.

func (MigrationBracket) ContainsRoutedVersion 

func (b MigrationBracket) ContainsRoutedVersion(rawKey, value, routeStart, routeEnd []byte, routeKey func([]byte) []byte) bool

ContainsRoutedVersion is the value-aware variant of ContainsRoutedKey. It is needed for legacy list metadata because old delta keys overlap byte-for-byte with base metadata keys whose user key begins with "d|".

type Route 

type Route struct {
	// RouteID is the durable identifier assigned by route catalog.
	// Zero means ephemeral/non-catalog routes.
	RouteID uint64
	// Start marks the inclusive beginning of the range.
	Start []byte
	// End marks the exclusive end of the range. nil means unbounded.
	End []byte
	// GroupID identifies the raft group for the range starting at Start.
	GroupID uint64
	// State tracks control-plane state for this route.
	State RouteState
	// StagedVisibilityActive allows serving reads to merge staged migration rows.
	StagedVisibilityActive bool
	// MigrationJobID identifies the active staged migration job.
	MigrationJobID uint64
	// MinWriteTSExclusive rejects writes at or below the migration cutover floor.
	MinWriteTSExclusive uint64
	// Load tracks the number of accesses served by this range.
	Load uint64
}

Route represents a mapping from a key range to a raft group. Ranges are right half-open intervals: [Start, End). Start is inclusive and End is exclusive. A nil End denotes an unbounded interval extending to positive infinity.

func RoutesFromCatalogSnapshot 

func RoutesFromCatalogSnapshot(snapshot CatalogSnapshot) ([]Route, error)

RoutesFromCatalogSnapshot validates and materializes the immutable route view contained in a durable catalog snapshot. Callers that need ownership as of an MVCC timestamp must use this view instead of the live Engine.

type RouteDescriptor 

type RouteDescriptor struct {
	RouteID                uint64
	Start                  []byte
	End                    []byte
	GroupID                uint64
	State                  RouteState
	ParentRouteID          uint64
	StagedVisibilityActive bool
	MigrationJobID         uint64
	MinWriteTSExclusive    uint64
	SplitAtHLC             uint64
}

RouteDescriptor is the durable representation of a route.

func CloneRouteDescriptor 

func CloneRouteDescriptor(route RouteDescriptor) RouteDescriptor

CloneRouteDescriptor returns a deep copy of route.

func DecodeRouteDescriptor 

func DecodeRouteDescriptor(raw []byte) (RouteDescriptor, error)

DecodeRouteDescriptor deserializes a route descriptor record.

type RouteHistorySnapshot 

type RouteHistorySnapshot struct {
	// contains filtered or unexported fields
}

RouteHistorySnapshot is a point-in-time view of the route catalog at a specific version. Returned by Engine.SnapshotAt for the M3 Composed-1 commit-time gate. Carries an immutable copy of the catalog's routes at the recorded version so a caller can resolve ownership without holding the Engine lock.

func (RouteHistorySnapshot) IntersectingRoutes 

func (s RouteHistorySnapshot) IntersectingRoutes(start, end []byte) []Route

IntersectingRoutes returns every route whose range intersects [start, end) in this snapshot. A nil end denotes +infinity.

func (RouteHistorySnapshot) OwnerOf 

func (s RouteHistorySnapshot) OwnerOf(key []byte) (uint64, bool)

OwnerOf returns the Raft group ID that owned key at this snapshot's version. Returns (0, false) when no route covers key (the pre-bootstrap state or an explicitly-uncovered range). Mirrors Engine.GetRoute's right-half-open interval semantics but against the historical snapshot, not the live engine state.

Routes are sorted by Start (recordHistorySnapshotLocked clones from e.routes, which Engine.UpdateRoute / routesFromCatalog keep sorted), so the scan can break the moment key < r.Start — every later route has a strictly greater Start and cannot cover key either. This matters because M3 puts OwnerOf on every txn commit's apply path (claude review on PR #894 — break-vs-continue lifts the worst-case scan from O(N) to "first non-covering gap" without changing the resolution semantics).

func (RouteHistorySnapshot) RouteOf 

func (s RouteHistorySnapshot) RouteOf(key []byte) (Route, bool)

RouteOf returns the route that covered key at this snapshot's version.

func (RouteHistorySnapshot) Version 

func (s RouteHistorySnapshot) Version() uint64

Version returns the catalog version this snapshot was recorded at.

type RouteQuery 

type RouteQuery struct {
	Start []byte
	End   []byte
	Exact bool
}

RouteQuery is one lookup for ResolveRoutesWithVersion. Exact resolves the single route containing Start; otherwise every route intersecting [Start, End) is returned.

type RouteState 

type RouteState byte

RouteState describes the control-plane state of a route. 

const (
	// RouteStateActive is a normal serving route.
	RouteStateActive RouteState = iota
	// RouteStateWriteFenced blocks writes during cutover.
	RouteStateWriteFenced
	// RouteStateMigratingSource means range data is being copied out.
	RouteStateMigratingSource
	// RouteStateMigratingTarget means range data is being copied in.
	RouteStateMigratingTarget
)

type SplitJob 

type SplitJob struct {
	JobID                            uint64
	SourceRouteID                    uint64
	SplitKey                         []byte
	TargetGroupID                    uint64
	Phase                            SplitJobPhase
	RetryPhase                       SplitJobPhase
	AbandonFromPhase                 SplitJobPhase
	SnapshotTS                       uint64
	SnapshotMinAdmittedTS            uint64
	WriteTrackerArmed                bool
	DeltaFloor                       uint64
	PostFenceDrainCompleted          bool
	FenceTS                          uint64
	CutoverVersion                   uint64
	CutoverReadFenceState            SplitJobBarrierState
	TargetStagedReadinessState       SplitJobBarrierState
	SourceCutoverReadFenceAckCursor  []byte
	TargetStagedReadinessAckCursor   []byte
	Cursor                           []byte
	MaxImportedTS                    uint64
	TargetPromotionDone              bool
	PromotionCompletedTS             uint64
	FenceCatalogVersion              uint64
	FenceAckCursor                   []byte
	SourceCutoverAckCursor           []byte
	SourceReadDrainCursor            []byte
	TargetClearedDescriptorAckCursor []byte
	BracketProgress                  []SplitJobBracketProgress
	SourceRetentionPinTS             uint64
	LastError                        string
	StartedAtMs                      int64
	UpdatedAtMs                      int64
	TerminalAtMs                     int64
}

SplitJob is the durable default-group state for a range split migration.

func AdvanceSameGroupNoop 

func AdvanceSameGroupNoop(job SplitJob, source RouteDescriptor, nowMs int64) (SplitJob, error)

AdvanceSameGroupNoop completes the PR4 same-group path without attempting data movement. Cross-group data copy is added by later M2 PRs.

func CloneSplitJob 

func CloneSplitJob(job SplitJob) SplitJob

func DecodeSplitJob 

func DecodeSplitJob(raw []byte) (SplitJob, error)

DecodeSplitJob deserializes a SplitJob record.

func InitializeSplitJobPlan 

func InitializeSplitJobPlan(job SplitJob, source RouteDescriptor, nowMs int64) (SplitJob, error)

InitializeSplitJobPlan validates the source route and seeds the job's bracket progress for the moving right child [SplitKey, source.End).

type SplitJobBarrierState 

type SplitJobBarrierState byte

SplitJobBarrierState is a restart witness for per-voter barrier progress. 

const (
	SplitJobBarrierNone SplitJobBarrierState = iota
	SplitJobBarrierArming
	SplitJobBarrierArmed
	SplitJobBarrierClearing
)

type SplitJobBracketProgress 

type SplitJobBracketProgress struct {
	BracketID         uint64
	Family            uint32
	ExportPhase       SplitJobExportPhase
	Cursor            []byte
	Done              bool
	ScannedBytes      uint64
	AcceptedRows      uint64
	LastAckedBatchSeq uint64
}

SplitJobBracketProgress is per-family export-bracket resume state.

func SplitJobBracketProgressForPlan 

func SplitJobBracketProgressForPlan(brackets []MigrationBracket, phase SplitJobExportPhase) []SplitJobBracketProgress

SplitJobBracketProgressForPlan creates durable per-bracket resume state for a SplitJob phase.

type SplitJobExportPhase 

type SplitJobExportPhase byte

SplitJobExportPhase records which export window a bracket is scanning. 

const (
	SplitJobExportPhaseNone SplitJobExportPhase = iota
	SplitJobExportPhaseBackfill
	SplitJobExportPhaseDeltaCopy
)

type SplitJobPhase 

type SplitJobPhase byte

SplitJobPhase is the durable phase of a split migration job. 

const (
	SplitJobPhaseNone SplitJobPhase = iota
	SplitJobPhasePlanned
	SplitJobPhaseBackfill
	SplitJobPhaseFence
	SplitJobPhaseDeltaCopy
	SplitJobPhaseCutover
	SplitJobPhaseCleanup
	SplitJobPhaseDone
	SplitJobPhaseFailed
	SplitJobPhaseAbandoning
	SplitJobPhaseAbandoned
)

Source Files

View all Source files

Directories

Path Synopsis

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.