verdict: mark a copy-and-swap run that blocked writes

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 |
 

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"`},

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
 	}

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(),

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

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

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

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

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')`)
+}

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