Help us shape the future of Go! Take the 2026 Go Developer Survey

storagereplacement

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Oct 29, 2026 License: Apache-2.0 Imports: 8 Imported by: 0

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

View Source 
const (
	CodeSourceRunning        = "replacement_source_running"
	CodeSourceOutcomeUnknown = "replacement_source_outcome_unknown"
	CodeActive               = "replacement_active"
	CodeTargetCreating       = "replacement_target_creating"
	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"
	CodeTargetUnavailable    = "replacement_target_unavailable"
	CodeIdempotencyConflict  = "replacement_idempotency_conflict"
	CodeSourceNotCurrent     = "replacement_source_not_current"
)

Stable machine-readable codes returned alongside API errors. Clients branch on these, so treat them as part of the public contract.

View Source 
const (
	MessageRetireDataSet        = "replacement.retire_dataset"
	MessageReplacementActivated = "replacement.activated"
)

Variables

View Source 
var (
	// ErrSourceRunning blocks replacement while creation work holds a live claim.
	ErrSourceRunning = errors.New("data set setup is still running")
	// ErrSourceOutcomeUnknown requires an observed result before replacement.
	ErrSourceOutcomeUnknown = errors.New("data set creation outcome is not confirmed")
	// ErrActiveReplacement means this data set is reserved by an unfinished
	// replacement; only stopped outgoing work permits another provider choice.
	ErrActiveReplacement = errors.New("data set already has an active replacement")

	// ErrTargetCreating means an earlier replacement of this replica may still
	// get its target storage service created on chain, so superseding it would
	// leave that service untracked.
	ErrTargetCreating = errors.New("an earlier replacement target is still being created")

	// 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")

	// ErrTargetUnavailable means the requested provider is not currently an
	// 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")

	// 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 Allowed 

func Allowed(from, to Status) bool

Allowed reports whether the state machine permits this transition. A status never transitions to itself; callers update fields in place instead.

func Code 

func Code(err error) string

Code maps a replacement error to its stable API code. It returns an empty string for errors that carry no client-facing code.

func ValidateCoordinateInput 

func ValidateCoordinateInput(input CoordinateInput) error

func ValidateRetireInput 

func ValidateRetireInput(input RetireInput) error

Types

type CoordinateInput 

type CoordinateInput struct {
	ReplacementID int64 `json:"replacement_id"`
	Generation    int64 `json:"generation"`
}

func ParseCoordinateInput 

func ParseCoordinateInput(task *model.Task) (CoordinateInput, error)

type ExecutionSnapshot 

type ExecutionSnapshot struct {
	ReplacementID   int64 `bun:"replacement_id"`
	SeedingComplete bool  `bun:"seeding_complete"`
	HasPending      bool  `bun:"has_pending"`
	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"
	FailureReasonTargetRejected FailureReason = "target_rejected"
)

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,identity"`
	ReplacementID int64 `bun:",notnull"`
	ContentID     int64 `bun:",notnull"`
	// TargetDataSetID identifies the target generation. Together with ContentID
	// it resolves exactly one bound copy without retaining a surrogate copy ID.
	// It is NOT NULL because a nullable column would make the composite foreign
	// key to the replacement's target skip validation whenever it was unset.
	TargetDataSetID int64      `bun:",notnull"`
	Status          ItemStatus `bun:"type:text,notnull,default:'pending'"`
	LastError       *string    `bun:"type:text,nullzero"`
	CreatedAt       time.Time  `bun:",nullzero,notnull"`
	UpdatedAt       time.Time  `bun:",nullzero,notnull"`
}

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.

func (*Item) BeforeAppendModel 

func (i *Item) BeforeAppendModel(_ context.Context, query bun.Query) error

BeforeAppendModel stamps the audit columns on insert. The database has no timestamp default, so every row is written with one encoding instead of two that sort against each other inside the same second.

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"
	// ItemStatusCopied means the target holds a committed readable copy.
	ItemStatusCopied ItemStatus = "copied"
	// ItemStatusCancelled means the content no longer needs migrating.
	ItemStatusCancelled ItemStatus = "cancelled"
	// ItemStatusAttention requires an operator decision before retirement.
	ItemStatusAttention ItemStatus = "attention"
)

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 NotRetryableError 

type NotRetryableError struct{ Message string }

NotRetryableError supplies the operator action for a stopped replacement.

func (*NotRetryableError) Error 

func (e *NotRetryableError) Error() string

func (*NotRetryableError) Unwrap 

func (e *NotRetryableError) Unwrap() error

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"
)

func PhaseFor 

func PhaseFor(status Status) Phase

PhaseFor maps a status to the work a coordinator should perform.

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,identity"`
	BucketID int64 `bun:",notnull"`
	// CopyIndex is the logical replica slot both generations belong to.
	CopyIndex            int              `bun:"type:integer,notnull"`
	SourceDataSetID      int64            `bun:",notnull"`
	TargetDataSetID      int64            `bun:",notnull"`
	SelectionMode        SelectionMode    `bun:"type:text,notnull"`
	RequestedProviderID  *types.OnChainID `bun:"type:text"`
	ClientRequestID      string           `bun:"type:text,notnull"`
	PriceListFingerprint string           `bun:"type:text,notnull"`
	Status               Status           `bun:"type:text,notnull"`
	WaitReason           *WaitReason      `bun:"type:text,nullzero"`
	FailureReason        *FailureReason   `bun:"type:text,nullzero"`
	LastError            *string          `bun:"type:text,nullzero"`
	// SeedCursorContentID advances through storage uploads in bounded batches so
	// no single transaction scales with retained bucket history.
	SeedCursorContentID int64     `bun:",notnull,default:0"`
	SeedingComplete     bool      `bun:",notnull,default:false"`
	TaskGeneration      int64     `bun:",notnull,default:1"`
	TaskID              *int64    `bun:",nullzero"`
	SupersededByID      *int64    `bun:",nullzero"`
	CreatedAt           time.Time `bun:",nullzero,notnull"`
	UpdatedAt           time.Time `bun:",nullzero,notnull"`

	// The current transaction is projected from the latest attempt; the epoch
	// belongs to the service and can be observed independently of that attempt.
	TerminationAttemptNo                int64   `bun:",scanonly"`
	TerminationTxHash                   *string `bun:",scanonly"`
	TerminationConfirmedTxHash          *string `bun:",scanonly"`
	TerminationReceiptStatus            *uint64 `bun:",scanonly"`
	TerminationEpoch                    *int64  `bun:",scanonly"`
	AbandonedTerminationAttemptNo       int64   `bun:",scanonly"`
	AbandonedTerminationTxHash          *string `bun:",scanonly"`
	AbandonedTerminationConfirmedTxHash *string `bun:",scanonly"`
	AbandonedTerminationReceiptStatus   *uint64 `bun:",scanonly"`
	AbandonedTerminationEpoch           *int64  `bun:",scanonly"`
}

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.

func (*Replacement) BeforeAppendModel 

func (r *Replacement) BeforeAppendModel(_ context.Context, query bun.Query) error

BeforeAppendModel stamps the audit columns on insert. The database has no timestamp default, so every row is written with one encoding instead of two that sort against each other inside the same second.

type ReplacementActivated 

type ReplacementActivated struct {
	BucketID  int64
	CopyIndex int
}

func (ReplacementActivated) MessageType 

func (ReplacementActivated) MessageType() string

type RetireDataSet 

type RetireDataSet struct {
	ReplacementID int64
	BindingID     int64
}

func (RetireDataSet) MessageType 

func (RetireDataSet) MessageType() string

type RetireInput 

type RetireInput struct {
	ReplacementID int64 `json:"replacement_id"`
	DataSetID     int64 `json:"data_set_id"`
	Generation    int64 `json:"generation"`
}

func ParseRetireInput 

func ParseRetireInput(task *model.Task) (RetireInput, error)

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 SourceOutcomeError 

type SourceOutcomeError struct{ Message string }

SourceOutcomeError supplies a safe explanation while preserving the 409 code.

func (*SourceOutcomeError) Error 

func (e *SourceOutcomeError) Error() string

func (*SourceOutcomeError) Unwrap 

func (e *SourceOutcomeError) Unwrap() error

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 automatic recovery stopped and the operator must 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 

func NextStates(from Status) []Status

NextStates lists the statuses reachable from one status.

func (Status) Active 

func (s Status) Active() bool

Active reports whether a coordinator should still advance this replacement.

func (Status) HoldsSource 

func (s Status) HoldsSource() bool

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 

func (s Status) Retryable() bool

Retryable reports whether the dedicated retry action accepts this status. Waiting work resumes on its own and is deliberately excluded.

func (Status) Terminal 

func (s Status) Terminal() bool

Terminal reports whether the replacement can never change again.

func (Status) Valid 

func (s Status) Valid() bool

Valid reports whether the value is a known status.

type Termination 

type Termination struct {
	bun.BaseModel `bun:"table:storage_data_set_terminations,alias:storage_data_set_termination"`

	ID            int64           `bun:",pk,autoincrement,identity"`
	ReplacementID int64           `bun:",notnull"`
	Role          TerminationRole `bun:"type:text,notnull"`
	// Exactly one of the data set columns is set, chosen by Role. Each carries a
	// composite foreign key back to the matching column on the replacement, so a
	// row cannot name a data set the replacement never held in that role.
	SourceDataSetID          *int64    `bun:",nullzero"`
	AbandonedTargetDataSetID *int64    `bun:",nullzero"`
	Epoch                    *int64    `bun:"type:bigint,nullzero"`
	CreatedAt                time.Time `bun:",nullzero,notnull"`
	UpdatedAt                time.Time `bun:",nullzero,notnull"`
}

Termination records the service's end epoch independently of transaction attempts.

func (*Termination) BeforeAppendModel 

func (t *Termination) BeforeAppendModel(_ context.Context, query bun.Query) error

BeforeAppendModel stamps the audit columns on insert. The database has no timestamp default, so every row is written with one encoding instead of two that sort against each other inside the same second.

type TerminationAttempt 

type TerminationAttempt struct {
	bun.BaseModel `bun:"table:storage_data_set_termination_attempts,alias:storage_data_set_termination_attempt"`

	ID              int64      `bun:",pk,autoincrement,identity"`
	TerminationID   int64      `bun:",notnull"`
	AttemptNo       int64      `bun:",notnull"`
	TaskID          *int64     `bun:",nullzero"`
	AttemptedAt     *time.Time `bun:",nullzero"`
	TxHash          *string    `bun:"type:text,nullzero"`
	ConfirmedTxHash *string    `bun:"type:text,nullzero"`
	ReceiptStatus   *uint64    `bun:"type:smallint"`
	// UnsentAt marks an admission proven not to put a transaction on chain.
	// Admitting the same attempt again, or recording its transaction, reopens
	// this row.
	UnsentAt  *time.Time `bun:",nullzero"`
	CreatedAt time.Time  `bun:",nullzero,notnull"`
	UpdatedAt time.Time  `bun:",nullzero,notnull"`

	ReplacementID int64           `bun:",scanonly"`
	Role          TerminationRole `bun:",scanonly"`
}

TerminationAttempt permanently retains one admitted termination request and its receipt. A reverted receipt permits a separately authorized attempt.

func (*TerminationAttempt) BeforeAppendModel 

func (a *TerminationAttempt) BeforeAppendModel(_ context.Context, query bun.Query) error

type TerminationRole 

type TerminationRole string

TerminationRole names which of a replacement's two data sets a termination ended. The source is terminated once migration is safe; the target is terminated only after the replacement is superseded and its target abandoned. 

const (
	TerminationRoleSource          TerminationRole = "source"
	TerminationRoleAbandonedTarget TerminationRole = "abandoned_target"
)

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

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.