kv

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: 40 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// DynamoTableMetaPrefix prefixes DynamoDB table metadata keys.
	DynamoTableMetaPrefix = "!ddb|meta|table|"
	// DynamoTableGenerationPrefix prefixes DynamoDB table generation keys.
	DynamoTableGenerationPrefix = "!ddb|meta|gen|"
	// DynamoItemPrefix prefixes DynamoDB item storage keys.
	DynamoItemPrefix = "!ddb|item|"
	// DynamoGSIPrefix prefixes DynamoDB GSI storage keys.
	DynamoGSIPrefix = "!ddb|gsi|"
)
View Source 
const HLCLogicalBits = hlcLogicalBits

HLCLogicalBits is the number of low bits in an HLC timestamp reserved for the in-memory logical counter (vs the upper bits which encode the Raft-agreed wall-clock millis). Exported so downstream tools — admin dashboard ISO-8601 formatting being the motivating case — can recover the physical half without hard-coding a magic number that silently drifts when the layout changes (Claude Issue 4 on PR #658).

View Source 
const MaxTSOBatchSize = maxHLCBatchSize

MaxTSOBatchSize is the largest consecutive timestamp window accepted by a TSO allocator or the Distribution.GetTimestamp RPC.

View Source 
const ObservedRouteVersionZero = ^uint64(0)

ObservedRouteVersionZero encodes a transaction pinned to catalog version 0. The protobuf field's literal zero remains the legacy/unpinned sentinel, so a real version-0 observation needs a distinct value. MaxUint64 is reserved for this internal encoding; normal catalog versions are far below this boundary.

View Source 
const (
	// TxnKeyPrefix is the common prefix shared by all transaction internal
	// key namespaces. All per-namespace prefixes below are derived from it.
	// NOTE: store/store.go duplicates this literal as txnInternalKeyPrefix
	// because an import cycle prevents store from importing kv.
	TxnKeyPrefix = "!txn|"
)
View Source 
const TxnMetaPrefix = txnMetaPrefix

TxnMetaPrefix is the key prefix used for transaction metadata mutations.

Variables

View Source 
var (
	ErrInvalidBackupPin     = errors.New("backup pin is invalid")
	ErrTooManyActiveBackups = errors.New("too many active backup pins")
)
View Source 
var (
	ErrBackupWireMalformed = errors.New("backup fsm wire payload is malformed")
	ErrBackupWireSubtype   = errors.New("backup fsm wire subtype is unknown")
)
View Source 
var (
	ErrCrossShardMutationBatchNotSupported = errors.New("cross-shard mutation batches are not supported")
	ErrReadRouteVersionUnavailable         = errors.New("read route version is not locally available")
	ErrFilesystemPlacementTargetNotFound   = errors.New("filesystem placement target group has no routable home slot")
)
View Source 
var (
	ErrTSOAllocatorRequired  = errors.New("tso: allocator is required")
	ErrTSOCoordinatorNil     = errors.New("tso: coordinator is required")
	ErrTSOClockNil           = errors.New("tso: coordinator clock is nil")
	ErrInvalidTSOBatchSize   = errors.New("tso: invalid batch size")
	ErrTSOPhaseDInactive     = errors.New("tso: phase D is not active")
	ErrTSOTimestampInvalid   = errors.New("tso: timestamp is not a durable phase-D allocation")
	ErrTSOTimestampPrePhaseD = errors.New("tso: timestamp predates phase D")
	ErrTSOReadVoucherLimit   = errors.New("tso: applied read timestamp voucher limit reached")
)
View Source 
var (
	ErrTSOStateMachineInvalidEntry      = errors.New("tso fsm: invalid entry")
	ErrTSOLegacyEncryptionEntryRejected = errors.New("tso fsm: legacy encryption entry rejected")
)
View Source 
var (
	ErrTSOGroupRequired       = errors.New("tso: dedicated raft group is required")
	ErrTSOStateRequired       = errors.New("tso: dedicated state machine is required")
	ErrTSOFloorProviderNeeded = errors.New("tso: authoritative commit floor provider is required")
	ErrTSONotLeader           = errors.New("tso: not leader")
	ErrTSOProtocolUnsupported = errors.New("tso: leader does not support durable timestamp windows")
)
View Source 
var (
	ErrInvalidTSOMode          = errors.New("tso: invalid runtime mode")
	ErrTSOModeRollback         = errors.New("tso: runtime mode rollback is prohibited")
	ErrUnsafeTSOModeTransition = errors.New("tso: runtime mode transition skips a required phase")
)
View Source 
var (
	ErrTxnMetaMissing        = errors.New("txn meta missing")
	ErrTxnInvalidMeta        = errors.New("txn meta invalid")
	ErrTxnLocked             = errors.New("txn locked")
	ErrTxnCommitTSRequired   = errors.New("txn commit ts required")
	ErrTxnAlreadyCommitted   = errors.New("txn already committed")
	ErrTxnAlreadyAborted     = errors.New("txn already aborted")
	ErrTxnPrimaryKeyRequired = errors.New("txn primary key required")
	// ErrTxnDedupRequiresSingleShard is returned when a transaction request
	// carries OperationGroup.PrevCommitTS (the option-2 one-phase dedup probe
	// key) but its mutations or read keys span shards. The 2PC log builders
	// encode only CommitTS, so silently honoring such a request would drop
	// the probe at the FSM and let the original duplicate-elements anomaly
	// reappear. See codex P2 in PR #796 and the design doc.
	ErrTxnDedupRequiresSingleShard = errors.New("txn dedup (prev_commit_ts) requires a single-shard write set")
	// ErrTxnSecondaryRouteShiftedAfterPrimaryCommit is returned by
	// dispatchMultiShardTxn when a per-secondary commit (after primary
	// has durably committed) fails its M3 verifyComposed1 check —
	// i.e. the route catalog moved between primary-COMMIT and the
	// secondary-COMMIT, the secondary's FSM rejects with a Composed-1
	// sentinel, and we cannot transparently recover (the prepared
	// lock lives at the old gid; the new owner per the catalog has no
	// commit record).  The 2PC contract is half-broken at this point:
	// the primary's write is durable but at least one secondary's
	// write is missing.  Surfacing this explicitly (rather than
	// swallowing per the original best-effort semantic OR silently
	// landing the write on a stale owner per a dropped-gate fix) is
	// the least bad outcome — the caller knows the txn state is
	// uncertain and can do application-level recovery.
	//
	// This is a DIFFERENT sentinel from ErrComposed1Violation by
	// design: the M4 retry path in dispatchTxnWithComposed1Retry
	// matches ErrComposed1Violation / ErrComposed1VersionGCd and
	// would otherwise loop here, re-prewriting against the same old
	// gid that already has the first attempt's prepared lock — pure
	// wasted work since the route catalog won't move backward.
	// codex P1 on 6202b964 (PR #900) raised the silent-partial-commit
	// hazard; codex P1 on d8487672 (PR #900) raised the symmetric
	// hazard of disabling the gate.  This sentinel resolves both by
	// keeping the gate active and surfacing the error fatally.
	ErrTxnSecondaryRouteShiftedAfterPrimaryCommit = errors.New("txn secondary commit failed after primary commit: route catalog shifted")
)
View Source 
var ErrBackupApply = errors.New("backup fsm apply failed")
View Source 
var ErrBackupSnapshotBlocked = errors.Wrap(
	raftengine.ErrSnapshotDeferred,
	"active backup pin blocks fsm snapshot",
)

ErrBackupSnapshotBlocked refuses an FSM snapshot while a backup pin is open. It is marked with raftengine.ErrSnapshotDeferred so the engine treats the refusal as "not now" and skips this snapshot round: without the mark the engine's run loop takes any Snapshot error as fatal and terminates the replica, so ordinary writes during a long backup would kill replicas instead of merely postponing compaction.

View Source 
var ErrBackupTimestampFenced = errors.New("backup timestamp fence rejects stale write")
View Source 
var ErrCeilingExpired = errors.New("hlc: physical/logical ceiling exhausted; refusing to issue persistence timestamp")

ErrCeilingExpired is returned by HLC.NextFenced() when the Raft-agreed physical ceiling has expired or the current in-memory logical window has been exhausted — i.e. issuing another fenced timestamp would require a physical millisecond above the committed ceiling. Callers MUST refuse to commit and propagate this to the client.

This implements HLC-4 precondition (iii) from docs/design/2026_05_28_implemented_tla_safety_spec.md §5.1: every persistence-grade ts allocation is gated on `wall_now < physicalCeiling`. The TLA+ MCHLC_gap.cfg counterexample at depth 5 demonstrates the safety property this enforces.

Pre-bootstrap (ceiling == 0) is intentionally NOT fenced: there is no prior leader to protect against and tests / demo clusters need to issue ts before the first RunHLCLeaseRenewal cycle. Strict bootstrap fencing is a follow-up consideration; the current semantics match the spec wherever the prior-leader hazard is real.

View Source 
var ErrComposed1VersionGCd = errors.New("composed-1: observed catalog version evicted from history ring; retry")

ErrComposed1VersionGCd is returned by verifyComposed1 when the txn's observed catalog version is no longer in the engine's retention ring — either because the FIFO ring evicted it (the txn lived longer than `routeHistoryDepth` versions worth of catalog churn) or because the version was never seen on this node. Surfaces to the coordinator as a retryable error: the caller's M4 retry path reads the current route cache and re-issues the txn with a fresh observedVer.

The not-found ⇒ hard-error semantics (rather than soft-pass) matters because a soft-pass would let the gate be bypassed exactly in the long-running-txn / high-churn cases where the cross-version-read hazard is most likely (design doc §4.3 + gemini medium + codex P2 on PR #870).

View Source 
var ErrComposed1Violation = errors.New("composed-1: route ownership shifted; retry on new owning group")

ErrComposed1Violation is returned by verifyComposed1 when the transaction's commit cannot proceed on this Raft group because the txn's read-set or write-set keys are not owned by this group at either the txn's observed catalog version (the spec-level §4.2(a) check) or the current catalog version observed by the FSM at apply time (the §4.4 cross-version-read fence). Surfaces to the coordinator as a retryable error: the M4 coordinator path re-reads the route cache, re-routes the txn, and re-issues it once on the new owning group.

Wrapped with errors.Wrapf at the call site to carry the per-key diagnostic (which key, which observed-version owner, which current-version owner) — the caller's retry path uses errors.Is(err, ErrComposed1Violation) to match.

View Source 
var ErrCrossShardTransactionNotSupported = errors.New("cross-shard transactions are not supported")
View Source 
var ErrInvalidHLCBatchSize = errors.New("hlc: invalid batch size")
View Source 
var ErrInvalidRequest = errors.New("invalid request")
View Source 
var ErrLeaderNotFound = errors.New("leader not found")
View Source 
var ErrLeaderProxyCircuitOpen = errors.New("leader proxy circuit open")

ErrLeaderProxyCircuitOpen indicates that recent attempts against the same leader identity failed and the shared forwarding breaker is suppressing a retry storm. It is transient: callers may retry after leader publication or the breaker backoff, but a single request must not busy-loop on it.

View Source 
var ErrNotImplemented = errors.New("not implemented")
View Source 
var ErrRouteWriteFenced = errors.New("route is write-fenced; retry after route migration")

ErrRouteWriteFenced is returned when a mutation targets a route that is in WriteFenced state during split migration. Callers should retry after routing catches up to the promoted owner.

View Source 
var ErrSnapshotHeaderInvalidLength = errors.New("snapshot header: invalid v2 length prefix")

ErrSnapshotHeaderInvalidLength indicates a v2 header whose length prefix is below the required minimum (cannot hold ceiling+cutover) or above the maxSnapshotHeaderPayload DoS bound.

View Source 
var ErrSnapshotHeaderUnknownMagic = errors.New("snapshot header: unknown EKVTHLC* magic")

ErrSnapshotHeaderUnknownMagic indicates an EKVTHLC* magic whose version byte the current binary does not recognise. The restore fail-closes; the operator must upgrade.

View Source 
var ErrTSOCommitFloorUnavailable = errors.New("tso: authoritative data-group commit floor is unavailable")
View Source 
var ErrUnknownRequestType = errors.New("unknown request type")

Functions

func DecodeObservedRouteVersion 

func DecodeObservedRouteVersion(observed uint64) (uint64, bool)

DecodeObservedRouteVersion converts OperationGroup.ObservedRouteVersion back to a catalog version and reports whether it was explicitly pinned.

func EncodeBackupExtendEntry 

func EncodeBackupExtendEntry(entry BackupExtendEntry) []byte

func EncodeBackupPinEntry 

func EncodeBackupPinEntry(entry BackupPinEntry) []byte

func EncodeBackupReleaseEntry 

func EncodeBackupReleaseEntry(entry BackupReleaseEntry) []byte

func EncodeBackupReserveEntry 

func EncodeBackupReserveEntry(entry BackupReserveEntry) []byte

func EncodeBackupUnreserveEntry 

func EncodeBackupUnreserveEntry(entry BackupUnreserveEntry) []byte

func EncodeObservedRouteVersion 

func EncodeObservedRouteVersion(version uint64) uint64

EncodeObservedRouteVersion converts a tracked catalog version into the wire value carried by OperationGroup. Version zero is left as the legacy unpinned zero until the version-zero sentinel is capability-gated across Raft members.

func EncodeTxnMeta 

func EncodeTxnMeta(m TxnMeta) []byte

func ExtractTxnUserKey 

func ExtractTxnUserKey(key []byte) []byte

ExtractTxnUserKey returns the logical user key embedded in a transaction- internal key such as !txn|lock|, !txn|cmt|, or !txn|meta|. It returns nil when the key does not use a transaction-internal namespace or is malformed.

func LeaseReadAllGroupsThrough 

func LeaseReadAllGroupsThrough(c Coordinator, ctx context.Context) error

LeaseReadAllGroupsThrough establishes the lease freshness bound across every shard group a multi-shard read can touch. When the coordinator owns multiple groups (AllGroupsLeaseReadableCoordinator) it fences all of them; otherwise it falls back to the single-group LeaseRead path so a single-group deployment still issues exactly one lease read. Adapter call sites use this for keyless reads (Scan, whole-table/GSI Query fallback) that the per-key LeaseReadForKey cannot route to one group.

func LeaseReadAllGroupsTimestampThrough 

func LeaseReadAllGroupsTimestampThrough(c Coordinator, ctx context.Context) (uint64, error)

LeaseReadAllGroupsTimestampThrough returns the greatest leader-side commit timestamp when the coordinator exposes it. Legacy coordinators still execute their all-group barrier and return a zero watermark to their caller.

func LeaseReadForGroupThrough 

func LeaseReadForGroupThrough(c Coordinator, ctx context.Context, groupID uint64, key []byte) (uint64, error)

LeaseReadGroupKey returns a representative key per distinct owning group for the supplied keys, so callers can issue one lease read per group rather than one per key. The returned slice preserves the order of first appearance. When the coordinator does not implement GroupRoutableCoordinator (single-group deployments) every distinct key is returned unchanged so the caller's per-key dedup still bounds the work. Keys that cannot be routed (group ID 0) are never collapsed — each is kept as its own representative so the lease check still runs and surfaces the routing failure. LeaseReadForGroupThrough runs the lease read against an already-resolved group when the coordinator supports it, falling back to the key-resolved path.

func LeaseReadForKeyThrough 

func LeaseReadForKeyThrough(c Coordinator, ctx context.Context, key []byte) (uint64, error)

LeaseReadForKeyThrough is the key-routed counterpart of LeaseReadThrough.

func LeaseReadGroupKeys 

func LeaseReadGroupKeys(c Coordinator, keys [][]byte) [][]byte

func LeaseReadThrough 

func LeaseReadThrough(c Coordinator, ctx context.Context) (uint64, error)

LeaseReadThrough is a helper that calls LeaseRead when the coordinator supports it, falling back to LinearizableRead otherwise. Adapter call sites use this so they don't have to repeat the type-assertion dance.

func MaxLatestCommitTS 

func MaxLatestCommitTS(ctx context.Context, st store.MVCCStore, keys [][]byte) (uint64, error)

MaxLatestCommitTS returns the maximum commit timestamp for the provided keys.

Missing keys are ignored. If any LatestCommitTS lookup returns an error, the error is returned to the caller.

func NewLeaderAdminProposer 

func NewLeaderAdminProposer(
	leader raftengine.LeaderView,
	local raftengine.Proposer,
	connCache *GRPCConnCache,
	opts ...LeaderAdminProposerOption,
) raftengine.Proposer

func NewTxnLockedError 

func NewTxnLockedError(key []byte) error

func NewTxnLockedErrorWithDetail 

func NewTxnLockedErrorWithDetail(key []byte, detail string) error

func NextTimestampAfterThrough 

func NextTimestampAfterThrough(ctx context.Context, coord Coordinator, startTS uint64, label string) (uint64, error)

NextTimestampAfterThrough allocates a timestamp strictly greater than startTS. It observes startTS into legacy HLC clocks before allocation so the fallback path preserves the previous OCC ordering contract.

func NextTimestampThrough 

func NextTimestampThrough(ctx context.Context, coord Coordinator, label string) (uint64, error)

NextTimestampThrough allocates a persistence-grade timestamp through coord's optional TSO allocator when available, falling back to the legacy HLC clock for older Coordinator implementations and tests.

func PrimaryKeyForElems 

func PrimaryKeyForElems(reqs []*Elem[OP]) []byte

PrimaryKeyForElems returns the primary key the coordinator derives for a single-shard one-phase txn over elems — the lexicographically smallest write key. Adapters that implement option-2 one-phase dedup must probe this exact key (it becomes the FSM's meta.PrimaryKey) so the adapter-side self-inflicted-conflict guard agrees with dedupProbeOnePhase. See docs/design/2026_06_03_implemented_dynamodb_onephase_dedup.md (R4).

func ReadSnapshotHeader 

func ReadSnapshotHeader(r *bufio.Reader) (ceiling, cutover uint64, err error)

ReadSnapshotHeader is the §3.2 read path. The caller wraps the input io.Reader in a *bufio.Reader and passes it here once, then MUST reuse the same *bufio.Reader for the inner-store restore — buffered bytes can sit in the bufio.Reader between calls; switching readers silently loses them.

Return contract:

  • v1 magic: (ceiling, 0, nil), magic + ceiling consumed.
  • v2 magic: (ceiling, cutover, nil), full v2 header consumed; trailing bytes above the parsed fields (forward-compat extension area) are consumed and discarded.
  • EKVTHLC* with an unknown version byte: (0, 0, ErrSnapshotHeaderUnknownMagic) — fail-closed, restore aborts.
  • v2 magic with a malformed length prefix: (0, 0, ErrSnapshotHeaderInvalidLength) — fail-closed.
  • Anything else (including streams shorter than 8 bytes): (0, 0, nil) and ALL bytes left in the *bufio.Reader for the inner-store path.

func RouteKey 

func RouteKey(key []byte) []byte

RouteKey normalizes internal keys (e.g., list metadata/items) to the logical user key used for shard routing.

func RouteKeyFilter 

func RouteKeyFilter(rangeStart, rangeEnd []byte) func([]byte) bool

RouteKeyFilter returns the migration export predicate for raw MVCC keys. rangeEnd nil or empty means +infinity, matching the route descriptor wire convention.

func RouteKeyFilterForGroup 

func RouteKeyFilterForGroup(rangeStart, rangeEnd []byte, sourceGroupID uint64, resolver PartitionResolver) func([]byte) bool

RouteKeyFilterForGroup returns the migration export predicate for a source route and group. Partition-resolved keyspaces such as HT-FIFO SQS are matched by resolver group instead of the byte-range route key.

func RunTSOModeFileReload 

func RunTSOModeFileReload(
	ctx context.Context,
	path string,
	interval time.Duration,
	controller *TSORuntimeController,
	logger *slog.Logger,
) error

RunTSOModeFileReload polls an atomically replaceable plain-text mode file. Invalid reads or transitions leave the current allocator untouched.

func StampGroupedMutationCommitTS 

func StampGroupedMutationCommitTS(grouped map[uint64][]*pb.Mutation, commitTS uint64) error

func StampMutationCommitTS 

func StampMutationCommitTS(muts []*pb.Mutation, commitTS uint64) error

StampMutationCommitTS patches every mutation that embeds the resolved transaction commit timestamp in its value. Offset zero disables patching; non-zero offsets are byte offsets into the mutation value.

func TxnLockedDetails 

func TxnLockedDetails(err error) ([]byte, string, bool)

func TxnSuccessMarkerKey 

func TxnSuccessMarkerKey(lockedKey []byte, startTS, commitTS uint64, primaryKey []byte) []byte

TxnSuccessMarkerKey builds the route-local transaction-success marker key used by the migration planner. Normal transaction traffic only writes this after the migration capability gate opens in a later PR.

func ValidateDurablePersistenceTimestamp 

func ValidateDurablePersistenceTimestamp(ctx context.Context, alloc TimestampAllocator, timestamp uint64, label string) error

ConfiguredTimestampAllocatorThrough returns the configured allocator without resolving runtime-mode decorators. It is used by long-lived internal servers that must keep a DynamicTimestampAllocator reference across later mode-file reloads while still falling back to HLC when that allocator reports legacy. ValidateDurablePersistenceTimestamp checks a caller-supplied persistence timestamp against the Phase-D durable contract.

It exists for receiver paths that accept a timestamp somebody else stamped -- Internal.Forward preserves a nonzero raw Request.Ts and an already-set transaction-meta CommitTS rather than allocating one -- and that therefore never reach ShardedCoordinator.prepareTxnCommitTimestamp, where the coordinator validates the timestamps it is handed. Without this a forwarding peer or a direct caller could persist AllocationFloor()+1 before group 0 allocates it, and the TSO would later issue the same value.

A zero timestamp means "unset"; the caller allocates one and that path is already validated. When Phase D is not in force this is a no-op.

func ValidateElemCommitTSPatches 

func ValidateElemCommitTSPatches(elems []*Elem[OP], commitTS uint64) error

func ValidateForwardedTxnCommitTimestamp 

func ValidateForwardedTxnCommitTimestamp(
	ctx context.Context,
	alloc TimestampAllocator,
	startTS uint64,
	commitTS uint64,
	resolution bool,
	label string,
) error

ValidateForwardedTxnCommitTimestamp validates the commit timestamp of a forwarded transaction whose caller already stamped it.

It is ValidateDurablePersistenceTimestamp with one carve-out: a commit timestamp that predates Phase D is accepted when the transaction's own start timestamp predates Phase D too.

That case is a legacy transaction still being resolved. A cross-shard transaction that began before the Phase-D marker can have unresolved intents when the marker applies, and resolving them replays the commit timestamp the primary already recorded (LockResolver.resolveExpiredLock -> applyTxnResolution). On a follower that replay travels through Internal.Forward, so rejecting it would leave the transaction partially resolved with its secondary keys locked, and the rollout does not require draining transactions before activating Phase D.

The carve-out cannot be used to claim a timestamp group 0 has not issued yet, which is what the check exists to prevent: both values sit at or below the Phase-D floor, and group 0 only ever issues above it. A timestamp beyond the allocation floor fails with a plain ErrTSOTimestampInvalid and is still rejected here.

func ValidateForwardedTxnStartTimestamp 

func ValidateForwardedTxnStartTimestamp(
	ctx context.Context,
	alloc TimestampAllocator,
	startTS uint64,
	label string,
) error

ValidateForwardedTxnStartTimestamp validates a start timestamp a forwarded transaction arrived with.

Internal.Forward keeps a nonzero Request.Ts rather than allocating one, and a PREPARE carries no transaction meta, so the commit-timestamp check never sees it -- yet handlePrepareRequest persists the intent at that value. A caller could otherwise submit AllocationFloor()+1, persist state at a timestamp group 0 has not issued, and let the TSO issue the same value later.

A start timestamp that predates Phase D is allowed: a transaction that began before the marker is still entitled to prepare and resolve its intents, and group 0 only ever issues above the floor, so nothing it issues can collide.

Types

type ActiveTimestampToken 

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

ActiveTimestampToken releases one tracked timestamp when the owning operation completes.

func (*ActiveTimestampToken) Release 

func (t *ActiveTimestampToken) Release()

type ActiveTimestampTracker 

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

ActiveTimestampTracker tracks in-flight read or transaction timestamps that must remain readable while background compaction is running.

func NewActiveTimestampTracker 

func NewActiveTimestampTracker(opts ...ActiveTimestampTrackerOption) *ActiveTimestampTracker

func (*ActiveTimestampTracker) ActiveBackupPinCount 

func (t *ActiveTimestampTracker) ActiveBackupPinCount() int

func (*ActiveTimestampTracker) ApplyExtendForGroup 

func (t *ActiveTimestampTracker) ApplyExtendForGroup(pinID BackupPinID, groupID uint64, deadline time.Time) error

func (*ActiveTimestampTracker) ApplyPinWithDeadlineForGroup 

func (t *ActiveTimestampTracker) ApplyPinWithDeadlineForGroup(pinID BackupPinID, groupID uint64, readTS uint64, deadline time.Time) error

func (*ActiveTimestampTracker) BackupPinDeadline 

func (t *ActiveTimestampTracker) BackupPinDeadline(pinID BackupPinID) (time.Time, bool)

func (*ActiveTimestampTracker) BackupPinDeadlineForGroup 

func (t *ActiveTimestampTracker) BackupPinDeadlineForGroup(pinID BackupPinID, groupID uint64) (time.Time, bool)

func (*ActiveTimestampTracker) ClearBackupPinsForGroup 

func (t *ActiveTimestampTracker) ClearBackupPinsForGroup(groupID uint64)

ClearBackupPinsForGroup drops every volatile backup pin recorded for one Raft group. Snapshot restore is the caller: a follower that applied a BackupPin but not its BackupRelease can catch up from a snapshot the leader took after the release, and that snapshot legitimately carries no pin state while both log entries are already compacted away. Keeping the old pin would block compaction and snapshots on that replica, and hold backup capacity, until its deadline lapsed.

func (*ActiveTimestampTracker) Close 

func (t *ActiveTimestampTracker) Close()

Close stops the backup-pin sweeper goroutine. It is safe to call more than once and on trackers that never started the sweeper.

func (*ActiveTimestampTracker) Extend 

func (t *ActiveTimestampTracker) Extend(pinID BackupPinID, deadline time.Time) error

func (*ActiveTimestampTracker) ExtendForGroup 

func (t *ActiveTimestampTracker) ExtendForGroup(pinID BackupPinID, groupID uint64, deadline time.Time) error

func (*ActiveTimestampTracker) Oldest 

func (t *ActiveTimestampTracker) Oldest() uint64

func (*ActiveTimestampTracker) OldestBackupForGroup 

func (t *ActiveTimestampTracker) OldestBackupForGroup(groupID uint64) uint64

OldestBackupForGroup returns the oldest live backup pin for groupID, excluding ordinary read pins. FSM snapshot generation uses this to avoid emitting a snapshot that would drop active backup retention state.

func (*ActiveTimestampTracker) OldestForGroup 

func (t *ActiveTimestampTracker) OldestForGroup(groupID uint64) uint64

OldestForGroup returns the oldest process-wide read pin or backup pin for groupID. Backup pins for other Raft groups do not constrain this group.

func (*ActiveTimestampTracker) Pin 

func (t *ActiveTimestampTracker) Pin(ts uint64) *ActiveTimestampToken

func (*ActiveTimestampTracker) PinWithDeadline 

func (t *ActiveTimestampTracker) PinWithDeadline(pinID BackupPinID, readTS uint64, deadline time.Time) error

func (*ActiveTimestampTracker) PinWithDeadlineForGroup 

func (t *ActiveTimestampTracker) PinWithDeadlineForGroup(pinID BackupPinID, groupID uint64, readTS uint64, deadline time.Time) error

func (*ActiveTimestampTracker) ReleaseBackupPin 

func (t *ActiveTimestampTracker) ReleaseBackupPin(pinID BackupPinID)

func (*ActiveTimestampTracker) ReleaseBackupPinForGroup 

func (t *ActiveTimestampTracker) ReleaseBackupPinForGroup(pinID BackupPinID, groupID uint64)

func (*ActiveTimestampTracker) SetBackupTimestampFloorObserver 

func (t *ActiveTimestampTracker) SetBackupTimestampFloorObserver(observer func(uint64))

SetBackupTimestampFloorObserver installs the process-local timestamp-cache invalidation callback invoked after a replicated backup pin is accepted.

type ActiveTimestampTrackerOption 

type ActiveTimestampTrackerOption func(*ActiveTimestampTracker)

func WithActiveTimestampTrackerLogger 

func WithActiveTimestampTrackerLogger(logger *slog.Logger) ActiveTimestampTrackerOption

func WithActiveTimestampTrackerMaxBackupPins 

func WithActiveTimestampTrackerMaxBackupPins(maxPins int) ActiveTimestampTrackerOption

func WithActiveTimestampTrackerSweepInterval 

func WithActiveTimestampTrackerSweepInterval(interval time.Duration) ActiveTimestampTrackerOption

type AllGroupsLeaseReadableCoordinator 

type AllGroupsLeaseReadableCoordinator interface {
	// LeaseReadAllGroups establishes the lease freshness bound on every
	// group the coordinator owns, failing closed on the first group that
	// cannot confirm its lease. The freshness bound is what a multi-shard
	// read relies on, so a returned error MUST abort the read.
	LeaseReadAllGroups(ctx context.Context) error
}

AllGroupsLeaseReadableCoordinator is the optional capability implemented by coordinators that own more than one Raft group and can establish the lease freshness bound on EVERY group in a single call. Multi-shard read handlers (Scan, GSI/whole-table Query) need this because the underlying scan visits all intersecting routes across all groups, whereas the plain LeaseRead only fences the default group. Single-group coordinators do not implement it: LeaseReadAllGroupsThrough falls back to LeaseRead so they still issue exactly one lease read.

type AllGroupsLeaseTimestampCoordinator 

type AllGroupsLeaseTimestampCoordinator interface {
	LeaseReadAllGroupsTimestamp(ctx context.Context) (uint64, error)
}

AllGroupsLeaseTimestampCoordinator extends the all-group barrier with the greatest commit timestamp applied by any fenced group leader.

type AppliedReadTimestampVoucher 

type AppliedReadTimestampVoucher interface {
	VouchAppliedReadTimestamp(uint64, AppliedReadTimestampVoucherRef) error
}

AppliedReadTimestampVoucher records an adapter-provided applied watermark. It is a process-local capability used only to distinguish audited adapter snapshots from arbitrary caller-supplied StartTS values during Phase D.

type AppliedReadTimestampVoucherRef 

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

AppliedReadTimestampVoucherRef is an opaque process-local dispatch capability. Only this package can mint a non-zero ref.

type AppliedReadTimestampVoucherRevoker 

type AppliedReadTimestampVoucherRevoker interface {
	RevokeAppliedReadTimestamp(uint64, AppliedReadTimestampVoucherRef)
}

AppliedReadTimestampVoucherRevoker removes an unused voucher registration. Decorators that forward VouchAppliedReadTimestamp should forward revocation too so pre-dispatch gates cannot leak inner-coordinator voucher entries.

type AppliedReadTimestampVoucherSupport 

type AppliedReadTimestampVoucherSupport interface {
	SupportsAppliedReadTimestampVoucher() bool
}

AppliedReadTimestampVoucherSupport lets coordinator decorators expose whether the inner coordinator can actually consume prepared vouchers.

type ApplyObserver 

type ApplyObserver interface {
	OnApply(op pb.Op, key []byte)
}

ApplyObserver receives a notification after a logical mutation is successfully applied by kvFSM. Implementations run inline on the Raft apply goroutine and must stay non-blocking.

type BackupExtendEntry 

type BackupExtendEntry struct {
	PinID    BackupPinID
	Deadline time.Time
}

type BackupKeyFilter 

type BackupKeyFilter func(key []byte) (bool, error)

BackupKeyFilter decides whether a key should be materialized by a value scanner. It runs after route ownership filtering but before reading values.

type BackupKeyScanner 

type BackupKeyScanner interface {
	Next(ctx context.Context) ([]byte, bool, error)
	Close() error
}

BackupKeyScanner is the count-only counterpart to BackupScanner. It pages through the same captured route set without materializing values.

func NewBackupKeyScanner 

func NewBackupKeyScanner(st *ShardStore, start []byte, end []byte, ts uint64, pageSize int) BackupKeyScanner

func NewBackupKeyScannerAtSnapshot 

func NewBackupKeyScannerAtSnapshot(st *ShardStore, snapshot BackupRouteSnapshot, ts uint64, pageSize int) BackupKeyScanner

NewBackupKeyScannerAtSnapshot creates a key-only scanner from a captured route view.

type BackupPinEntry 

type BackupPinEntry struct {
	PinID    BackupPinID
	ReadTS   uint64
	Deadline time.Time
}

type BackupPinID 

type BackupPinID [16]byte

func (BackupPinID) IsZero 

func (id BackupPinID) IsZero() bool

func (BackupPinID) String 

func (id BackupPinID) String() string

type BackupReleaseEntry 

type BackupReleaseEntry struct {
	PinID BackupPinID
}

type BackupReserveEntry 

type BackupReserveEntry = BackupPinEntry

BackupReserveEntry is committed through one deterministic Raft group before fan-out. Its group-zero tracker record serializes the cluster-wide pin cap.

type BackupRouteSnapshot 

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

BackupRouteSnapshot is an immutable route view shared by every scan in one logical backup. Keeping it separate from a scanner lets BeginBackup count keys and StreamBackup materialize values from the same ownership view even when the live route catalog changes between those RPCs.

func BackupRouteSnapshotWithScanGroups 

func BackupRouteSnapshotWithScanGroups(snapshot BackupRouteSnapshot, groupIDs []uint64) BackupRouteSnapshot

BackupRouteSnapshotWithScanGroups returns a snapshot that also scans the supplied Raft groups for resolver-owned partitioned keys even when the durable byte-range catalog has no route for those groups.

func CaptureBackupRouteSnapshotAt 

func CaptureBackupRouteSnapshotAt(ctx context.Context, catalog *distribution.CatalogStore, ts uint64) (BackupRouteSnapshot, error)

CaptureBackupRouteSnapshotAt reads the durable distribution catalog at ts. The caller must pass the CatalogStore bound to the catalog owner group; using a normally routed ShardStore can split the version read from the route-row scan when those reserved prefixes route to different groups.

type BackupScanner 

type BackupScanner interface {
	Next(ctx context.Context) (*store.KVPair, bool, error)
	Close() error
}

BackupScanner pages through ShardStore.ScanKeysAt without holding store locks across pages, then materializes each key at the pinned read timestamp.

func NewBackupScanner 

func NewBackupScanner(st *ShardStore, start []byte, end []byte, ts uint64, pageSize int) BackupScanner

func NewBackupScannerAtSnapshot 

func NewBackupScannerAtSnapshot(st *ShardStore, snapshot BackupRouteSnapshot, ts uint64, pageSize int) BackupScanner

NewBackupScannerAtSnapshot creates a value scanner from a captured route view.

func NewFilteredBackupScannerAtSnapshot 

func NewFilteredBackupScannerAtSnapshot(
	st *ShardStore,
	snapshot BackupRouteSnapshot,
	ts uint64,
	pageSize int,
	keyFilter BackupKeyFilter,
) BackupScanner

NewFilteredBackupScannerAtSnapshot creates a value scanner that skips filtered-out keys before materializing values.

type BackupUnreserveEntry 

type BackupUnreserveEntry = BackupReleaseEntry

BackupUnreserveEntry releases the group-zero capacity reservation.

type BatchAllocator 

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

BatchAllocator serves local timestamps from immutable windows fetched from a TSOAllocator. The hot path is lock-free: callers claim a slot with atomic Add on the currently published window.

func NewBatchAllocator 

func NewBatchAllocator(tso TSOAllocator, batchSize int) (*BatchAllocator, error)

func (*BatchAllocator) Invalidate 

func (b *BatchAllocator) Invalidate()

func (*BatchAllocator) Next 

func (b *BatchAllocator) Next(ctx context.Context) (uint64, error)

func (*BatchAllocator) NextAfter 

func (b *BatchAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*BatchAllocator) PhaseDActive 

func (b *BatchAllocator) PhaseDActive() bool

func (*BatchAllocator) PhaseDRequired 

func (b *BatchAllocator) PhaseDRequired() bool

func (*BatchAllocator) ValidateDurableTimestamp 

func (b *BatchAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error

type ConfiguredTimestampAllocatorProvider 

type ConfiguredTimestampAllocatorProvider interface {
	ConfiguredTimestampAllocator() TimestampAllocator
}

ConfiguredTimestampAllocatorProvider lets coordinator decorators expose the configured allocator without resolving runtime-mode wrappers. Long-lived services use this to keep a DynamicTimestampAllocator reference while the active mode is still legacy.

type Coordinate 

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

func NewCoordinatorWithEngine 

func NewCoordinatorWithEngine(txm Transactional, engine raftengine.Engine, opts ...CoordinatorOption) *Coordinate

func (*Coordinate) Clock 

func (c *Coordinate) Clock() *HLC

func (*Coordinate) Close 

func (c *Coordinate) Close() error

Close releases any engine-side registrations (currently the leader-loss callback) held by this Coordinate. It is safe to call on a nil receiver and multiple times. Owners whose lifetime matches the engine's do not need to call Close; owners who discard the Coordinate before closing the engine MUST.

func (*Coordinate) Dispatch 

func (c *Coordinate) Dispatch(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, error)

func (*Coordinate) EngineGroupIDForKey 

func (c *Coordinate) EngineGroupIDForKey(_ []byte) uint64

EngineGroupIDForKey makes Coordinate satisfy GroupRoutableCoordinator. A Coordinate fronts exactly one Raft group, so every key maps to the same group and batched lease checks collapse to a single read.

func (*Coordinate) GroupLeadership 

func (c *Coordinate) GroupLeadership(_ uint64) (bool, uint64)

GroupLeadership reports leadership for Coordinate's single Raft group.

func (*Coordinate) IsLeader 

func (c *Coordinate) IsLeader() bool

func (*Coordinate) IsLeaderAcceptingWrites 

func (c *Coordinate) IsLeaderAcceptingWrites() bool

IsLeaderAcceptingWrites reports whether this node is leader and not currently transferring leadership. Background proposers should gate on this to avoid piling up dropped proposals while a transfer is in flight.

func (*Coordinate) IsLeaderForKey 

func (c *Coordinate) IsLeaderForKey(_ []byte) bool

func (*Coordinate) IsTimestampLeader 

func (c *Coordinate) IsTimestampLeader() bool

func (*Coordinate) LeadershipForKey 

func (c *Coordinate) LeadershipForKey(_ []byte) (bool, uint64)

LeadershipForKey reports local leadership and the current Raft term for the single group that owns every key handled by Coordinate.

func (*Coordinate) LeaseRead 

func (c *Coordinate) LeaseRead(ctx context.Context) (uint64, error)

LeaseRead returns a read fence backed by a leader-local lease when available, falling back to a full LinearizableRead when no fast path is live or the engine does not implement LeaseProvider.

The PRIMARY lease path is maintained inside the engine from ongoing MsgAppResp / MsgHeartbeatResp traffic, so that path does not rely on callers sampling time.Now() before the slow path to "extend" a lease afterwards. The earlier pre-read sampling was racy under congestion: if a LinearizableRead took longer than LeaseDuration, the extension would land already expired and the lease never warmed up. The engine-driven anchor is refreshed every heartbeat independent of read latency.

The SECONDARY caller-side lease remains as a rollout fallback, still populated by the original pre-read sampling; it covers the narrow window between startup and the first quorum heartbeat round landing on the engine.

The returned index is the engine's current applied index (fast path) or the index returned by LinearizableRead (slow path). Callers that resolve timestamps via store.LastCommitTS may discard the value.

func (*Coordinate) LeaseReadForKey 

func (c *Coordinate) LeaseReadForKey(ctx context.Context, key []byte) (uint64, error)

func (*Coordinate) LinearizableRead 

func (c *Coordinate) LinearizableRead(ctx context.Context) (uint64, error)

func (*Coordinate) LinearizableReadForKey 

func (c *Coordinate) LinearizableReadForKey(ctx context.Context, key []byte) (uint64, error)

func (*Coordinate) Next 

func (c *Coordinate) Next(ctx context.Context) (uint64, error)

Next makes Coordinate usable as a TimestampAllocator for adapter helpers.

func (*Coordinate) NextAfter 

func (c *Coordinate) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*Coordinate) ObserveForwardedRequests 

func (c *Coordinate) ObserveForwardedRequests(reqs []*pb.Request)

ObserveForwardedRequests records leader-side sampling evidence for writes that entered through a follower and were committed via Internal.Forward.

func (*Coordinate) ProposeHLCLease 

func (c *Coordinate) ProposeHLCLease(ctx context.Context, ceilingMs int64) error

ProposeHLCLease proposes a new physical ceiling to the Raft cluster. Only the current leader should call this; followers silently ignore proposals from non-leaders via Raft's leader-only write guarantee.

A successful propose is a quorum-acked Raft commit, exactly the same confirmation Dispatch relies on, so it also warms the leader-local read lease. The lease-extension base (dispatchStart) and the invalidation generation are sampled BEFORE the propose, mirroring refreshLeaseAfterDispatch: the window can only ever be SHORTER than the true safety window, and a leader-loss callback that fires during the propose advances the generation so extend refuses to resurrect a stale lease. This is the background warm-up that flattens the read-only lease-expiry sawtooth on idle-write workloads -- no extra goroutine, no change to the lease window/duration semantics.

On a leadership-loss propose error the lease is invalidated eagerly, mirroring refreshLeaseAfterDispatch's error branch exactly: when Propose returns the loss before the async RegisterLeaderLossCallback fires, a stale-warm lease must not survive on a non-leader node for the callback latency window. Non-leadership errors (no quorum, validation) are NOT leadership signals and must not tear down a warm lease -- doing so would force every read onto the slow path.

func (*Coordinate) RaftLeader 

func (c *Coordinate) RaftLeader() string

RaftLeader returns the current leader's address as known by this node.

func (*Coordinate) RaftLeaderForKey 

func (c *Coordinate) RaftLeaderForKey(_ []byte) string

func (*Coordinate) RecoverHLCLease 

func (c *Coordinate) RecoverHLCLease(ctx context.Context) error

func (*Coordinate) RevokeAppliedReadTimestamp 

func (c *Coordinate) RevokeAppliedReadTimestamp(uint64, AppliedReadTimestampVoucherRef)

RevokeAppliedReadTimestamp is a no-op for the single-group coordinator.

func (*Coordinate) RunHLCLeaseRenewal 

func (c *Coordinate) RunHLCLeaseRenewal(ctx context.Context)

RunHLCLeaseRenewal runs a background loop that periodically proposes a new physical ceiling to the Raft cluster while this node is the leader.

The ceiling is set to now + hlcPhysicalWindowMs and is renewed every hlcRenewalInterval. While renewals keep succeeding, the committed ceiling and NextFenced fail-closed check prevent timestamp issuance after the safe window has expired.

RunHLCLeaseRenewal blocks until ctx is cancelled; call it in a goroutine.

func (*Coordinate) SetHLCLeaseRenewalBlocker 

func (c *Coordinate) SetHLCLeaseRenewalBlocker(blocked func() bool)

SetHLCLeaseRenewalBlocker installs a predicate that suppresses background HLC lease-renewal proposals while it returns true.

func (*Coordinate) SupportsAppliedReadTimestampVoucher 

func (c *Coordinate) SupportsAppliedReadTimestampVoucher() bool

func (*Coordinate) TimestampAllocator 

func (c *Coordinate) TimestampAllocator() TimestampAllocator

TimestampAllocator exposes the configured allocator to coordinator decorators without widening the Coordinator interface.

func (*Coordinate) VerifyLeader 

func (c *Coordinate) VerifyLeader(ctx context.Context) error

func (*Coordinate) VerifyLeaderForKey 

func (c *Coordinate) VerifyLeaderForKey(ctx context.Context, _ []byte) error

func (*Coordinate) VouchAppliedReadTimestamp 

func (c *Coordinate) VouchAppliedReadTimestamp(uint64, AppliedReadTimestampVoucherRef) error

VouchAppliedReadTimestamp is a no-op for the single-group coordinator. The sharded coordinator consumes vouchers before cross-group StartTS validation.

func (*Coordinate) WithSampler 

func (c *Coordinate) WithSampler(s keyviz.Sampler, routeID uint64) *Coordinate

WithSampler wires a keyviz sampler onto the single-group coordinator. routeID must be the catalog route ID covering the single group; a zero routeID disables sampling so callers cannot accidentally emit unroutable rows.

func (*Coordinate) WithSamplerRouteResolver 

func (c *Coordinate) WithSamplerRouteResolver(s keyviz.Sampler, resolve func(key []byte) (uint64, bool)) *Coordinate

WithSamplerRouteResolver wires keyviz sampling with a route lookup evaluated for every observed key. Single-group deployments need this after a range split, when one fixed startup RouteID no longer covers the keyspace.

type CoordinateResponse 

type CoordinateResponse struct {
	CommitIndex uint64
	CommitTS    uint64
}

func DispatchWithReadTimestamp 

func DispatchWithReadTimestamp(
	ctx context.Context,
	coord Coordinator,
	reqs *OperationGroup[OP],
) (*CoordinateResponse, error)

DispatchWithReadTimestamp dispatches an OCC operation under the reusable applied-read capability bound by ReadTimestamp.WithDispatchVoucher. Each call reserves one distinct token tied to that bound capability immediately before dispatching, so a same-valued StartTS from another request cannot consume it and overlapping dispatches cannot revoke each other's authorization.

type Coordinator 

type Coordinator interface {
	Dispatch(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, error)
	IsLeader() bool
	VerifyLeader(ctx context.Context) error
	LinearizableRead(ctx context.Context) (uint64, error)
	RaftLeader() string
	IsLeaderForKey(key []byte) bool
	VerifyLeaderForKey(ctx context.Context, key []byte) error
	RaftLeaderForKey(key []byte) string
	Clock() *HLC
}

func WithKeyVizLabel 

func WithKeyVizLabel(c Coordinator, label keyviz.Label) Coordinator

WithKeyVizLabel returns a Coordinator wrapper that stamps all dispatches and key-routed lease reads with the supplied adapter label.

type CoordinatorOption 

type CoordinatorOption func(*Coordinate)

CoordinatorOption is a functional option for Coordinate constructors.

func WithHLC 

func WithHLC(hlc *HLC) CoordinatorOption

WithHLC sets a pre-created HLC on the coordinator. Use this together with NewKvFSMWithHLC so the FSM and coordinator share the same clock instance: the FSM advances physicalCeiling on every applied HLC lease entry, and the coordinator reads it inside Next() to floor new timestamps above the previous leader's committed window.

func WithLeaseReadObserver 

func WithLeaseReadObserver(observer LeaseReadObserver) CoordinatorOption

WithLeaseReadObserver wires a LeaseReadObserver onto a Coordinate. This is the mechanism monitoring uses to surface the lease-hit ratio panel on the Redis hot-path dashboard (see the "Hot Path" row in monitoring/grafana/dashboards/elastickv-redis-summary.json).

Typed-nil guard: a caller passing a typed-nil pointer (e.g. `var o *myObserver; WithLeaseReadObserver(o)`) produces an interface value that is NOT equal to nil under the normal `!= nil` check, yet invoking ObserveLeaseRead would panic. Normalise here with reflect.Value.IsNil so the hot-path nil check in LeaseRead stays a single branch on a real nil interface.

func WithTSOAllocator 

func WithTSOAllocator(alloc TimestampAllocator) CoordinatorOption

WithTSOAllocator routes coordinator-owned persistence timestamps through a TSO-compatible allocator. When unset, the coordinator keeps using its shared HLC directly.

type CutoverSource 

type CutoverSource interface {
	RaftEnvelopeCutoverIndex() uint64
}

CutoverSource is the writer-side view of the Phase-2 envelope cutover. kvFSMSnapshot consults it once per snapshot to decide v1 vs v2 layout. A nil source means "always v1" (matches the Phase-0/Phase-1 posture and every pre-8a code path).

type DurableTimestampValidator 

type DurableTimestampValidator interface {
	ValidateDurableTimestamp(context.Context, uint64) error
}

DurableTimestampValidator verifies that a timestamp belongs to the durable post-Phase-D allocation range owned by the dedicated TSO group.

type DynamicTimestampAllocator 

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

DynamicTimestampAllocator atomically publishes an optional allocator. A nil current allocator deliberately means "use the coordinator's legacy HLC path"; callers must resolve it through TimestampAllocatorThrough rather than calling Next directly.

func NewDynamicTimestampAllocator 

func NewDynamicTimestampAllocator(initial TimestampAllocator) *DynamicTimestampAllocator

func (*DynamicTimestampAllocator) Invalidate 

func (d *DynamicTimestampAllocator) Invalidate()

func (*DynamicTimestampAllocator) Next 

func (d *DynamicTimestampAllocator) Next(ctx context.Context) (uint64, error)

func (*DynamicTimestampAllocator) NextAfter 

func (d *DynamicTimestampAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*DynamicTimestampAllocator) PhaseDActive 

func (d *DynamicTimestampAllocator) PhaseDActive() bool

func (*DynamicTimestampAllocator) PhaseDRequired 

func (d *DynamicTimestampAllocator) PhaseDRequired() bool

func (*DynamicTimestampAllocator) ValidateDurableTimestamp 

func (d *DynamicTimestampAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error

type Elem 

type Elem[T OP] struct {
	Op    T
	Key   []byte
	Value []byte
	// CommitTSValueOffset, when non-zero, asks the coordinator or forwarded
	// leader to stamp the resolved transaction commit timestamp into Value at
	// this byte offset before committing the mutation.
	CommitTSValueOffset uint64
	// GroupID optionally pins this mutation to a shard group. Zero preserves
	// normal key-based routing.
	GroupID uint64
}

Elem is an element of a transaction.

type EncryptionApplier 

type EncryptionApplier interface {
	ApplyRegistration(p fsmwire.RegistrationPayload) error
	ApplyBootstrap(raftIdx uint64, p fsmwire.BootstrapPayload) error
	ApplyRotation(raftIdx uint64, p fsmwire.RotationPayload) error
}

EncryptionApplier owns the side-effects an encryption FSM entry must persist on apply: keystore mutation, sidecar update, and writer-registry insert. Stage 4 ships the dispatch seam and HaltApply propagation; Stage 5/6/7 will provide a concrete implementation that

  • KEK-unwraps the wrapped DEK and calls Keystore.Set
  • mutates the local sidecar (Active.{Storage,Raft}, keys map, raft_envelope_cutover_index) via the §5.1 crash-durable WriteSidecar protocol
  • inserts writer-registry rows under the §4.1 `!encryption|writers|<dek_id>|<uint16(node_id)>` Pebble key

The separation lets Stage 4 land the byte-tag dispatch + halt machinery without depending on the Stage 7 writer-registry storage layer or the Stage 5 admin RPC plumbing.

All three methods may return an error wrapped with encryption.ErrEncryptionApply to halt the apply loop. The kvFSM dispatcher converts any non-nil return into a haltApplyResponse so internal/raftengine/etcd's HaltApply seam recognises it.

The raftIdx parameter on ApplyBootstrap and ApplyRotation is the Raft entry index of the entry being applied. The applier persists this as sidecar.RaftAppliedIndex inside the same WriteSidecar fsync that mutates the keys[] map, so the §9.1 ErrSidecarBehindRaftLog startup guard can compare the sidecar's last-witnessed index against the engine's AppliedIndex on the next process start. ApplyRegistration does NOT take an index because writer-registry inserts do not touch the sidecar (§5.5 OpRegistration is intentionally excluded from the audit predicate).

type FSM 

type FSM interface {
	raftengine.StateMachine
}

func NewKvFSMWithHLC 

func NewKvFSMWithHLC(store store.MVCCStore, hlc *HLC, opts ...FSMOption) FSM

NewKvFSMWithHLC creates a KV FSM that updates hlc.physicalCeiling whenever a HLC lease entry is applied. The caller must pass the same *HLC instance to the coordinator so both sides share the agreed physical ceiling.

Optional FSMOption arguments configure additional handlers (see WithEncryption). Existing callers without options keep the pre-Stage-4 behaviour byte-for-byte.

func NewKvFSMWithHLCAndTracker 

func NewKvFSMWithHLCAndTracker(store store.MVCCStore, hlc *HLC, tracker *ActiveTimestampTracker, opts ...FSMOption) FSM

type FSMCompactRuntime 

type FSMCompactRuntime struct {
	GroupID      uint64
	StatusReader RaftStatusProvider
	Store        store.MVCCStore
}

type FSMCompactor 

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

func NewFSMCompactor 

func NewFSMCompactor(runtimes []FSMCompactRuntime, opts ...FSMCompactorOption) *FSMCompactor

func (*FSMCompactor) Run 

func (c *FSMCompactor) Run(ctx context.Context) error

func (*FSMCompactor) SyncOnce 

func (c *FSMCompactor) SyncOnce(ctx context.Context) error

type FSMCompactorOption 

type FSMCompactorOption func(*FSMCompactor)

func WithFSMCompactorActiveTimestampTracker 

func WithFSMCompactorActiveTimestampTracker(tracker *ActiveTimestampTracker) FSMCompactorOption

func WithFSMCompactorInterval 

func WithFSMCompactorInterval(interval time.Duration) FSMCompactorOption

func WithFSMCompactorLSMBackpressureLimits 

func WithFSMCompactorLSMBackpressureLimits(maxL0Files int64, maxDebtBytes uint64) FSMCompactorOption

func WithFSMCompactorLSMBackpressureSublevelLimit 

func WithFSMCompactorLSMBackpressureSublevelLimit(maxL0Sublevels int32) FSMCompactorOption

func WithFSMCompactorLeaderCooldown 

func WithFSMCompactorLeaderCooldown(cooldown time.Duration) FSMCompactorOption

func WithFSMCompactorLeaderTimeout 

func WithFSMCompactorLeaderTimeout(timeout time.Duration) FSMCompactorOption

func WithFSMCompactorLogger 

func WithFSMCompactorLogger(logger *slog.Logger) FSMCompactorOption

func WithFSMCompactorRetentionWindow 

func WithFSMCompactorRetentionWindow(window time.Duration) FSMCompactorOption

func WithFSMCompactorTimeout 

func WithFSMCompactorTimeout(timeout time.Duration) FSMCompactorOption

func WithFSMCompactorTimeoutBackoff 

func WithFSMCompactorTimeoutBackoff(backoff time.Duration) FSMCompactorOption

type FSMOption 

type FSMOption func(*kvFSM)

FSMOption configures a *kvFSM at construction. Stage 4 introduces WithEncryption; future stages may add more.

func WithActiveTimestampTracker 

func WithActiveTimestampTracker(tracker *ActiveTimestampTracker) FSMOption

func WithApplyObserver 

func WithApplyObserver(observer ApplyObserver) FSMOption

WithApplyObserver registers an observer for successful logical mutations. Nil observers are ignored so callers can pass optional wiring directly.

func WithCutoverSource 

func WithCutoverSource(src CutoverSource) FSMOption

WithCutoverSource installs the Stage 8a §3.3 writer-side view of the Phase-2 envelope cutover index. The FSM consults it once per Snapshot call to decide v1 vs v2 layout. Pass nil (or omit the option) to keep the pre-8a posture where every snapshot is v1.

func WithEncryption 

func WithEncryption(applier EncryptionApplier) FSMOption

WithEncryption installs the EncryptionApplier the kvFSM dispatches opcodes 0x03 / 0x04 / 0x05 to. Pass nil (or omit the option entirely) to leave the FSM in the Stage-4-default fail-closed state where any encryption opcode halts the apply loop via ErrEncryptionApply.

func WithRouteHistory 

func WithRouteHistory(routes RouteHistory, shardGroupID uint64) FSMOption

WithRouteHistory installs the M2 Composed-1 versioned-snapshot provider and the FSM's owning shard group ID. Both fields are consumed by the M3 verifyComposed1 gate. At M2 the values are stored but not consulted (M3 wires the check); a caller that constructs a kvFSM without this option remains "unpinned" — the M3 gate will short-circuit and behave exactly like the pre-feature FSM.

shardGroupID MUST match the Raft group ID this FSM serves — the gate uses it as the "this group" value when comparing against the historical owner-of-key resolution. Zero is reserved for the not-wired case.

See docs/design/2026_05_29_implemented_composed1_cross_group_commit_guard.md §M2 + §4.2 prerequisite block.

type GRPCConnCache 

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

GRPCConnCache reuses gRPC connections per address. gRPC itself handles reconnection on transient failures; we only force a re-dial if the conn has already been closed (Shutdown).

func (*GRPCConnCache) Close 

func (c *GRPCConnCache) Close() error

func (*GRPCConnCache) ConnFor 

func (c *GRPCConnCache) ConnFor(addr string) (*grpc.ClientConn, error)

type GroupLeaderRoutableCoordinator 

type GroupLeaderRoutableCoordinator interface {
	IsLeaderForGroup(groupID uint64) bool
	RaftLeaderForGroup(groupID uint64) string
}

GroupLeaderRoutableCoordinator is satisfied by coordinators that can answer leadership for an already-resolved group id, so a read fence does not have to re-derive the group from a representative key.

type GroupRoutableCoordinator 

type GroupRoutableCoordinator interface {
	// EngineGroupIDForKey returns the owning group ID, or 0 when the
	// key cannot be routed.
	EngineGroupIDForKey(key []byte) uint64
}

GroupRoutableCoordinator is the optional capability implemented by coordinators that can resolve the owning Raft group of a key without any I/O. Callers that need to lease-check a set of keys use it to deduplicate by group: keys that resolve to the same group share one lease read instead of issuing one per key. Single-group coordinators do not implement it, so callers must fall back to per-key dedup.

type HLC 

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

HLC implements a hybrid logical clock where the physical part is agreed upon via Raft consensus and the logical counter is managed purely in memory.

Layout (ms | logical): 

high 48 bits: wall clock milliseconds since Unix epoch  ← Raft-agreed physical part
low 16 bits : logical counter                           ← in-memory only

Physical ceiling (前半, consensus): 

The leader periodically commits a HLC lease entry to the Raft log that
establishes an upper bound for the physical timestamp (physicalCeiling).
All nodes apply this entry via the FSM, advancing their local ceiling.
When a new leader is elected it inherits the committed ceiling from the
FSM state so it never issues timestamps below the previous leader's window.

Logical counter (後半, memory): 

The 16-bit counter increments purely in memory on every Next() call within
the same millisecond. It resets to 0 whenever wall time advances, and
overflows by bumping the wall millisecond by one. No Raft round-trip is
needed for logical counter advancement.

func NewHLC 

func NewHLC() *HLC

func (*HLC) Current 

func (h *HLC) Current() uint64

Current returns the last issued or observed HLC value without advancing it. If no timestamp has been generated yet, it returns 0.

func (*HLC) Next 

func (h *HLC) Next() uint64

Next returns the next hybrid logical timestamp, ignoring the HLC-4 physical-ceiling fence. Kept for non-persistence callers (diagnostics, identifiers, retry IDs, tests, demo wiring) and for adapter sites whose fail-closed migration is not yet completed.

NEW persistence-grade allocations (everything that ends up as a startTS / commitTS, an MVCC write timestamp, or a lease/expiry boundary) MUST go through NextFenced instead — Next bypasses the HLC-4 (iii) ceiling fence and can therefore issue a timestamp inside a stale leader window after lease renewal stops.

Physical part (upper 48 bits): derived from the wall clock, but never less than the Raft-agreed physicalCeiling so that a newly elected leader cannot issue timestamps that collide with the previous leader's window.

Logical part (lower 16 bits): a pure in-memory counter that increments without any Raft round-trip. It resets to 0 whenever the physical millisecond advances and overflows by bumping the physical millisecond by one.

func (*HLC) NextBatchFenced 

func (h *HLC) NextBatchFenced(n int) (uint64, error)

NextBatchFenced reserves n consecutive HLC timestamps in one atomic update. It enforces the same physical-ceiling fence as NextFenced and returns the first timestamp in the reserved window: [base, base+n-1].

func (*HLC) NextFenced 

func (h *HLC) NextFenced() (uint64, error)

NextFenced returns the next hybrid logical timestamp with the HLC-4 (iii) physical-ceiling fence enforced. ALL persistence-grade allocations (startTS, commitTS, MVCC write ts, lease/expiry bounds) MUST go through this entry point so that an expired-ceiling allocation fails closed instead of silently issuing a ts that could collide with a subsequent leader's window after renewal catches up.

Fence semantics:

  • ceiling == 0 (pre-bootstrap, no prior leader): no fence, identical to Next.
  • ceiling > 0 AND wall_now >= ceiling: returns (0, ErrCeilingExpired).
  • ceiling > 0 AND wall_now < ceiling: floor wall at ceiling, then proceed.
  • ceiling > 0 AND the next timestamp's physical part would exceed ceiling: returns (0, ErrCeilingExpired) and waits for a fresh committed lease.

The TLA+ proof for this lives in tla/hlc/MCHLC_gap.cfg (HLC-4 counterexample, depth 5) — see docs/design/2026_05_28_implemented_tla_safety_spec.md §5.1.

func (*HLC) NextFencedRejections 

func (h *HLC) NextFencedRejections() uint64

NextFencedRejections returns the cumulative count of NextFenced() calls that returned ErrCeilingExpired since process start. The monitoring layer reads this on a fixed interval and exports it as a Prometheus counter so operators can alert on the rate.

A non-zero value here means the HLC-4 (i) bounded-skew assumption (`MaxClockSkewMs < HlcPhysicalWindowMs` — see docs/design/2026_05_28_implemented_tla_safety_spec.md §5.1) has at some point been violated by enough margin that wall_now caught up to physicalCeiling and the fence fired — typically because the leader's lease renewal stopped (network partition, GC pause, …) for longer than `hlcPhysicalWindowMs`.

func (*HLC) Observe 

func (h *HLC) Observe(ts uint64)

Observe bumps the local clock if a higher timestamp is seen.

func (*HLC) PhysicalCeiling 

func (h *HLC) PhysicalCeiling() int64

PhysicalCeiling returns the last Raft-committed physical ceiling in Unix milliseconds. Returns 0 if no ceiling has been established yet.

func (*HLC) SetPhysicalCeiling 

func (h *HLC) SetPhysicalCeiling(ms int64)

SetPhysicalCeiling atomically advances the Raft-agreed physical ceiling. It is called by the FSM whenever a HLC lease entry is applied to the log. The ceiling is monotonically increasing: calls with a smaller value are silently ignored.

type LeaderAdminProposerOption 

type LeaderAdminProposerOption func(*leaderAdminProposer)

func WithLeaderAdminToken 

func WithLeaderAdminToken(token string) LeaderAdminProposerOption

type LeaderProxy 

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

LeaderProxy forwards transactional requests to the current raft leader when the local node is not the leader.

func NewLeaderProxyForShardGroup 

func NewLeaderProxyForShardGroup(g *ShardGroup, opts ...TransactionOption) *LeaderProxy

NewLeaderProxyForShardGroup wires a LeaderProxy whose proposer consults g.raftPayloadWrap on every call, so SetRaftPayloadWrap becomes the hot-swap surface for the raft envelope cutover.

Use this in preference to NewLeaderProxyWithEngine(g.Engine, ...) for any ShardGroup that participates in the encryption cutover pipeline — without the dynamic wrap, post-cutover writes would land cleartext at index > cutoverIndex and halt the apply loop on strict-> unwrap. The non-wrap-aware constructor remains as a convenience for shard groups that opt out of encryption (test fixtures, transient groups).

Contract: g MUST be non-nil. The constructor takes the address of g.raftPayloadWrap so a nil receiver is an immediate nil deref — caller bug, not silently swallowed. Returning nil here would only defer the panic to the first sg.Txn.Commit call site (see main.go's buildShardGroups), with worse diagnostics; CLAUDE.md's "don't validate for scenarios that can't happen at internal boundaries" applies.

func NewLeaderProxyWithEngine 

func NewLeaderProxyWithEngine(engine raftengine.Engine, opts ...TransactionOption) *LeaderProxy

func (*LeaderProxy) Abort 

func (p *LeaderProxy) Abort(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)

func (*LeaderProxy) Close 

func (p *LeaderProxy) Close() error

func (*LeaderProxy) Commit 

func (p *LeaderProxy) Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)

type LeaderRoutedStore 

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

LeaderRoutedStore is an MVCCStore wrapper that serves reads from the local store only when leadership is verified; otherwise it proxies reads to the current leader via gRPC.

This is intended for single-raft-group deployments where the underlying store itself is not leader-aware (e.g. *store.MVCCStore).

Writes and maintenance operations are delegated to the local store.

func NewLeaderRoutedStore 

func NewLeaderRoutedStore(local store.MVCCStore, coordinator Coordinator) *LeaderRoutedStore

func (*LeaderRoutedStore) AllowExactScanFallbackAfterPhysicalLimit 

func (s *LeaderRoutedStore) AllowExactScanFallbackAfterPhysicalLimit(ctx context.Context, start []byte, _ []byte, visibleLimit, physicalLimit int, _ uint64, _ bool) bool

func (*LeaderRoutedStore) ApplyMutations 

func (s *LeaderRoutedStore) ApplyMutations(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS uint64) error

func (*LeaderRoutedStore) ApplyMutationsRaft 

func (s *LeaderRoutedStore) ApplyMutationsRaft(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS uint64) error

ApplyMutationsRaft forwards to the local store's raft-apply variant. See store.MVCCStore for the durability contract.

func (*LeaderRoutedStore) ApplyMutationsRaftAt 

func (s *LeaderRoutedStore) ApplyMutationsRaftAt(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error

ApplyMutationsRaftAt forwards to the local store's raft-entry-index- aware variant so the underlying pebbleStore can bundle metaAppliedIndex with the mutation. See PR #910 design §2.

func (*LeaderRoutedStore) Close 

func (s *LeaderRoutedStore) Close() error

func (*LeaderRoutedStore) CommittedVersionAt 

func (s *LeaderRoutedStore) CommittedVersionAt(ctx context.Context, key []byte, commitTS uint64) (bool, error)

CommittedVersionAt gates the exact-timestamp existence probe so client reads through this wrapper get a fresh authoritative answer even on a deposed leader. The FSM apply path is NOT affected — it holds the raw local store (not a LeaderRoutedStore), so its deterministic probe never goes through this method. The option-2 reuse path (RedisServer.resolveReuseLength) DOES call this and needs the authoritative answer to preserve the pending.length fast-path (returning the per-our-commit length rather than the leader's current Len) when our prior attempt actually committed.

Two-path strategy, mirroring how LatestCommitTS uses a lease fast-path and a proxy slow-path:

  • We are the leader with a valid lease (leaderOKForKey is true): the local replica is up-to-date by the lease invariant; read local.
  • Not leader (deposed or never): there is no RawCommittedVersionAt RPC to proxy to, so use the coordinator's LinearizableRead to submit a Raft ReadIndex — that protocol forwards to the current leader and waits until our local applied index has caught up to the leader's commit point. After that, a local probe sees every committed version of this key (including any landed at commitTS). If the read-index fails (no leader reachable, ctx canceled), fall back to (false, nil); the adapter's resolveReuseLength then re-reads via the already-leader-fenced ScanAt/GetAt, returning the leader's current Len — a valid serialization, just not the per-our-commit value.

func (*LeaderRoutedStore) Compact 

func (s *LeaderRoutedStore) Compact(ctx context.Context, minTS uint64) error

func (*LeaderRoutedStore) DeleteAt 

func (s *LeaderRoutedStore) DeleteAt(ctx context.Context, key []byte, commitTS uint64) error

func (*LeaderRoutedStore) DeletePrefixAt 

func (s *LeaderRoutedStore) DeletePrefixAt(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS uint64) error

func (*LeaderRoutedStore) DeletePrefixAtRaft 

func (s *LeaderRoutedStore) DeletePrefixAtRaft(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS uint64) error

DeletePrefixAtRaft forwards to the local store's raft-apply variant.

func (*LeaderRoutedStore) DeletePrefixAtRaftAt 

func (s *LeaderRoutedStore) DeletePrefixAtRaftAt(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS, appliedIndex uint64) error

DeletePrefixAtRaftAt forwards to the local store's raft-entry- index-aware variant. See PR #910 design §2 "why both leaves".

func (*LeaderRoutedStore) ExistsAt 

func (s *LeaderRoutedStore) ExistsAt(ctx context.Context, key []byte, ts uint64) (bool, error)

func (*LeaderRoutedStore) ExpireAt 

func (s *LeaderRoutedStore) ExpireAt(ctx context.Context, key []byte, expireAt uint64, commitTS uint64) error

func (*LeaderRoutedStore) ExportVersions 

func (s *LeaderRoutedStore) ExportVersions(ctx context.Context, opts store.ExportVersionsOptions) (store.ExportVersionsResult, error)

func (*LeaderRoutedStore) GetAt 

func (s *LeaderRoutedStore) GetAt(ctx context.Context, key []byte, ts uint64) ([]byte, error)

func (*LeaderRoutedStore) GetAtWithReadFence 

func (s *LeaderRoutedStore) GetAtWithReadFence(ctx context.Context, key []byte, ts uint64, groupID uint64, readRouteVersion uint64) ([]byte, error)

func (*LeaderRoutedStore) GlobalLastCommitTS 

func (s *LeaderRoutedStore) GlobalLastCommitTS(ctx context.Context) uint64

GlobalLastCommitTS returns the most recently committed HLC timestamp from the authoritative leader. On the leader this is the local LastCommitTS. On a follower the method issues a lightweight RPC (RawLatestCommitTS with an empty key) so callers obtain a non-stale snapshot — critical for ConsistentRead semantics where followers must not serve reads at a stale local watermark. Falls back to the local LastCommitTS on any error.

func (*LeaderRoutedStore) ImportVersions 

func (s *LeaderRoutedStore) ImportVersions(ctx context.Context, opts store.ImportVersionsOptions) (store.ImportVersionsResult, error)

func (*LeaderRoutedStore) LastAppliedIndex 

func (s *LeaderRoutedStore) LastAppliedIndex() (uint64, bool, error)

LastAppliedIndex forwards to the local store when it implements raftengine.AppliedIndexReader. Defensive: in production today the kvFSM holds a *pebbleStore directly (not a LeaderRoutedStore — that wrapper is used by adapter/server code for read routing, not by the FSM apply path); so this forward is currently dead code for the cold-start skip optimisation. We add it anyway because future refactors might wrap the FSM's store, and a silent no-op there would degrade the optimisation to full-restore-always with no failure signal.

(0, false, nil) returns are the strictly-additive fallback — either the wrapper has no local, the local does not implement the reader, or the local reports missing/truncated. The caller in internal/raftengine/etcd/wal_store.go (Branch 3) treats all of these as "fall back to full restore", which is correct.

func (*LeaderRoutedStore) LastCommitTS 

func (s *LeaderRoutedStore) LastCommitTS() uint64

func (*LeaderRoutedStore) LatestCommitTS 

func (s *LeaderRoutedStore) LatestCommitTS(ctx context.Context, key []byte) (uint64, bool, error)

func (*LeaderRoutedStore) LatestCommitTSWithReadFence 

func (s *LeaderRoutedStore) LatestCommitTSWithReadFence(ctx context.Context, key []byte, readRouteVersion uint64) (uint64, bool, error)

func (*LeaderRoutedStore) MigrationHLCFloor 

func (s *LeaderRoutedStore) MigrationHLCFloor(ctx context.Context, jobID uint64) (uint64, error)

func (*LeaderRoutedStore) PutAt 

func (s *LeaderRoutedStore) PutAt(ctx context.Context, key []byte, value []byte, commitTS uint64, expireAt uint64) error

func (*LeaderRoutedStore) PutWithTTLAt 

func (s *LeaderRoutedStore) PutWithTTLAt(ctx context.Context, key []byte, value []byte, commitTS uint64, expireAt uint64) error

func (*LeaderRoutedStore) Restore 

func (s *LeaderRoutedStore) Restore(buf io.Reader) error

func (*LeaderRoutedStore) RetireMigration 

func (s *LeaderRoutedStore) RetireMigration(ctx context.Context, jobID uint64) error

func (*LeaderRoutedStore) ReverseScanAt 

func (s *LeaderRoutedStore) ReverseScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error)

func (*LeaderRoutedStore) ReverseScanAtPhysicalLimit 

func (s *LeaderRoutedStore) ReverseScanAtPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64) ([]*store.KVPair, bool, error)

func (*LeaderRoutedStore) ScanAt 

func (s *LeaderRoutedStore) ScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error)

func (*LeaderRoutedStore) ScanAtPhysicalLimit 

func (s *LeaderRoutedStore) ScanAtPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64) ([]*store.KVPair, bool, error)

func (*LeaderRoutedStore) ScanAtWithReadFence 

func (s *LeaderRoutedStore) ScanAtWithReadFence(ctx context.Context, start []byte, end []byte, limit int, ts uint64, reverse bool, groupID uint64, readRouteVersion uint64, routeStart []byte, routeEnd []byte) ([]*store.KVPair, error)

func (*LeaderRoutedStore) ScanKeysAt 

func (s *LeaderRoutedStore) ScanKeysAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([][]byte, error)

func (*LeaderRoutedStore) ScanKeysAtWithReadFence 

func (s *LeaderRoutedStore) ScanKeysAtWithReadFence(ctx context.Context, start []byte, end []byte, limit int, ts uint64, groupID uint64, readRouteVersion uint64) ([][]byte, error)

func (*LeaderRoutedStore) SetDurableAppliedIndex 

func (s *LeaderRoutedStore) SetDurableAppliedIndex(idx uint64) error

SetDurableAppliedIndex forwards to the local store when it implements raftengine.AppliedIndexWriter. Symmetric defensive no-op when the local store does not expose the writer seam — see LastAppliedIndex doc-comment.

func (*LeaderRoutedStore) Snapshot 

func (s *LeaderRoutedStore) Snapshot() (store.Snapshot, error)

func (*LeaderRoutedStore) WriteConflictCountsByPrefix 

func (s *LeaderRoutedStore) WriteConflictCountsByPrefix() map[string]uint64

WriteConflictCountsByPrefix delegates to the local MVCC store. The leader-routed wrapper does not add cross-group conflict detection of its own, so the node-local view IS the authoritative view.

type LeaderRoutedTSOAllocator 

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

LeaderRoutedTSOAllocator serves local requests on the TSO leader and sends follower requests to the leader address published by the group-0 engine. It re-resolves that address after transient errors so a leadership change does not pin a BatchAllocator refill to a stale endpoint.

func NewLeaderRoutedTSOAllocator 

func NewLeaderRoutedTSOAllocator(
	local TSOAllocator,
	leader raftengine.LeaderView,
	opts ...LeaderRoutedTSOAllocatorOption,
) (*LeaderRoutedTSOAllocator, error)

func (*LeaderRoutedTSOAllocator) Close 

func (a *LeaderRoutedTSOAllocator) Close() error

func (*LeaderRoutedTSOAllocator) IsLeader 

func (a *LeaderRoutedTSOAllocator) IsLeader() bool

func (*LeaderRoutedTSOAllocator) Next 

func (a *LeaderRoutedTSOAllocator) Next(ctx context.Context) (uint64, error)

func (*LeaderRoutedTSOAllocator) NextAfter 

func (a *LeaderRoutedTSOAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*LeaderRoutedTSOAllocator) NextBatch 

func (a *LeaderRoutedTSOAllocator) NextBatch(ctx context.Context, n int) (uint64, error)

func (*LeaderRoutedTSOAllocator) NextBatchAfter 

func (a *LeaderRoutedTSOAllocator) NextBatchAfter(ctx context.Context, n int, min uint64) (uint64, error)

func (*LeaderRoutedTSOAllocator) PhaseDActive 

func (a *LeaderRoutedTSOAllocator) PhaseDActive() bool

func (*LeaderRoutedTSOAllocator) PhaseDRequired 

func (a *LeaderRoutedTSOAllocator) PhaseDRequired() bool

func (*LeaderRoutedTSOAllocator) RunLeaseRenewal 

func (a *LeaderRoutedTSOAllocator) RunLeaseRenewal(ctx context.Context)

func (*LeaderRoutedTSOAllocator) ValidateDurableTimestamp 

func (a *LeaderRoutedTSOAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error

func (*LeaderRoutedTSOAllocator) ValidateShadowTimestamp 

func (a *LeaderRoutedTSOAllocator) ValidateShadowTimestamp(ctx context.Context, min uint64) (TSOReservation, error)

type LeaderRoutedTSOAllocatorOption 

type LeaderRoutedTSOAllocatorOption func(*LeaderRoutedTSOAllocator)

func WithTSOCutoverActivation 

func WithTSOCutoverActivation() LeaderRoutedTSOAllocatorOption

func WithTSOObserver 

func WithTSOObserver(observer TSOObserver) LeaderRoutedTSOAllocatorOption

func WithTSOPhaseDActivation 

func WithTSOPhaseDActivation() LeaderRoutedTSOAllocatorOption

func WithTSORoutedClock 

func WithTSORoutedClock(clock *HLC) LeaderRoutedTSOAllocatorOption

type LeaseReadObserver 

type LeaseReadObserver interface {
	// ObserveLeaseRead is called with hit=true when the lease fast path
	// served the read from local AppliedIndex, or hit=false when the
	// coordinator fell back to a full LinearizableRead (expired lease,
	// engine reported non-leader, or leader-loss callback raced with
	// the request).
	ObserveLeaseRead(hit bool)
}

LeaseReadObserver records lease-read fast-path vs slow-path outcomes without coupling kv to a concrete monitoring backend. It is called once per LeaseRead invocation that actually evaluates the lease (the initial type-assertion/LeaseDuration==0 short-circuits are NOT counted because they indicate the engine does not participate in lease reads at all).

Implementations MUST be safe for concurrent use and MUST NOT block; the observer is invoked on the Redis GET hot path.

type LeaseReadableCoordinator 

type LeaseReadableCoordinator interface {
	LeaseRead(ctx context.Context) (uint64, error)
	LeaseReadForKey(ctx context.Context, key []byte) (uint64, error)
}

LeaseReadableCoordinator is the optional capability implemented by coordinators that participate in the leader-local lease read path (see docs/design/2026_04_20_implemented_lease_read.md). Callers that want lease reads should type-assert to this interface and fall back to LinearizableRead when the assertion fails, following the same pattern as raftengine.LeaseProvider. Keeping the lease methods OFF the Coordinator interface avoids breaking existing external implementations that predate the lease-read feature.

type LocalTSOAllocator 

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

func NewLocalTSOAllocator 

func NewLocalTSOAllocator(coord tsoCoordinator, opts ...LocalTSOAllocatorOption) (*LocalTSOAllocator, error)

func (*LocalTSOAllocator) IsLeader 

func (a *LocalTSOAllocator) IsLeader() bool

func (*LocalTSOAllocator) Next 

func (a *LocalTSOAllocator) Next(ctx context.Context) (uint64, error)

func (*LocalTSOAllocator) NextAfter 

func (a *LocalTSOAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*LocalTSOAllocator) NextBatch 

func (a *LocalTSOAllocator) NextBatch(ctx context.Context, n int) (uint64, error)

func (*LocalTSOAllocator) NextBatchAfter 

func (a *LocalTSOAllocator) NextBatchAfter(ctx context.Context, n int, min uint64) (uint64, error)

func (*LocalTSOAllocator) RunLeaseRenewal 

func (a *LocalTSOAllocator) RunLeaseRenewal(ctx context.Context)

type LocalTSOAllocatorOption 

type LocalTSOAllocatorOption func(*LocalTSOAllocator)

func WithTSOLeaderPollInterval 

func WithTSOLeaderPollInterval(interval time.Duration) LocalTSOAllocatorOption

type LockResolver 

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

LockResolver periodically scans for expired transaction locks and resolves them. This handles the case where secondary commit fails and leaves orphaned locks that no read path would discover (e.g., cold keys).

func NewLockResolver 

func NewLockResolver(ss *ShardStore, groups map[uint64]*ShardGroup, log *slog.Logger) *LockResolver

NewLockResolver creates and starts a background lock resolver.

func (*LockResolver) Close 

func (lr *LockResolver) Close()

Close stops the background resolver and waits for it to finish.

type MutationWriteGate 

type MutationWriteGate interface {
	EnsureMutationsWriteAllowed([]*pb.Mutation, uint64) error
}

MutationWriteGate rejects raw mutations whose commit timestamp lands at or below the owning route's migration write floor. Follower-forwarded writes are stamped on the leader, outside the coordinator that owns the route table, so the leader-side RPC handler needs this to re-apply the same check.

type OP 

type OP int

OP is an operation type. 

const (
	Put OP = iota
	Del
	// DelPrefix deletes all visible keys matching the prefix stored in Key.
	// An empty Key means "all keys". Transaction-internal keys are excluded.
	DelPrefix
)

Operation types.

type OperationGroup 

type OperationGroup[T OP] struct {
	Elems []*Elem[T]
	IsTxn bool
	// KeyVizLabel tags this operation group for KeyViz attribution.
	// The zero value is the legacy unlabeled route-only view.
	KeyVizLabel keyviz.Label
	// StartTS is a logical timestamp captured at transaction begin.
	// It is ignored for non-transactional groups.
	StartTS uint64
	// CommitTS optionally pins the transaction commit timestamp.
	// Coordinators choose one automatically when this is zero.
	CommitTS uint64
	// PrevCommitTS carries the commit timestamp of a failed previous attempt
	// of the same single-shard transaction (option-2 one-phase idempotency
	// dedup). It is set only on a retry that reuses the prior attempt's write
	// set, and only flows to the one-phase apply path, where the FSM probes
	// whether that attempt already landed and no-ops the apply if so. Zero on
	// first attempts and on every non-retry caller. See
	// docs/design/2026_05_21_proposed_txn_secondary_idempotency.md.
	PrevCommitTS uint64
	// ReadKeys carries the transaction's read set so the FSM can validate
	// read-write conflicts atomically with the commit.
	ReadKeys [][]byte
	// ObservedRouteVersion is the encoded durable catalog version this
	// transaction's read set was captured at (typically set on BeginTxn
	// from distribution.Engine.Version()). Zero means "unpinned"; the
	// version-0 sentinel is decoded for compatibility but is not emitted
	// until every Raft member advertises support. M3 of the Composed-1 design
	// (docs/design/2026_05_29_implemented_composed1_cross_group_commit_guard.md)
	// will gate the FSM apply path on this version so a route shift
	// between BeginTxn and Commit is caught before it can produce a
	// G1c anomaly across a cross-group MoveRange / SplitRange.
	ObservedRouteVersion uint64
}

OperationGroup is a group of operations that should be executed atomically.

type PartitionResolver 

type PartitionResolver interface {
	ResolveGroup(key []byte) (uint64, bool)

	// RecognisesPartitionedKey reports whether the key SHAPE is
	// one this resolver is responsible for. Implementations
	// answer based on prefix / structural inspection only — the
	// answer must NOT depend on any in-memory mapping that could
	// drift out of sync, otherwise the router cannot reliably
	// fail-closed for unresolved-but-recognised keys.
	RecognisesPartitionedKey(key []byte) bool
}

PartitionResolver maps a key to its owning Raft group when the key belongs to a partition-scheme keyspace (e.g. SQS HT-FIFO, where each (queue, partition) pair lives on a different group). ShardRouter consults the resolver before falling through to the byte-range engine, so partition routing can override the default shard-range layout without breaking the engine's non-overlapping- cover invariant.

Implementations must be safe for concurrent use — ResolveGroup is called on the request hot path. Returning (0, false) for a key the resolver does not recognise lets the router fall through to the engine. Returning (0, false) for a key the resolver DOES recognise (the partitioned shape matches but the queue is not in the map, or the partition is out of range) is also valid; the router uses RecognisesPartitionedKey to distinguish "not partitioned, fall through" from "partitioned but unresolved, fail closed".

The fail-closed split matters under partition-map drift / a partial rollout: without it, an unresolved partitioned key would silently land on the engine's SQS-catalog default group (because routeKey normalises every !sqs|... key to !sqs|route|global) instead of surfacing a routing error.

type ProposalObserver 

type ProposalObserver interface {
	ObserveProposalFailure()
}

ProposalObserver records raft proposal failures for operational metrics.

type RaftMember 

type RaftMember struct {
	NodeID   string
	Address  string
	Suffrage string
}

RaftMember describes one member of the Raft group that owns a key. The adapter layer uses this read-only view for peer-local side channels whose payloads deliberately do not enter the Raft log.

type RaftMembershipCoordinator 

type RaftMembershipCoordinator interface {
	RaftMembers(ctx context.Context) ([]RaftMember, error)
	RaftMembersForKey(ctx context.Context, key []byte) ([]RaftMember, error)
}

RaftMembershipCoordinator is the optional capability implemented by coordinators that can expose both cluster-wide and per-key Raft membership. It does not establish a read fence; callers must use it only for peer discovery and keep correctness decisions behind their own quorum checks.

type RaftPayloadWrapper 

type RaftPayloadWrapper func(payload []byte) ([]byte, error)

RaftPayloadWrapper transforms an FSM payload into a §4.2 raft envelope just before submission to the engine. The Stage 3 default (when no wrapper is installed on a coordinator) is identity — payloads pass through unchanged. Stage 6's cluster-flag pipeline installs an active wrapper, sourced from the sidecar's currently- active raft DEK and a writer-registry-backed nonce factory.

Implementations MUST be safe to call concurrently from many goroutines: the coordinator may invoke this on every concurrent proposal. Encryption-state transitions (Phase 1 → Phase 2 cutover) publish a fresh closure via atomic.Pointer so the wrapper observes one consistent (cipher, key_id, nonce_factory) tuple per call.

type RaftStatusProvider 

type RaftStatusProvider interface {
	Status() raftengine.Status
}

type RaftTSOAllocator 

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

RaftTSOAllocator reserves timestamp windows on the dedicated TSO leader. A window is returned only after its inclusive end has committed to Raft. Failed proposals may leak a local window, but they can never expose an uncommitted timestamp or make a later leader reuse a returned timestamp.

func NewRaftTSOAllocator 

func NewRaftTSOAllocator(group *ShardGroup, clock *HLC, opts ...RaftTSOAllocatorOption) (*RaftTSOAllocator, error)

func (*RaftTSOAllocator) AllocationFloor 

func (a *RaftTSOAllocator) AllocationFloor() uint64

func (*RaftTSOAllocator) CutoverActive 

func (a *RaftTSOAllocator) CutoverActive() bool

CutoverActive reports the durable cutover marker. It sits beside PhaseDActive so a caller holding this allocator can tell which markers are already committed without reaching for the state machine itself -- the distribution server needs exactly that to know whether a requested activation would change anything.

func (*RaftTSOAllocator) IsLeader 

func (a *RaftTSOAllocator) IsLeader() bool

func (*RaftTSOAllocator) Next 

func (a *RaftTSOAllocator) Next(ctx context.Context) (uint64, error)

func (*RaftTSOAllocator) NextAfter 

func (a *RaftTSOAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*RaftTSOAllocator) NextBatch 

func (a *RaftTSOAllocator) NextBatch(ctx context.Context, n int) (uint64, error)

func (*RaftTSOAllocator) NextBatchAfter 

func (a *RaftTSOAllocator) NextBatchAfter(ctx context.Context, n int, min uint64) (uint64, error)

func (*RaftTSOAllocator) PhaseDActive 

func (a *RaftTSOAllocator) PhaseDActive() bool

func (*RaftTSOAllocator) PhaseDFloor 

func (a *RaftTSOAllocator) PhaseDFloor() uint64

func (*RaftTSOAllocator) PhaseDRequired 

func (a *RaftTSOAllocator) PhaseDRequired() bool

func (*RaftTSOAllocator) ReserveBatchAfter 

func (a *RaftTSOAllocator) ReserveBatchAfter(
	ctx context.Context,
	n int,
	min uint64,
	activateCutover bool,
	activatePhaseD bool,
) (TSOReservation, error)

ReserveBatchAfter serializes floor discovery, the one-way cutover marker, and window reservation under the TSO leader. A returned window is always above both the caller's minimum and every authoritative data-group commit observed when this leader term first serves a request.

func (*RaftTSOAllocator) RunLeaseRenewal 

func (a *RaftTSOAllocator) RunLeaseRenewal(ctx context.Context)

func (*RaftTSOAllocator) ValidateDurableTimestamp 

func (a *RaftTSOAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error

type RaftTSOAllocatorOption 

type RaftTSOAllocatorOption func(*RaftTSOAllocator)

func WithTSOCutoverFloorProvider 

func WithTSOCutoverFloorProvider(provider TSOCutoverFloorProvider) RaftTSOAllocatorOption

type ReadFenceTarget 

type ReadFenceTarget struct {
	GroupID uint64
	Key     []byte
}

ReadFenceTarget names one Raft group a read fence must cover, together with a representative key inside it.

GroupID is carried explicitly rather than left to be re-derived from Key, because that derivation is lossy. Redis wide-column rows can be reachable through two routes: the logical owner of the decoded user key, and a legacy route still holding rows under the raw !hs|fld| prefix. Their representative keys both normalize through routeKey to the same user key, so a consumer that re-resolves by bytes collapses the two groups into one and silently leaves the legacy group unfenced. A zero GroupID means "resolve from Key" and is what plain per-command keys use.

func LeaseReadGroupTargets 

func LeaseReadGroupTargets(c Coordinator, targets []ReadFenceTarget) []ReadFenceTarget

LeaseReadGroupTargets collapses targets to one per Raft group, preferring the group id a target already carries over re-deriving it from the key bytes. A target with GroupID 0 falls back to key resolution, which is what plain per-command keys need.

The distinction matters for Redis wide-column ranges: the owner target and the legacy raw-prefix target normalize to the same user key, so resolving both from bytes would dedup them into a single group and leave the legacy group unfenced.

type ReadTimestamp 

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

ReadTimestamp is the adapter-side result of beginning a transaction snapshot. When it represents an applied pre-Phase-D watermark, it also carries a process-local capability. The ReadTimestamp is intentionally reusable by adapter helpers that perform several writes against the same audited read boundary: each DispatchWithReadTimestamp call mints a distinct one-use coordinator voucher, and unused prepared vouchers are revoked after that dispatch attempt. The capability cannot be constructed outside this package because both the timestamp and voucher state are private.

func BeginReadTimestampThrough 

func BeginReadTimestampThrough(
	ctx context.Context,
	coord Coordinator,
	legacyTimestamp uint64,
	label string,
) (ReadTimestamp, error)

BeginReadTimestampThrough preserves the caller's applied-snapshot watermark. Once Phase D is requested, this boundary activates it before validation. An applied pre-D watermark receives a bounded one-use coordinator voucher; arbitrary caller timestamps remain subject to group-0 numeric validation. The returned timestamp must be used for every read and OperationGroup.StartTS.

func (ReadTimestamp) Timestamp 

func (t ReadTimestamp) Timestamp() uint64

func (ReadTimestamp) WithDispatchVoucher 

func (t ReadTimestamp) WithDispatchVoucher(ctx context.Context) context.Context

WithDispatchVoucher binds this read timestamp's process-local capability to ctx. A timestamp without a voucher is still bound so it shadows any parent capability instead of accidentally inheriting authority for an older read.

type RegistrationGate 

type RegistrationGate struct {
	// Barrier is the open-or-closed channel described above. A nil
	// Barrier (the zero value, or an explicitly nil field) means
	// "never armed" → ungated; awaitRegistration checks for nil first
	// so a nil channel is never received on (which would block forever).
	Barrier <-chan struct{}
	// StorageEnvelopeActive and ActiveStorageKeyID read the process-wide
	// encryption StateCache. A write is gated only when the §7.1 cutover
	// has fired AND a storage DEK is active — i.e. the write would land
	// encrypted and thus emit a nonce under this node's identity.
	StorageEnvelopeActive func() bool
	ActiveStorageKeyID    func() (uint32, bool)
}

RegistrationGate carries the Stage 7a §4.1 registration-before- first-write barrier into the coordinator. main.go owns the encryption StateCache and the registration goroutine and supplies this; kv stays decoupled from internal/encryption by taking a plain channel + predicate closures rather than the StateCache type.

Barrier three-state (per the 7a design §3.2):

  • nil → no registration pending (Phase 0 / skip / off) → ungated
  • open (non-nil, not closed) → registration in flight → mutating encrypted writes block on it
  • closed → registration committed → ungated (fast path)

type RouteHistory 

type RouteHistory interface {
	// SnapshotAt returns the route catalog at the given catalog
	// version.  Returns (zero, false) when the version is outside
	// the ring (either evicted by depth, or in the future).  The
	// M3 gate maps the not-found case to ErrComposed1VersionGCd.
	SnapshotAt(version uint64) (RouteSnapshot, bool)
	// Current returns the route catalog snapshot at the engine's
	// current catalog version.  Returns (zero, false) when the
	// engine has no history (bare-struct case used by some test
	// seams).  The M3 cross-version fence uses this to compare
	// the txn's observed-version owner against the current
	// owner — a mismatch is the §3 codex P1 trace.
	Current() (RouteSnapshot, bool)
}

RouteHistory is the kv-side interface to the route catalog's versioned-snapshot ring. *distribution.Engine satisfies it via WrapDistributionEngine. Defined in the kv package so kvFSM does not have to import a concrete type for the field; the M3 verifyComposed1 gate uses only SnapshotAt + Current + the returned snapshot's OwnerOf, so the interface stays minimal.

func WrapDistributionEngine 

func WrapDistributionEngine(e *distribution.Engine) RouteHistory

WrapDistributionEngine adapts a *distribution.Engine so it satisfies the kv.RouteHistory interface that kvFSM's M2 Composed-1 plumbing consumes.

The adapter is a thin two-hop boxing: kv.RouteHistory.SnapshotAt returns a kv.RouteSnapshot interface; distribution.Engine.SnapshotAt returns a concrete distribution.RouteHistorySnapshot struct. Go's structural interface satisfaction is byte-equivalent on return types, so we cannot have *distribution.Engine satisfy kv.RouteHistory directly — and moving the interface to the distribution package would create an import cycle (kv already imports distribution via kv/sharded_coordinator.go).

Production wiring in main.go uses WrapDistributionEngine to install the engine as the FSM's route-history provider; tests that want to mock kv.RouteHistory bypass the wrapper entirely and implement the kv interface directly.

See docs/design/2026_05_29_implemented_composed1_cross_group_commit_guard.md §M2.

type RouteSnapshot 

type RouteSnapshot interface {
	// Version returns the catalog version this snapshot was
	// recorded at.
	Version() uint64
	// OwnerOf returns the Raft group ID that owned key at this
	// snapshot's version.  (0, false) when no route covered key.
	OwnerOf(key []byte) (uint64, bool)
	// WriteFencedForKey reports whether key is currently inside a
	// WriteFenced route in this snapshot.
	WriteFencedForKey(key []byte) bool
	// WriteFencedIntersects reports whether [start, end) intersects
	// any WriteFenced route in this snapshot.
	WriteFencedIntersects(start, end []byte) bool
}

RouteSnapshot is the historical view of the route catalog at a specific version. Returned by RouteHistory.SnapshotAt; the M3 gate uses Version + OwnerOf to compare against the FSM's shardGroupID.

type ShadowTimestampAllocator 

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

ShadowTimestampAllocator serializes each legacy candidate through group 0 before returning it. Candidates at or below a prior TSO floor are discarded and retried; once the durable cutover marker is active, the allocator returns the reserved TSO timestamp directly so rolling restarts cannot mix sources.

func NewShadowTimestampAllocator 

func NewShadowTimestampAllocator(
	legacy *HLC,
	shadow TSOShadowReservationAllocator,
	logger *slog.Logger,
	opts ...ShadowTimestampAllocatorOption,
) (*ShadowTimestampAllocator, error)

func (*ShadowTimestampAllocator) Close 

func (a *ShadowTimestampAllocator) Close() error

func (*ShadowTimestampAllocator) Next 

func (a *ShadowTimestampAllocator) Next(ctx context.Context) (uint64, error)

func (*ShadowTimestampAllocator) NextAfter 

func (a *ShadowTimestampAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*ShadowTimestampAllocator) PhaseDActive 

func (a *ShadowTimestampAllocator) PhaseDActive() bool

func (*ShadowTimestampAllocator) PhaseDRequired 

func (a *ShadowTimestampAllocator) PhaseDRequired() bool

func (*ShadowTimestampAllocator) ValidateDurableTimestamp 

func (a *ShadowTimestampAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error

type ShadowTimestampAllocatorOption 

type ShadowTimestampAllocatorOption func(*ShadowTimestampAllocator)

func WithTSOShadowCutoverState 

func WithTSOShadowCutoverState(state tsoCutoverState) ShadowTimestampAllocatorOption

func WithTSOShadowObserver 

func WithTSOShadowObserver(observer TSOObserver) ShadowTimestampAllocatorOption

type ShardGroup 

type ShardGroup struct {
	Engine raftengine.Engine
	Store  store.MVCCStore
	Txn    Transactional
	// TSOState is set only for reserved group 0. Keeping the applied floor and
	// cutover marker next to the group's engine lets the leader allocator read
	// consensus-owned state without treating the shared HLC mirror as durable.
	TSOState *TSOStateMachine
	// contains filtered or unexported fields
}

func (*ShardGroup) BeginCutoverBarrier 

func (g *ShardGroup) BeginCutoverBarrier() <-chan struct{}

BeginCutoverBarrier opens the §7.1 step-1 quiescence barrier on this shard group's proposer chain. Returns a channel that closes when all in-flight user Propose calls drain; the typical caller uses WaitInflightDrained which composes context cancellation.

Forwards to *dynamicWrappedProposer when present. When the proposer is the bare engine (raw Engine fallback in test fixtures), returns a pre-closed channel so callers that don't distinguish barrier-capable from -incapable proposers can drive the same state-machine shape against either.

6E-2d wiring: every leader's EnableRaftEnvelope handler calls this on each ShardGroup that participates in the cutover before proposing the cutover entry. After return, the proposer's dynamicWrappedProposer.Propose rejects fresh user calls with raftengine.ErrEnvelopeCutoverInProgress.

func (*ShardGroup) EndCutoverBarrier 

func (g *ShardGroup) EndCutoverBarrier()

EndCutoverBarrier closes the §7.1 step-6 barrier on this shard group's proposer chain. Idempotent against barrier-incapable proposers (no-op). Callers MUST pair each BeginCutoverBarrier with exactly one EndCutoverBarrier (the EnableRaftEnvelope handler uses defer).

func (*ShardGroup) Proposer 

func (g *ShardGroup) Proposer() raftengine.Proposer

Proposer returns the wrap-aware proposer chain installed by NewLeaderProxyForShardGroup, or the raw Engine when the constructor was bypassed (legacy / test fixtures that build ShardGroup via struct literal). Direct shard proposals (HLC lease renewal, future cipher rotations, etc.) MUST go through this getter rather than g.Engine.Propose so the Stage 6E-2c dynamic wrap path applies — a direct g.Engine Propose call would bypass the wrap pointer and let post-cutover writes land cleartext above the raft-envelope cutover, halting the apply loop on §6.3 strict-> unwrap (codex P2 round-1).

func (*ShardGroup) RaftPayloadWrap 

func (g *ShardGroup) RaftPayloadWrap() RaftPayloadWrapper

RaftPayloadWrap returns the currently-installed wrap closure, or nil if the wrap is inactive. Primarily intended for tests and diagnostics; production proposers consult the underlying atomic.Pointer directly (see dynamicWrappedProposer).

func (*ShardGroup) SetLeaderReadToken 

func (g *ShardGroup) SetLeaderReadToken(token string)

SetLeaderReadToken configures the bearer token used by forwarded lease-read RPCs. An empty token preserves explicitly configured insecure admin mode.

func (*ShardGroup) SetRaftPayloadWrap 

func (g *ShardGroup) SetRaftPayloadWrap(wrap RaftPayloadWrapper)

SetRaftPayloadWrap publishes wrap as the active raft envelope closure for this shard group. Passing nil clears the wrap (the proposer reverts to cleartext pass-through). Safe to call from any goroutine; the next Propose / ProposeAdmin observes the new state via the proposer's atomic.Pointer.Load.

This is the sole supported way to install or rotate the wrap closure on a running coordinator. Stage 6E-2d's EnableRaftEnvelope handler will call this on every leader when the cutover entry commits.

func (*ShardGroup) WaitInflightDrained 

func (g *ShardGroup) WaitInflightDrained(ctx context.Context) error

WaitInflightDrained blocks until the in-flight Propose counter drops to 0 after BeginCutoverBarrier ran on this ShardGroup, or ctx fires. Returns nil on drain or when the proposer is barrier- incapable (degraded fast-path so test fixtures don't deadlock the handler). Wraps ctx.Err() on cancellation.

type ShardRouter 

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

ShardRouter routes requests to multiple raft groups based on key ranges.

Cross-shard transactions are not supported. They require distributed coordination (for example, 2PC) to ensure atomicity.

Non-transactional request batches may still partially succeed across shards.

func NewShardRouter 

func NewShardRouter(e *distribution.Engine) *ShardRouter

NewShardRouter creates a new router.

func (*ShardRouter) Abort 

func (s *ShardRouter) Abort(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)

Abort dispatches aborts to the correct raft group.

func (*ShardRouter) Commit 

func (s *ShardRouter) Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)

func (*ShardRouter) Get 

func (s *ShardRouter) Get(ctx context.Context, key []byte) ([]byte, error)

Get retrieves a key routed to the correct shard.

func (*ShardRouter) Register 

func (s *ShardRouter) Register(group uint64, tm Transactional, st store.MVCCStore)

Register associates a raft group ID with its transactional manager and store.

func (*ShardRouter) ResolveGroup 

func (s *ShardRouter) ResolveGroup(rawKey []byte) (uint64, bool)

ResolveGroup tries the partition resolver first (when installed), then falls through to the byte-range engine. Exposed at package scope so ShardedCoordinator's per-key helpers (groupForKey, routeAndGroupForKey, engineGroupIDForKey, groupMutations) can consult the same dispatch path Commit / Abort / Get use — without it those helpers would bypass the resolver and partitioned-FIFO traffic would silently mis-route through 2PC and the read paths.

The resolver runs on the RAW key before any user-key normalization. SQS keys in particular are collapsed to !sqs|route|global by routeKey to keep the engine's per-shard layout simple, but that collapse hides the partitioned-prefix information the resolver needs (issue: codex P1 / gemini high on PR #715). The engine still sees the post-normalization key, so legacy routing (catalog → !sqs|route|global → default group) stays unchanged.

Fail-closed for recognised-but-unresolved keys: when the resolver recognises a partitioned shape (RecognisesPartitionedKey == true) but cannot resolve the queue/partition pair (ResolveGroup returns ok=false), the router refuses to fall through to the engine. Otherwise the engine would route the partitioned key to !sqs|route|global's default group, silently mis-routing HT-FIFO traffic during partition-map drift or partial rollout (codex P1 round 2 on PR #715).

Returns (0, false) when neither the resolver nor the engine recognises the key. Caller surfaces this as an "unknown group" error so a partitioned-prefix key whose queue is missing from the resolver map fails closed rather than landing on whichever engine-default group happens to cover the raw bytes.

func (*ShardRouter) WithPartitionResolver 

func (s *ShardRouter) WithPartitionResolver(r PartitionResolver) *ShardRouter

WithPartitionResolver installs a partition-keyspace resolver that is consulted before the byte-range engine on every dispatch. A nil resolver clears any previously-installed resolver. Returns the receiver so callers can chain.

Intended for use during startup, before the router begins handling requests. Interface assignment in Go is not atomic, so a call that races with a concurrent ResolveGroup in resolveGroup may produce a torn read; callers must wire the resolver once during construction (parseRuntimeConfig → NewShardedCoordinator → WithPartitionResolver) and treat any post-startup re-assignment as undefined behaviour.

type ShardStore 

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

ShardStore routes MVCC reads to shard-specific stores and proxies to leaders when needed.

func NewShardStore 

func NewShardStore(engine *distribution.Engine, groups map[uint64]*ShardGroup) *ShardStore

NewShardStore creates a sharded MVCC store wrapper.

func (*ShardStore) AllowExactScanFallbackAfterPhysicalLimit 

func (s *ShardStore) AllowExactScanFallbackAfterPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, _ uint64, _ bool) bool

func (*ShardStore) ApplyMutations 

func (s *ShardStore) ApplyMutations(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS uint64) error

ApplyMutations applies a batch of mutations to the correct shard store.

All mutations must belong to the same shard. Cross-shard mutation batches are not supported.

func (*ShardStore) ApplyMutationsRaft 

func (s *ShardStore) ApplyMutationsRaft(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS uint64) error

ApplyMutationsRaft is the raft-apply variant; see store.MVCCStore for the durability contract. Only the FSM may call this method.

func (*ShardStore) ApplyMutationsRaftAt 

func (s *ShardStore) ApplyMutationsRaftAt(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error

ApplyMutationsRaftAt is the raft-entry-index-aware variant. Threads appliedIndex through to the single owning shard so the leaf can bundle metaAppliedIndex with the mutation. See PR #910 design §2.

func (*ShardStore) CaptureBackupRouteSnapshot 

func (s *ShardStore) CaptureBackupRouteSnapshot(start []byte, end []byte) BackupRouteSnapshot

CaptureBackupRouteSnapshot captures route ownership and scan bounds once.

func (*ShardStore) Close 

func (s *ShardStore) Close() error

func (*ShardStore) CommittedVersionAt 

func (s *ShardStore) CommittedVersionAt(ctx context.Context, key []byte, commitTS uint64) (bool, error)

CommittedVersionAt routes the exact-timestamp existence probe to the owning group's local store, gated on the same lease-aware leader check GetAt uses, so a deposed node that has not yet applied a freshly- committed entry does not silently return false to a client read. The FSM apply path is NOT affected — it holds the per-shard store directly (not ShardStore) and runs the probe on the deterministic local replica it is writing to. The option-2 reuse path (RedisServer.resolveReuseLength) goes through this wrapper, so during leader churn the probe must answer authoritatively or defer to a leader-routed re-read.

There is no RawCommittedVersionAt RPC to proxy to; when we are not the linearizable leader for the group we return (false, nil) and let the caller fall back to derived reads (resolveListMeta uses ScanAt/GetAt, which ARE leader-fenced / proxied per group). The fallback returns the leader's current Len — a valid serialization — at the cost of the pending.length fast-path during churn. Mirrors LeaderRoutedStore's fix for codex P1 #796.

func (*ShardStore) Compact 

func (s *ShardStore) Compact(ctx context.Context, minTS uint64) error

func (*ShardStore) DeleteAt 

func (s *ShardStore) DeleteAt(ctx context.Context, key []byte, commitTS uint64) error

func (*ShardStore) DeletePrefixAt 

func (s *ShardStore) DeletePrefixAt(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS uint64) error

DeletePrefixAt applies a prefix delete to every shard in the store.

func (*ShardStore) DeletePrefixAtRaft 

func (s *ShardStore) DeletePrefixAtRaft(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS uint64) error

DeletePrefixAtRaft is the raft-apply variant of DeletePrefixAt.

func (*ShardStore) DeletePrefixAtRaftAt 

func (s *ShardStore) DeletePrefixAtRaftAt(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS, appliedIndex uint64) error

DeletePrefixAtRaftAt is the raft-entry-index-aware variant. The caller's raft entry index applies only to the local group whose FSM is driving this apply; on a multi-group ShardStore, fanning the SAME index across other groups would corrupt their metaAppliedIndex. The single-group case (the common case for an FSM-local DeletePrefixAtRaft path) gets the correct bundling; the multi-group broadcast case is treated as "passive" — peer groups receive the prefix-delete without a meta-key bump (their own raft applies will catch up the index on the next mutation).

In practice the FSM call sites that issue raft-DeletePrefix operate against a single group's store; the multi-group ShardStore is the receiver only when an aggregate (admin / coordinator) path is replaying a global FLUSHALL, which is not raft-applied.

func (*ShardStore) ExistsAt 

func (s *ShardStore) ExistsAt(ctx context.Context, key []byte, ts uint64) (bool, error)

func (*ShardStore) ExpireAt 

func (s *ShardStore) ExpireAt(ctx context.Context, key []byte, expireAt uint64, commitTS uint64) error

func (*ShardStore) ExportVersions 

func (s *ShardStore) ExportVersions(context.Context, store.ExportVersionsOptions) (store.ExportVersionsResult, error)

func (*ShardStore) FilesystemGroupForHome 

func (s *ShardStore) FilesystemGroupForHome(homeSlot uint64, inode uint64) (uint64, bool)

FilesystemGroupForHome resolves the group that owns one file-home route.

func (*ShardStore) FilesystemGroupIDs 

func (s *ShardStore) FilesystemGroupIDs() []uint64

FilesystemGroupIDs returns every physical group that may retain filesystem chunks, including stale copies no longer owned by the current route catalog.

func (*ShardStore) GetAt 

func (s *ShardStore) GetAt(ctx context.Context, key []byte, ts uint64) ([]byte, error)

func (*ShardStore) GetAtWithReadFence 

func (s *ShardStore) GetAtWithReadFence(ctx context.Context, key []byte, ts uint64, groupID uint64, readRouteVersion uint64) ([]byte, error)

func (*ShardStore) GetGroupAt 

func (s *ShardStore) GetGroupAt(ctx context.Context, groupID uint64, key []byte, ts uint64) ([]byte, error)

GetGroupAt reads a key from the explicitly selected Raft group. It is for keyspaces whose owner is resolved outside the byte-range engine (for example SQS HT-FIFO's (queue, partition) resolver).

func (*ShardStore) GlobalCommittedTimestampFloor 

func (s *ShardStore) GlobalCommittedTimestampFloor(ctx context.Context) (uint64, error)

GlobalCommittedTimestampFloor returns a strict, leader-fenced maximum over every data group. Unlike GlobalLastCommitTS helpers used by best-effort read snapshots, this method never falls back to a stale local follower watermark: inability to reach any group's authoritative leader fails the TSO term initialization closed.

func (*ShardStore) GroupCommittedTimestampFloor 

func (s *ShardStore) GroupCommittedTimestampFloor(ctx context.Context, groupID uint64) (uint64, error)

GroupCommittedTimestampFloor returns the local group's watermark only after a ReadIndex fence proves this node is still that group's leader. It is the server-side contract for remote TSO term initialization; callers must not substitute a node-global or follower-local watermark.

func (*ShardStore) ImportVersions 

func (s *ShardStore) ImportVersions(context.Context, store.ImportVersionsOptions) (store.ImportVersionsResult, error)

func (*ShardStore) LastAppliedIndex 

func (s *ShardStore) LastAppliedIndex() (uint64, bool, error)

LastAppliedIndex aggregates the durable applied-index across every shard group, returning the MIN over all groups that report one.

MIN is the right aggregator because the kvFSM is per-shard in production — each shard's FSM independently asks "is MY group's applied index at least as fresh as MY group's snapshot?" — and ShardStore is NEVER used as the FSM's f.store in production today (the FSM holds a *pebbleStore directly; ShardStore is the coordinator-facing fanout wrapper). This method exists as a defensive forward in case a future refactor uses ShardStore from the apply path; reporting MIN guarantees the cold-start skip gate would refuse to skip whenever ANY group lags, matching the conservative "over-restore beats under-restore" rule (PR #910 design §4).

(0, false, nil) when no group reports a value — strictly-additive fallback per design §4.

func (*ShardStore) LastCommitTS 

func (s *ShardStore) LastCommitTS() uint64

func (*ShardStore) LatestCommitTS 

func (s *ShardStore) LatestCommitTS(ctx context.Context, key []byte) (uint64, bool, error)

func (*ShardStore) LatestCommitTSGroupWithReadFence 

func (s *ShardStore) LatestCommitTSGroupWithReadFence(ctx context.Context, key []byte, groupID uint64, readRouteVersion uint64) (uint64, bool, error)

func (*ShardStore) LatestCommitTSWithReadFence 

func (s *ShardStore) LatestCommitTSWithReadFence(ctx context.Context, key []byte, readRouteVersion uint64) (uint64, bool, error)

func (*ShardStore) LocalStoreForKey 

func (s *ShardStore) LocalStoreForKey(key []byte) (store.MVCCStore, bool)

LocalStoreForKey returns this process's store for the key's owning group without a leader fence or network proxy. It is reserved for node-local auxiliary state such as content-addressed S3 chunk blobs; replicated state must continue through the normal ShardStore or Coordinator paths.

func (*ShardStore) LocalStores 

func (s *ShardStore) LocalStores() []store.MVCCStore

LocalStores returns every process-local shard store in stable group order. It is used by node-local auxiliary maintenance that must recover state after snapshot restore without leader routing.

func (*ShardStore) MigrationHLCFloor 

func (s *ShardStore) MigrationHLCFloor(context.Context, uint64) (uint64, error)

func (*ShardStore) NewBackupKeyScanner 

func (s *ShardStore) NewBackupKeyScanner(start []byte, end []byte, ts uint64, pageSize int) BackupKeyScanner

func (*ShardStore) NewBackupKeyScannerAtSnapshot 

func (s *ShardStore) NewBackupKeyScannerAtSnapshot(snapshot BackupRouteSnapshot, ts uint64, pageSize int) BackupKeyScanner

func (*ShardStore) NewBackupScanner 

func (s *ShardStore) NewBackupScanner(start []byte, end []byte, ts uint64, pageSize int) BackupScanner

func (*ShardStore) NewBackupScannerAtSnapshot 

func (s *ShardStore) NewBackupScannerAtSnapshot(snapshot BackupRouteSnapshot, ts uint64, pageSize int) BackupScanner

func (*ShardStore) NewFilteredBackupScannerAtSnapshot 

func (s *ShardStore) NewFilteredBackupScannerAtSnapshot(
	snapshot BackupRouteSnapshot,
	ts uint64,
	pageSize int,
	keyFilter BackupKeyFilter,
) BackupScanner

func (*ShardStore) PutAt 

func (s *ShardStore) PutAt(ctx context.Context, key []byte, value []byte, commitTS uint64, expireAt uint64) error

func (*ShardStore) PutWithTTLAt 

func (s *ShardStore) PutWithTTLAt(ctx context.Context, key []byte, value []byte, commitTS uint64, expireAt uint64) error

func (*ShardStore) ReadFenceGroupKeysForRange 

func (s *ShardStore) ReadFenceGroupKeysForRange(start []byte, end []byte) [][]byte

ReadFenceGroupKeysForRange returns one representative routing key for each Raft group that ScanAt can visit for [start, end).

Prefer ReadFenceTargetsForRange: this form drops the group identity and is kept for callers that only need the key bytes.

func (*ShardStore) ReadFenceRouteVersion 

func (s *ShardStore) ReadFenceRouteVersion() uint64

ReadFenceRouteVersion returns the route catalog version paired with ReadFenceGroupKeysForRange so callers can discard reads whose route set changed before the scan completed.

func (*ShardStore) ReadFenceTargetsForRange 

func (s *ShardStore) ReadFenceTargetsForRange(start []byte, end []byte) []ReadFenceTarget

ReadFenceTargetsForRange returns one target per Raft group that ScanAt can visit for [start, end). It uses the same route expansion as ScanAt, so callers that take a snapshot outside ShardStore can fence every intersecting group before reading.

func (*ShardStore) ReadRouteVersion 

func (s *ShardStore) ReadRouteVersion() uint64

func (*ShardStore) ResolveFilesystemHomeSlot 

func (s *ShardStore) ResolveFilesystemHomeSlot(targetGroup uint64, inode uint64) (uint64, error)

ResolveFilesystemHomeSlot finds a home token whose file route belongs to targetGroup. It derives candidates from current route boundaries and verifies each candidate against the live catalog before returning it.

func (*ShardStore) Restore 

func (s *ShardStore) Restore(_ io.Reader) error

func (*ShardStore) RetireMigration 

func (s *ShardStore) RetireMigration(context.Context, uint64) error

func (*ShardStore) ReverseScanAt 

func (s *ShardStore) ReverseScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error)

func (*ShardStore) ReverseScanAtPhysicalLimit 

func (s *ShardStore) ReverseScanAtPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64) ([]*store.KVPair, bool, error)

func (*ShardStore) ReverseScanGroupAt 

func (s *ShardStore) ReverseScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error)

ReverseScanGroupAt reverse-scans a range on the explicitly selected Raft group.

func (*ShardStore) ScanAt 

func (s *ShardStore) ScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error)

ScanAt scans keys across shards at the given timestamp. When the caller has already fenced every group to applied_index >= f(ts), as BeginBackup does, the result is consistent across groups. Without that fence, ranges spanning multiple shards are best-effort because each shard may have a different Raft apply position.

func (*ShardStore) ScanAtPhysicalLimit 

func (s *ShardStore) ScanAtPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64) ([]*store.KVPair, bool, error)

func (*ShardStore) ScanAtWithReadFence 

func (s *ShardStore) ScanAtWithReadFence(ctx context.Context, start []byte, end []byte, limit int, ts uint64, reverse bool, groupID uint64, readRouteVersion uint64, routeStart []byte, routeEnd []byte) ([]*store.KVPair, error)

func (*ShardStore) ScanGroupAt 

func (s *ShardStore) ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error)

ScanGroupAt scans a range on the explicitly selected Raft group. It is for keyspaces whose owner is resolved outside the byte-range engine (for example SQS HT-FIFO's (queue, partition) resolver). Normal callers should use ScanAt so range scans keep following the distribution engine's route table.

func (*ShardStore) ScanGroupKeysAt 

func (s *ShardStore) ScanGroupKeysAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([][]byte, error)

ScanGroupKeysAt scans keys on the explicitly selected Raft group without materializing values over proxy links.

func (*ShardStore) ScanKeysAt 

func (s *ShardStore) ScanKeysAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([][]byte, error)

func (*ShardStore) ScanKeysAtWithReadFence 

func (s *ShardStore) ScanKeysAtWithReadFence(ctx context.Context, start []byte, end []byte, limit int, ts uint64, groupID uint64, readRouteVersion uint64) ([][]byte, error)

func (*ShardStore) SetDurableAppliedIndex 

func (s *ShardStore) SetDurableAppliedIndex(idx uint64) error

SetDurableAppliedIndex broadcasts the bump to every group store that exposes the writer seam.

This is purely defensive — in production today the FSM holds a *pebbleStore directly; ShardStore is never f.store. Were it ever wired through the FSM apply path, broadcasting the same idx across groups would corrupt their per-group metaAppliedIndex semantics (each group has its own raft log with its own entry numbering). For that hypothetical, the test convention from DeletePrefixAtRaftAt applies: tests MUST pass idx=0 to opt out, or not use ShardStore as the writer at all. Returns the first per-group error.

func (*ShardStore) Snapshot 

func (s *ShardStore) Snapshot() (store.Snapshot, error)

func (*ShardStore) ValidateBackupSnapshotAt 

func (s *ShardStore) ValidateBackupSnapshotAt(ctx context.Context, snapshot BackupRouteSnapshot, ts uint64, pageSize int) error

ValidateBackupSnapshotAt resolves committed or rolled-back transaction locks and fails closed while any prepared transaction remains pending at the backup cut. The scan covers lock-only inserts that have no visible user key.

func (*ShardStore) VersionExistsAtOrBeforeGroupWithReadFence 

func (s *ShardStore) VersionExistsAtOrBeforeGroupWithReadFence(ctx context.Context, key []byte, groupID uint64, ts uint64, readRouteVersion uint64) (bool, bool, error)

VersionExistsAtOrBeforeGroupWithReadFence serves the remote half of routeHasVersionAtOrBeforeRemote. The second bool reports whether this node could answer authoritatively; a replica that is no longer the group leader says no rather than guessing.

func (*ShardStore) WithPartitionResolver 

func (s *ShardStore) WithPartitionResolver(r PartitionResolver) *ShardStore

WithPartitionResolver installs the same partition-keyspace resolver used by ShardedCoordinator. ShardStore keeps normal byte-range routing for ordinary calls, but backup scanners need the resolver to decide which physical group owns partition-routed keys discovered while scanning every group.

func (*ShardStore) WriteConflictCountsByPrefix 

func (s *ShardStore) WriteConflictCountsByPrefix() map[string]uint64

WriteConflictCountsByPrefix aggregates OCC conflict counts across every shard group owned by this ShardStore. Per-shard counts share the same "<kind>|<key_prefix>" label schema, so a simple sum gives the node-wide view. The result is always non-nil.

type ShardedCoordinator 

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

ShardedCoordinator routes operations to shard-specific raft groups. It issues timestamps via a shared HLC and uses ShardRouter to dispatch.

func NewShardedCoordinator 

func NewShardedCoordinator(engine *distribution.Engine, groups map[uint64]*ShardGroup, defaultGroup uint64, clock *HLC, st store.MVCCStore) *ShardedCoordinator

NewShardedCoordinator builds a coordinator for the provided shard groups. The defaultGroup is used for non-keyed leader checks.

func (*ShardedCoordinator) Clock 

func (c *ShardedCoordinator) Clock() *HLC

func (*ShardedCoordinator) Close 

func (c *ShardedCoordinator) Close() error

Close releases per-shard engine-side registrations. Idempotent.

func (*ShardedCoordinator) Dispatch 

func (c *ShardedCoordinator) Dispatch(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, error)

func (*ShardedCoordinator) EngineGroupIDForKey 

func (c *ShardedCoordinator) EngineGroupIDForKey(key []byte) uint64

EngineGroupIDForKey reports the Raft group ID that owns key, or 0 when the key cannot be routed. Callers that batch lease checks across many keys use it to collapse keys sharing a group into a single lease read (see GroupRoutableCoordinator). It performs no I/O — only an in-memory router lookup — so it is safe on the read hot path.

func (*ShardedCoordinator) EnsureMutationsWriteAllowed 

func (c *ShardedCoordinator) EnsureMutationsWriteAllowed(muts []*pb.Mutation, commitTS uint64) error

EnsureMutationsWriteAllowed exposes the route-floor check to the leader-side Internal.Forward handler. It is the same predicate the local stamping path applies, so a forwarded write cannot reach Raft under a floor that a locally-stamped write would have been rejected by.

func (*ShardedCoordinator) GroupLeadership 

func (c *ShardedCoordinator) GroupLeadership(groupID uint64) (bool, uint64)

GroupLeadership reports local leadership and Raft term for groupID.

func (*ShardedCoordinator) IsLeader 

func (c *ShardedCoordinator) IsLeader() bool

func (*ShardedCoordinator) IsLeaderForGroup 

func (c *ShardedCoordinator) IsLeaderForGroup(groupID uint64) bool

IsLeaderForGroup reports local leadership for an already-resolved group, skipping the key -> route -> group derivation IsLeaderForKey performs. Read fencing needs this: a fence target's group is known when the target is built, and re-deriving it from the representative key can land on a different group (see ReadFenceTarget).

func (*ShardedCoordinator) IsLeaderForKey 

func (c *ShardedCoordinator) IsLeaderForKey(key []byte) bool

func (*ShardedCoordinator) IsTimestampLeader 

func (c *ShardedCoordinator) IsTimestampLeader() bool

func (*ShardedCoordinator) LeadershipForKey 

func (c *ShardedCoordinator) LeadershipForKey(key []byte) (bool, uint64)

LeadershipForKey reports local leadership and Raft term for key's group.

func (*ShardedCoordinator) LeaseRead 

func (c *ShardedCoordinator) LeaseRead(ctx context.Context) (uint64, error)

LeaseRead routes through the default group's lease. See Coordinate.LeaseRead for semantics.

func (*ShardedCoordinator) LeaseReadAllGroups 

func (c *ShardedCoordinator) LeaseReadAllGroups(ctx context.Context) error

LeaseReadAllGroups establishes the lease freshness bound on every configured all-shard data group. Multi-shard reads (Scan, GSI/whole-table Query) visit all intersecting data routes across all groups (see ShardStore.ScanAt), so fencing only the default group would let those reads sample a snapshot on a non-default group without the freshness bound. It fails closed on the first group that cannot confirm its lease, since a partially-fenced read is exactly the stale read this guards against. Group iteration order is deterministic but correctness does not depend on it because every configured group must succeed.

func (*ShardedCoordinator) LeaseReadAllGroupsTimestamp 

func (c *ShardedCoordinator) LeaseReadAllGroupsTimestamp(ctx context.Context) (uint64, error)

LeaseReadAllGroupsTimestamp fences every data-group leader and returns the maximum commit timestamp observed after those barriers.

func (*ShardedCoordinator) LeaseReadForGroup 

func (c *ShardedCoordinator) LeaseReadForGroup(ctx context.Context, groupID uint64) (uint64, error)

LeaseReadForGroup establishes the lease freshness bound on an already-resolved group. Read fencing knows the group when it builds a ReadFenceTarget, and re-deriving it from the representative key can select a different group, so the fence path uses this instead of LeaseReadForKey.

func (*ShardedCoordinator) LeaseReadForKey 

func (c *ShardedCoordinator) LeaseReadForKey(ctx context.Context, key []byte) (uint64, error)

LeaseReadForKey performs the lease check on the shard group that owns key. Each group maintains its own lease since each group has independent leadership and term.

func (*ShardedCoordinator) LinearizableRead 

func (c *ShardedCoordinator) LinearizableRead(ctx context.Context) (uint64, error)

func (*ShardedCoordinator) LinearizableReadForKey 

func (c *ShardedCoordinator) LinearizableReadForKey(ctx context.Context, key []byte) (uint64, error)

func (*ShardedCoordinator) LocalLeaderGroupIDs 

func (c *ShardedCoordinator) LocalLeaderGroupIDs() []uint64

LocalLeaderGroupIDs returns the configured data shard groups this process currently leads. Background maintenance uses this to avoid issuing whole-keyspace scans from every node and proxying those scans back to the same hot leaders.

func (*ShardedCoordinator) Next 

func (c *ShardedCoordinator) Next(ctx context.Context) (uint64, error)

Next makes ShardedCoordinator usable as a TimestampAllocator for adapter helpers that need to allocate persistence-grade timestamps outside Dispatch.

func (*ShardedCoordinator) NextAfter 

func (c *ShardedCoordinator) NextAfter(ctx context.Context, min uint64) (uint64, error)

func (*ShardedCoordinator) ObserveForwardedRequests 

func (c *ShardedCoordinator) ObserveForwardedRequests(reqs []*pb.Request)

ObserveForwardedRequests records committed leader-side sampling evidence for writes that entered through a shard follower and were committed via Internal.Forward on this shard leader.

func (*ShardedCoordinator) ObserveTimestampFloor 

func (c *ShardedCoordinator) ObserveTimestampFloor(ts uint64)

ObserveTimestampFloor advances the process clock past a replicated backup cut and invalidates any cached TSO batch that could still contain values at or below it. Claims returned before invalidation remain protected by the FSM's durable backup timestamp floor.

func (*ShardedCoordinator) ProposeHLCLease 

func (c *ShardedCoordinator) ProposeHLCLease(ctx context.Context, ceilingMs int64) error

func (*ShardedCoordinator) RaftLeader 

func (c *ShardedCoordinator) RaftLeader() string

func (*ShardedCoordinator) RaftLeaderForGroup 

func (c *ShardedCoordinator) RaftLeaderForGroup(groupID uint64) string

RaftLeaderForGroup returns the Raft leader address for an already-resolved group, the group-keyed counterpart of RaftLeaderForKey.

func (*ShardedCoordinator) RaftLeaderForKey 

func (c *ShardedCoordinator) RaftLeaderForKey(key []byte) string

func (*ShardedCoordinator) RaftMembers 

func (c *ShardedCoordinator) RaftMembers(ctx context.Context) ([]RaftMember, error)

RaftMembers returns every configured Raft endpoint across all local groups. Endpoints, rather than node IDs, are deduplicated because one process can expose a distinct listener per group. The stable sort keeps capability-cache fingerprints deterministic despite map iteration order.

func (*ShardedCoordinator) RaftMembersForKey 

func (c *ShardedCoordinator) RaftMembersForKey(ctx context.Context, key []byte) ([]RaftMember, error)

RaftMembersForKey returns a stable value copy of the current configuration for the key's owning group. It is intentionally membership-only: callers do not gain access to the mutable ShardGroup or its local store.

func (*ShardedCoordinator) RecoverHLCLease 

func (c *ShardedCoordinator) RecoverHLCLease(ctx context.Context) error

renewHLCLease proposes a fresh physical ceiling on one shard group and, on a successful (quorum-acked) propose, warms that group's read lease.

The lease-extension base (start) and the invalidation generation are sampled BEFORE the propose so the warm-up mirrors leaseRefreshingTxn's success branch exactly: the window can only be SHORTER than the true safety window, and a leader-loss callback that fires during the propose advances the generation so extend refuses to resurrect a stale lease. The propose is the SAME quorum confirmation a client write goes through, so warming on its success cannot widen the lease-read freshness window beyond what a write on this group would. This is the background warm-up that flattens the read-only lease-expiry sawtooth for the default group on idle-write workloads; the lease window/duration semantics are unchanged.

On a leadership-loss propose error the group lease is invalidated eagerly, mirroring leaseRefreshingTxn's error branch exactly: when Propose returns the loss before the async RegisterLeaderLossCallback fires, a stale-warm lease must not survive on a non-leader node for the callback latency window. Non-leadership errors (no quorum, validation) are NOT leadership signals and must not tear down a warm lease -- doing so would force every read onto the slow path.

func (*ShardedCoordinator) RevokeAppliedReadTimestamp 

func (c *ShardedCoordinator) RevokeAppliedReadTimestamp(timestamp uint64, ref AppliedReadTimestampVoucherRef)

RevokeAppliedReadTimestamp removes one prepared voucher that did not reach ShardedCoordinator dispatch validation, for example because an outer coordinator decorator rejected the dispatch first.

func (*ShardedCoordinator) RunHLCLeaseRenewal 

func (c *ShardedCoordinator) RunHLCLeaseRenewal(ctx context.Context)

RunHLCLeaseRenewal periodically proposes a new physical ceiling to every shard group's Raft cluster while this node is that group's leader. This mirrors the single-shard Coordinate.RunHLCLeaseRenewal behaviour while avoiding the multi-shard gap where a non-default group leader could issue timestamps without renewing the shared HLC's Raft-backed ceiling.

RunHLCLeaseRenewal blocks until ctx is cancelled; call it in a goroutine.

func (*ShardedCoordinator) SetHLCLeaseRenewalBlocker 

func (c *ShardedCoordinator) SetHLCLeaseRenewalBlocker(blocked func() bool)

SetHLCLeaseRenewalBlocker installs a predicate that suppresses background HLC lease-renewal proposals while it returns true.

func (*ShardedCoordinator) SupportsAppliedReadTimestampVoucher 

func (c *ShardedCoordinator) SupportsAppliedReadTimestampVoucher() bool

func (*ShardedCoordinator) TimestampAllocator 

func (c *ShardedCoordinator) TimestampAllocator() TimestampAllocator

TimestampAllocator exposes the configured allocator to coordinator decorators without widening the Coordinator interface.

func (*ShardedCoordinator) VerifyLeader 

func (c *ShardedCoordinator) VerifyLeader(ctx context.Context) error

func (*ShardedCoordinator) VerifyLeaderForKey 

func (c *ShardedCoordinator) VerifyLeaderForKey(ctx context.Context, key []byte) error

func (*ShardedCoordinator) VouchAppliedReadTimestamp 

func (c *ShardedCoordinator) VouchAppliedReadTimestamp(timestamp uint64, ref AppliedReadTimestampVoucherRef) error

VouchAppliedReadTimestamp records one use of an audited adapter watermark. The bounded map prevents abandoned requests from growing process memory.

func (*ShardedCoordinator) WithAllShardGroups 

func (c *ShardedCoordinator) WithAllShardGroups(groupIDs ...uint64) *ShardedCoordinator

WithAllShardGroups restricts whole-keyspace operations to the supplied data groups. When unset, the coordinator preserves the legacy behaviour and uses every group it owns.

func (*ShardedCoordinator) WithKeyVizLabelsEnabled 

func (c *ShardedCoordinator) WithKeyVizLabelsEnabled(enabled bool) *ShardedCoordinator

func (*ShardedCoordinator) WithLeaseReadObserver 

func (c *ShardedCoordinator) WithLeaseReadObserver(observer LeaseReadObserver) *ShardedCoordinator

WithLeaseReadObserver wires a LeaseReadObserver onto a ShardedCoordinator. Applied after construction because the NewShardedCoordinator signature is already heavily overloaded; see Coordinate.WithLeaseReadObserver for the equivalent option on the single-group coordinator, including the typed-nil guard rationale.

func (*ShardedCoordinator) WithPartitionResolver 

func (c *ShardedCoordinator) WithPartitionResolver(r PartitionResolver) *ShardedCoordinator

WithPartitionResolver wires a PartitionResolver onto the coordinator's underlying ShardRouter. The resolver runs before the byte-range engine on every dispatch, so partition-keyspace schemes (e.g. SQS HT-FIFO) can override the default shard layout without breaking the engine's non-overlapping-cover invariant.

Applied after construction for the same reason as the other With* options on this type — NewShardedCoordinator is already heavily overloaded. Passing a nil resolver clears any previously- installed resolver.

func (*ShardedCoordinator) WithRegistrationGate 

func (c *ShardedCoordinator) WithRegistrationGate(g *RegistrationGate) *ShardedCoordinator

WithRegistrationGate wires the Stage 7a first-write barrier onto the coordinator. Applied after construction for the same reason as the other With* options. A nil gate (or a gate with a nil Barrier) leaves every write ungated — the encryption-off / no-pending- registration posture.

func (*ShardedCoordinator) WithSampler 

func (c *ShardedCoordinator) WithSampler(s keyviz.Sampler) *ShardedCoordinator

WithSampler wires a keyviz.Sampler onto a ShardedCoordinator. The coordinator calls sampler.Observe at dispatch entry — once per resolved (RouteID, mutation key) pair — to feed the key visualizer heatmap (design doc §5.1). Applied after construction for the same reason as WithLeaseReadObserver: NewShardedCoordinator is already heavily overloaded.

Passing a nil interface value is supported and disables sampling (the call site guards against it). Passing a typed-nil *keyviz.MemSampler also works because Observe is nil-safe by contract.

func (*ShardedCoordinator) WithTSOAllocator 

func (c *ShardedCoordinator) WithTSOAllocator(alloc TimestampAllocator) *ShardedCoordinator

WithTSOAllocator routes sharded-coordinator timestamp issuance through a TSO-compatible allocator. Existing deployments keep the legacy shared-HLC path unless this is wired explicitly.

func (*ShardedCoordinator) WithTSOCutoverState 

func (c *ShardedCoordinator) WithTSOCutoverState(state interface {
	CutoverActive() bool
	PhaseDActive() bool
}) *ShardedCoordinator

WithTSOCutoverState wires the durable group-0 migration state. The state is read dynamically so a marker applied after startup changes renewal and validation behavior without a process-local mode race.

func (*ShardedCoordinator) WithTimestampGroup 

func (c *ShardedCoordinator) WithTimestampGroup(groupID uint64) *ShardedCoordinator

WithTimestampGroup pins timestamp issuance leadership to one Raft group. Callers should only enable this once a data-shard leader can redirect timestamp allocation to that group; otherwise data leaders would stop being able to commit writes when they do not also lead the timestamp group.

func (*ShardedCoordinator) WithTimestampGroupCandidate 

func (c *ShardedCoordinator) WithTimestampGroupCandidate(groupID uint64) *ShardedCoordinator

WithTimestampGroupCandidate records the group that owns timestamp issuance once the dedicated allocator takes over, without pinning it yet.

A deployment can start in legacy warm-up and advance to Phase D later, either by reloading --tsoModeFile or by another node proposing the marker. The warm-up contract forbids pinning up front: while persistence timestamps still come from the data groups' local HLC path, narrowing lease renewal to the timestamp group would let a group-0 quorum loss expire healthy data groups' ceilings. But leaving the group unknown is worse once Phase D activates -- shouldRenewHLCGroup then retires every data group and finds no timestamp group to renew instead, so every ceiling expires and issuance fails with ErrCeilingExpired. Recording the candidate here lets effectiveTimestampGroup pin it exactly when PhaseDActive flips, with no runtime mutation of the coordinator.

type TSOAllocator 

type TSOAllocator interface {
	Next(ctx context.Context) (uint64, error)
	NextBatch(ctx context.Context, n int) (uint64, error)
	IsLeader() bool
	RunLeaseRenewal(ctx context.Context)
}

TSOAllocator issues globally monotonic timestamps. NextBatch returns the first timestamp in a consecutive window [base, base+n-1].

type TSOCutoverFloorProvider 

type TSOCutoverFloorProvider interface {
	GlobalCommittedTimestampFloor(context.Context) (uint64, error)
}

TSOCutoverFloorProvider supplies the highest commit timestamp that the dedicated TSO must exceed before a leader term can issue a window.

type TSODurableStateObserver 

type TSODurableStateObserver interface {
	ObserveTSODurableState(cutoverActive, phaseDActive bool)
}

TSODurableStateObserver is the narrow slice of TSOObserver the state machine needs. Keeping it separate lets the FSM publish without depending on the allocation-latency surface.

type TSOMode 

type TSOMode uint32

TSOMode is the process-local timestamp issuance mode. Its ordering is part of the migration contract: runtime reload may only move to the next value. 

const (
	TSOModeLegacy TSOMode = iota
	TSOModeShadow
	TSOModeCutover
	TSOModePhaseD
)

func ParseTSOMode 

func ParseTSOMode(raw string) (TSOMode, error)

func ReadTSOModeFile 

func ReadTSOModeFile(path string) (TSOMode, error)

func (TSOMode) String 

func (m TSOMode) String() string

type TSOObserver 

type TSOObserver interface {
	ObserveTSORequest(operation, path, outcome string, duration time.Duration)
	ObserveTSOShadowComparison(result string, divergence uint64)
	ObserveTSOMode(mode string)
	ObserveTSOModeReload(result string)
	ObserveTSODurableState(cutoverActive, phaseDActive bool)
}

TSOObserver is the bounded-cardinality operational surface for production allocation latency, shadow divergence, durable state, and mode reloads.

type TSOPhaseDState 

type TSOPhaseDState interface {
	PhaseDActive() bool
	PhaseDRequired() bool
}

TSOPhaseDState exposes the one-way Phase-D state to coordinators and adapter migration helpers without coupling them to TSOStateMachine.

type TSOReservation 

type TSOReservation struct {
	Base                    uint64
	Count                   int
	PreviousAllocationFloor uint64
	CutoverActive           bool
	PhaseDActive            bool
	PhaseDFloor             uint64
}

TSOReservation describes one group-0-serialized timestamp window. PreviousAllocationFloor is sampled before this request's reservation and is used by shadow migration to reject overlapping legacy candidates.

type TSOReservationAllocator 

type TSOReservationAllocator interface {
	ReserveBatchAfter(context.Context, int, uint64, bool, bool) (TSOReservation, error)
}

TSOReservationAllocator is the migration-aware extension exposed by the dedicated group leader. Ordinary callers continue to use TSOAllocator.

type TSORuntimeController 

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

TSORuntimeController owns the process-local mode while treating group-0's durable markers as authoritative. InitialMode may start at any phase for a restart, but ApplyMode requires adjacent, one-way runtime transitions.

func NewTSORuntimeController 

func NewTSORuntimeController(cfg TSORuntimeControllerConfig) (*TSORuntimeController, error)

func (*TSORuntimeController) Allocator 

func (c *TSORuntimeController) Allocator() *DynamicTimestampAllocator

func (*TSORuntimeController) ApplyMode 

func (c *TSORuntimeController) ApplyMode(requested TSOMode) error

func (*TSORuntimeController) CurrentMode 

func (c *TSORuntimeController) CurrentMode() TSOMode

func (*TSORuntimeController) EffectiveMode 

func (c *TSORuntimeController) EffectiveMode(requested TSOMode) TSOMode

type TSORuntimeControllerConfig 

type TSORuntimeControllerConfig struct {
	Clock       *HLC
	Routed      *LeaderRoutedTSOAllocator
	State       TSORuntimeState
	BatchSize   int
	InitialMode TSOMode
	Logger      *slog.Logger
	Observer    TSOObserver
}

type TSORuntimeState 

type TSORuntimeState interface {
	CutoverActive() bool
	PhaseDActive() bool
}

TSORuntimeState is the consensus-owned one-way migration state.

type TSOShadowReservationAllocator 

type TSOShadowReservationAllocator interface {
	ValidateShadowTimestamp(context.Context, uint64) (TSOReservation, error)
}

type TSOStateMachine 

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

TSOStateMachine is the minimal state machine for the dedicated timestamp group. It accepts HLC lease-renewal entries plus explicit allocation-floor entries. The HLC is only a volatile mirror; snapshots are sourced from the TSO FSM's own applied state so unrelated shard-group lease renewals cannot advance group-0 state outside the group-0 log.

func NewTSOStateMachine 

func NewTSOStateMachine(hlc *HLC, opts ...TSOStateMachineOption) *TSOStateMachine

NewTSOStateMachine constructs the dedicated TSO FSM over the shared HLC.

func (*TSOStateMachine) AllocationFloor 

func (f *TSOStateMachine) AllocationFloor() uint64

AllocationFloor returns the highest timestamp window end applied by the dedicated TSO group. It is consensus-owned state, unlike HLC.Current().

func (*TSOStateMachine) Apply 

func (f *TSOStateMachine) Apply(data []byte) any

func (*TSOStateMachine) CutoverActive 

func (f *TSOStateMachine) CutoverActive() bool

CutoverActive reports whether production issuance has durably crossed the one-way migration marker. The marker cannot be cleared without a separate cluster-wide rollback protocol.

func (*TSOStateMachine) IsVolatileOnlyPayload 

func (f *TSOStateMachine) IsVolatileOnlyPayload(payload []byte) bool

func (*TSOStateMachine) PhaseDActive 

func (f *TSOStateMachine) PhaseDActive() bool

PhaseDActive reports whether the compatibility window has been durably closed. Once active, data-shard HLC renewal and caller-supplied cross-shard timestamps may no longer use legacy issuance semantics.

func (*TSOStateMachine) PhaseDFloor 

func (f *TSOStateMachine) PhaseDFloor() uint64

PhaseDFloor is the highest allocation floor that existed when Phase D was activated. Only timestamps reserved strictly above it are valid M7 durable read/start allocations.

func (*TSOStateMachine) Restore 

func (f *TSOStateMachine) Restore(r io.Reader) error

func (*TSOStateMachine) SetDurableStateObserver 

func (f *TSOStateMachine) SetDurableStateObserver(observer TSODurableStateObserver)

SetDurableStateObserver installs the observer after construction. Wiring happens where the metrics registry is in scope rather than at the group builder, which several test harnesses construct without one.

func (*TSOStateMachine) Snapshot 

func (f *TSOStateMachine) Snapshot() (raftengine.Snapshot, error)

type TSOStateMachineOption 

type TSOStateMachineOption func(*TSOStateMachine)

TSOStateMachineOption configures optional state-machine wiring.

func WithTSODurableStateObserver 

func WithTSODurableStateObserver(observer TSODurableStateObserver) TSOStateMachineOption

WithTSODurableStateObserver publishes cutover / phase-D gauges whenever the markers are applied or restored.

type TimestampAfterAllocator 

type TimestampAfterAllocator interface {
	NextAfter(ctx context.Context, min uint64) (uint64, error)
}

type TimestampAllocator 

type TimestampAllocator interface {
	Next(ctx context.Context) (uint64, error)
}

TimestampAllocator is the minimal timestamp source the coordinators need. TSOAllocator and BatchAllocator both satisfy it; keeping this interface narrow lets production use a batched TSO while tests can inject a tiny fake.

func ConfiguredTimestampAllocatorThrough 

func ConfiguredTimestampAllocatorThrough(coord Coordinator) (TimestampAllocator, bool)

func TimestampAllocatorThrough 

func TimestampAllocatorThrough(coord Coordinator) (TimestampAllocator, bool)

TimestampAllocatorThrough returns the currently active allocator behind a coordinator or coordinator decorator. A runtime allocator in legacy mode is intentionally hidden so normal write paths use the coordinator HLC fallback.

type TimestampAllocatorProvider 

type TimestampAllocatorProvider interface {
	TimestampAllocator() TimestampAllocator
}

TimestampAllocatorProvider lets coordinator decorators preserve access to the configured allocator without making the decorator itself an allocator.

type TransactionManager 

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

func NewTransactionWithProposer 

func NewTransactionWithProposer(proposer raftengine.Proposer, opts ...TransactionOption) *TransactionManager

func (*TransactionManager) Abort 

func (t *TransactionManager) Abort(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)

func (*TransactionManager) Close 

func (t *TransactionManager) Close()

Close signals the TransactionManager to stop and drains any pending raw commit items, sending each an error so callers are not blocked forever.

func (*TransactionManager) Commit 

func (t *TransactionManager) Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)

type TransactionOption 

type TransactionOption func(*TransactionManager)

func WithProposalObserver 

func WithProposalObserver(observer ProposalObserver) TransactionOption

WithProposalObserver records raft.Apply failures without coupling kv to a concrete monitoring backend.

type TransactionResponse 

type TransactionResponse struct {
	CommitIndex uint64
}

type Transactional 

type Transactional interface {
	Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)
	Abort(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error)
}

Transactional is the kv-internal interface that fronts the raft propose path. Implementations (TransactionManager, LeaderProxy, ShardRouter, leaseRefreshingTxn) thread the caller's context end-to-end so a Redis / gRPC / S3 / SQS handler's deadline reaches Propose / VerifyLeader without being silently dropped to context.Background. See PR #748 / design doc 2026_05_10_implemented_kv_ctx_plumbing.md for the rationale; the prior signatures lived behind `verifyLeaderEngine`'s 5 s safety bound (#745), which is preserved as the no-ctx defense-in-depth fallback.

type TxnLockDrainEntry 

type TxnLockDrainEntry struct {
	LockKey      []byte
	UserKey      []byte
	StartTS      uint64
	TTLExpireAt  uint64
	PrimaryKey   []byte
	IsPrimaryKey bool
}

TxnLockDrainEntry describes one prepared transaction lock that still belongs to a migration route. The lock key is cloned so callers can safely retain it across drain ticks.

func PendingTxnLocksInRoute 

func PendingTxnLocksInRoute(ctx context.Context, st store.MVCCStore, routeStart, routeEnd []byte, ts uint64, limit int) ([]TxnLockDrainEntry, error)

PendingTxnLocksInRoute scans the txn-lock namespace and filters each lock by routeKey(userKey). It intentionally does not bracket the scan by txnLockKey(routeStart)/txnLockKey(routeEnd): txn locks are sorted by raw user key while migration routes are defined in route-key space.

type TxnLockedError 

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

func (*TxnLockedError) Error 

func (e *TxnLockedError) Error() string

func (*TxnLockedError) Unwrap 

func (e *TxnLockedError) Unwrap() error

type TxnMeta 

type TxnMeta struct {
	PrimaryKey   []byte
	LockTTLms    uint64
	CommitTS     uint64
	PrevCommitTS uint64
}

TxnMeta is embedded into transactional raft log requests via a synthetic mutation (key prefix "!txn|meta|"). It is not persisted in the MVCC store.

PrevCommitTS is the commit timestamp of a failed previous attempt of the same single-shard transaction. It is set only on a retry, and only carries the one-phase idempotency dedup probe (option 2): at apply, the FSM checks whether the previous attempt's write set already landed at exactly this timestamp and, if so, no-ops the apply instead of re-applying. Because it only needs the V2 wire format, EncodeTxnMeta keeps emitting V1 whenever PrevCommitTS is zero (every non-retry path), so the default wire format is unchanged. See docs/design/2026_05_21_proposed_txn_secondary_idempotency.md.

func DecodeTxnMeta 

func DecodeTxnMeta(b []byte) (TxnMeta, error)

Source Files

View all Source files

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.