Documentation
¶
Overview ¶
Package storagereplacement owns the operator-approved provider replacement record: its state machine, wait reasons, stable API codes, and task payload contract. It holds no database or provider dependencies.
Index ¶
- Constants
- Variables
- func AbandonedTargetTaskKey(replacementID int64) string
- func Allowed(from, to Status) bool
- func Code(err error) string
- func IsCoordinatorTask(taskType model.TaskType, stage *string) bool
- func MigrateTaskKey(replacementID int64) string
- func NewAbandonedTargetTask(replacementID, bucketID int64, maxRetries int, scheduledAt time.Time) *model.Task
- func NewMigratePayload(replacementID int64) map[string]any
- func NewMigrateTask(replacementID, bucketID int64, versionID string, maxRetries int, ...) *model.Task
- func NewRetirePayload(replacementID int64) map[string]any
- func NewRetireTask(replacementID, bucketID int64, maxRetries int, scheduledAt time.Time) *model.Task
- func ParseRetirePayload(task *model.Task) (int64, error)
- func RetireTaskKey(replacementID int64) string
- type ClaimToken
- type ExecutionSnapshot
- type FailureReason
- type Item
- type ItemStatus
- type MigratePayload
- type Phase
- type ProgressSnapshot
- type Replacement
- type SelectionMode
- type Status
- type WaitReason
Constants ¶
const ( CodeActive = "replacement_active" CodeSuperseded = "replacement_superseded" CodeNotRetryable = "replacement_not_retryable" CodeTaskRunning = "replacement_task_running" CodeTargetInvalid = "replacement_target_invalid" CodeTargetInUse = "replacement_target_in_use" CodeNoEligibleProvider = "replacement_no_eligible_provider" CodeIdempotencyConflict = "replacement_idempotency_conflict" CodeSourceNotCurrent = "replacement_source_not_current" // CodeTaskRetryUnsupported is returned by the generic exhausted-task retry // endpoint when the task belongs to a replacement. CodeTaskRetryUnsupported = "replacement_task_retry_unsupported" )
Stable machine-readable codes returned alongside API errors. Clients branch on these, so treat them as part of the public contract.
const ( // StageMigrate advances the replacement control plane through the uploader; // item transfers run on the dedicated replacement worker. StageMigrate = "replace_provider" // StageRetire runs the retirement safety gate through the storage cleanup // worker. StageRetire = "retire_data_set" // StageRetireAbandonedTarget ends the service of a target that a later // confirmation replaced before it ever took over the slot. StageRetireAbandonedTarget = "retire_abandoned_target" )
Variables ¶
var ( // ErrActiveReplacement means the source data set already has a replacement // that has not reached a terminal state. ErrActiveReplacement = errors.New("data set already has an active replacement") // ErrSuperseded means a later confirmation took ownership of this work. ErrSuperseded = errors.New("replacement has been superseded") // ErrNotRetryable means the replacement is progressing or finished, so the // dedicated retry action does not apply. ErrNotRetryable = errors.New("replacement is not in a retryable state") // ErrTaskRunning means a coordinator task still holds this replacement. ErrTaskRunning = errors.New("replacement task is still running") // ErrInvalidTarget means the requested provider cannot serve as the target, // for example because it is the source itself. ErrInvalidTarget = errors.New("requested provider cannot replace this data set") // ErrTargetInUse means the requested provider already owns another current // data set in the same bucket. ErrTargetInUse = errors.New("requested provider already serves this bucket") // ErrNoEligibleProvider means automatic selection found no provider the // bucket has not already used. ErrNoEligibleProvider = errors.New("no eligible replacement provider is available") // active PDP-capable replacement candidate. ErrTargetUnavailable = errors.New("requested provider is not available for replacement") // ErrIdempotencyConflict means a client request id was reused with different // replacement parameters. ErrIdempotencyConflict = errors.New("replacement idempotency key conflicts with an earlier request") // ErrSourceNotCurrent means the data set no longer owns its replica slot, // so replacing it would not change where writes go. ErrSourceNotCurrent = errors.New("data set is not the current replica") // ErrItemCancelled means this migration item no longer has executable work. ErrItemCancelled = errors.New("replacement item is no longer executable") // ErrItemDeferred means the item cannot run yet but must be revisited. The // coordinator moves on to other content rather than blocking on it. ErrItemDeferred = errors.New("replacement item is waiting for its source") // ErrPrematureComplete means a retirement safety gate still reports a // blocker. Repositories return it even when called outside the worker. ErrPrematureComplete = errors.New("replacement cannot complete while a safety gate blocks it") // ErrIllegalTransition means the requested state change is not in the // state machine. ErrIllegalTransition = errors.New("illegal replacement state transition") )
Functions ¶
func AbandonedTargetTaskKey ¶
AbandonedTargetTaskKey identifies the cleanup of a target a later confirmation abandoned. It is a separate coordinator because it retires the opposite generation and answers a different safety question.
func Allowed ¶
Allowed reports whether the state machine permits this transition. A status never transitions to itself; callers update fields in place instead.
func Code ¶
Code maps a replacement error to its stable API code. It returns an empty string for errors that carry no client-facing code.
func IsCoordinatorTask ¶
IsCoordinatorTask reports whether a task belongs to a replacement. Generic exhausted-task retry uses it to refuse work that must resume through the dedicated replacement action.
func MigrateTaskKey ¶
MigrateTaskKey identifies the single migration coordinator for one replacement. Item concurrency is owned by the durable replacement queue, not by this control-plane task.
func NewAbandonedTargetTask ¶
func NewAbandonedTargetTask(replacementID, bucketID int64, maxRetries int, scheduledAt time.Time) *model.Task
NewAbandonedTargetTask builds the coordinator that ends the paid service of a target no confirmation uses any more. Without it the abandoned service keeps costing money after a later confirmation takes over.
func NewMigratePayload ¶
NewMigratePayload builds the coordinator payload.
func NewMigrateTask ¶
func NewMigrateTask(replacementID, bucketID int64, versionID string, maxRetries int, scheduledAt time.Time) *model.Task
NewMigrateTask builds the singleton migration coordinator for one replacement. It prepares the target, seeds durable items, observes their bounded execution state, and hands completed migration to retirement.
func NewRetirePayload ¶
NewRetirePayload builds the retirement coordinator payload.
func NewRetireTask ¶
func NewRetireTask(replacementID, bucketID int64, maxRetries int, scheduledAt time.Time) *model.Task
NewRetireTask builds the singleton retirement coordinator for one replacement. It runs on the storage cleanup worker because terminating a service is a destructive lifecycle action, not upload work.
func ParseRetirePayload ¶
ParseRetirePayload decodes a retirement coordinator payload.
func RetireTaskKey ¶
RetireTaskKey identifies the single retirement coordinator for one replacement. It retires the generation being replaced.
Types ¶
type ClaimToken ¶
ClaimToken fences lifecycle updates made by one item worker lease.
type ExecutionSnapshot ¶
type ExecutionSnapshot struct {
ReplacementID int64 `bun:"replacement_id"`
SeedingComplete bool `bun:"seeding_complete"`
ItemsTotal int `bun:"items_total"`
ItemsCopied int `bun:"items_copied"`
HasPending bool `bun:"has_pending"`
HasActive bool `bun:"has_active"`
HasRetrying bool `bun:"has_retrying"`
HasWaitingSource bool `bun:"has_waiting_source"`
HasFailed bool `bun:"has_failed"`
}
ExecutionSnapshot is the bounded coordinator view of one replacement. It intentionally reports presence rather than item counts; full aggregates are reserved for operator-facing progress reads.
type FailureReason ¶
type FailureReason string
FailureReason identifies a permanent failure whose recovery action differs from retrying the same approved target.
const (
FailureReasonTargetInUse FailureReason = "target_in_use"
)
func (FailureReason) Valid ¶
func (r FailureReason) Valid() bool
Valid reports whether the value is a known permanent failure reason.
type Item ¶
type Item struct {
bun.BaseModel `bun:"table:storage_replacement_items,alias:storage_replacement_item"`
ID int64 `bun:",pk,autoincrement"`
ReplacementID int64 `bun:",notnull"`
UploadID int64 `bun:",notnull"`
// TargetCopyID is the concrete copy row on the target generation. Tasks
// address it directly so they can never write the wrong generation.
TargetCopyID *int64 `bun:",nullzero"`
Status ItemStatus `bun:",notnull"`
Attempts int `bun:",notnull,default:0"`
ScheduledAt time.Time `bun:",nullzero,notnull,default:current_timestamp"`
RetryCount int `bun:",notnull,default:0"`
// MaxRetries is nullable only for rows created before the durable item
// queue migration. Startup recovery initializes it exactly once.
MaxRetries *int `bun:",nullzero"`
ClaimedAt *time.Time `bun:",nullzero"`
LeaseUntil *time.Time `bun:",nullzero"`
LastError *string `bun:",nullzero"`
CreatedAt time.Time `bun:",nullzero,notnull,default:current_timestamp"`
UpdatedAt time.Time `bun:",nullzero,notnull,default:current_timestamp"`
}
Item is one unit of migration work. Items are keyed by storage upload, not by object version, so content shared by many versions is copied once.
type ItemStatus ¶
type ItemStatus string
ItemStatus is the lifecycle of one unit of migration work. An item is keyed by stored content, not by object version, so shared content migrates once.
const ( // ItemStatusPending is seeded work not yet attempted. ItemStatusPending ItemStatus = "pending" // ItemStatusRunning is an item leased by a replacement worker. ItemStatusRunning ItemStatus = "running" // ItemStatusRetrying is temporarily failed work waiting for its persisted // retry schedule. It remains retirement-blocking. ItemStatusRetrying ItemStatus = "retrying" // ItemStatusWaitingSource means no readable copy and no cached content is // available yet. The coordinator moves on and revisits it later. ItemStatusWaitingSource ItemStatus = "waiting_source" // ItemStatusCopied means the target holds a committed readable copy. ItemStatusCopied ItemStatus = "copied" // ItemStatusCancelled means the content no longer needs migrating. ItemStatusCancelled ItemStatus = "cancelled" // ItemStatusFailed exhausted its item-level retry budget. Other items may // continue, but the source cannot retire until an operator retries it. ItemStatusFailed ItemStatus = "failed" )
func (ItemStatus) Blocking ¶
func (s ItemStatus) Blocking() bool
Blocking reports whether the item prevents the source from being retired.
func (ItemStatus) Executable ¶
func (s ItemStatus) Executable() bool
Executable reports whether an item is still active or recoverable.
func (ItemStatus) Valid ¶
func (s ItemStatus) Valid() bool
Valid reports whether the value is a known item status.
type MigratePayload ¶
type MigratePayload struct {
ReplacementID int64
}
MigratePayload is the persisted state of the migration coordinator. The durable item queue owns all transfer state.
func ParseMigratePayload ¶
func ParseMigratePayload(task *model.Task) (MigratePayload, error)
ParseMigratePayload decodes a migration coordinator payload. These tasks only ever exist after the upgrade that introduced them, so the replacement ID is required rather than inferred.
type Phase ¶
type Phase string
Phase is the worker responsibility a status maps to. It keeps the coordinator dispatch and the state machine from drifting apart.
const ( // PhasePrepare creates the approved target service and activates it. PhasePrepare Phase = "prepare" // PhaseMigrate copies stored content to the target. PhaseMigrate Phase = "migrate" // PhaseRetire runs the safety gate and terminates the old service. PhaseRetire Phase = "retire" // PhaseNone means no coordinator should act. PhaseNone Phase = "none" )
type ProgressSnapshot ¶
type ProgressSnapshot struct {
ReplacementID int64
Phase Phase
SeedingComplete bool
ItemsTotal int
ItemsProcessed int
ItemsCopied int
ItemsNoLongerNeeded int
ItemsPending int
ItemsActive int
ItemsAttention int
ItemsRetrying int
ItemsWaitingSource int
ItemsFailed int
Percent *int
NextRetryAt *time.Time
}
ProgressSnapshot is the UI-neutral aggregate for one replacement. Total is provisional until SeedingComplete is true.
type Replacement ¶
type Replacement struct {
bun.BaseModel `bun:"table:storage_replacements,alias:storage_replacement"`
ID int64 `bun:",pk,autoincrement"`
BucketID int64 `bun:",notnull"`
// CopyIndex is the logical replica slot both generations belong to.
CopyIndex int `bun:",notnull"`
SourceDataSetID int64 `bun:",notnull"`
TargetDataSetID int64 `bun:",notnull"`
SelectionMode SelectionMode `bun:",notnull"`
RequestedProviderID *types.OnChainID `bun:"type:text"`
ClientRequestID string `bun:",notnull"`
Status Status `bun:",notnull"`
WaitReason *WaitReason `bun:",nullzero"`
FailureReason *FailureReason `bun:",nullzero"`
LastError *string `bun:",nullzero"`
// ItemsTotal and ItemsCopied are maintained inside the transactions that
// seed and complete items, so progress never needs a history-sized count.
ItemsTotal int `bun:",notnull,default:0"`
ItemsCopied int `bun:",notnull,default:0"`
// SeedCursorUploadID advances through storage uploads in bounded batches so
// no single transaction scales with retained bucket history.
SeedCursorUploadID int64 `bun:",notnull,default:0"`
SeedingComplete bool `bun:",notnull,default:false"`
// StateVersion fences item-worker pause requests from a later coordinator
// recovery. LastDispatchedAt is the durable fairness cursor used by the
// global item queue.
StateVersion int64 `bun:",notnull,default:1"`
LastDispatchedAt *time.Time `bun:",nullzero"`
// TerminationEpoch is recorded before the old service is treated as
// terminated, so a crash between termination and observation re-reads it
// instead of terminating twice.
TerminationTxHash *string `bun:",nullzero"`
TerminationEpoch *int64 `bun:",nullzero"`
TerminationObservedAt *time.Time `bun:",nullzero"`
// AbandonedTerminationEpoch records termination of a superseded target.
// It is separate from the source termination fields above because the two
// services belong to opposite generations.
AbandonedTerminationTxHash *string `bun:",nullzero"`
AbandonedTerminationEpoch *int64 `bun:",nullzero"`
AbandonedTerminationObservedAt *time.Time `bun:",nullzero"`
SupersededByID *int64 `bun:",nullzero"`
ConfirmedAt time.Time `bun:",nullzero,notnull,default:current_timestamp"`
CreatedAt time.Time `bun:",nullzero,notnull,default:current_timestamp"`
UpdatedAt time.Time `bun:",nullzero,notnull,default:current_timestamp"`
}
Replacement is one operator-approved provider replacement for a single bucket replica slot. It is created only by an explicit confirmation and authorizes a new paid service, the topology switch, migration, and termination of the old service once the safety gate passes.
type SelectionMode ¶
type SelectionMode string
SelectionMode records how the operator chose the target provider.
const ( // SelectionModeAutomatic picks an eligible provider the bucket has not used. SelectionModeAutomatic SelectionMode = "automatic" // SelectionModeManual uses exactly the provider the operator supplied. SelectionModeManual SelectionMode = "manual" )
func (SelectionMode) Valid ¶
func (m SelectionMode) Valid() bool
Valid reports whether the value is a known selection mode.
type Status ¶
type Status string
Status is the lifecycle of one approved replacement.
const ( // StatusPreparingTarget means the approved target service is being created. // Writes still go to the source. StatusPreparingTarget Status = "preparing_target" // StatusMigrating means the target owns the slot for new writes and stored // content is being copied across. StatusMigrating Status = "migrating" // StatusWaiting means progress is blocked on a recoverable dependency. It // consumes no retry budget and is never a failure. StatusWaiting Status = "waiting" // StatusRetiring means migration finished and the source is going through // the retirement safety gate. StatusRetiring Status = "retiring" // StatusCleanupAttention means retirement cannot proceed without an // operator decision. Automatic task retry is suppressed. StatusCleanupAttention Status = "cleanup_attention" // StatusFailed means the work exhausted its retries and needs the operator // to retry it from the Data Sets surface. StatusFailed Status = "failed" // StatusCompleted means the source service was terminated and observed as // terminated. StatusCompleted Status = "completed" // StatusSuperseded means a later confirmation replaced this one. StatusSuperseded Status = "superseded" )
func NextStates ¶
NextStates lists the statuses reachable from one status.
func OnTaskExhausted ¶
OnTaskExhausted maps a status to the state a terminally exhausted coordinator task leaves behind. Retirement goes to operator attention rather than failed because its remaining work is a cleanup decision, not a retryable copy. The second return value reports whether any change is warranted.
func (Status) HoldsSource ¶
HoldsSource reports whether the replacement still owns its source data set, which is what stops a second replacement from starting on the same source.
func (Status) Retryable ¶
Retryable reports whether the dedicated retry action accepts this status. Waiting work resumes on its own and is deliberately excluded.
type WaitReason ¶
type WaitReason string
WaitReason explains a recoverable pause so an operator can tell "nothing is wrong yet" from "something needs me". It is only meaningful while the status is StatusWaiting.
const ( // WaitReasonReadableSource means no replica and no cached copy can supply // the content this item needs. WaitReasonReadableSource WaitReason = "readable_source" // WaitReasonTarget means the approved target provider is unreachable. WaitReasonTarget WaitReason = "target" // WaitReasonTargetCreating means the paid target service has not been // created yet. WaitReasonTargetCreating WaitReason = "target_creating" // WaitReasonTargetWritable means the created service is not writable yet. WaitReasonTargetWritable WaitReason = "target_writable" // WaitReasonFunding means the wallet cannot yet pay for the target service. WaitReasonFunding WaitReason = "funding" // WaitReasonProvider means a provider call failed in a way that resolves on // its own. WaitReasonProvider WaitReason = "provider" // WaitReasonTerminationEpoch means the chain has not yet reached the epoch // at which the old service ends. WaitReasonTerminationEpoch WaitReason = "termination_epoch" // WaitReasonSourceWrites means a write to the old provider is still in // flight, so it cannot be retired. WaitReasonSourceWrites WaitReason = "source_writes" // WaitReasonCoverage means some retained version is not yet readable on the // new provider. WaitReasonCoverage WaitReason = "coverage" )
func (WaitReason) Message ¶
func (r WaitReason) Message() string
Message is the operator-facing explanation of the pause. It says what is being waited on, never an internal state name.
func (WaitReason) Valid ¶
func (r WaitReason) Valid() bool
Valid reports whether the value is a known wait reason.
Source Files