README
¶
Runtime
The runtime package exists to reduce duplication between the three runners: migration, move and datasync. Each runner has a runner.go that drives the same lifecycle. They began as copy-paste forks, so they kept separate copies of the same plumbing. Those copies drift apart. A fix made in one runner is easy to miss in the other two, and nothing fails to compile when it is missed. (See "Keeping the runner triplet in sync" in AGENTS.md.)
Code moves here when it meets both conditions:
- It is shared. At least two of the runners have a copy, and the copies differ only in which of the runner's own fields they read.
- It is runner plumbing. It is not a subsystem. The copier, applier, checksum, change feed, checkpoint table and sentinel each have their own package.
runtimeonly wires them into a run, so it imports them and they never import it.
A copy whose difference between runners is real is not forced into one shape. The difference is passed in instead: a callback, a field of a config struct, or an optional value. runtime never imports a runner package.
The name matches the standard library's runtime package. A file that needs both must import one of them under an alias.
What lives here
| Type or function | What it replaced | Used by |
|---|---|---|
Snapshot, Source |
Two copies of the periodic Status() block, Progress() and its throttle status |
migration, move |
FatalGate |
Two copies of the fatal change-feed handler (change.ClientConfig.CancelFunc) |
migration, move |
Lifecycle |
Three copies of the cancel function a Run invocation publishes, the opening lines of Run, the evidence behind Result(), and the order of Close (Teardown) |
migration, move, datasync |
SharedThrottler |
Two copies of the mutex-guarded throttler that setup resolves | migration, move |
RecordCopyCompleted |
Three identical copies of the copy aggregate reported when the copy ends | migration, move, datasync |
RetryCutover |
Two copies of the cutover attempt loop, one with a backoff and joined attempt errors and one without | migration, move |
Snapshot and Source
Migration and move walk the same states with the same subsystems. Their status block and Progress report are therefore built once, from a Snapshot. The runner builds a new Snapshot on every call. The Snapshot reads the state once, so every field of one report describes the same state.
Source is a struct of functions that read the runner's subsystems: copier, applier, checker, feeds and throttler. They are functions because setup assigns the subsystems while an API caller may already be polling. A Snapshot calls only the functions that the current state reports on. By the time the run is in that state, setup has assigned those subsystems. Each runner fills in Source inline in its snapshot method, so no runner declares an adapter type. SentinelSchema is optional. A move leaves it nil, because its sources can span several schemas. Snapshot reports load throttling during the sentinel wait only while Checker.ContinuousActive() reports an active verification pass; interval waits report no throttling.
FatalGate
This is the handler a runner wires to every change feed it starts. On the first fatal condition, Trip does the following, once:
- Moves the run to
status.ErrCleanup. - Logs the advice for the operator.
- Drops the checkpoint, unless the reason leaves the checkpoint resumable (
change.FatalReason.PreservesCheckpoint). - Cancels the run with a
status.FatalAbortcause.
Once the run has reached cutover, Trip does nothing. The runner supplies its noun, its checkpoint drop and its cancel function in a FatalTarget.
Datasync does not use FatalGate. It has no cutover, it always keeps its checkpoint, and it records the cause itself (recordFatal).
Lifecycle
Lifecycle holds what a Run invocation shares with the methods other goroutines call while it runs:
- the function that cancels the run, used by
Cancel,Abort,Closeand the fatal handler; - the correctness evidence the run leaves behind (
status.WorkflowResult).
Run starts with:
ctx, end := r.lifecycle.Begin(ctx)
defer end(&retErr)
end does three things, in this order:
- It replaces a
context.Canceledresult with the fatal-abort cause (status.AbortCause). - It records the evidence that the error carries.
- It cancels the context.
The evidence is recorded after the substitution, so it is the evidence of the error Run actually returns: a fatal-abort cause can carry status.ErrDurableMutation or status.ErrOwnershipAmbiguous, and context.Canceled never does. Cancelling last does not change the outcome. A context keeps the first cause it was cancelled with, so cancel(nil) cannot hide a fatal cause that is already set.
Lifecycle is a named field and is not embedded. Each runner forwards its own Cancel, Abort and Result to it. As a result, the runner's public API does not gain MarkDurableMutation, SetTerminalOwnership or RecordError. Datasync uses only the cancel side; it reports no Result. Tests that drive phases without calling Run use SetCancel.
Teardown is the body of each runner's Close. It cancels the run (an operator cancel), waits for the status and checkpoint goroutines, then runs every step the runner passes, in order, and joins their errors. A failed step does not skip the rest. The wait comes first so a late checkpoint write cannot reach a closed pool. The steps stay with each runner because each owns different resources.
SharedThrottler
This holds the throttler that setup resolves partway through. Progress and the change feed's UnderLoad callback may already be reading it at that point. Datasync reads its own load signal under progMu, so it does not use this.
RecordCopyCompleted
This reports the copy aggregate (rows and chunks) when the copy ends. A resumed run's chunker restores its row count from the checkpoint but starts its chunk count at zero. RecordCopyCompleted subtracts the restored rows so that the two counts cover the same invocation.
RetryCutover
This runs the cutover's attempt loop for migration and move. It does the following:
- Runs up to
CutoverRetry.Attemptsattempts. A value below 1 is one attempt. - Checks the context before every attempt. Between attempts it sleeps a backoff that starts at
CutoverInitialBackoff(100ms) and doubles up toCutoverMaxBackoff(10s). The sleep ends early on cancel. - Records every failed attempt's error as
attempt N: <error>. The error it returns joins them, so an operator sees every attempt, not only the last.
What happens after a failed attempt is the runner's. Its StopFunc is passed that attempt's error only, never the history, and returns a verdict: retry, succeeded (the runner proved the cutover committed) or failed (with an optional error appended after the history, such as status.ErrOwnershipAmbiguous). The runner also supplies what a cancel returns and what running out of attempts returns. Migration uses both to check whether a rename whose outcome was unknown committed after all. Move uses the defaults: the history joined with the context's error, and the history.
The loop does not cover move's rename retries under the source locks (renameRetryWait). Those happen inside one attempt, after the traffic switch, and must not release the locks.
What does not live here (yet)
These parts of the runners are still duplicated or still differ.
DumpCheckpointin migration and move. The two share a skeleton, including the rule for which states may persist checksum evidence. They differ only in the position and in the record fields.invalidateChecksumWatermark. It runs raw SQL that belongs inpkg/checkpoint.
When you find another copy, extract it here, or into the subsystem package it belongs to. Then add a row to the table above and to the shared-table section of AGENTS.md.
Documentation
¶
Overview ¶
Package runtime holds the runner pieces that pkg/migration, pkg/move and pkg/datasync would otherwise each keep a copy of, and let drift apart.
The finite runners (migration and move) walk the same states with the same subsystems, so their periodic status block and Progress report are built here from a Snapshot rather than from two copies of the same switch. FatalGate is the state transition both make when their change feed fails, and SharedThrottler the throttler both resolve while already being polled. All three runners use Lifecycle, the cancel function and correctness evidence of a Run invocation, and RecordCopyCompleted.
The name matches the standard library's runtime package. A file that needs both must import one under an alias.
Index ¶
- Constants
- func RecordCopyCompleted(tracker *status.Tracker, c copier.Copier, rowsAtResume uint64)
- func RetryCutover(ctx context.Context, cfg CutoverRetry, ...) error
- type CutoverRetry
- type CutoverVerdict
- type FatalGate
- type FatalTarget
- type Lifecycle
- func (l *Lifecycle) Begin(ctx context.Context) (context.Context, func(*error))
- func (l *Lifecycle) Cancel(cause error)
- func (l *Lifecycle) MarkDurableMutation()
- func (l *Lifecycle) RecordError(err error)
- func (l *Lifecycle) Result() status.WorkflowResult
- func (l *Lifecycle) SetCancel(cancel context.CancelCauseFunc)
- func (l *Lifecycle) SetTerminalOwnership(o status.WorkflowTerminalOwnership)
- func (l *Lifecycle) Teardown(wait func(), steps ...func() error) error
- type SharedThrottler
- type Snapshot
- type Source
- type StopFunc
Constants ¶
const ( // CutoverInitialBackoff is the delay before the second cutover attempt. // It doubles before each later attempt, up to CutoverMaxBackoff. Without a // backoff a contended table lock produces a tight retry loop that worsens // the contention; the delay gives its source time to clear. CutoverInitialBackoff = 100 * time.Millisecond // CutoverMaxBackoff caps the delay between cutover attempts. CutoverMaxBackoff = 10 * time.Second )
Variables ¶
This section is empty.
Functions ¶
func RecordCopyCompleted ¶
RecordCopyCompleted reports to tracker the copy aggregate settled during this Run invocation. The chunker restores its settled row count from the checkpoint while its chunk count starts afresh, so rowsAtResume, the row count it restored, is subtracted to keep the two figures on the same invocation. It does nothing until c has a chunker.
Used by all three runners, including datasync.
func RetryCutover ¶ added in v0.19.0
func RetryCutover( ctx context.Context, cfg CutoverRetry, attempt func(ctx context.Context, i int) error, stop StopFunc, onCancel func(history []error) error, exhausted func(history []error) error, ) error
RetryCutover runs a cutover's attempt loop: it counts the attempts, sleeps an exponential backoff between them, records every failed attempt's error prefixed "attempt N:", and returns them joined, so an operator debugging a flapping cutover sees the full failure history rather than only the last try.
Which failures are retried is the runner's: stop classifies each failed attempt. A done ctx, checked before every attempt and during the backoff, returns onCancel(history); nil onCancel returns the history joined with ctx.Err(). When the attempts run out, RetryCutover returns exhausted(history); nil exhausted returns the history joined. history is the failed attempts' prefixed errors, oldest first, in a slice the callback owns (empty when no attempt has failed). Join it with what the callback adds (errors.Join(append(history, extra...)...)), so the result unwraps to one error per attempt.
Types ¶
type CutoverRetry ¶ added in v0.19.0
type CutoverRetry struct {
// Attempts is the number of attempts. A value below 1 is one attempt.
Attempts int
// InitialBackoff and MaxBackoff shape the delay between attempts. A zero
// value takes CutoverInitialBackoff or CutoverMaxBackoff. MaxBackoff caps
// every delay, the first one included.
InitialBackoff, MaxBackoff time.Duration
// Logger logs each attempt and failure. Nil logs to slog.Default().
Logger *slog.Logger
// Sleep waits d, or returns ctx's error when ctx is done first. Nil is a
// timer that stops early on cancel. It is a test seam.
Sleep func(ctx context.Context, d time.Duration) error
}
CutoverRetry configures RetryCutover.
type CutoverVerdict ¶ added in v0.19.0
type CutoverVerdict int
CutoverVerdict is a StopFunc's decision about a failed cutover attempt.
const ( // CutoverVerdictRetry runs the next attempt, after the backoff. When no attempts // are left, RetryCutover returns what the exhausted callback returns. CutoverVerdictRetry CutoverVerdict = iota // CutoverVerdictSucceeded ends the loop with a nil error: the runner has proven // that the cutover committed even though the attempt reported an error. CutoverVerdictSucceeded // CutoverVerdictFailed ends the loop without another attempt. CutoverVerdictFailed )
type FatalGate ¶
type FatalGate struct {
// contains filtered or unexported fields
}
FatalGate is the change feed's fatal-condition handler (change.ClientConfig.CancelFunc) shared by the finite runners. The zero value is ready to use. A runner owns one gate for its lifetime and wires every change feed it starts to it, so a burst of fatal events from several feed goroutines is realistic: the gate acts on the first and ignores the rest.
Datasync does not use it. It has no cutover to guard and always keeps its checkpoint (see change.FatalReason.PreservesCheckpoint).
func (*FatalGate) Trip ¶
func (g *FatalGate) Trip(reason change.FatalReason, t FatalTarget) bool
Trip aborts the run for reason. It returns false, and does nothing, once the run has reached cutover: the tables have been swapped, so there is nothing left to abort. Otherwise it returns true, and on the first call only:
- sets the state to status.ErrCleanup,
- logs reason's operator advice (change.FatalReason.Advice),
- drops the checkpoint unless reason leaves it resumable (change.FatalReason.PreservesCheckpoint). A schema change, for example, would otherwise block the run from proceeding permanently; starting again is the better choice. A dead stream preserves it: that is exactly the failure checkpoint resume exists to recover from.
- cancels the run with a status.FatalAbort cause.
Doing these once keeps the side effects from racing each other and the runner's Close.
type FatalTarget ¶
type FatalTarget struct {
// Noun names the run in log lines and the abort error ("migration", "move").
Noun string
Tracker *status.Tracker
Logger *slog.Logger
// DropCheckpoint invalidates the run's checkpoint. It must return nil
// without doing anything when setup has not created the checkpoint table
// yet: a feed can fail during early setup.
DropCheckpoint func(context.Context) error
// Cancel cancels the run with cause. Run returns the cause (see
// status.AbortCause), so the abort is reported as the failure it is and
// not as an operator cancellation.
Cancel func(cause error)
}
FatalTarget is the run a FatalGate aborts.
type Lifecycle ¶
type Lifecycle struct {
// contains filtered or unexported fields
}
Lifecycle is the state a runner's Run invocation shares with the methods an API caller or a background goroutine calls while it runs: the function that cancels it, and the correctness evidence (status.WorkflowResult) it leaves behind. The zero value is ready to use. A runner owns one as a named field and forwards its own Cancel, Abort and Result to it, so the runner's public API does not grow the methods that record evidence.
func (*Lifecycle) Begin ¶
Begin starts a Run invocation. It derives a context that Cancel cancels, publishes its cancel function, and clears the evidence the previous invocation left. Run must defer the returned function with the address of its named error result:
ctx, end := r.lifecycle.Begin(ctx) defer end(&retErr)
end replaces a context.Canceled result with a fatal-abort cause (see status.AbortCause), then records the evidence the result carries (see RecordError), then cancels the context. Recording after the substitution records the evidence of the error Run returns, which a fatal-abort cause can carry and context.Canceled cannot. Cancelling last changes nothing: a context keeps the first cause it was cancelled with.
func (*Lifecycle) Cancel ¶
Cancel cancels the running invocation with cause. A nil cause is an operator cancellation: Run returns context.Canceled. A cause marked with status.FatalAbort is returned by Run instead. It does nothing before Begin (early setup, or test paths that bypass Run).
func (*Lifecycle) MarkDurableMutation ¶
func (l *Lifecycle) MarkDurableMutation()
MarkDurableMutation records that this invocation changed something that a retry cannot undo.
func (*Lifecycle) RecordError ¶
RecordError records the evidence err carries: status.ErrDurableMutation and status.ErrOwnershipAmbiguous.
func (*Lifecycle) Result ¶
func (l *Lifecycle) Result() status.WorkflowResult
Result returns the evidence retained from the most recent invocation.
func (*Lifecycle) SetCancel ¶
func (l *Lifecycle) SetCancel(cancel context.CancelCauseFunc)
SetCancel replaces the cancel function. Begin calls it; tests that drive a runner's phases without Run call it to observe or supply cancellation.
func (*Lifecycle) SetTerminalOwnership ¶
func (l *Lifecycle) SetTerminalOwnership(o status.WorkflowTerminalOwnership)
SetTerminalOwnership records which side owns the table or traffic when the invocation ends.
func (*Lifecycle) Teardown ¶ added in v0.19.0
Teardown is the shared order of a runner's Close. It cancels the run as an operator cancel (Cancel(nil)), waits for the run's background tasks, then runs every step and joins their errors. wait is the status.WatchTask join; nil means none started. The wait comes before any step, so a late checkpoint write cannot land after Close has returned or reach a closed pool. Every step runs even when an earlier one fails, so one failed close does not leak the rest. Steps run in order: a runner closes its change feeds before the throttler their UnderLoad callback reads, and the pools last.
type SharedThrottler ¶
type SharedThrottler struct {
// contains filtered or unexported fields
}
SharedThrottler holds the throttler a runner resolves during setup. Setup assigns it partway through, while an API caller may already be polling Progress, which reports whether the run is paused, and a change feed's UnderLoad callback may already be reading it. The zero value holds nil.
func (*SharedThrottler) Get ¶
func (s *SharedThrottler) Get() throttler.Throttler
Get returns the resolved throttler, or nil if setup has not resolved one.
func (*SharedThrottler) Set ¶
func (s *SharedThrottler) Set(t throttler.Throttler)
Set publishes the resolved throttler.
type Snapshot ¶
type Snapshot struct {
// Noun names the workflow in the status header ("migration", "move").
Noun string
State status.State
Tracker *status.Tracker
// Tables is the copy chunker's per-table progress. Progress and Status
// both derive their copy figures from it, so the API and the log block
// report one measure: settled rows against the tables' cardinality
// estimates, kept past the end of the copy.
Tables []status.TableProgress
Checkpoint *status.LastCheckpoint
Resumed bool
Source Source
}
Snapshot is what a runner reports on. The runner builds one per call, and the state is read once when it does, so every field of a report describes the same state rather than whichever state each field happened to observe.
func (*Snapshot) Status ¶
Status returns the periodic report on the whole run: a header line plus one indented row per subsystem (see status.Block). It absorbs what used to be separate periodic lines from the change feeds (flushes, rotations) and the checkpoint dumper, which each ran on their own interval — see github.com/block/spirit/issues/329.
func (*Snapshot) ThrottleStatus ¶
func (s *Snapshot) ThrottleStatus() status.ThrottleStatus
ThrottleStatus reports only signals honored by the work currently running. Checksum passes honor load signals; interval waits and cutover are unpaced. The sentinel wait only reports pacing while a continuous checksum runs.
type Source ¶
type Source struct {
Copier func() copier.Copier
Applier func() applier.Applier
Checker func() checksum.Checker
Feeds func() []change.Source
Throttler func() throttler.Throttler
// SentinelSchema, when set and not empty, adds the sentinel table's name
// to the sentinel row. A move's sources can span several schemas, so it
// leaves it nil.
SentinelSchema func() string
}
Source gives a Snapshot the runner's subsystems. Setup assigns them while an API caller may already be polling, so they are read through functions, and a Snapshot only calls the ones the current state reports on, which are the ones setup has finished assigning. Every function but SentinelSchema must be set.
type StopFunc ¶ added in v0.19.0
type StopFunc func(err error) (verdict CutoverVerdict, extra error)
StopFunc classifies the error of the attempt that just failed. It is passed that attempt's error only, never the errors of earlier attempts, so a marker that an earlier attempt carried cannot stop a later one.
For CutoverVerdictFailed, extra is appended after the attempt history in the error RetryCutover returns; nil appends nothing. extra is ignored for the other verdicts.
Source Files