Documentation
¶
Overview ¶
Package backfill moves data after a schema change, in the background (zero-impact deploys, G6b). A migration changes schema only: a table-wide UPDATE inside one holds row locks in a long transaction, bloats the table and holds up the migration job before every rollout (migrate_lint_test.go refuses one from migration 0110 on, unless the file carries "-- da-ack: backfill-in-migration <reason>").
A Job is a named, idempotent step over a table by keyset: given the cursor it processes at most `limit` rows after it and says where it got to. The Runner runs jobs on one pod at a time (a Postgres advisory lock, internal/singleton), a batch at a time, each batch in its own short transaction together with the update of the job's row in `backfills`, so the data and the progress commit or roll back as one. A drain or a crash between batches loses nothing and repeats nothing; the next pod (or the next pass) resumes from the stored cursor.
Rules for a job (DA's rulings; the review checks them):
- the new value is computed INSIDE the UPDATE statement from the row's own current columns (SET new = f(old) WHERE new IS DISTINCT FROM f(old)), never read in Go and written back: a recorder update that lands between the read and the write would be overwritten with a stale value. On the recorder's tables (drives, charges, vehicle_state, ...) a job writes only new or still-unfilled columns, never a column the recorder keeps writing;
- idempotent (a second run over a row changes nothing), bounded (LIMIT; one batch finishes well inside the statement timeout) and key-ordered;
- never assumed done by readers: code that reads what a backfill fills treats the old shape as valid until `done`, and a "-- contract:" removal of the old shape waits for it via contract-after;
- registered by a DA-acked PR that carries a row-count test on a fixture.
It runs on the system pool (the one that serves sweeps, the monitor's writes and migrations), never the RLS request pool. A job is a maintenance job like the sample thinner: it touches every user's rows in key order and calls nothing outside Postgres, so it is not a group owner's write (the fence lint lists it as such).
Bound on the recorder: a batch takes its row locks in key order and holds them until it commits, so a recorder write to one of those rows waits for the rest of the batch. A batch runs under statement_timeout and lock_timeout of 5 s each (a 1000-row keyed UPDATE takes milliseconds), so the recorder's worst wait behind a backfill is about 5 s. A batch that hits either timeout (SQLSTATE 55P03, 57014) is ordinary contention: it rolls back with its progress and is retried after the pause (rivolt_backfill_retries_total); only five in a row, or any other error, mark the job failed.
Registration order: a job is registered only in a release AFTER the one whose writers (the recorder, the API) fill the new column for every new row; Fill.WriterRelease names that tag and is required. A row written behind the cursor by a writer that does not yet fill the column stays unfilled while the job reports done, and nothing walks back to it. A Fill's key (Key plus MoreKeys, a composite key walked as a row comparison) must be exactly the columns of one unique index, in its column order (checked before the first batch), so no row is skipped at a batch boundary.
Pausing: RIVOLT_BACKFILL=0 stops every job (the runner is on by default, with an empty registry); a row in `flags` named backfill.<job>.paused with value {"paused": true} stops that one job, checked before every batch (a malformed row also pauses it: the safe reading). Progress is kept either way. Default pace: 1000 rows per batch and one batch per 10 seconds (about 3 hours per million rows), per job via BatchSize and Interval.
Index ¶
Constants ¶
const ( DefaultBatch = 1000 DefaultPause = 10 * time.Second DefaultIdle = 10 * time.Minute DefaultFailedRetry = 15 * time.Minute DefaultLockTimeout = 5 * time.Second DefaultStmtTimeout = 5 * time.Second // a 1000-row keyed UPDATE takes milliseconds: 5 s is a failure, not a slow batch )
Defaults of a Runner.
const ( StatePending = "pending" StateRunning = "running" StateDone = "done" StateFailed = "failed" )
States of a job's row.
Variables ¶
var RecorderTables = map[string]bool{"drives": true, "charges": true, "vehicles": true, "vehicle_state": true, "charges_capped_backup": true, "charges_deleted_backup": true}
RecorderTables are the tables the live recorder keeps writing (and the repair tools touch): a backfill there must be a Fill, so what it may write is declared and enforced in the SQL. Keep in step with db.RepairWriteTables and the partitioned vehicle_state.
Functions ¶
This section is empty.
Types ¶
type Fill ¶
type Fill struct {
// Table and Key (the primary key column) name what is walked; KeyType is its SQL type for the cursor cast (uuid, bigint, integer, text,
// timestamptz).
Table string
Key string
KeyType string
// MoreKeys are the further columns of a composite key, after Key, in order (vehicle_state is keyed by vehicle_id, at). The cursor then covers the
// whole key (row comparison, `(a, b) > (x, y)`), so rows that share a leading value are walked in full. Key plus MoreKeys must be exactly the
// columns of one unique index, or the job is refused.
MoreKeys []FillKey
Columns []FillColumn
// WriterRelease is the final tag (vX.Y.Z) of the release whose writers (the recorder, the API) already fill these columns on every new row.
// A job is registered only in a later release: a row written behind the cursor by an old writer is never revisited, yet the job would report
// done. The registration PR shows the writer live in that tag.
WriterRelease string
}
Fill is the ready-made job body for the usual backfill: fill new columns of an existing table from the row's own columns, in key order, only where they are still empty. The UPDATE is built here, so what a job may write is fixed by its declaration and enforced in the SQL itself:
- only the declared columns are SET, and each one only where its Empty predicate holds (`CASE WHEN <empty> THEN <expr> ELSE col END`, with the same predicate in the WHERE), so a value the recorder (or a user) has written since is never overwritten, not even by a batch that read the row earlier: the row is locked (FOR UPDATE) and the predicate is re-evaluated against its current contents;
- the new value is computed inside the statement from the row's own columns (Expr), never read in Go and written back;
- rows are taken in key order after the cursor, a batch at a time, and a row another transaction holds is waited for (bounded by the batch's lock_timeout; a timeout fails the batch, which is retried), never skipped: skipping would leave it behind the cursor for good.
A job on a recorder table (RecorderTables) must be a Fill: Validate refuses a raw Step there.
type FillColumn ¶
type FillColumn struct {
Name string
// Expr is the SQL expression for the new value, over the row's own columns (referred to as t.<col>) and constants.
Expr string
// Empty is the predicate that says the column is still unfilled; "" is `t.<Name> IS NULL`.
Empty string
}
FillColumn is one column a Fill writes.
type FillKey ¶
type FillKey struct{ Name, Type string }
FillKey is one further column of a composite key.
type Job ¶
type Job struct {
Name string
// Table is the table the job writes (set from the Fill; a raw Step declares it). A job on a recorder table must be a Fill.
Table string
// Fill, when set, builds the job's Step (see Fill): the declared columns, written only where still empty, computed inside the UPDATE.
Fill *Fill
// BatchSize is the rows per batch; 0 is DefaultBatch. Interval is the wait between this job's batches; 0 is the runner's Pause.
BatchSize int
Interval time.Duration
// Step processes one batch inside tx (already bounded by lock_timeout and statement_timeout) after cursor ("" at the start).
Step func(ctx context.Context, tx *sql.Tx, cursor string, limit int) (Result, error)
}
Job is one backfill. Name is permanent (it keys the progress row): never rename a job that has started.
type Metrics ¶
Metrics are the backfill series. Register Collectors with the metrics registry; a nil *Metrics ignores every call.
func (*Metrics) Collectors ¶
func (m *Metrics) Collectors() []prometheus.Collector
Collectors are what the metrics registry registers.
type Result ¶
type Result struct {
// Next is the cursor to resume from (the last key processed); unchanged when the batch found nothing.
Next string
// Rows is how many rows the batch changed.
Rows int
// Done: nothing is left after Next. A job that cannot tell returns Done when a batch changes no row.
Done bool
}
Result is what one batch did.
type Runner ¶
type Runner struct {
DB *sql.DB
Jobs []Job
Log *slog.Logger
// Pause is the wait between batches when the job sets no Interval (DefaultPause); Idle the wait between passes when every job is done or failed (DefaultIdle);
// FailedRetry how long a failed job waits before it is tried again (DefaultFailedRetry).
Pause time.Duration
Idle time.Duration
FailedRetry time.Duration
Metrics *Metrics
// LockTimeout and StatementTimeout bound one batch (5 s each by default). They are also the longest a recorder write can wait behind a batch:
// a batch holds its row locks until it commits, so the worst case for the recorder is StatementTimeout, about 5 s.
LockTimeout time.Duration
StatementTimeout time.Duration
// Now and Sleep are for tests.
Now func() time.Time
Sleep func(ctx context.Context, d time.Duration)
// contains filtered or unexported fields
}
Runner runs the jobs. The zero value of every field but DB and Jobs is usable.
Source Files