diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 1075dbac..46fc942c 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -431,7 +431,7 @@ backend ends in exactly one of three verdicts: | Verdict | What it means | Orchestrator action | | --- | --- | --- | -| `outcome: executed-via-copy-and-swap` | The shadow was built, filled, verified, and swapped in under the source's name; `detail` says how many rows and chunks the verification pass covered, whether the retained source was dropped or kept under its `_old` name, whether the run first reconciled a lost slot, and — for a quiesced run — that the source's writes were fenced for the run's length, since exit 0 otherwise reads as online-safe. A run whose done row is saved is this outcome even when what came after failed — `OnCheckpoint` handed the done row, or the release of the lock session; the detail then names that error. The error is nil | Done. Exit 0 at the CLI; the outcome token is distinct from `executed-natively` so an adapter can tell the two paths apart in its record | +| `outcome: executed-via-copy-and-swap` | The shadow was built, filled, verified, and swapped in under the source's name; `detail` says how many rows and chunks the verification pass covered, whether the retained source was dropped or kept under its `_old` name, whether the run first reconciled a lost slot, and — for a quiesced run — that the source's writes were fenced for the run's length, since exit 0 otherwise reads as online-safe; `writes_blocked: true` says the same for a machine, on this outcome and on a failed run whose fence went on: one that left its fence standing, or that was abandoned after the fence went on. A blocked-writes run that stopped before its fence went on, for example at a table lock another session holds, reports `false`, since the table kept taking writes; its detail says so and that running the statement again starts from the beginning. A run whose done row is saved is this outcome even when what came after failed — `OnCheckpoint` handed the done row, or the release of the lock session; the detail then names that error. The error is nil | Done. Exit 0 at the CLI; the outcome token is distinct from `executed-natively` so an adapter can tell the two paths apart in its record | | `outcome: refused`, `reason: copy-and-swap-refused` | The route's own gates declined: the preflight found a table shape or an environment it will not run against, or a shadow operation refused mid-run to keep an invariant. `cause` is the route's token unchanged — a `copy-and-swap-…` cause from the preflight, a `shadow-…`, `cutover-…`, `resume-…`, or `copy-…` cause from the shadow operations — and `class` is its routing class. The error is nil | Route on `cause` with the two tables above and [refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause). A preflight refusal touched nothing. A refusal raised after the row was saved — a `cutover-…` or `resume-…` cause — leaves the row, the shadow, and the run's hold on the source in place: the write fence of a quiesced run, or the replication slot of a live run, which pins WAL until it is dropped; the cause says whether the next run clears it or the run is abandoned first. A missing privilege is not this reason: it is the same `insufficient-privileges` refusal the native route reports, with the exact `GRANT` in `detail` | | `outcome: failed`, with `code` | The run stopped on an error that is not a refusal; the error is returned alongside the verdict. `code` is `cancelled-by-caller` when the caller's context ended, `invariant-violation` when the engine caught a breach of its own contract, `data-fault` when the server rejected the rows as the statement would shape them — a `*pgconn.PgError` in SQLSTATE class `22` or `23`: a value the new type will not hold, a constraint the copied rows break — and `execution-failed` for everything else — a divergence, an unsettled chunk, a rolled-back swap, a lost lock session, exhausted lock retries, a slot past its lag ceiling | `detail` says what the stop left and what to do with it. A data fault would stop the same way on every resume, so the Runner abandons the run itself before returning (`*schemachange.AbandonedError`): the shadow is dropped, the fence is taken off the source or the slot is dropped, and the checkpoint row is deleted — nothing is held; correct the data or the statement, then run it again without any acknowledgement, and the run starts from the beginning. Every other stop leaves the checkpoint row at its last save with the run's hold on the source — the fence or the slot: a divergence (`*checksum.DivergenceError`) is wedged in `verifying`, and a data fault whose abandon itself failed (the detail says the run "could not be abandoned") is held the same way; for both the detail says to correct the data or the statement and then start the run over with the restart acknowledgement (`--restart SCHEMA.TABLE`; `CopySwapOptions.Restart`), which drops the shadow, takes the fence off the source or drops the slot, deletes the row, and runs the change from the start. Any other stop says that running the same statement again resumes from the row. The `Run` outcome table above is the finer map. Match the wrapped error with `errors.Is`/`errors.As` for the finer routing — the verdict's `code` is the coarse class | diff --git a/internal/cli/color_test.go b/internal/cli/color_test.go index f25fff55..7f95c80e 100644 --- a/internal/cli/color_test.go +++ b/internal/cli/color_test.go @@ -183,6 +183,7 @@ func fullVerdict(t *testing.T) verdict.Verdict { Forced: true, AllowedDestructive: true, BlockingPassthrough: true, + WritesBlocked: true, LockTimeout: "3s", StatementTimeout: "10m", StatisticsRefresh: &verdict.StatisticsRefresh{Columns: []string{"c"}, Pending: `ANALYZE "t"`}, diff --git a/internal/cli/dryrun_text.go b/internal/cli/dryrun_text.go index f052f64f..4726ae16 100644 --- a/internal/cli/dryrun_text.go +++ b/internal/cli/dryrun_text.go @@ -177,7 +177,7 @@ func rewriteRequiredRefusalText(ps plan.Statement) string { // the run needs the free-disk declaration. func writeExecutionNotes(w *stickyWriter, ps plan.Statement, runner string) int { if ps.Backend == router.BackendCopyAndSwap { - w.diag("note", "", runner+" will run it as a copy-and-swap: build a shadow table with the change applied, copy the rows into it while streaming live writes, verify, then swap the two under a brief lock (needs --free-disk; --quiesced when nothing writes the table)") + w.diag("note", "", runner+" will run it as a copy-and-swap: build a shadow table with the change applied, copy the rows into it while streaming live writes, verify, then swap the two under a brief lock (needs --free-disk and logical decoding on the server; --block-writes instead blocks the table's writes until the swap (offline mode))") w.diag("note", "", "the route's preflight — free disk, logical decoding, triggers, dependent views, publication membership — is not part of this dry run; the run reads it before anything executes and refuses with nothing run when it fails") return 1 } diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index 244c01ea..e3cb74ef 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -304,6 +304,7 @@ func TestMigrateRunsTypeRewriteViaCopyAndSwap(t *testing.T) { assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) assert.Equal(t, schema+".t", v.Table) assert.False(t, v.Forced, "the copy-and-swap route is the planned path, not an override") + assert.True(t, v.WritesBlocked, "the JSON verdict says the table was offline for writes while the run held it") var typ string require.NoError(t, pool.QueryRow(t.Context(), diff --git a/internal/cli/verdict_text.go b/internal/cli/verdict_text.go index dc2f48e4..81e2928c 100644 --- a/internal/cli/verdict_text.go +++ b/internal/cli/verdict_text.go @@ -48,6 +48,9 @@ func writeVerdictText(out io.Writer, pal palette, v verdict.Verdict) error { if v.Forced { fmt.Fprintf(&b, "\n %s the submitted form ran as-is (force acknowledged)", pal.bold("forced:")) } + if v.WritesBlocked { + fmt.Fprintf(&b, "\n %s writes to the table were refused (55006) while the run held it", pal.bold("blocked:")) + } if v.AllowedDestructive { fmt.Fprintf(&b, "\n %s the statement's data loss was acknowledged (allow-destructive)", pal.bold("allowed:")) } diff --git a/pkg/migrate/copy_swap.go b/pkg/migrate/copy_swap.go index 21d60494..35f98f14 100644 --- a/pkg/migrate/copy_swap.go +++ b/pkg/migrate/copy_swap.go @@ -213,12 +213,16 @@ func copyAndSwap(ctx context.Context, pool *pgxpool.Pool, st statement.Statement logger.Debug("copy-and-swap preflight passed", "decodes_wal", target.DecodesWAL(), "keep_old", opts.CopySwap.KeepOld) - quiesced := opts.CopySwap.BlockWrites != "" - if quiesced { + hold := holdSlot + if opts.CopySwap.BlockWrites != "" { + hold = holdFence auditBlockWrites(opts.audit(), st) } result, err := runner.Run(ctx, pool, target, st) - return copySwapOutcome(st, quiesced, result, err) + if hold == holdFence && err != nil && result.Checkpoint.Phase != checkpoint.PhaseDone { + hold, err = fenceAfterStop(ctx, store, st, err) + } + return copySwapOutcome(st, hold, result, err) } // copySwapOutcome folds what the Runner returned into Run's contract. A @@ -229,19 +233,20 @@ func copyAndSwap(ctx context.Context, pool *pgxpool.Pool, st statement.Statement // and neither is a failed change, so the verdict says what the engine // tables say and names the error in its detail for the operator to find. // Any other error is folded by copySwapRunVerdict. -func copySwapOutcome(st statement.Statement, quiesced bool, result schemachange.Result, err error) (verdict.Verdict, error) { +func copySwapOutcome(st statement.Statement, hold sourceHold, result schemachange.Result, err error) (verdict.Verdict, error) { if err != nil && result.Checkpoint.Phase != checkpoint.PhaseDone { - return copySwapRunVerdict(st, quiesced, err) + return copySwapRunVerdict(st, hold, err) } - detail := copySwapDetail(result, quiesced) + detail := copySwapDetail(result, hold == holdFence) if err != nil { detail += "; the change is complete and its done row saved, but the run then returned: " + err.Error() } return verdict.Verdict{ - Outcome: verdict.OutcomeExecutedViaCopyAndSwap, - Statement: st.SQL(), - Table: qualified(st), - Detail: detail, + Outcome: verdict.OutcomeExecutedViaCopyAndSwap, + Statement: st.SQL(), + Table: qualified(st), + Detail: detail, + WritesBlocked: hold == holdFence, }, nil } @@ -290,7 +295,7 @@ func classifyCopySwapPreflight(err error) (verdict.Refusal, string, error) { // for every other stop the checkpoint row stays at its last save, and what // the rerun does with it depends on the error and on which hold the run // keeps on the source while it is stopped. -func copySwapRunVerdict(st statement.Statement, quiesced bool, err error) (verdict.Verdict, error) { +func copySwapRunVerdict(st statement.Statement, hold sourceHold, err error) (verdict.Verdict, error) { if cause := schemachange.RefusalCauseOf(err); cause != "" { r, ok := shadowRefusal(cause) if !ok { @@ -300,11 +305,12 @@ func copySwapRunVerdict(st statement.Statement, quiesced bool, err error) (verdi return refusedVerdict(st, r, err.Error()), nil } return verdict.Verdict{ - Outcome: verdict.OutcomeFailed, - Code: string(copySwapFailureCode(err)), - Statement: st.SQL(), - Table: qualified(st), - Detail: "the copy-and-swap run stopped: " + err.Error() + "; " + copySwapStoppedRemedy(quiesced, err), + Outcome: verdict.OutcomeFailed, + Code: string(copySwapFailureCode(err)), + Statement: st.SQL(), + Table: qualified(st), + Detail: "the copy-and-swap run stopped: " + err.Error() + "; " + copySwapStoppedRemedy(hold, err), + WritesBlocked: hold == holdFence, }, err } @@ -325,14 +331,19 @@ func copySwapRunVerdict(st statement.Statement, quiesced bool, err error) (verdi // other stop is resumed by running the statement again. In each case the // detail names the hold the run keeps or kept on the source, since that is // what the operator is paying for in the meantime: the write fence of a -// quiesced run, or the replication slot of a live run, which pins WAL on -// the server until it is dropped. -func copySwapStoppedRemedy(quiesced bool, err error) string { - hold := "its replication slot stays and pins WAL on the server" +// run that blocked writes, or the replication slot of a live run, which +// pins WAL on the server until it is dropped. A blocked-writes run that +// stopped before its fence went on left neither, and saved no row to +// resume from, so the detail says the table kept taking writes. +func copySwapStoppedRemedy(hold sourceHold, err error) string { + if hold == holdNone { + return "the run stopped before it fenced the source, so the table kept taking writes and no checkpoint row was saved; running the same statement again starts from the beginning" + } + kept := "its replication slot stays and pins WAL on the server" release := "drops the replication slot" released := "its replication slot is dropped" - if quiesced { - hold = "the source stays fenced against writes" + if hold == holdFence { + kept = "the source stays fenced against writes" release = "takes the write fence off the source" released = "the write fence is off the source" } @@ -348,7 +359,7 @@ func copySwapStoppedRemedy(quiesced bool, err error) string { if schemachange.IsDataFault(err) { return "the server rejected the rows as the statement would shape them and the run could not be abandoned, so running the same statement again would copy the same rows and stop here again; " + startOver } - return "any checkpoint row it saved remains and " + hold + " until the same statement is run again, which resumes from the row" + return "any checkpoint row it saved remains and " + kept + " until the same statement is run again, which resumes from the row" } // copySwapFailureCode names the executor outcome code for an error out of diff --git a/pkg/migrate/copy_swap_abandon_integration_test.go b/pkg/migrate/copy_swap_abandon_integration_test.go index 214ed16a..03e25595 100644 --- a/pkg/migrate/copy_swap_abandon_integration_test.go +++ b/pkg/migrate/copy_swap_abandon_integration_test.go @@ -34,6 +34,7 @@ func TestRunAbandonsACopyAndSwapRunOnADataFault(t *testing.T) { assert.Equal(t, f.schema+".orders", v.Table) assert.Contains(t, v.Detail, "the run was abandoned") assert.Contains(t, v.Detail, "the write fence is off the source") + assert.True(t, v.WritesBlocked, "the copy the run abandoned ran behind its fence") assert.NotContains(t, v.Detail, "--restart", "nothing remains for a restart to discard") assert.False(t, f.relationExists(t, schemachange.ShadowName(f.schema, "orders")), "the shadow is dropped") diff --git a/pkg/migrate/copy_swap_block_writes.go b/pkg/migrate/copy_swap_block_writes.go index 9d2bc49c..4becca09 100644 --- a/pkg/migrate/copy_swap_block_writes.go +++ b/pkg/migrate/copy_swap_block_writes.go @@ -1,13 +1,69 @@ package migrate import ( + "context" + "errors" "fmt" "log/slog" + "github.com/block/pg-sprite/pkg/checkpoint" + "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/schemachange" "github.com/block/pg-sprite/pkg/statement" ) +// sourceHold is what a copy-and-swap run holds on its source while it is +// stopped short of done, and so what its verdict says the operator is +// paying for until the run is resumed or started over. +type sourceHold int + +const ( + // holdSlot is a live run's replication slot, which pins WAL on the + // server until it is dropped. + holdSlot sourceHold = iota + // holdFence is the write fence of a run that blocked the table's + // writes: every write is refused with SQLSTATE 55006 while it stands. + holdFence + // holdNone is a blocked-writes run that stopped before its fence went + // on: the table kept taking writes and nothing stands on it. + holdNone +) + +// fenceAfterStop reports the hold a blocked-writes run that stopped short +// of done left on the source. It reads what the engine guarantees rather +// than the mode the run was asked for, so a run that failed before it +// fenced the table does not report the table as offline: +// +// - a run abandoned on a data fault had its fence on, since the copy it +// abandoned runs behind the fence; the abandon has taken it off again; +// - a checkpoint row for the table means its fence stands: a fresh run +// fences the source before it writes the row and takes the fence back +// off when the row does not land, and a resumed run stands on a row +// whose fence the stopped run left; +// - no row means the run stopped before its fence went on. +// +// The read outlives the caller's context, since a cancelled run is a usual +// stop, but is bounded. When it fails, the hold is reported as the fence, +// the claim that sends the operator to look rather than one that tells +// them the table is untouched, and the read's error joins the run's. +func fenceAfterStop(ctx context.Context, store *checkpoint.Store, st statement.Statement, runErr error) (sourceHold, error) { + var abandoned *schemachange.AbandonedError + if errors.As(runErr, &abandoned) { + return holdFence, runErr + } + readCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), dbconn.DefaultConnectTimeout) + defer cancel() + _, err := store.Inspect(readCtx, st.Schema(), st.Table()) + switch { + case err == nil: + return holdFence, runErr + case errors.Is(err, checkpoint.ErrNotFound), errors.Is(err, checkpoint.ErrTableMissing): + return holdNone, runErr + default: + return holdFence, errors.Join(runErr, fmt.Errorf("read whether the stopped run fenced %s: %w", qualified(st), err)) + } +} + // checkBlockWritesAck confirms that the block-writes acknowledgement names the // resolved target table, exactly as the force and restart acknowledgements // must. A block-writes run refuses every write to the table for its whole diff --git a/pkg/migrate/copy_swap_block_writes_integration_test.go b/pkg/migrate/copy_swap_block_writes_integration_test.go index 3b66b18d..fe1c6c3e 100644 --- a/pkg/migrate/copy_swap_block_writes_integration_test.go +++ b/pkg/migrate/copy_swap_block_writes_integration_test.go @@ -1,11 +1,15 @@ package migrate import ( + "context" + "errors" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/block/pg-sprite/pkg/checkpoint" + "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/schemachange" "github.com/block/pg-sprite/pkg/statement" "github.com/block/pg-sprite/pkg/verdict" @@ -32,3 +36,31 @@ func TestRunRejectsAMismatchedBlockWritesAck(t *testing.T) { f.exec(t, `INSERT INTO {schema}.orders (qty, note, sku) VALUES (1, 'after', 'sku1')`) assert.Equal(t, "integer", f.columnType(t, "orders", "qty"), "nothing was executed") } + +// A blocked-writes run that stops before its fence goes on does not report +// the table as offline. Another session holds the table's lock, so the run +// passes its preflight and then stops at the lock, before it builds a +// shadow or fences the source. The verdict is a failure whose +// writes_blocked is false, because the table kept taking writes, and no +// checkpoint row is left to resume from. +func TestBlockWritesRunStoppedBeforeItsFenceDoesNotReportWritesBlocked(t *testing.T) { + f := newCopySwapFixture(t) + f.createOrders(t) + holder, err := dbconn.AcquireTableLock(t.Context(), f.cfg, f.schema, "orders") + require.NoError(t, err) + t.Cleanup(func() { assert.NoError(t, holder.Release(context.WithoutCancel(t.Context()))) }) + + v, err := f.run(t, `ALTER TABLE {schema}.orders ALTER COLUMN qty TYPE bigint`, f.options()) + + var held *dbconn.TableLockHeldError + require.ErrorAs(t, err, &held, "the run stops at the lock another session holds") + assert.Equal(t, verdict.OutcomeFailed, v.Outcome) + assert.False(t, v.WritesBlocked, "the run stopped before its fence went on") + store, err := checkpoint.NewStore(f.pool, checkpoint.Options{}) + require.NoError(t, err) + _, err = store.Inspect(t.Context(), f.schema, "orders") + assert.True(t, errors.Is(err, checkpoint.ErrNotFound) || errors.Is(err, checkpoint.ErrTableMissing), + "no checkpoint row was saved: %v", err) + assert.False(t, f.relationExists(t, schemachange.ShadowName(f.schema, "orders")), "no shadow was built") + f.exec(t, `INSERT INTO {schema}.orders (qty, note, sku) VALUES (1, 'during', 'sku1')`) +} diff --git a/pkg/migrate/copy_swap_integration_test.go b/pkg/migrate/copy_swap_integration_test.go index d6432af5..8dce823c 100644 --- a/pkg/migrate/copy_swap_integration_test.go +++ b/pkg/migrate/copy_swap_integration_test.go @@ -340,6 +340,7 @@ func TestCopyAndSwapStopsWhereTheHostSaysAndResumes(t *testing.T) { assert.Equal(t, verdict.OutcomeFailed, v.Outcome) assert.Equal(t, string(executor.CodeExecutionFailed), v.Code) assert.Contains(t, v.Detail, "resumes from the row") + assert.True(t, v.WritesBlocked, "the stopped run left its fence on the source") assert.Equal(t, checkpoint.PhaseVerifying, f.inspectRow(t).Phase, "the row stands where the host stopped") assert.True(t, f.hasColumn(t, "orders", "note"), "the source is untouched while the run is stopped") diff --git a/pkg/migrate/copy_swap_test.go b/pkg/migrate/copy_swap_test.go index 36a5974a..91637d95 100644 --- a/pkg/migrate/copy_swap_test.go +++ b/pkg/migrate/copy_swap_test.go @@ -92,7 +92,7 @@ func TestCopySwapRunVerdict(t *testing.T) { t.Run("a shadow refusal is refused with its cause under the class it declares", func(t *testing.T) { refusal := &schemachange.RefusalError{Cause: schemachange.CauseNameTaken, Detail: "a relation wears the shadow's name"} - v, err := copySwapRunVerdict(st, false, fmt.Errorf("run billing.orders: %w", refusal)) + v, err := copySwapRunVerdict(st, holdSlot, fmt.Errorf("run billing.orders: %w", refusal)) require.NoError(t, err) assert.Equal(t, verdict.OutcomeRefused, v.Outcome) assert.Equal(t, verdict.ReasonCopyAndSwapRefused, v.Reason) @@ -104,7 +104,7 @@ func TestCopySwapRunVerdict(t *testing.T) { t.Run("a divergence is a failure whose remedy is to start the run over", func(t *testing.T) { divergence := &checksum.DivergenceError{Report: checksum.Report{Chunks: 3, Mismatches: []checksum.Mismatch{{}}}} runErr := fmt.Errorf("run billing.orders: %w", divergence) - v, err := copySwapRunVerdict(st, false, runErr) + v, err := copySwapRunVerdict(st, holdSlot, runErr) assert.ErrorIs(t, err, runErr) assert.Equal(t, verdict.OutcomeFailed, v.Outcome) assert.Equal(t, string(executor.CodeExecutionFailed), v.Code) @@ -112,13 +112,15 @@ func TestCopySwapRunVerdict(t *testing.T) { assert.Contains(t, v.Detail, "start the run over") assert.Contains(t, v.Detail, "drops the replication slot", "a live run's hold is its slot") assert.NotContains(t, v.Detail, "resumes") + assert.False(t, v.WritesBlocked, "a live run never took the table offline for writes") }) t.Run("a quiesced divergence names the fence as the hold to release", func(t *testing.T) { divergence := &checksum.DivergenceError{Report: checksum.Report{Chunks: 1, Mismatches: []checksum.Mismatch{{}}}} - v, _ := copySwapRunVerdict(st, true, divergence) + v, _ := copySwapRunVerdict(st, holdFence, divergence) assert.Contains(t, v.Detail, "takes the write fence off the source") assert.NotContains(t, v.Detail, "replication slot") + assert.True(t, v.WritesBlocked, "the failed verdict says the run took the table offline for writes") }) t.Run("a value the new type will not hold is an abandoned run coded as a data fault", func(t *testing.T) { @@ -130,7 +132,7 @@ func TestCopySwapRunVerdict(t *testing.T) { // --restart to discard. tooLong := &pgconn.PgError{Code: "22001", Message: "value too long for type character varying(10)"} runErr := &schemachange.AbandonedError{Schema: "billing", Table: "orders", Fault: fmt.Errorf("copy chunk: %w", tooLong)} - v, err := copySwapRunVerdict(st, true, runErr) + v, err := copySwapRunVerdict(st, holdFence, runErr) assert.ErrorIs(t, err, runErr) assert.Equal(t, verdict.OutcomeFailed, v.Outcome) assert.Equal(t, string(executor.CodeDataFault), v.Code) @@ -146,7 +148,7 @@ func TestCopySwapRunVerdict(t *testing.T) { // default fills from NULL on the first chunk. notNull := &pgconn.PgError{Code: "23502", Message: "null value in column violates not-null constraint"} runErr := &schemachange.AbandonedError{Schema: "billing", Table: "orders", Fault: fmt.Errorf("copy chunk: %w", notNull)} - v, _ := copySwapRunVerdict(st, false, runErr) + v, _ := copySwapRunVerdict(st, holdSlot, runErr) assert.Equal(t, string(executor.CodeDataFault), v.Code) assert.Contains(t, v.Detail, "the run was abandoned") assert.Contains(t, v.Detail, "its replication slot is dropped", "a live run's hold was its slot") @@ -160,7 +162,7 @@ func TestCopySwapRunVerdict(t *testing.T) { // names the fault, since that is what the operator corrects. tooLong := &pgconn.PgError{Code: "22001", Message: "value too long for type character varying(10)"} runErr := errors.Join(fmt.Errorf("copy chunk: %w", tooLong), errors.New("abandon run on billing.orders: drop shadow: connection reset")) - v, err := copySwapRunVerdict(st, true, runErr) + v, err := copySwapRunVerdict(st, holdFence, runErr) assert.ErrorIs(t, err, runErr) assert.Equal(t, string(executor.CodeDataFault), v.Code) assert.Contains(t, v.Detail, "could not be abandoned") @@ -173,13 +175,13 @@ func TestCopySwapRunVerdict(t *testing.T) { // 57P01 admin_shutdown: the server went away under the copy; the // rows are not at fault and the next run carries on from the row. shutdown := &pgconn.PgError{Code: "57P01", Message: "terminating connection due to administrator command"} - v, _ := copySwapRunVerdict(st, false, fmt.Errorf("copy chunk: %w", shutdown)) + v, _ := copySwapRunVerdict(st, holdSlot, fmt.Errorf("copy chunk: %w", shutdown)) assert.Contains(t, v.Detail, "resumes from the row") assert.NotContains(t, v.Detail, "abandon") }) t.Run("a stop a rerun clears says the slot stays until it resumes", func(t *testing.T) { - v, err := copySwapRunVerdict(st, false, fmt.Errorf("copy: %w", context.Canceled)) + v, err := copySwapRunVerdict(st, holdSlot, fmt.Errorf("copy: %w", context.Canceled)) require.Error(t, err) assert.Contains(t, v.Detail, "replication slot stays") assert.Contains(t, v.Detail, "resumes from the row") @@ -187,23 +189,32 @@ func TestCopySwapRunVerdict(t *testing.T) { }) t.Run("a quiesced stop a rerun clears says the fence stays until it resumes", func(t *testing.T) { - v, _ := copySwapRunVerdict(st, true, fmt.Errorf("copy: %w", context.Canceled)) + v, _ := copySwapRunVerdict(st, holdFence, fmt.Errorf("copy: %w", context.Canceled)) assert.Contains(t, v.Detail, "fenced against writes") assert.Contains(t, v.Detail, "resumes from the row") }) + t.Run("a blocked-writes run stopped before its fence names no hold and resumes nothing", func(t *testing.T) { + v, _ := copySwapRunVerdict(st, holdNone, fmt.Errorf("acquire table lock: %w", context.Canceled)) + assert.False(t, v.WritesBlocked, "the table kept taking writes") + assert.Contains(t, v.Detail, "kept taking writes") + assert.Contains(t, v.Detail, "starts from the beginning") + assert.NotContains(t, v.Detail, "fenced against writes") + assert.NotContains(t, v.Detail, "resumes from the row") + }) + t.Run("a cancelled run is coded as cancelled by the caller", func(t *testing.T) { - v, _ := copySwapRunVerdict(st, false, fmt.Errorf("copy: %w", context.Canceled)) + v, _ := copySwapRunVerdict(st, holdSlot, fmt.Errorf("copy: %w", context.Canceled)) assert.Equal(t, string(executor.CodeCancelledByCaller), v.Code) }) t.Run("a breach of the shadow's invariants is coded as an invariant violation", func(t *testing.T) { - v, _ := copySwapRunVerdict(st, false, fmt.Errorf("gate: %w", schemachange.ErrInvariantViolation)) + v, _ := copySwapRunVerdict(st, holdSlot, fmt.Errorf("gate: %w", schemachange.ErrInvariantViolation)) assert.Equal(t, string(executor.CodeInvariantViolation), v.Code) }) t.Run("a rolled-back swap is an execution failure", func(t *testing.T) { - v, _ := copySwapRunVerdict(st, false, schemachange.ErrCutoverRolledBack) + v, _ := copySwapRunVerdict(st, holdSlot, schemachange.ErrCutoverRolledBack) assert.Equal(t, string(executor.CodeExecutionFailed), v.Code) }) } @@ -225,7 +236,7 @@ func TestCopySwapOutcomeTreatsADoneRowAsExecuted(t *testing.T) { Verification: checksum.Report{Rows: 7, Chunks: 1}, OldTableDropped: true, } - v, err := copySwapOutcome(st, false, done, releaseErr) + v, err := copySwapOutcome(st, holdSlot, done, releaseErr) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) assert.Empty(t, v.Code) @@ -235,14 +246,23 @@ func TestCopySwapOutcomeTreatsADoneRowAsExecuted(t *testing.T) { t.Run("a done row without an error carries no release wording", func(t *testing.T) { done := schemachange.Result{Checkpoint: checkpoint.Checkpoint{Phase: checkpoint.PhaseDone}} - v, err := copySwapOutcome(st, false, done, nil) + v, err := copySwapOutcome(st, holdSlot, done, nil) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) assert.NotContains(t, v.Detail, "the run then returned") + assert.False(t, v.WritesBlocked, "a live run's verdict does not claim the table was offline") + }) + + t.Run("a block-writes done row is marked writes-blocked", func(t *testing.T) { + done := schemachange.Result{Checkpoint: checkpoint.Checkpoint{Phase: checkpoint.PhaseDone}} + v, err := copySwapOutcome(st, holdFence, done, nil) + require.NoError(t, err) + assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) + assert.True(t, v.WritesBlocked, "the executed verdict says the table was offline for writes while the run held it") }) t.Run("an error with the zero result is a failure", func(t *testing.T) { - v, err := copySwapOutcome(st, false, schemachange.Result{}, releaseErr) + v, err := copySwapOutcome(st, holdSlot, schemachange.Result{}, releaseErr) assert.ErrorIs(t, err, releaseErr) assert.Equal(t, verdict.OutcomeFailed, v.Outcome) }) diff --git a/pkg/verdict/verdict.go b/pkg/verdict/verdict.go index eaaa3142..4a07ec5b 100644 --- a/pkg/verdict/verdict.go +++ b/pkg/verdict/verdict.go @@ -535,6 +535,15 @@ type Verdict struct { Table string `json:"table,omitempty"` // Detail is the human explanation: why refused, or what committed. Detail string `json:"detail,omitempty"` + // WritesBlocked is true on an executed or failed copy-and-swap verdict + // whose run blocked the table's writes (offline mode, --block-writes): + // the write fence went on, and every write that reached it was refused + // with SQLSTATE 55006. On a failed verdict it is set only when the fence + // went on, and Detail says whether it still stands; a blocked-writes run + // that stopped before its fence went on reports false, since the table + // kept taking writes. False for every other verdict, including a live + // run's. + WritesBlocked bool `json:"writes_blocked,omitempty"` // SaferIdiom is a native alternative to the refused statement, when one // exists (e.g. CREATE INDEX CONCURRENTLY, ADD CONSTRAINT ... NOT VALID). SaferIdiom string `json:"safer_idiom,omitempty"` @@ -654,6 +663,9 @@ func (v Verdict) String() string { if v.Forced { b.WriteString("\n forced: the submitted form ran as-is (force acknowledged)") } + if v.WritesBlocked { + b.WriteString("\n blocked: writes to the table were refused (55006) while the run held it") + } if v.AllowedDestructive { b.WriteString("\n allowed: the statement's data loss was acknowledged (allow-destructive)") }