copier: commit a collided chunk around the held-off rows and retry only those

docs/invariants.md

@@ -279,22 +279,36 @@ which the change lands and the catch-up converges). *Enforced* for the copier's
 of a unique value that moves between rows while its chunk is in flight: the chunk's insert
 collides (SQLSTATE `23505`, or `23P01` for an exclusion constraint) with the shadow row that
 still holds the value — a landed row the flush has yet to rewrite, or a row of the same chunk
-the copier itself read before the swap — and the copier runs the chunk again after a bounded
-wait (`copier.Options.CollisionAttempts` × `CollisionBackoff`, linear), counting each
-collision, until the flush has moved the value and the chunk lands or the attempts run out and
-the server's error is returned with no cause of its own (a copier integration test holds a
-colliding row across several attempts and removes it; a catch-up convergence test runs
-unique-key swaps throughout the copy). *Enforced* for a primary-key move whose row carries a
+the copier itself read before the swap — and the copier commits the chunk around the colliding
+rows: a second pass inserts every row of the chunk whose values no shadow row holds and
+settles the chunk with the held-off keys pending (`copier.Position.Pending`), so from then on
+the applier writes every other key of the chunk as landed and keeps only the pending keys in
+flight; each later pass, after a wait one `copier.Options.CollisionBackoff` longer than the one
+before, inserts only the keys still pending, never the range again, so a value that moves
+into the range after the second pass is the applier's to write and cannot make the chunk
+collide anew. The pending keys only shrink — the flush rewrites the row holding a pending
+row's value and the next pass lands it, or the source deletes the pending row and the next
+pass no longer reads it — and the chunk lands when they are gone; a chunk whose pending keys
+outlast `copier.Options.CollisionAttempts` passes stops the copy with the server's error and
+no cause of its own, the chunk still in flight with its keys pending so a checkpoint records
+none of it as copied and the resume clear removes the rows committed around them (a copier
+integration test holds a colliding row, asserts the chunk commits around it with that key
+alone pending, and removes the row; a catch-up convergence test runs unique-key swaps
+throughout the copy; a second runs primary-key moves faster than one per backoff from before
+the first chunk is read, so every move lands in the last chunk while it is in flight, and
+converges within the attempts). *Enforced* for a primary-key move whose row carries a
 unique secondary value, from a landed key to a key in flight: the drain flushes the landed old
 key's delete ahead of the image (the pair rule above), the retried chunk's insert of the row
 under the new key then lands, and the image, flushed once its chunk has landed, completes from
 the source row because the old key's shadow row is gone (a drain unit test asserts the landed
 old key flushes alone as a delete marker with the image deferred and flushed by the next drain;
 a catch-up test holds the new key's chunk in flight by hand, asserts the old key's shadow row
-is deleted while it is, and converges once the chunk lands). The chunk-level retry remains
-bounded: a chunk whose key range keeps receiving unique values from landed rows faster than one
-move per catch-up cycle collides on every attempt with a different row, and the copy stops with
-the server's error once the attempts run out. *Test obligation:* a
+is deleted while it is, and converges once the chunk lands). The passes a chunk makes stay
+bounded whatever the rate of moves: a unique value that keeps arriving in a chunk's key range
+from landed rows reaches it after the pass that committed around the first collision, and is
+then the applier's to write, never the chunk's; only the keys held off at that pass are
+retried, and the copy stops with the server's error only when one of those keys' values is
+still held by another shadow row after the last pass. *Test obligation:* a
 marker-bearing UPDATE for a key inside an in-flight chunk asserts the flush waits for the chunk
 and the row is then completed from the copied shadow row, never an absent-row abort; a
 marker-bearing UPDATE for a key whose row the source then deleted before its chunk was read,

pkg/applier/catchup_collision_integration_test.go

@@ -1,6 +1,7 @@
 package applier_test
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"sync"
@@ -70,6 +71,67 @@ func TestCatchupConvergesWhenTheCopierCollidesWithAnUnflushedMove(t *testing.T)
 	f.assertConverged(t, run.shadow)
 }
 
+// keyMoveOnLanding is the copier's clock. On the first reading after the
+// shadow holds a key the source still has, it gives that key a fresh
+// primary key from the identity sequence, as the workload's key move does,
+// once: the moved row lands in the last chunk carrying a uniq value the
+// landed row's shadow copy still holds until a flush deletes it. The key is
+// picked from the shadow rather than named up front because a concurrent
+// workload may already have moved any given key. Every reading returns the
+// wall clock.
+type keyMoveOnLanding struct {
+	t *testing.T
+	f catchupFixture
+	// shadow is the quoted shadow table, set before the copier runs.
+	shadow string
+
+	mu    sync.Mutex
+	moved bool
+	err   error
+}
+
+func (m *keyMoveOnLanding) Now() time.Time {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if !m.moved && m.err == nil {
+		m.moved, m.err = m.tryMove()
+	}
+	return time.Now()
+}
+
+func (m *keyMoveOnLanding) tryMove() (bool, error) {
+	ctx := m.t.Context()
+	source := m.f.table.Qualified()
+	q := fmt.Sprintf(`
+		UPDATE %[1]s
+		SET id = nextval(pg_get_serial_sequence($1, 'id'))
+		WHERE id = (
+			SELECT s.id
+			FROM %[2]s s
+			JOIN %[1]s src ON src.id = s.id
+			ORDER BY s.id
+			LIMIT 1
+		)`, source, m.shadow)
+	tag, err := m.f.pool.Exec(ctx, q, source)
+	if err != nil {
+		return false, fmt.Errorf("move a landed key: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// result reports whether the move was committed.
+func (m *keyMoveOnLanding) result() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.err != nil {
+		return m.err
+	}
+	if !m.moved {
+		return errors.New("no key landed while the copier ran, so no key was moved")
+	}
+	return nil
+}
+
 // moveOnLanding is the copier's clock. On the first reading after the shadow
 // holds key landed, it commits statements against the source in one
 // transaction, once; each statement's %s is the source table. Every reading

pkg/applier/catchup_integration_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"sync"
 	"testing"
 	"time"
 
@@ -482,6 +483,72 @@ func TestCatchupConvergesUnderUniqueKeyMovesDuringTheCopy(t *testing.T) {
 	f.assertConverged(t, run.shadow)
 }
 
+// Primary keys move to fresh identity values from before the copy claims
+// its first chunk, faster than the copier's collision backoff: every move
+// lands in the last chunk, [max, +∞), under a uniq value a landed row still
+// holds in the shadow until the next drain flushes its delete. The chunk's
+// first attempt collides; the attempts after it commit the rows the shadow
+// admits and run again only for the rows it refused, so a move that
+// arrives between attempts cannot send the whole chunk back to the start,
+// the chunk lands within the bound, and the shadow converges (CO-4).
+//
+// The moves run many times faster than the backoff but well below the
+// rate at which the copier lands collided chunks, so the copy gains on the
+// frontier the moves push upward and finishes; a load that mints keys
+// faster than the copier lands them keeps the last chunk open for ever.
+//
+// Whether a random move hits a landed key before the copy reaches the last
+// chunk depends on timing, so the test makes one such move itself, on the
+// first landing, and holds the flushes until the copier has collided on
+// it: the collision happens on every run. The load keeps moving keys
+// throughout, and the flushes resume as soon as the collision is seen.
+func TestCatchupConvergesUnderKeyMovesFasterThanTheCollisionBackoff(t *testing.T) {
+	f := newCatchupFixture(t, 1500)
+	move := &keyMoveOnLanding{t: t, f: f}
+	run := startCatchupWith(t, f, catchupSetup{copierClock: move})
+	move.shadow = pgx.Identifier{f.table.Schema, run.shadow.ShadowTable()}.Sanitize()
+	load := testutil.StartLoad(t, f.pool, f.table, testutil.LoadSpec{
+		Seed: 7, Workers: 2, RatePerSecond: 30,
+		Mix: testutil.Mix{KeyMove: 1},
+	})
+
+	run.awaitStatus(t, "discard changes to keys the copier has not claimed", func(s applier.Status) bool {
+		return s.Discarded > 0
+	})
+	resume, err := run.catchup.Pause(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(resume)
+
+	copied := make(chan error, 1)
+	var wg sync.WaitGroup
+	wg.Go(func() { copied <- run.copier.Run(t.Context(), f.pool) })
+	t.Cleanup(wg.Wait)
+
+	const collisionDeadline = 15 * time.Second
+	require.Eventually(t, func() bool { return run.copier.Collisions() >= 1 }, collisionDeadline, 10*time.Millisecond,
+		"the last chunk should collide on the uniq value the moved landed key still holds in the shadow")
+	require.NoError(t, move.result())
+
+	resume()
+	const copyDeadline = 30 * time.Second
+	select {
+	case err := <-copied:
+		require.NoError(t, err, "the copy lands every chunk while the moves keep arriving")
+	case <-time.After(copyDeadline):
+		require.FailNow(t, "the copy did not finish after the flushes resumed")
+	}
+	assert.Equal(t, copier.NewWatermark(math.MaxInt64), run.copier.Position().Cut, "every key is landed")
+	t.Logf("chunk passes the copier counted after a collision: %d", run.copier.Collisions())
+	run.awaitStatus(t, "apply moved images and the deletes under their old keys", func(s applier.Status) bool {
+		return s.Applied > 0
+	})
+	summary := run.quiesce(t, load)
+	assert.Positive(t, summary.KeyMoves)
+
+	run.stop(t)
+	f.assertConverged(t, run.shadow)
+}
+
 // A second Run on the same catch-up is refused: its stream is single-use.
 func TestCatchupRunsOnce(t *testing.T) {
 	f := newCatchupFixture(t, 10)

pkg/copier/copier.go

@@ -45,14 +45,18 @@ type Options struct {
 	StatementTimeout time.Duration
 	// Chunker sizes the chunks.
 	Chunker ChunkerOptions
-	// CollisionAttempts is the number of times a chunk whose insert the
-	// server refused on a unique index or an exclusion constraint is run
-	// before the copy stops with that refusal. Each attempt reads the source
-	// afresh, so a value a source change moved between keys lands once the
-	// applier has flushed the move's other half into the shadow.
+	// CollisionAttempts is the number of passes a chunk whose insert the
+	// server refused on a unique index or an exclusion constraint is given
+	// before the copy stops with that refusal: the refused insert, then one
+	// pass that commits every row of the chunk but those whose values other
+	// shadow rows hold, then passes over only the rows held off. Each pass
+	// reads the source afresh, so a value a source change moved between
+	// keys lands once the applier has flushed the move's other half into
+	// the shadow.
 	CollisionAttempts int
-	// CollisionBackoff is the wait before a collided chunk's second attempt;
-	// each later wait is one backoff longer.
+	// CollisionBackoff is the wait before a collided chunk's first pass over
+	// the rows it held off; each later wait is one backoff longer. The pass
+	// that commits the rest of the chunk waits for nothing.
 	CollisionBackoff time.Duration
 	// Clock times each chunk for the chunker's chunk-time throttling
 	// feedback (docs/copy-and-swap-design.md#d12--throttle-by-chunk-time-and-slot-lag).
@@ -135,7 +139,12 @@ type Copier struct {
 	// sql is the one insert statement every chunk runs, frozen at
 	// construction so no worker builds SQL, and searchPath the SET LOCAL
 	// its transaction runs first so the statement resolves as it should.
+	// skipSQL and keysSQL are the statements a chunk whose insert collided
+	// runs instead: over its range, committing around the rows held off,
+	// then over the held-off keys alone.
 	sql        string
+	skipSQL    string
+	keysSQL    string
 	searchPath string
 	// clearSQL removes one batch of shadow rows above the resume watermark.
 	clearSQL string
@@ -189,6 +198,8 @@ func NewCopier(target preflight.CopySwapTarget, shadow Shadow, lock *dbconn.Tabl
 		chunker:    chunker,
 		opts:       opts,
 		sql:        copySQL(target, shadow),
+		skipSQL:    copySkipSQL(target, shadow),
+		keysSQL:    copyKeysSQL(target, shadow),
 		searchPath: copySearchPath(shadow),
 		clearSQL:   clearAboveSQL(target, shadow),
 		ledger:     newLedger(from),
@@ -245,12 +256,16 @@ func (c *Copier) Position() Position {
 // and continues, so a caller can back off and resume behind a busy table.
 //
 // A chunk whose insert the server refuses on a unique index or an
-// exclusion constraint (SQLSTATE 23505 or 23P01) is run again after a wait,
-// up to Options.CollisionAttempts times: the value it collided on was moved
-// between keys by a source change the applier has yet to flush, and a later
-// attempt's snapshot sees the shadow after that flush. A collision no
-// attempt outlasts returns the server's error wrapped, the chunk in flight,
-// and Collisions counts every refused attempt.
+// exclusion constraint (SQLSTATE 23505 or 23P01) is given up to
+// Options.CollisionAttempts passes: the value it collided on was moved
+// between keys by a source change the applier has yet to flush. The next
+// pass commits every row of the chunk but those whose values other shadow
+// rows hold, and the chunk's Position entry then names those keys as
+// Pending, so the applier writes the rest of the chunk while later passes,
+// each after a wait, copy only the held-off rows once the flush has
+// rewritten the rows holding their values. A collision no pass outlasts
+// returns the server's error wrapped, the chunk in flight with its pending
+// keys, and Collisions counts every pass that left a row held off.
 //
 // Cancellation reaches the server as a closed connection, so a statement
 // that was running keeps running until it finishes or hits the transaction's
@@ -321,13 +336,13 @@ func (c *Copier) finish(cause error) error {
 }
 
 // work is one worker's loop: claim a chunk, copy it in its own transaction
-// (again, after a wait, when the insert collided with a shadow row the
-// applier has yet to clear), land it, and tell the chunker how long the
-// committed attempt took. It returns nil when the chunker has no chunk left
-// and the first error otherwise; a cancelled context ends the loop with the
-// context's error, which Run resolves to the cancellation's cause. A chunk
-// whose copy failed is left in flight: it did not land, so its keys must
-// not read as landed.
+// (in further passes, around and then after a wait for the rows whose
+// values collided with shadow rows the applier has yet to clear), land it,
+// and tell the chunker how long the pass that committed the range took. It
+// returns nil when the chunker has no chunk left and the first error
+// otherwise; a cancelled context ends the loop with the context's error,
+// which Run resolves to the cancellation's cause. A chunk whose copy failed
+// is left in flight: it did not land, so its keys must not read as landed.
 func (c *Copier) work(ctx context.Context, pool *pgxpool.Pool) error {
 	for ctx.Err() == nil {
 		chunk, ok, err := c.claim(ctx, pool)
@@ -382,3 +397,16 @@ func (c *Copier) land(chunk Chunk, inserted int64) error {
 	}
 	return nil
 }
+
+// settle records that a chunk has committed inserted rows in all and still
+// owes the rows under pending, so Position classifies its other keys as
+// landed while the chunk stays in flight (CO-4).
+func (c *Copier) settle(chunk Chunk, inserted int64, pending []int64) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	// INV: CO-4
+	if !c.ledger.settle(chunk, inserted, pending) {
+		return refuse(CauseChunksInFlight, nil, "chunk [%d, %d] settled with %d rows pending was not in flight", chunk.Lower(), chunk.Upper(), len(pending))
+	}
+	return nil
+}

pkg/copier/copy_chunk.go

@@ -41,17 +41,35 @@ const (
 
 // copyChunk copies one chunk in its own guarded transaction: it inserts the
 // chunk's live rows into the shadow without overwriting any row already
-// there and returns the number of rows the insert added. A copy that
-// evaluates conversion expressions confirms their resolution once more
-// after the insert and before the commit (CO-9), so a catalog change that
-// committed between the guard's check and the insert rolls the chunk back.
-// The transaction is READ COMMITTED whatever the server or role
-// configures, so that repeated read takes a snapshot after the insert's;
-// the insert is one statement and relies on no snapshot beyond its own.
+// there and returns the number of rows the insert added. The insert is one
+// statement and relies on no snapshot beyond its own.
 func (c *Copier) copyChunk(ctx context.Context, pool *pgxpool.Pool, chunk Chunk) (int64, error) {
+	var tag pgconn.CommandTag
+	err := c.chunkTx(ctx, pool, chunk, func(tx pgx.Tx) error {
+		var err error
+		tag, err = tx.Exec(ctx, c.sql, chunk.Lower(), chunk.Upper())
+		if err != nil {
+			return fmt.Errorf("copy chunk [%d, %d] of %s.%s into %s: %w", chunk.Lower(), chunk.Upper(), c.shadow.Schema(), c.shadow.SourceTable(), c.shadow.ShadowTable(), err)
+		}
+		return nil
+	})
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
+
+// chunkTx runs write, one chunk's insert into the shadow, in its own guarded
+// transaction and commits it. A copy that evaluates conversion expressions
+// confirms their resolution once more after the write and before the
+// commit (CO-9), so a catalog change that committed between the guard's
+// check and the write rolls the chunk back. The transaction is READ
+// COMMITTED whatever the server or role configures, so that repeated read
+// takes a snapshot after the write's.
+func (c *Copier) chunkTx(ctx context.Context, pool *pgxpool.Pool, chunk Chunk, write func(tx pgx.Tx) error) error {
 	tx, err := pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted, AccessMode: pgx.ReadWrite})
 	if err != nil {
-		return 0, fmt.Errorf("begin copy of chunk [%d, %d]: %w", chunk.Lower(), chunk.Upper(), err)
+		return fmt.Errorf("begin copy of chunk [%d, %d]: %w", chunk.Lower(), chunk.Upper(), err)
 	}
 	defer func() {
 		// Redundant safety closer: after a successful Commit this returns
@@ -60,19 +78,18 @@ func (c *Copier) copyChunk(ctx context.Context, pool *pgxpool.Pool, chunk Chunk)
 		_ = tx.Rollback(context.WithoutCancel(ctx))
 	}()
 	if err := c.guard(ctx, tx, lockRowExclusive); err != nil {
-		return 0, err
+		return err
 	}
-	tag, err := tx.Exec(ctx, c.sql, chunk.Lower(), chunk.Upper())
-	if err != nil {
-		return 0, fmt.Errorf("copy chunk [%d, %d] of %s.%s into %s: %w", chunk.Lower(), chunk.Upper(), c.shadow.Schema(), c.shadow.SourceTable(), c.shadow.ShadowTable(), err)
+	if err := write(tx); err != nil {
+		return err
 	}
 	if err := c.confirmResolution(ctx, tx); err != nil {
-		return 0, err
+		return err
 	}
 	if err := tx.Commit(ctx); err != nil {
-		return 0, fmt.Errorf("commit chunk [%d, %d] of %s.%s: %w", chunk.Lower(), chunk.Upper(), c.target.Schema(), c.target.Table(), err)
+		return fmt.Errorf("commit chunk [%d, %d] of %s.%s: %w", chunk.Lower(), chunk.Upper(), c.target.Schema(), c.target.Table(), err)
 	}
-	return tag.RowsAffected(), nil
+	return nil
 }
 
 // guard prepares the transaction every write into the shadow runs in: it