reconcile: clear the shadow above a partial watermark before the recopy pass

docs/invariants.md

@@ -202,7 +202,8 @@ transaction; chunks are cut and registered under one lock, and the ledger refuse
 does not start just above the frontier (`copy-frontier-broken`, a typed `copier.RefusalCause`),
 so the frontier never runs ahead of an unregistered chunk; a chunk is registered in flight before its transaction begins and leaves the in-flight set
 only when it commits — a chunk whose transaction did not commit stays in flight, so its keys
-never read as landed; a resumed copy first deletes every shadow row above W in bounded batches,
+never read as landed; a resumed copy, and a reconcile before its recopy pass over a partial
+watermark (`copier.ClearAbove`, ST-4), first deletes every shadow row above W in bounded batches,
 each in its own guarded transaction, before its first chunk is cut — the whole shadow when
 nothing landed, since a zero watermark says nothing about what an earlier run left above it — so
 every key above W is genuinely uncut when the applier starts discarding for it; the first clear
@@ -1098,7 +1099,9 @@ dies restarts as a reconcile, and the slot it is about to make is owned by the r
 moment it exists (ST-3) — replaces whatever slot bears the derived name with a fresh one through
 `decode.ReplaceSlot`, keeping the publication in place, since the source's writers look it up as
 they run and one dropped under them fails the write, and
-— when the watermark says anything landed — runs one `checksum` pass at or below the watermark
+— when the watermark says anything landed — clears the shadow above the watermark through
+`copier.ClearAbove`, as the copier that resumes from it would first (CO-4), so no shadow row
+lies outside the pass that follows, then runs one `checksum` pass at or below the watermark
 under the `recopy` policy (D14), recopying only the chunks the gap touched without a reread,
 since the source is live and the new slot carries every write that lands after the recopy,
 then writes the row back to `copying` or `catching_up` by the watermark with the new slot's
@@ -1112,21 +1115,26 @@ refuses-replaced-relations-before-its-marker,
 leaves-the-slot-it-created-owned-by-the-row, resumes-in-copying-below-a-partial-watermark,
 refuses-a-row-of-another-source-shape, stops-when-the-lock-is-lost-while-the-drop-waits,
 refuses-a-foreign-slot-of-the-name, refuses-the-slot-while-another-backend-holds-it,
-reads-a-slot-the-server-reports-lost-as-reconcile, and
-survives-a-unique-value-moved-during-its-recopy tests). The recopy pass compares a chunk and
+reads-a-slot-the-server-reports-lost-as-reconcile,
+survives-a-unique-value-moved-during-its-recopy, and
+clears-the-shadow-above-a-partial-watermark-before-its-recopy tests). The recopy pass compares a chunk and
 recopies it in separate statements against a live source, so a source that moves a unique
 value from a row in a chunk the compare read equal to a row in a differing chunk, between the
 two, makes the recopy collide with a shadow row the compare left alone; the one recopy
 transaction rolls back, the pass compares again, and the recopy cut from that comparison
 covers both chunks and lands (CO-3). A source that moves a value into the differing chunks
 before every recopy exhausts the pass's bounded attempts; the pass then ends in the last
-collision's error with the row still in `reconciling` and the shadow unchanged, and the next
-attempt restarts the reconcile — closed, never a shadow with a row the source lacks. *Known
-limit:* on resume the reconcile runs before the copier clears the shadow rows above a partial
-watermark, and the pass compares only through that watermark, so a unique value the source
-moves out of a shadow row above it into a differing chunk collides on every attempt of every
-reconcile until the source moves it back; the run stays in `reconciling`, closed, and
-`--restart` (`CopySwapOptions.Restart`) starts it over. *Source:*
+collision's error with the row still in `reconciling`, no recopy committed, and the watermark
+unmoved, and the next attempt restarts the reconcile — closed, never a shadow with a row the
+source lacks. The pass compares only through the watermark, which is why the reconcile clears
+the shadow above a partial watermark before it, and before any slot command so a new slot
+retains WAL only while the pass runs: a unique value the source moved out of a shadow row
+above the watermark into a differing chunk below it would otherwise collide with a row no
+comparison can reach, on every attempt of every reconcile, until the source moved it back. A
+clear that fails, a lock timeout behind a writer still holding the shadow included, ends the
+attempt before the marker with the row and its watermark as they were
+(retries-after-the-clear-fence-times-out test); with nothing landed there is no pass and no
+clear (with-nothing-landed-leaves-the-shadow-alone test). *Source:*
 [low-level-design § failover](low-level-design.md#failover-during-migration-what-survives-and-what-doesnt).
 
 ### ST-5 — The swap is gated on a fidelity checklist, not just the checksum

pkg/checksum/recopy_retry.go

@@ -52,7 +52,9 @@ func collided(err error) bool {
 // no other. Comparing again clears a collision only when the colliding row
 // lies inside the comparison: a shadow row above the watermark the pass
 // reads through is never compared, so a collision with one repeats on
-// every attempt and the pass ends in its error. The Outcome returned
+// every attempt and the pass ends in its error — which is why a caller
+// that compares through a partial watermark clears the shadow above it
+// first (copier.ClearAbove). The Outcome returned
 // carries the comparison the committed recopy acted on, the repairs it
 // committed, and the attempts that collided, with the error when the last
 // one did; a recopy that fails for

pkg/copier/resume.go

@@ -7,9 +7,37 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 
+	"github.com/block/pg-sprite/pkg/dbconn"
 	"github.com/block/pg-sprite/pkg/preflight"
 )
 
+// ClearAbove removes every shadow row whose key lies above from, as a
+// resumed Run does before it cuts its first chunk, and nothing else. It is
+// for a caller that compares the shadow against the source through from
+// before any copier resumes — reconcile mode (ST-4) — since a shadow row
+// above the watermark is outside such a comparison yet still holds its
+// values: a unique value the source has since moved out of it into a chunk
+// the comparison finds differing would make every recopy of that chunk
+// collide with it, and no comparison through from can clear the collision.
+// With the shadow clear above from, the rows the comparison reads are the
+// only rows the shadow has, and the copier that resumes from the same
+// watermark lands the rest. ClearAbove refuses what NewCopier refuses and
+// bounds each batch as a chunk transaction is bounded. Of opts it honors
+// LockTimeout, StatementTimeout, Chunker (the batch size) and Clock; the
+// clear runs one batch at a time and reports no progress, so Workers and
+// Tracker have no effect. It runs under the
+// lock session's Bind context, so a lost lock cancels the batch in flight
+// (LK-1). A complete watermark has nothing above it and clears nothing.
+func ClearAbove(ctx context.Context, pool *pgxpool.Pool, target preflight.CopySwapTarget, shadow Shadow, lock *dbconn.TableLockSession, from Watermark, opts Options) error {
+	c, err := NewCopier(target, shadow, lock, from, opts)
+	if err != nil {
+		return err
+	}
+	ctx, unbind := c.lock.Bind(ctx)
+	defer unbind()
+	return c.clearAbove(ctx, pool, from)
+}
+
 // clearAbove removes every shadow row whose key lies above the resume
 // watermark before the resumed copy cuts its first chunk. The earlier run's
 // unlanded chunks may have committed rows there, and the applier treats

pkg/schemachange/reconcile.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/block/pg-sprite/pkg/checkpoint"
 	"github.com/block/pg-sprite/pkg/checksum"
+	"github.com/block/pg-sprite/pkg/copier"
 	"github.com/block/pg-sprite/pkg/dbconn"
 	"github.com/block/pg-sprite/pkg/decode"
 	"github.com/block/pg-sprite/pkg/preflight"
@@ -18,7 +19,8 @@ import (
 // Reconciler is reconcile mode (ST-4): the recovery for a run whose
 // replication slot is gone or lost, keeping the shadow and the copier's
 // watermark and replacing only what the slot carried. One Reconciler
-// serves one built shadow; its verifier is built up front so a reconcile
+// serves one built shadow; its verifier is built up front, and the options
+// its clear of the shadow runs under are validated then, so a reconcile
 // never starts with options that cannot run a pass.
 type Reconciler struct {
 	cfg      dbconn.Config
@@ -27,6 +29,10 @@ type Reconciler struct {
 	target   preflight.CopySwapTarget
 	shadow   BuiltShadow
 	verifier *checksum.Verifier
+	// clearOpts bounds the clear of the shadow above a partial watermark
+	// as the pass's options bound the pass: the same timeouts, the pass's
+	// chunk size as the batch size.
+	clearOpts copier.Options
 }
 
 // Reconciled is what a completed reconcile hands back: the slot change
@@ -66,7 +72,23 @@ func NewReconciler(cfg dbconn.Config, lock *dbconn.TableLockSession, store *chec
 	if err != nil {
 		return nil, err
 	}
-	return &Reconciler{cfg: cfg, lock: lock, store: store, target: target, shadow: shadow, verifier: verifier}, nil
+	clearOpts := clearOptions(opts)
+	if err := clearOpts.Validate(); err != nil {
+		return nil, err
+	}
+	return &Reconciler{cfg: cfg, lock: lock, store: store, target: target, shadow: shadow, verifier: verifier, clearOpts: clearOpts}, nil
+}
+
+// clearOptions bounds the reconcile's clear of the shadow as its pass is
+// bounded. The copier's worker count has no bearing on a clear, which
+// runs one batch at a time, and is left to its default.
+func clearOptions(opts checksum.Options) copier.Options {
+	return copier.Options{
+		LockTimeout:      opts.LockTimeout,
+		StatementTimeout: opts.StatementTimeout,
+		Chunker:          opts.Chunker,
+		Clock:            opts.Clock,
+	}
 }
 
 // Reconcile recovers from slot loss for the run cp records, in an order a
@@ -75,26 +97,36 @@ func NewReconciler(cfg dbconn.Config, lock *dbconn.TableLockSession, store *chec
 //  1. The source and shadow are confirmed to be the relations the built
 //     shadow proved, so a row whose tables have been swapped or replaced
 //     under their names is refused before anything is written.
-//  2. The row is saved in PhaseReconciling, naming the slot the reconcile
+//  2. When anything has landed, the shadow is cleared above the row's
+//     watermark, as the copier that resumes from it would clear it first
+//     (CO-4): the rows the earlier run's unlanded chunks committed there
+//     lie outside the pass that follows, and one still holding a unique
+//     value the source has since moved into a chunk the pass recopies
+//     would collide with every recopy of it, with no comparison through
+//     the watermark able to clear the collision. A complete watermark has
+//     nothing above it. The clear runs before any slot command, so a new
+//     slot retains WAL only while the pass runs; a crash after it leaves
+//     the row in its earlier phase, and clearing again is harmless.
+//  3. The row is saved in PhaseReconciling, naming the slot the reconcile
 //     will own under the target's derived name, so a resume that finds
 //     it reconciles again rather than trusting whatever slot it finds,
 //     and the reaper sees the slot as the row's.
-//  3. The slot under the derived name is dropped — a slot already gone is
+//  4. The slot under the derived name is dropped — a slot already gone is
 //     success, a half-made one from an earlier attempt goes too.
-//  4. A new slot is created under the same name. Every change committed
+//  5. A new slot is created under the same name. Every change committed
 //     after its consistent point replays when the stream is reopened.
-//  5. When anything has landed, one recopy pass runs through the row's
-//     watermark under checksum.DivergenceRecopy: every chunk that differs
-//     is recopied from the source, which is where the changes the lost
-//     slot never delivered are, and is not read again. The source keeps
-//     taking writes throughout; a write that lands in a chunk after its
-//     recopy is behind the new slot's consistent point and replays when
-//     the stream reopens, so the pass has nothing to prove about it. A
-//     unique value the source moves from a chunk the pass read equal into
-//     a differing chunk, between the comparison and the recopy, makes the
-//     recopy collide with the shadow row still holding it; the pass then
-//     compares again and recopies both chunks, a bounded number of times.
-//  6. The row is saved out of PhaseReconciling: PhaseCatchingUp when the
+//  6. Then one recopy pass runs through the row's watermark under
+//     checksum.DivergenceRecopy: every chunk that differs is recopied
+//     from the source, which is where the changes the lost slot never
+//     delivered are, and is not read again. The source keeps taking
+//     writes throughout; a write that lands in a chunk after its recopy is
+//     behind the new slot's consistent point and replays when the stream
+//     reopens, so the pass has nothing to prove about it. A unique value
+//     the source moves from a chunk the pass read equal into a differing
+//     chunk, between the comparison and the recopy, makes the recopy
+//     collide with the shadow row still holding it; the pass then compares
+//     again and recopies both chunks, a bounded number of times.
+//  7. The row is saved out of PhaseReconciling: PhaseCatchingUp when the
 //     watermark is complete, PhaseCopying when the copier still has keys
 //     to land, with the new slot's consistent point as the last applied
 //     position.
@@ -105,7 +137,11 @@ func NewReconciler(cfg dbconn.Config, lock *dbconn.TableLockSession, store *chec
 // this target and shadow, is refused. On an error the slot the attempt
 // created, if any, stays under the row's name, so the reaper leaves it for
 // the next attempt to drop, and the row stays in PhaseReconciling for that
-// attempt.
+// attempt. The clear can end an attempt before any slot command, with the
+// copier's own refusals (a lost lock, a replaced relation, conversion
+// drift) or with an untyped lock timeout when a writer still holds the
+// shadow past lock_timeout; either leaves the row and the watermark as
+// they were, and a later attempt clears again. Retry the timeout.
 //
 // The whole sequence runs under the lock session's context: a lock lost
 // at any point cancels the statement in flight — a drop still waiting out
@@ -137,6 +173,12 @@ func (r *Reconciler) reconcile(ctx context.Context, pool *pgxpool.Pool, cp check
 	if err := r.confirmShadowRelations(ctx, pool); err != nil {
 		return Reconciled{}, err
 	}
+	// The clear needs only the lock and the shadow, so it runs before any
+	// slot command: a new slot then retains WAL only while the pass runs,
+	// and a clear that fails ends the attempt with no slot made.
+	if err := r.clearAbove(ctx, pool, cp); err != nil {
+		return Reconciled{}, err
+	}
 	// INV: ST-4 — the durable marker precedes every change to the slot.
 	name := r.target.DecodingName()
 	marked := cp
@@ -216,9 +258,21 @@ func (r *Reconciler) confirmShadowRelations(ctx context.Context, pool *pgxpool.P
 	return nil
 }
 
-// recopy runs the recopy pass through the row's watermark, or no pass
-// when nothing has landed: a shadow with no rows has nothing the lost
-// slot could have left stale.
+// clearAbove clears the shadow above the row's watermark, or does nothing
+// when nothing has landed: with no watermark there is no pass, so no row
+// above one to collide with, and whatever the earlier run committed is the
+// resumed copier's to clear.
+func (r *Reconciler) clearAbove(ctx context.Context, pool *pgxpool.Pool, cp checkpoint.Checkpoint) error {
+	if !cp.Watermark.Valid() {
+		return nil
+	}
+	// INV: CO-4 — every shadow row the pass does not compare is removed
+	// before the pass, so none can hold a value a recopy puts back.
+	return copier.ClearAbove(ctx, pool, r.target, r.shadow, r.lock, cp.Watermark, r.clearOpts)
+}
+
+// recopy runs the recopy pass through the row's watermark, or nothing
+// when nothing has landed.
 func (r *Reconciler) recopy(ctx context.Context, pool *pgxpool.Pool, cp checkpoint.Checkpoint) (checksum.Outcome, error) {
 	if !cp.Watermark.Valid() {
 		return checksum.Outcome{}, nil