copier, applier: a cancel with a cause still reads as a cancellation

docs/schemabot-integration.md

@@ -340,7 +340,7 @@ the slot under the row (decoding target):
 | `ErrCutoverRolledBack` / `ErrLockRetriesExhausted` | The swap did not commit; the source is still live and the row is in `cutover` | Retry `Run`: the row is answered by `InspectSwapped`, which sends the run back through verify and gate to a fresh swap attempt |
 | `ErrTableLockLost` (`shadow-lock-lost`) | The lock session dropped mid-run; whatever was in flight was cancelled with it | Retry `Run` once the database is reachable; the row says where to continue |
 | An error wrapped `after the checkpoint in phase …` | The row saved; `AfterCheckpoint` returned an error after it | The row and the run are consistent; retry `Run` when the hook's cause is cleared. A hook that fails at the first save leaves the shadow, since the row records it. A hook that fails at the done save stopped nothing: `Run` returns the completed `Result` beside the error, with `Result.Checkpoint.Phase` in `PhaseDone`, and a retry would start a new change |
-| `context.Canceled` / `DeadlineExceeded` | The caller stopped the run; a cancelled copy saves what its committed chunks landed before returning | Retry `Run`; the copy continues from the saved watermark |
+| `context.Canceled` / `DeadlineExceeded` | The caller stopped the run; a cancelled copy saves what its committed chunks landed before returning. A caller that cancelled with a cause of its own (`context.WithCancelCause`) gets both: the error answers `errors.Is` for the cause and for `context.Canceled` | Retry `Run`; the copy continues from the saved watermark |
 
 A refusal from either read leaves the row and the slot as they were. A `Reconcile` that fails
 after its marker leaves the row in `reconciling`, and `InspectResume` answers that row as
@@ -444,7 +444,7 @@ backend ends in exactly one of three verdicts:
 | --- | --- | --- |
 | `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 run that blocked writes (`--block-writes`) — 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 run that blocked writes, 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 — `pool-too-small` when the pool holds fewer than `migrate.CopySwapMinConns(Copier)` connections — refused before the preflight with nothing created, the error matching `errors.Is(err, executor.ErrPoolTooSmall)`; resize the pool rather than retry — 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 |
+| `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 — whatever cause the caller cancelled it with: the code says only that the context ended, so an adapter that cancels with reasons of its own (a lost lease, a failed health gate) tells them apart with `errors.Is(err, <its cause>)` on the returned error, `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 — `pool-too-small` when the pool holds fewer than `migrate.CopySwapMinConns(Copier)` connections — refused before the preflight with nothing created, the error matching `errors.Is(err, executor.ErrPoolTooSmall)`; resize the pool rather than retry — 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 |
 
 The adapter supplies the route's policy on `migrate.Options.CopySwap`: the connection the
 pool was dialed with (the route opens dedicated sessions — the table lock, the copy workers,

internal/cli/signal.go

@@ -14,10 +14,12 @@ import (
 const exitCodeInterrupted = 128 + int(syscall.SIGINT)
 
 // errInterrupted is the cancellation cause recorded when an operator's
-// signal stops a run. It wraps context.Canceled, so a caller that reads
-// the cause (context.Cause) sees the same ordinary cancellation as one
-// that reads ctx.Err(), and the run's verdict is cancelled-by-caller
-// whichever phase the signal lands in.
+// signal stops a run. The copier and the catch-up report any cause as a
+// cancellation, so the run's verdict is cancelled-by-caller whichever
+// phase the signal lands in. Wrapping context.Canceled as well keeps the
+// cause readable as a cancellation to code that reads it directly
+// (context.Cause), and lets the engine report it as it stands rather than
+// appending the cancellation a second time.
 var errInterrupted = fmt.Errorf("interrupted by signal: %w", context.Canceled)
 
 // Interruptible derives a context that an operator's SIGINT or SIGTERM

internal/cli/signal_integration_test.go

@@ -25,8 +25,8 @@ const copyStartedDeadline = 30 * time.Second
 
 // An operator's first signal while the copy is running stops the run
 // through its own stop path and reports the cancellation as the caller's:
-// the copier returns the context's cause, and the cause must still read
-// as a cancellation, or the verdict says the run broke.
+// the copier's error carries the signal's cause and reads as a
+// cancellation, or the verdict says the run broke.
 func TestMigrateSignalDuringCopyReportsCancelledByCaller(t *testing.T) {
 	url := testutil.StartPostgres(t)
 	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})

pkg/applier/catchup.go

@@ -249,10 +249,13 @@ func (c *Catchup) Confirmed() decode.LSN {
 //
 // Run never returns nil: a catch-up has no end of its own, so its owner
 // stops it by ending ctx once the shadow is close enough to cut over, and
-// Run returns the cause ctx ended with, wrapped, once the stream reports
-// it. A stopped catch-up leaves the stream failed and the slot at the last
-// confirmed position, which is where the next stream resumes. A cycle
-// error that is not the stop itself travels with the stop, so a caller
+// Run returns the stop, wrapped, once the stream reports it. The stop
+// answers errors.Is for ctx.Err() (context.Canceled or
+// context.DeadlineExceeded), and also for the cause when ctx was cancelled
+// with one (context.WithCancelCause). A stopped catch-up leaves the stream
+// failed and the slot at the last confirmed position, which is where the
+// next stream resumes. A cycle error that is not the stop itself travels
+// with the stop, so a caller
 // checks the failure sentinels — ErrInvariantViolation, ErrTableLockLost,
 // ErrBatchDeferred — before it reads the stop as clean. The whole run is
 // bound to the table lock session, so losing the lock ends the cycle in
@@ -308,13 +311,33 @@ func stopped(ctx context.Context, schema, table string, err error) error {
 	if ctx.Err() == nil {
 		return err
 	}
-	stop := fmt.Errorf("catch-up on %s.%s stopped: %w", schema, table, context.Cause(ctx))
+	stop := fmt.Errorf("catch-up on %s.%s stopped: %w", schema, table, endedWith(ctx))
 	if errors.Is(err, ctx.Err()) {
 		return stop
 	}
 	return fmt.Errorf("%w; the cycle it stopped in failed: %w", stop, err)
 }
 
+// endedWith is why an ended ctx ended. A caller that cancelled with a cause
+// of its own (context.WithCancelCause) gets that cause wrapped together
+// with ctx.Err(), so the error reads as the cause and as a cancellation
+// (errors.Is(err, context.Canceled)) alike; a caller that cancelled
+// plainly, or whose deadline passed, gets ctx.Err() as it stands, and so
+// does a cause that already wraps it.
+//
+// It reads only that ctx ended, whoever ended it: a cancellation with a
+// cause always reads as a stop. An owner that cancels a context it derived
+// for this package because something failed must report that failure
+// itself, as the table lock session's loss and the runner's capture and
+// save failures are; otherwise the failure reads as the caller's stop.
+func endedWith(ctx context.Context) error {
+	cause := context.Cause(ctx)
+	if errors.Is(cause, ctx.Err()) {
+		return cause
+	}
+	return fmt.Errorf("%w (%w)", cause, ctx.Err())
+}
+
 // cycle is one gather, drain, flush, and confirm. Its status is recorded
 // once, at the end, so a poll never pairs this cycle's counters with the
 // last cycle's positions.

pkg/applier/catchup_test.go

@@ -94,14 +94,42 @@ func TestStoppedKeepsACycleErrorThatIsNotTheStop(t *testing.T) {
 	receive := fmt.Errorf("receive from slot: %w", context.Canceled)
 	err := stopped(ctx, "s", "t", receive)
 	assert.ErrorIs(t, err, cause)
-	assert.EqualError(t, err, "catch-up on s.t stopped: owner is ready to cut over", "the receive error is the stop itself")
+	assert.ErrorIs(t, err, context.Canceled, "a stop with the owner's own cause is still a cancellation")
+	assert.EqualError(t, err, "catch-up on s.t stopped: owner is ready to cut over (context canceled)", "the receive error is the stop itself")
 
 	err = stopped(ctx, "s", "t", breach)
 	assert.ErrorIs(t, err, cause, "the stop is still reported")
 	assert.ErrorIs(t, err, ErrInvariantViolation, "the breach travels with it")
 	assert.ErrorIs(t, err, breach)
 }
 
+// An ended ctx is reported by its cause; a cause the owner chose is wrapped
+// with ctx.Err() so the error is also the cancellation it is, while a plain
+// cancel or a passed deadline is reported as it stands, not doubled.
+func TestEndedWithKeepsACancellationReadableAsOne(t *testing.T) {
+	plain, cancel := context.WithCancel(t.Context())
+	cancel()
+	assert.Equal(t, context.Canceled, endedWith(plain))
+
+	expired, cancelExpired := context.WithDeadline(t.Context(), time.Unix(0, 0))
+	defer cancelExpired()
+	<-expired.Done()
+	assert.Equal(t, context.DeadlineExceeded, endedWith(expired))
+
+	caused, cancelCaused := context.WithCancelCause(t.Context())
+	cause := errors.New("owner is ready to cut over")
+	cancelCaused(cause)
+	err := endedWith(caused)
+	assert.ErrorIs(t, err, cause)
+	assert.ErrorIs(t, err, context.Canceled)
+	assert.EqualError(t, err, "owner is ready to cut over (context canceled)")
+
+	interrupted := fmt.Errorf("interrupted by signal: %w", context.Canceled)
+	wrapped, cancelWrapped := context.WithCancelCause(t.Context())
+	cancelWrapped(interrupted)
+	assert.Same(t, interrupted, endedWith(wrapped), "a cause that is already a cancellation is reported as it stands")
+}
+
 // The copy has landed when its cut frontier is complete and no claimed
 // chunk is unlanded; a complete frontier with a chunk in flight, or a
 // frontier short of the end, still has a chunk that can resolve a refusal.

pkg/applier/pause.go

@@ -20,10 +20,12 @@ import (
 // that waits for the catch-up to confirm past its own work resumes first,
 // since the confirmation needs the flushes. Pause waits for a flush in
 // flight to commit or fail, or for ctx to end; a second Pause waits for the
-// first's resume; resume is safe to call more than once. Pause before Run
-// takes the hold at once, and a Run started under it flushes nothing until
-// resume. Status reports the hold as FlushesHeld from Pause's return until
-// the resume.
+// first's resume; resume is safe to call more than once. A Pause whose ctx
+// ends first takes no hold and returns an error that answers errors.Is for
+// ctx.Err(), and also for the cause when ctx was cancelled with one. Pause
+// before Run takes the hold at once, and a Run started under it flushes
+// nothing until resume. Status reports the hold as FlushesHeld from Pause's
+// return until the resume.
 func (c *Catchup) Pause(ctx context.Context) (resume func(), err error) {
 	if err := c.holdFlushes(ctx); err != nil {
 		return nil, err
@@ -61,7 +63,7 @@ func (c *Catchup) holdFlushes(ctx context.Context) error {
 	case <-c.flushes:
 		return nil
 	case <-ctx.Done():
-		return fmt.Errorf("pause the catch-up on %s.%s: %w", c.flusher.target.Schema(), c.flusher.target.Table(), context.Cause(ctx))
+		return fmt.Errorf("pause the catch-up on %s.%s: %w", c.flusher.target.Schema(), c.flusher.target.Table(), endedWith(ctx))
 	}
 }
 

pkg/applier/pause_test.go

@@ -2,6 +2,7 @@ package applier
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -31,3 +32,22 @@ func TestCatchupPauseTakesAFreeGateOnAnEndedContext(t *testing.T) {
 		c.releaseFlushes()
 	}
 }
+
+// A Pause that waits behind a flush in flight and whose context is cancelled
+// with a cause takes no hold and returns an error that carries the cause
+// and still reads as a cancellation, so a live repair stopped this way is
+// reported as the caller's stop.
+func TestCatchupPauseCancelledWithACauseReadsAsCancelled(t *testing.T) {
+	c := &Catchup{flushes: newFlushGate(), flusher: &Flusher{}}
+	require.True(t, c.tryFlushes(), "a flush in flight holds the gate")
+	ctx, cancel := context.WithCancelCause(t.Context())
+	ownerStop := errors.New("owner is ready to cut over")
+	cancel(ownerStop)
+
+	_, err := c.Pause(ctx)
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ownerStop, "the owner's cause is carried")
+	assert.ErrorIs(t, err, context.Canceled, "a cancel with a cause is still a cancellation")
+	assert.False(t, c.Status().FlushesHeld)
+}