From 722dec2146e35879e2bbba609fcb765fe5bc1167 Mon Sep 17 00:00:00 2001 From: Tommaso Barbugli Date: Wed, 12 Aug 2026 22:43:45 +0200 Subject: [PATCH] sequences: advance them alone, by a million, and check there is room The cutover's sequence step is now a command of its own, for a migration that moves traffic before it moves the database. It reads each selected sequence's next value on the source and sets the target's copy that far past it. Values are set absolutely rather than advanced, so running it again is harmless and the cutover redoes it against the source's final values. The gap it leaves is a million rather than a thousand. That gap is the only thing between what the source allocates while it is still serving and a key the target has already handed out, and a thousand of them is a few seconds of a busy table. A gap that size makes a sequence running out of room a real failure mode, so preflight reports one now. Less room left than --sequence-offset is an error: setval refuses a value past the bound, so the cutover would fail at its sequence step no matter how willing the operator is. Under ten million values left is a warning, because how much is enough depends on how fast the application allocates and how long the source keeps serving, which only the operator knows. Only the sequences the cutover will set are considered, and the headroom arithmetic is unsigned: sequence bounds span the whole of int64, and subtracting two of them as signed values overflows and reports a sequence with room to spare as exhausted. Co-authored-by: Cursor --- README.md | 59 +++++-- internal/app/app.go | 48 +++++- internal/app/app_test.go | 7 + internal/cli/cli.go | 3 + internal/cli/cli_test.go | 27 ++++ internal/config/config.go | 2 + internal/cutover/cutover.go | 11 +- internal/cutover/cutover_integration_test.go | 6 +- internal/preflight/preflight.go | 147 ++++++++++++++++++ .../preflight/preflight_integration_test.go | 91 +++++++++++ internal/preflight/preflight_test.go | 85 ++++++++++ 11 files changed, 464 insertions(+), 22 deletions(-) create mode 100644 internal/cli/cli_test.go diff --git a/README.md b/README.md index 23ccb95..6b8bd17 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,12 @@ both servers, and atomically writes `cutover-report.json`. It moves data and metadata and judges neither: freezing application writes and deciding the copy is good enough to serve are the operator's, done before it runs. +`pgmigrate sequences` runs that one step on its own, from `follow` onwards, for a +cutover that moves traffic before it moves the database. Sequences are set +absolutely, so it is rerunnable and `cutover` redoes it against the source's final +values. `--sequence-offset` is the room the source keeps: whatever it allocates +beyond that collides with the target. + ## Documentation This README is the documentation. [Design considerations](#design-considerations-why-oh-why) @@ -69,7 +75,7 @@ explains why the mechanism is what it is and what each choice cost, [Limitations](#limitations) what the tool does not do. Test patterns and environment controls are in [test/README.md](test/README.md). -There are five commands: +There are six commands: | command | what it does | |---|---| @@ -77,6 +83,7 @@ There are five commands: | `run` | starts or resumes the migration, and waits in `follow` until cutover completes | | `status` | reads local state only, so it is safe to run beside `run` | | `verify` | samples each table against the target and checks what replication wrote | +| `sequences` | advances target sequences alone, so the target can take writes before the cutover | | `cutover` | performs the rerunnable, durably stepped cutover | Every command takes `--dir`. All but `status` also need source and target @@ -177,9 +184,9 @@ step log trimmed: "completed_at": "2026-08-11T07:40:10.07482Z", "end_position": "0/1BE9D08", "sequences": [ - {"schema": "e2e", "name": "order_id_seq", "source_value": 1143, "target_value": 2143, "is_called": true} + {"schema": "e2e", "name": "order_id_seq", "source_value": 1143, "target_value": 1001143, "is_called": true} ], - "configuration": {"sequence_offset": 1000, "values": {"workers": "16"}}, + "configuration": {"sequence_offset": 1000000, "values": {"workers": "16"}}, "steps": [] } ``` @@ -188,9 +195,10 @@ step log trimmed: version, and one built from a checkout names the commit it came from and whether that tree was clean. `end_position` is the boundary the target was drained through, and it is the line between what this migration carried and what it did -not: anything the source wrote after it stayed behind. `sequences` records that `order_id_seq` was left 1000 ahead -of the source, so the application cannot collide with an existing key. `steps` is -the durable log the cutover resumed against. +not: anything the source wrote after it stayed behind. `sequences` records that +`order_id_seq` was left 1,000,000 ahead of the source, so the application cannot +collide with an existing key. `steps` is the durable log the cutover resumed +against. ## Installing pgmigrate @@ -216,10 +224,17 @@ others are ignored. `--source` and `--target` default to `PGMIGRATE_SOURCE` and ### pgmigrate preflight Inventories the selected tables and checks server versions, logical-replication -settings, replica identity, WAL headroom, collations, extensions, target state, -privileges, and client-tool versions. Findings are persisted in the migration -directory, so `status` shows them later. Warnings block until acknowledged, and -`run` repeats the same checks with the same gate. +settings, replica identity, sequence headroom, WAL headroom, collations, +extensions, target state, privileges, and client-tool versions. Findings are +persisted in the migration directory, so `status` shows them later. Warnings block +until acknowledged, and `run` repeats the same checks with the same gate. + +Sequence headroom is checked against `--sequence-offset`, for every sequence a +selected table owns or draws a column default from. A sequence with fewer than ten +million values left before its maximum, or its minimum when it counts down, is a +warning: what remains has to cover both databases until traffic moves. One with +less room than the offset is an error, because `setval` refuses a value past the +bound and the cutover would fail at its sequence step. | flag | default | what it does | |---|---|---| @@ -232,6 +247,7 @@ directory, so `status` shows them later. Warnings block until acknowledged, and | `--pg-dump ` | found on `PATH` | `pg_dump` executable, whose version is checked here | | `--pg-restore ` | found on `PATH` | `pg_restore` executable, whose version is checked here | | `--wal-sample-duration ` | `1m` | how long to sample the source WAL rate when judging slot retention headroom | +| `--sequence-offset ` | `1000000` | the gap the cutover will leave, which is the room each selected sequence is checked for | | `--workers ` | host CPU count | index-build concurrency the tuning plan is sized for, so preflight reports the plan `run` would apply | | `--skip-target-tuning` | false | report no tuning plan, because the run will not tune | | `--target-memory ` | estimated from `shared_buffers` | target memory the plan is sized against, for example `64GB` | @@ -306,6 +322,28 @@ standard error. A named divergence, or a table stopped early, exits non-zero. | `--verify-converge-timeout ` | `1m` | how long a row that appears to differ is given to settle against a fixed WAL position before it is reported | | `--verify-cdc-rows ` | `100000` | applier-recorded keys per table checked alongside the heap sample. `0` falls back to the default | +### pgmigrate sequences + +Runs the cutover's sequence step on its own, from the `follow` phase onwards, for +a cutover that moves traffic before it moves the database. It reads each selected +sequence's next value on the source and sets the target's copy that far past it, so +the target can accept writes while the source is still serving. Every sequence a +selected table owns or draws a column default from is included. The results are +written to standard output as JSON. + +Values are set absolutely rather than advanced, so running it again is harmless, +and `cutover` runs the same step against the source's final values. The offset is +the room the source keeps: whatever it allocates beyond that collides with what the +target has already handed out, so size it above what the source can consume before +traffic moves. + +| flag | default | what it does | +|---|---|---| +| `--dir ` | required | migration state directory holding the schema selection | +| `--source ` | `PGMIGRATE_SOURCE` | source connection string | +| `--target ` | `PGMIGRATE_TARGET` | target connection string | +| `--sequence-offset ` | `1000000` | values each target sequence is set past the source's. `0` leaves no gap, which is only safe once the source will never allocate again | + ### pgmigrate cutover Performs the cutover as a sequence of durably recorded steps: validate the @@ -327,6 +365,7 @@ the result, and decide for yourself. | `--source ` | `PGMIGRATE_SOURCE` | source connection string | | `--target ` | `PGMIGRATE_TARGET` | target connection string | | `--endpos ` | the boundary cutover emits | explicit inclusive end position, for advanced use. Must resolve to an exact durable transaction or boundary | +| `--sequence-offset ` | `1000000` | values each target sequence is set past the source's; see [pgmigrate sequences](#pgmigrate-sequences) | | `--no-cleanup` | false | retain the source replication objects and target migration metadata. Target tuning and target replica identities are still reverted, because the target is about to serve production | ## Dependencies diff --git a/internal/app/app.go b/internal/app/app.go index e63448e..87542d3 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -183,7 +183,7 @@ func (a App) Preflight(ctx context.Context, cfg config.Config) error { RequiredExtensions: preflightSelection.Extensions, AcknowledgeWarnings: cfg.AckWarnings, AllowCollationChange: cfg.AllowCollationChange, PGDumpPath: cfg.PGDumpPath, PGRestorePath: cfg.PGRestorePath, - WALSampleDuration: cfg.WALSampleDuration, + SequenceOffset: cfg.SequenceOffset, WALSampleDuration: cfg.WALSampleDuration, } if err := applyTuningPreflight(cfg, &preflightConfig); err != nil { return err @@ -402,7 +402,7 @@ func (a App) Run(ctx context.Context, cfg config.Config) (runErr error) { SourceDSN: cfg.Source, TargetDSN: cfg.Target, Tables: toPreflight(tables), AcknowledgeWarnings: cfg.AckWarnings, AllowCollationChange: cfg.AllowCollationChange, PGDumpPath: cfg.PGDumpPath, PGRestorePath: cfg.PGRestorePath, - WALSampleDuration: cfg.WALSampleDuration, + SequenceOffset: cfg.SequenceOffset, WALSampleDuration: cfg.WALSampleDuration, } if err := applyTuningPreflight(cfg, &preflightConfig); err != nil { return err @@ -2184,6 +2184,45 @@ func (a App) Verify(ctx context.Context, cfg config.Config) error { return nil } +// Sequences advances the target's sequences without cutting over, so the target +// can accept writes while the source still holds the traffic. Cutover runs the +// same step again against the source's final values. +func (a App) Sequences(ctx context.Context, cfg config.Config) error { + if cfg.SequenceOffset < 0 { + return errors.New("sequence offset must not be negative") + } + store, err := state.OpenControl(ctx, cfg.Dir) + if err != nil { + return err + } + defer store.Close() + if err := validateTargetIdentity(ctx, cfg, store); err != nil { + return err + } + migration, err := store.Migration(ctx) + if err != nil { + return err + } + switch migration.Phase { + case state.PhaseFollow, state.PhaseDrained, state.PhaseCutover, state.PhaseComplete: + default: + return fmt.Errorf("sequences requires follow phase or later, current phase is %s", migration.Phase) + } + schemaSelection, err := loadSchemaSelection(ctx, store) + if err != nil { + return err + } + selected := make([]cutover.Sequence, len(schemaSelection.DependentRelations)) + for i, sequence := range schemaSelection.DependentRelations { + selected[i] = cutover.Sequence{Schema: sequence.Schema, Name: sequence.Name} + } + results, err := cutover.SynchronizeSequences(ctx, connector(cfg.Source), connector(cfg.Target), cfg.SequenceOffset, selected) + if err != nil { + return err + } + return json.NewEncoder(a.output()).Encode(results) +} + func (a App) Cutover(ctx context.Context, cfg config.Config) error { store, err := state.OpenControl(ctx, cfg.Dir) if err != nil { @@ -2254,8 +2293,9 @@ func (a App) Cutover(ctx context.Context, cfg config.Config) error { } report, err := cutover.Run(ctx, cutover.Config{ Source: connector(cfg.Source), Target: connector(cfg.Target), State: store, Dir: cfg.Dir, - WaitDrain: waitDrain, - Sequences: selectedSequences, + WaitDrain: waitDrain, + Sequences: selectedSequences, + SequenceOffset: cfg.SequenceOffset, EmitBoundary: func(ctx context.Context) (string, error) { conn, err := pgx.Connect(ctx, cfg.Source) if err != nil { diff --git a/internal/app/app_test.go b/internal/app/app_test.go index be5dd46..e142ef4 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -386,3 +386,10 @@ func TestCommentSelectionAndLookupUseTheEntryNamespace(t *testing.T) { t.Errorf("materialized view comment lookup = %v, %v", args, err) } } + +func TestSequencesRejectsNegativeOffset(t *testing.T) { + err := App{}.Sequences(context.Background(), config.Config{SequenceOffset: -1}) + if err == nil || !strings.Contains(err.Error(), "must not be negative") { + t.Errorf("Sequences with a negative offset = %v, want a negative-offset error", err) + } +} diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 4117181..3cc37f6 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -51,6 +51,8 @@ func NewRootCommand() *cobra.Command { flags.DurationVar(&cfg.StatusWatch, "watch", 0, "refresh status at this interval") flags.BoolVar(&cfg.NoCleanup, "no-cleanup", false, "retain replication and target metadata") flags.StringVar(&cfg.EndPosition, "endpos", "", "explicit cutover end LSN") + flags.Int64Var(&cfg.SequenceOffset, "sequence-offset", cfg.SequenceOffset, + "values each target sequence is set past the source's, leaving the source room to keep allocating") flags.DurationVar(&cfg.WALSampleDuration, "wal-sample-duration", cfg.WALSampleDuration, "source WAL-rate sample duration") flags.DurationVar(&cfg.SegmentPruneInterval, "segment-prune-interval", cfg.SegmentPruneInterval, "minimum interval between applied CDC segment pruning") flags.BoolVar(&cfg.RetryBaseCopy, "retry-base-copy", false, "restart the base copy even though the last attempts failed the same way") @@ -77,6 +79,7 @@ func NewRootCommand() *cobra.Command { newDatabaseCommand("run", "Start or resume a migration", &cfg, application.Run), newStateCommand("status", "Show migration progress", &cfg, false, application.Status), newStateCommand("verify", "Verify source and target data", &cfg, true, application.Verify), + newStateCommand("sequences", "Advance target sequences past the source", &cfg, true, application.Sequences), newStateCommand("cutover", "Finalize a migration for cutover", &cfg, true, application.Cutover), ) diff --git a/internal/cli/cli_test.go b/internal/cli/cli_test.go new file mode 100644 index 0000000..91910a0 --- /dev/null +++ b/internal/cli/cli_test.go @@ -0,0 +1,27 @@ +package cli + +import ( + "slices" + "strings" + "testing" +) + +func TestSequencesIsItsOwnCommand(t *testing.T) { + root := NewRootCommand() + var names []string + for _, command := range root.Commands() { + names = append(names, command.Name()) + } + if !slices.Contains(names, "sequences") { + t.Fatalf("root commands = %s, want one named sequences", strings.Join(names, ", ")) + } + offset := root.PersistentFlags().Lookup("sequence-offset") + if offset == nil { + t.Fatal("sequence-offset flag is missing") + } + // Cutover leaves the source room to keep allocating, and so must a + // standalone run: a zero default would hand both databases the same values. + if offset.DefValue != "1000000" { + t.Errorf("sequence-offset defaults to %s, want 1000000", offset.DefValue) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 6a4b9ee..07ea8c3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -35,6 +35,7 @@ type Config struct { StatusWatch time.Duration NoCleanup bool EndPosition string + SequenceOffset int64 WALSampleDuration time.Duration SegmentPruneInterval time.Duration RetryBaseCopy bool @@ -118,6 +119,7 @@ func FromEnvironment() Config { RestoreJobs: max(1, runtime.NumCPU()/2), WALSampleDuration: time.Minute, SegmentPruneInterval: time.Minute, + SequenceOffset: 1_000_000, VerifyWorkers: 1, VerifySampleRows: 1_000_000, diff --git a/internal/cutover/cutover.go b/internal/cutover/cutover.go index 1972d86..29cfdfd 100644 --- a/internal/cutover/cutover.go +++ b/internal/cutover/cutover.go @@ -105,9 +105,6 @@ func Run(ctx context.Context, cfg Config) (Report, error) { cfg.WaitDrain == nil || cfg.Cleanup == nil || strings.TrimSpace(cfg.Dir) == "" { return Report{}, errors.New("source, target, state, directory, drain, and cleanup are required") } - if cfg.SequenceOffset == 0 { - cfg.SequenceOffset = 1000 - } if cfg.SequenceOffset < 0 { return Report{}, errors.New("sequence offset must not be negative") } @@ -180,7 +177,7 @@ func Run(ctx context.Context, cfg Config) (Report, error) { } if err := runStep(ctx, cfg.State, stepSequences, func() (string, error) { - sequences, err := synchronizeSequences(ctx, cfg.Source, cfg.Target, cfg.SequenceOffset, cfg.Sequences) + sequences, err := SynchronizeSequences(ctx, cfg.Source, cfg.Target, cfg.SequenceOffset, cfg.Sequences) report.Sequences = sequences data, _ := json.Marshal(sequences) return string(data), err @@ -292,7 +289,11 @@ func runStep(ctx context.Context, store State, name string, action func() (strin return nil } -func synchronizeSequences( +// SynchronizeSequences sets each selected target sequence to the source's value +// plus offset. It is absolute, so rerunning it is safe, and the offset is the +// room the source has left to keep allocating: run it before the source stops +// and anything it allocates beyond the offset collides with the target. +func SynchronizeSequences( ctx context.Context, sourceConnect, targetConnect Connector, offset int64, diff --git a/internal/cutover/cutover_integration_test.go b/internal/cutover/cutover_integration_test.go index f96ffc6..3dff882 100644 --- a/internal/cutover/cutover_integration_test.go +++ b/internal/cutover/cutover_integration_test.go @@ -20,7 +20,7 @@ func TestPostgres17SequenceSynchronization(t *testing.T) { CREATE SCHEMA "odd""schema"; CREATE SEQUENCE "odd""schema"."never called"; CREATE SEQUENCE "odd""schema"."called"; - CREATE SEQUENCE "odd""schema"."descending" INCREMENT BY -1 MINVALUE -10000 MAXVALUE -1 START -1` + CREATE SEQUENCE "odd""schema"."descending" INCREMENT BY -1 MINVALUE -2000000 MAXVALUE -1 START -1` if _, err := source.Exec(ctx, ddl); err != nil { t.Fatal(err) } @@ -44,7 +44,7 @@ func TestPostgres17SequenceSynchronization(t *testing.T) { {Schema: `odd"schema`, Name: "called"}, {Schema: `odd"schema`, Name: "descending"}, } - results, err := synchronizeSequences(ctx, connect(sourceInstance.URI), connect(targetInstance.URI), 1000, selected) + results, err := SynchronizeSequences(ctx, connect(sourceInstance.URI), connect(targetInstance.URI), 1_000_000, selected) if err != nil { t.Fatal(err) } @@ -61,7 +61,7 @@ func TestPostgres17SequenceSynchronization(t *testing.T) { if err := target.QueryRow(ctx, `SELECT nextval('"odd""schema"."descending"')`).Scan(&descending); err != nil { t.Fatal(err) } - if never != 1010 || called != 1021 || descending != -1021 { + if never != 1_000_010 || called != 1_000_021 || descending != -1_000_021 { t.Fatalf("next sequence values = %d, %d, %d", never, called, descending) } } diff --git a/internal/preflight/preflight.go b/internal/preflight/preflight.go index 11617e4..0b20670 100644 --- a/internal/preflight/preflight.go +++ b/internal/preflight/preflight.go @@ -74,6 +74,10 @@ type Config struct { AllowCollationChange bool PGDumpPath string PGRestorePath string + // SequenceOffset is how far past the source's values the cutover will set the + // target's sequences, which is the room each one has to have left for that to + // be possible at all. + SequenceOffset int64 // WALSampleDuration controls the WAL-rate sample. Zero uses one minute. WALSampleDuration time.Duration // WALRetentionDuration is the period of generated WAL that must fit within @@ -150,6 +154,9 @@ func RunConnections(ctx context.Context, source, target *pgx.Conn, cfg Config) ( runCheck(add, "replica-identity", func() ([]Finding, error) { return checkReplicaIdentity(ctx, source, cfg.Tables) }) + runCheck(add, "source-sequences", func() ([]Finding, error) { + return checkSequences(ctx, source, cfg.Tables, cfg.SequenceOffset) + }) runCheck(add, "target-empty", func() ([]Finding, error) { return checkTargetEmpty(ctx, target) }) @@ -449,6 +456,146 @@ func replicaIdentityFindings(relations []replident.Relation) []Finding { return findings } +// sequenceHeadroomWarning is the room a sequence has to have left before +// preflight stops mentioning it. It is generous on purpose: the number that +// matters is how many values the application will consume between now and the +// moment traffic moves, nobody knows that in advance, and a sequence with ten +// million values left is one an operator should have heard about while there was +// still time to widen it. +const sequenceHeadroomWarning = 10_000_000 + +// sequenceState is one source sequence a selected table owns or draws its column +// default from, which is exactly the set the cutover advances. +type sequenceState struct { + OID uint32 + Schema string + Name string + Current int64 + Min int64 + Max int64 + Increment int64 + Cycles bool +} + +func (s sequenceState) descending() bool { return s.Increment < 0 } + +// headroom is how many values the sequence can still hand out before it reaches +// the bound it moves toward. The arithmetic is unsigned because sequence bounds +// span the whole of int64: a sequence sitting at a large negative value under a +// large positive maximum has room to spare, and the signed subtraction of the two +// would overflow and report it as exhausted. +func (s sequenceState) headroom() uint64 { + if s.descending() { + if s.Current <= s.Min { + return 0 + } + return uint64(s.Current) - uint64(s.Min) + } + if s.Current >= s.Max { + return 0 + } + return uint64(s.Max) - uint64(s.Current) +} + +func checkSequences(ctx context.Context, conn *pgx.Conn, tables []Table, offset int64) ([]Finding, error) { + oids := make([]uint32, len(tables)) + for i, table := range tables { + oids[i] = table.OID + } + // The dependency predicate is the one the run uses to build the dump + // selection, so this reports on the sequences the cutover will actually set. + // last_value is null for a sequence never read from, whose next value is its + // start, and for one the migration role cannot read, where the start is the + // best available guess and the privilege checks report the real problem. + rows, err := conn.Query(ctx, ` + SELECT c.oid, s.schemaname, s.sequencename, COALESCE(s.last_value, s.start_value), + s.min_value, s.max_value, s.increment_by, s.cycle + FROM pg_catalog.pg_sequences s + JOIN pg_catalog.pg_namespace n ON n.nspname=s.schemaname + JOIN pg_catalog.pg_class c ON c.relnamespace=n.oid AND c.relname=s.sequencename AND c.relkind='S' + WHERE EXISTS ( + SELECT 1 FROM pg_catalog.pg_depend d + WHERE d.classid='pg_catalog.pg_class'::regclass AND d.objid=c.oid + AND d.refclassid='pg_catalog.pg_class'::regclass AND d.refobjid=ANY($1::oid[]) + ) OR EXISTS ( + SELECT 1 FROM pg_catalog.pg_attrdef ad + JOIN pg_catalog.pg_depend d + ON d.classid='pg_catalog.pg_attrdef'::regclass AND d.objid=ad.oid + WHERE ad.adrelid=ANY($1::oid[]) + AND d.refclassid='pg_catalog.pg_class'::regclass AND d.refobjid=c.oid + ) + ORDER BY s.schemaname,s.sequencename`, oids) + if err != nil { + return nil, err + } + defer rows.Close() + var sequences []sequenceState + for rows.Next() { + var sequence sequenceState + if err := rows.Scan(&sequence.OID, &sequence.Schema, &sequence.Name, &sequence.Current, + &sequence.Min, &sequence.Max, &sequence.Increment, &sequence.Cycles); err != nil { + return nil, err + } + sequences = append(sequences, sequence) + } + if err := rows.Err(); err != nil { + return nil, err + } + return sequenceFindings(sequences, offset), nil +} + +// sequenceFindings reports every sequence too close to exhaustion for the cutover +// to leave the source room to keep allocating. +// +// The severity split follows what PostgreSQL will do. A sequence with less room +// left than the offset is an error: setval rejects a value past the bound, so the +// cutover would fail at its sequence step no matter how willing the operator is. +// Anything else short of the threshold is a warning, because how much room is +// enough depends on how fast the application allocates and how long the source +// keeps serving, which only the operator knows. +func sequenceFindings(sequences []sequenceState, offset int64) []Finding { + var findings []Finding + for _, sequence := range sequences { + room := sequence.headroom() + blocks := offset > 0 && room < uint64(offset) + if !blocks && room >= sequenceHeadroomWarning { + continue + } + qualified := fmt.Sprintf("%s.%s", sequence.Schema, sequence.Name) + bound, wrap := fmt.Sprintf("maximum %d", sequence.Max), sequence.Min + if sequence.descending() { + bound, wrap = fmt.Sprintf("minimum %d", sequence.Min), sequence.Max + } + cycles := "" + if sequence.Cycles { + cycles = fmt.Sprintf(" It cycles, so rather than failing there it wraps to %d and hands "+ + "out values the target already has.", wrap) + } + id := fmt.Sprintf("sequence-%d-headroom", sequence.OID) + if blocks { + findings = append(findings, errorFinding(id, "sequence", fmt.Sprintf( + "%s has %d values left before its %s, fewer than the %d pgmigrate sets the target's "+ + "copy ahead by, so setval would be rejected and the cutover would fail at its "+ + "sequence step.%s Lower --sequence-offset below what is left, or give the sequence "+ + "room with ALTER SEQUENCE, widening the column to bigint where the type is the limit", + qualified, room, bound, offset, cycles, + ))) + continue + } + findings = append(findings, Finding{ + ID: id, Kind: "sequence", Severity: SeverityWarning, + Message: fmt.Sprintf( + "%s has %d values left before its %s. pgmigrate sets the target's copy %d past the "+ + "source's while the source keeps allocating from where it is, so what is left has to "+ + "cover both until traffic moves.%s Widen it with ALTER SEQUENCE, lower "+ + "--sequence-offset, or acknowledge this with --ack-warnings", + qualified, room, bound, offset, cycles, + ), + }) + } + return findings +} + func checkTargetEmpty(ctx context.Context, conn *pgx.Conn) ([]Finding, error) { var count int err := conn.QueryRow(ctx, ` diff --git a/internal/preflight/preflight_integration_test.go b/internal/preflight/preflight_integration_test.go index 685f759..6236c22 100644 --- a/internal/preflight/preflight_integration_test.go +++ b/internal/preflight/preflight_integration_test.go @@ -91,6 +91,97 @@ func roleDSN(t testing.TB, uri, user, password string) string { return parsed.String() } +// TestPG17SequenceHeadroom checks the headroom report against real sequences: a +// sequence with less room left than --sequence-offset stops the migration because +// the cutover's setval would be rejected, one merely running low is +// acknowledgeable, and widening it with the ALTER the finding recommends clears +// it. Sequences no selected table owns or defaults from are none of pgmigrate's +// business and stay unreported. +func TestPG17SequenceHeadroom(t *testing.T) { + source := pgtest.Start(t, 17) + target := pgtest.Start(t, 17) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + conn := source.Connect(t) + for _, statement := range []string{ + "CREATE SEQUENCE low MAXVALUE 6000000", + "CREATE SEQUENCE tight MAXVALUE 2147483647", + "CREATE SEQUENCE unselected MAXVALUE 10", + `CREATE TABLE selected ( + id bigserial PRIMARY KEY, + tag integer DEFAULT nextval('low'), + note integer DEFAULT nextval('tight'))`, + "SELECT setval('low',2000000)", + "SELECT setval('tight',2147000000)", + } { + if _, err := conn.Exec(ctx, statement); err != nil { + t.Fatalf("%s: %v", statement, err) + } + } + oids := make(map[string]uint32, 4) + for _, name := range []string{"selected", "selected_id_seq", "low", "tight", "unselected"} { + var oid uint32 + if err := conn.QueryRow(ctx, "SELECT $1::regclass::oid", name).Scan(&oid); err != nil { + t.Fatalf("resolve %s: %v", name, err) + } + oids[name] = oid + } + + tool := filepath.Join(t.TempDir(), "pg-tool") + if err := os.WriteFile(tool, []byte("#!/bin/sh\necho 'pg_dump (PostgreSQL) 17.1'\n"), 0o700); err != nil { + t.Fatal(err) + } + run := func() (preflight.Result, map[string]preflight.Finding) { + result, err := preflight.Run(ctx, preflight.Config{ + SourceDSN: source.URI, TargetDSN: target.URI, + Tables: []preflight.Table{{OID: oids["selected"], Schema: "public", Name: "selected"}}, + PGDumpPath: tool, PGRestorePath: tool, WALSampleDuration: 10 * time.Millisecond, + SequenceOffset: 1_000_000, AcknowledgeWarnings: true, + }) + if err != nil { + t.Fatalf("preflight: %v", err) + } + byID := make(map[string]preflight.Finding, len(result.Findings)) + for _, finding := range result.Findings { + byID[finding.ID] = finding + } + return result, byID + } + headroom := func(name string) string { + return fmt.Sprintf("sequence-%d-headroom", oids[name]) + } + + result, byID := run() + if result.Allowed { + t.Error("a sequence with less room than the offset was acknowledgeable") + } + if finding := byID[headroom("tight")]; finding.Severity != preflight.SeverityError || + !strings.Contains(finding.Message, "public.tight") { + t.Errorf("tight finding = %+v, want an error naming the sequence", finding) + } + if finding := byID[headroom("low")]; finding.Severity != preflight.SeverityWarning || + !strings.Contains(finding.Message, "4000000 values left") { + t.Errorf("low finding = %+v, want a warning counting what is left", finding) + } + for _, name := range []string{"selected_id_seq", "unselected"} { + if finding, reported := byID[headroom(name)]; reported { + t.Errorf("%s was reported: %+v", name, finding) + } + } + + if _, err := conn.Exec(ctx, "ALTER SEQUENCE tight MAXVALUE 9223372036854775807"); err != nil { + t.Fatal(err) + } + result, byID = run() + if _, reported := byID[headroom("tight")]; reported || !result.Allowed { + t.Errorf("after the recommended ALTER: allowed=%v findings=%+v", result.Allowed, result.Findings) + } + if _, reported := byID[headroom("low")]; !reported { + t.Errorf("the acknowledgeable warning disappeared: %+v", result.Findings) + } +} + func TestPG17MandatoryPreflight(t *testing.T) { source := pgtest.Start(t, 17) target := pgtest.Start(t, 17) diff --git a/internal/preflight/preflight_test.go b/internal/preflight/preflight_test.go index e568e7a..19f1a61 100644 --- a/internal/preflight/preflight_test.go +++ b/internal/preflight/preflight_test.go @@ -2,6 +2,7 @@ package preflight import ( "fmt" + "math" "strings" "testing" "time" @@ -344,6 +345,90 @@ func TestReplicaIdentityFindingsNoLongerOfferTheRemovedFlag(t *testing.T) { } } +// TestSequenceFindingsReportWhatIsLeftBeforeExhaustion pins the severity split. +// A sequence that cannot fit the offset is an error because setval will refuse +// the value, and one merely running low is a warning because only the operator +// knows how fast the application allocates. +func TestSequenceFindingsReportWhatIsLeftBeforeExhaustion(t *testing.T) { + const offset = 1_000_000 + findings := sequenceFindings([]sequenceState{ + {OID: 1, Schema: "app", Name: "roomy", Current: 1000, Max: 1 << 62, Increment: 1}, + {OID: 2, Schema: "app", Name: "low", Current: 2_000_000, Max: 6_000_000, Increment: 1}, + {OID: 3, Schema: "app", Name: "serial4", Current: 2_147_000_000, Max: 2_147_483_647, Increment: 1}, + {OID: 4, Schema: "app", Name: "descending", Current: -90, Min: -1000, Max: -1, Increment: -1}, + {OID: 5, Schema: "app", Name: "cycling", Current: 900, Min: 1, Max: 1000, Increment: 1, Cycles: true}, + }, offset) + byID := make(map[string]Finding, len(findings)) + for _, finding := range findings { + byID[finding.ID] = finding + } + if len(findings) != 4 { + t.Fatalf("findings = %+v, want one per sequence short of room only", findings) + } + if _, reported := byID["sequence-1-headroom"]; reported { + t.Error("a sequence with 2^62 values left was reported") + } + + low := byID["sequence-2-headroom"] + if low.Severity != SeverityWarning || low.Kind != "sequence" { + t.Fatalf("low finding = %+v, want a sequence warning --ack-warnings can consent to", low) + } + // The operator decides by the numbers: what is left, what the bump needs, and + // which knob moves which. + for _, want := range []string{ + "app.low", "4000000 values left", "maximum 6000000", "1000000 past", "--sequence-offset", + "--ack-warnings", "ALTER SEQUENCE", + } { + if !strings.Contains(low.Message, want) { + t.Errorf("warning omits %q:\n%s", want, low.Message) + } + } + + exhausted := byID["sequence-3-headroom"] + if exhausted.Severity != SeverityError { + t.Fatalf("severity with less room than the offset = %q, want error: setval would be rejected", + exhausted.Severity) + } + for _, want := range []string{"483647 values left", "setval would be rejected", "bigint"} { + if !strings.Contains(exhausted.Message, want) { + t.Errorf("error omits %q:\n%s", want, exhausted.Message) + } + } + + // A descending sequence runs out at its minimum, and the offset it is set by + // is negative, so the room that matters is below it rather than above. + descending := byID["sequence-4-headroom"] + if descending.Severity != SeverityError || + !strings.Contains(descending.Message, "910 values left before its minimum -1000") { + t.Errorf("descending finding = %+v", descending) + } + + if cycling := byID["sequence-5-headroom"].Message; !strings.Contains(cycling, "wraps to 1") { + t.Errorf("cycling message does not say it hands out values again:\n%s", cycling) + } +} + +// TestSequenceHeadroomSpansTheWholeOfInt64 covers the bounds a signed +// subtraction cannot: the difference between them overflows int64 and would +// report a sequence with every value still available as exhausted. +func TestSequenceHeadroomSpansTheWholeOfInt64(t *testing.T) { + widest := sequenceState{Current: math.MinInt64, Min: math.MinInt64, Max: math.MaxInt64, Increment: 1} + if room := widest.headroom(); room != math.MaxUint64 { + t.Errorf("headroom() = %d, want %d", room, uint64(math.MaxUint64)) + } + if findings := sequenceFindings([]sequenceState{widest}, math.MaxInt64); len(findings) != 0 { + t.Errorf("findings = %+v, want none for a sequence at its minimum with all of int64 left", findings) + } + spent := sequenceState{Current: math.MaxInt64, Max: math.MaxInt64, Increment: 1} + if room := spent.headroom(); room != 0 { + t.Errorf("exhausted headroom() = %d, want 0", room) + } + descended := sequenceState{Current: math.MinInt64, Min: math.MinInt64, Increment: -1} + if room := descended.headroom(); room != 0 { + t.Errorf("exhausted descending headroom() = %d, want 0", room) + } +} + func TestGateAggregatesSeverityAndAcknowledgements(t *testing.T) { findings := []Finding{ {ID: "info", Severity: SeverityInfo},