applier: flush a landed old key ahead of its in-flight moved image

docs/invariants.md

@@ -230,11 +230,17 @@ not one the copier's fence — or the verifier's repair fence — waits on; that
 `pkg/applier` `Buffer.Drain` — every
 buffered key is judged against one `Position` snapshot: landed flushes, uncut is discarded,
 in-flight stays buffered until a later drain finds its chunk landed, and a key-moving UPDATE's
-two entries are judged per key, except that once neither is uncut the image and its old key's
-entry wait together and flush together, so the flush never deletes or copies the old key's row
-while an image still needs it for completion (three-way rule, discard-all before the first claim,
-deferred entry keeps merging, straddling move, pair waits in either direction, uncut half judged
-alone, deferral spreading through a shared old key and along a chain of pairs); `Drain`
+two entries are judged per key, except that once neither is uncut the image waits for its old
+key's entry — an old key in flight, or invisible, holds its image back so the flush never
+deletes or copies the old key's row while the chunk that will copy it is still landing — while a
+landed old key's entry flushes ahead of an image in flight for its chunk (an invisible image
+still holds its old key's entry back): the delete must not wait for the image's chunk, because
+when the moved row carries a unique value the chunk cannot land while the old key's shadow row
+holds it, and the image, once its chunk lands, completes from the source row instead (three-way
+rule, discard-all before the first claim, deferred entry keeps merging,
+straddling move, pair waits for the old key, landed old key flushes ahead of its in-flight image,
+uncut half judged alone, deferral spreading through a shared old key and along a chain of
+pairs); `Drain`
 judges against a `Position` read after the last `Add`, and returns a `Batch` whose `Deferred`
 count is what a catch-up is waiting on the copier for and whose `CompleteFirst` names, as
 copies of its `Entries`, the moved, marker-bearing images the flush must complete — from the old
@@ -278,11 +284,17 @@ wait (`copier.Options.CollisionAttempts` × `CollisionBackoff`, linear), countin
 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). *Open:* a primary-key move whose row carries a unique
-secondary value, from a landed key to a key in flight: the drain's pair rule defers the old
-key's delete until the chunk lands, and the chunk cannot land while the old key's row holds the
-value, so the two wait on each other until the copier's attempts run out; the convergence test
-for key moves starts its load after the copy has landed. *Test obligation:* a
+unique-key swaps throughout the copy). *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
 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_integration_test.go

@@ -348,14 +348,10 @@ func TestCatchupConvergesUnderMixedLoad(t *testing.T) {
 }
 
 // A row's primary key moves to a fresh identity value, so the change is an
-// image under a new key paired with a delete under the old one. The pair
-// travels together through the drain, and the flush completes the image's
-// unchanged blob from the old key's row before it deletes that row (CO-4,
-// D13). The workload starts after the copy has landed every key: the moved
-// row keeps its unique key, so a move from a landed key to a key in flight
-// collides on the copier's insert with the row still under the old key, and
-// the pair rule defers that key's delete until the chunk lands — the two
-// would wait on each other until the copier's collision attempts ran out.
+// image under a new key paired with a delete under the old one. With the
+// copy landed, both halves are landed: the pair flushes together, and the
+// flush completes the image's unchanged blob from the old key's row before
+// it deletes that row (CO-4, D13).
 func TestCatchupConvergesUnderKeyMoves(t *testing.T) {
 	f := newCatchupFixture(t, 1500)
 	run := startCatchup(t, f, applier.CatchupOptions{})

pkg/applier/catchup_stall_integration_test.go

@@ -97,6 +97,48 @@ func TestCatchupRequeuesADeferredBatchAndConverges(t *testing.T) {
 	f.assertConverged(t, run.shadow)
 }
 
+// Row 7 moves to key 1000, keeping its unique value, while the chunk that
+// holds key 1000 is in flight. The delete under the landed key 7 flushes
+// ahead of the image: the copier's insert of the row under key 1000 would
+// collide with the shadow row under key 7 for as long as that row stood,
+// so holding the delete back for the image's chunk would wait on a chunk
+// that is waiting on it. Once the chunk lands the image flushes alone,
+// completes its unchanged blob from the source row since the shadow row
+// under key 7 is gone, and the shadow converges (CO-4, D13).
+func TestCatchupFlushesALandedOldKeyAheadOfItsInFlightImage(t *testing.T) {
+	f := newCatchupFixture(t, 20)
+	chunk, err := copier.NewChunk(21, math.MaxInt64)
+	require.NoError(t, err)
+	inFlight := allLanded()
+	inFlight.InFlight = []copier.Chunk{chunk}
+	positions := &switchPosition{pos: inFlight}
+	run := startCatchupWith(t, f, catchupSetup{positions: positions})
+
+	q := f.table.Qualified()
+	_, err = f.pool.Exec(t.Context(), `UPDATE `+q+` SET id = 1000 WHERE id = 7`)
+	require.NoError(t, err)
+
+	run.awaitStatus(t, "delete the row under key 7 while key 1000's chunk is in flight", func(s applier.Status) bool {
+		return s.Applied == 1 && s.Deferred == 1
+	})
+	shadowTable := pgx.Identifier{f.table.Schema, run.shadow.ShadowTable()}.Sanitize()
+	var sevens int
+	require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT count(*) FROM `+shadowTable+` WHERE id = 7`).Scan(&sevens))
+	assert.Zero(t, sevens, "the shadow row under the old key is gone before the image's chunk lands")
+	deferred := run.status(t)
+	assert.Equal(t, 1, deferred.Buffered, "the image waits for its chunk")
+	assert.Zero(t, deferred.BatchesDeferred, "the delete flushed, it was not refused")
+
+	positions.set(allLanded())
+	end := f.currentWALLSN(t)
+	run.awaitStatus(t, "flush the image once its chunk has landed", func(s applier.Status) bool {
+		return s.Confirmed >= end && s.Buffered == 0
+	})
+	assert.Equal(t, uint64(2), run.status(t).Applied, "the delete and then the image")
+	run.stop(t)
+	f.assertConverged(t, run.shadow)
+}
+
 // The schema change narrows a uniquely indexed column, and after the copy
 // has landed every key a write gives one row a value that is distinct on
 // the source and collides in the shadow's type. Nothing is in flight, so no

pkg/applier/drain.go

@@ -22,9 +22,9 @@ type Batch struct {
 	// the copier's own read of those chunks will see the change.
 	Discarded int
 	// Deferred counts the entries still buffered because their keys are
-	// inside an in-flight chunk, or travel with one that is or with one
-	// that is invisible. This is the number a stalled catch-up is waiting
-	// on the copier for.
+	// inside an in-flight chunk, or are moved images whose old key's entry
+	// is in flight or invisible, or travel with an invisible image. This is
+	// the number a stalled catch-up is waiting on the copier for.
 	Deferred int
 	// Held counts the entries still buffered because a flush held them (see
 	// Buffer.Hold) — with a completion, or because their shadow row was
@@ -103,13 +103,21 @@ func (b Batch) CompleteFirst() []Entry {
 // batch's OldestFirstLSN until the flush commits, and rebuilds them.
 //
 // An image that moved from an old key, and the entry still buffered at that
-// old key, travel together once neither is uncut: if either is in flight or
-// invisible both wait, otherwise both flush. The flush completes the image's
+// old key, travel together once neither is uncut: the image waits while the
+// old key's entry is in flight or invisible, and the old key's entry waits
+// while the image is invisible. The flush completes the image's
 // unchanged-TOAST markers before it deletes anything (D13), and that reading
-// must not race the deletion or copy of the old key's row; Batch.CompleteFirst
-// names the images it applies to. An image whose completion is pending
-// (Buffer.Hold) stays buffered whatever its key's state, until Release lets
-// the completion in or an event drops it.
+// must not race the copy of the old key's row or a transaction no snapshot
+// shows yet; Batch.CompleteFirst names the images it applies to. A landed
+// old key's entry does not wait for an image whose own chunk is in flight:
+// the row's unique values are still under the old key in the shadow, and
+// the chunk's insert of the row under the new key collides with them until
+// that entry flushes, so holding it back would wait on a chunk that is
+// waiting on it. The image, flushed once its chunk lands, finds the old
+// key's shadow row gone and completes from the source row under its own key
+// instead, held until the stream passes the read (D13). An image whose
+// completion is pending (Buffer.Hold) stays buffered whatever its key's
+// state, until Release lets the completion in or an event drops it.
 func (b *Buffer) Drain(pos copier.Position, snap Snapshot) Batch {
 	// INV: CO-4
 	state := make(map[int64]copier.KeyState, len(b.entries))
@@ -123,7 +131,7 @@ func (b *Buffer) Drain(pos copier.Position, snap Snapshot) Batch {
 			invisible[key] = struct{}{}
 		}
 	}
-	b.deferPairs(state)
+	b.deferPairs(state, invisible)
 	var batch Batch
 	b.hasOldest = false
 	for key, e := range b.entries {
@@ -163,13 +171,15 @@ type movePair struct {
 	image, old int64
 }
 
-// deferPairs marks both halves of every travelling pair in flight when either
-// half is. Two images may share one old key, so a deferral can spread through
-// that key to another pair; the passes repeat until one changes nothing.
-// Every pass that continues moves at least one key to in flight, so the work
-// is bounded by the number of pairs squared, not by the buffer: the pairs are
-// collected once, and a buffer with no moves costs one scan.
-func (b *Buffer) deferPairs(state map[int64]copier.KeyState) {
+// deferPairs marks both halves of every travelling pair in flight when the
+// pair must wait together (see waitsTogether). Two images may share one old
+// key, so a deferral can spread through that key to another pair; the passes
+// repeat until one changes nothing. Every pass that continues moves at least
+// one key to in flight, so the work is bounded by the number of pairs
+// squared, not by the buffer: the pairs are collected once, and a buffer
+// with no moves costs one scan. invisible names the keys whose entries are
+// in flight for their transaction's visibility rather than for a chunk.
+func (b *Buffer) deferPairs(state map[int64]copier.KeyState, invisible map[int64]struct{}) {
 	var pairs []movePair
 	for key, e := range b.entries {
 		if e.OldKey == nil {
@@ -182,7 +192,8 @@ func (b *Buffer) deferPairs(state map[int64]copier.KeyState) {
 	for changed := true; changed; {
 		changed = false
 		for _, p := range pairs {
-			if !waitsTogether(state[p.image], state[p.old]) {
+			_, imageInvisible := invisible[p.image]
+			if !waitsTogether(state[p.image], state[p.old], imageInvisible) {
 				continue
 			}
 			if state[p.image] != copier.KeyInFlight || state[p.old] != copier.KeyInFlight {
@@ -195,10 +206,14 @@ func (b *Buffer) deferPairs(state map[int64]copier.KeyState) {
 }
 
 // waitsTogether reports whether a moved image and its old key's entry must
-// both be deferred: neither is uncut, and at least one is in flight.
-func waitsTogether(image, old copier.KeyState) bool {
+// both be deferred: neither is uncut, and the old key's entry is in flight —
+// for its chunk or for its visibility — or the image is invisible. An image
+// in flight for its chunk alone does not hold a landed old key's entry back
+// (see Drain): the chunk cannot land while that entry's row still holds the
+// moved row's unique values in the shadow.
+func waitsTogether(image, old copier.KeyState, imageInvisible bool) bool {
 	if image == copier.KeyUncut || old == copier.KeyUncut {
 		return false
 	}
-	return image == copier.KeyInFlight || old == copier.KeyInFlight
+	return old == copier.KeyInFlight || imageInvisible
 }

pkg/applier/drain_test.go

@@ -115,31 +115,49 @@ func TestDrainKeyMoveStraddlingTheFrontier(t *testing.T) {
 	assert.Equal(t, 0, b.Len())
 }
 
-// The two halves of a move travel together once neither is uncut: a landed
-// image waits for the in-flight old key, and a landed old key waits for the
-// in-flight image. Neither flushes alone.
-func TestDrainKeyMovePairWaitsTogether(t *testing.T) {
-	cases := map[string]struct {
-		from, to int64
-	}{
-		"old key in flight, image landed": {from: 15, to: 25},
-		"old key landed, image in flight": {from: 25, to: 15},
-	}
-	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			b := NewBuffer()
-			require.NoError(t, b.Add(keyMove(10, tc.from, tc.to, col("label", "a"), marker("doc"))))
-
-			batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible())
-			assert.Empty(t, batch.Entries)
-			assert.Equal(t, 0, batch.Discarded)
-			assert.Equal(t, 2, batch.Deferred)
-			assert.Equal(t, 2, b.Len())
-
-			batch = b.Drain(position(t, 30), AllVisible())
-			assert.Equal(t, sortedPair(tc.from, tc.to), keys(batch.Entries), "both halves flush in one batch once the chunk lands")
-		})
-	}
+// A landed image waits for its in-flight old key: the flush must not copy
+// the old key's row while the image still needs it for completion, so
+// neither half flushes alone, and both flush once the chunk lands.
+func TestDrainKeyMovePairWaitsForTheOldKey(t *testing.T) {
+	b := NewBuffer()
+	require.NoError(t, b.Add(keyMove(10, 15, 25, col("label", "a"), marker("doc"))))
+
+	batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible())
+	assert.Empty(t, batch.Entries)
+	assert.Equal(t, 0, batch.Discarded)
+	assert.Equal(t, 2, batch.Deferred)
+	assert.Equal(t, 2, b.Len())
+
+	batch = b.Drain(position(t, 30), AllVisible())
+	assert.Equal(t, []int64{15, 25}, keys(batch.Entries), "both halves flush in one batch once the chunk lands")
+}
+
+// A landed old key does not wait for its in-flight image: the delete at the
+// old key flushes now, so the shadow row holding the moved row's unique
+// values is gone before the chunk's insert of the row under its new key
+// runs again — holding it back would wait on a chunk that is waiting on it
+// (CO-4). The image stays buffered, marker intact, until its own chunk
+// lands, and is then named for completion, which the flush reads from the
+// source row under its key once the old key's shadow row is absent.
+func TestDrainLandedOldKeyFlushesAheadOfItsInFlightImage(t *testing.T) {
+	b := NewBuffer()
+	require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "a"), marker("doc"))))
+
+	batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible())
+	assert.Equal(t, []int64{25}, keys(batch.Entries), "the delete marker at the landed old key flushes alone")
+	assert.Equal(t, DeleteMarker, batch.Entries[0].Kind)
+	assert.Empty(t, batch.CompleteFirst(), "nothing in this batch needs the old key's row")
+	assert.Equal(t, 0, batch.Discarded)
+	assert.Equal(t, 1, batch.Deferred, "the image at 15 waits for its own chunk")
+	assert.Equal(t, 1, b.Len())
+	image := entry(t, b, 15)
+	assert.Equal(t, Image, image.Kind)
+	assert.True(t, image.HasMarker(), "the deferred image keeps its marker for a later completion")
+
+	batch = b.Drain(position(t, 30), AllVisible())
+	assert.Equal(t, []int64{15}, keys(batch.Entries), "the image flushes once its chunk lands")
+	assert.Equal(t, []int64{15}, keys(batch.CompleteFirst()), "and is named for completion")
+	assert.Equal(t, 0, b.Len())
 }
 
 // An uncut half is discarded on its own and does not hold the other back: a
@@ -177,18 +195,39 @@ func TestDrainKeyMoveUncutHalfIsJudgedAlone(t *testing.T) {
 }
 
 // A deferral spreads through a shared old key: two images that both left key
-// 25 wait when either of them is in flight, even though 25 itself has landed.
+// 25 wait when the entry at 25 does, and an invisible image holds it. An
+// image in flight for its chunk alone does not: the landed entry at 25 and
+// the other image flush, and only the in-flight image waits.
 func TestDrainDeferralSpreadsThroughSharedOldKey(t *testing.T) {
-	b := NewBuffer()
-	require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "first"))))
-	require.NoError(t, b.Add(insert(20, 25, col("label", "reinserted"))))
-	require.NoError(t, b.Add(keyMove(30, 25, 28, col("label", "second"))))
+	t.Run("one image invisible", func(t *testing.T) {
+		b := NewBuffer()
+		require.NoError(t, b.Add(inTx(keyMove(10, 25, 15, col("label", "first")), 600)))
+		require.NoError(t, b.Add(inTx(insert(20, 25, col("label", "reinserted")), 650)))
+		require.NoError(t, b.Add(inTx(keyMove(30, 25, 28, col("label", "second")), 700)))
+		require.NoError(t, b.Add(inTx(insert(40, 25, col("label", "reinserted again")), 710)))
+
+		// 710 has committed and 700 is still in progress: the entry at 25 is
+		// visible on its own, and only the image at 28 is not.
+		snap := Snapshot{read: true, xmin: 700, xmax: 711, inProgress: map[uint64]struct{}{700: {}}}
+		batch := b.Drain(position(t, 30), snap)
+		assert.Empty(t, batch.Entries, "28 is invisible, so 25 waits, so 15 waits")
+		assert.Equal(t, 0, batch.Discarded)
+		assert.Equal(t, 1, batch.Invisible)
+		assert.Equal(t, 2, batch.Deferred)
+		assert.Equal(t, 3, b.Len())
+	})
+	t.Run("one image in flight", func(t *testing.T) {
+		b := NewBuffer()
+		require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "first"))))
+		require.NoError(t, b.Add(insert(20, 25, col("label", "reinserted"))))
+		require.NoError(t, b.Add(keyMove(30, 25, 28, col("label", "second"))))
 
-	batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible())
-	assert.Empty(t, batch.Entries, "15 is in flight, so 25 waits, so 28 waits")
-	assert.Equal(t, 0, batch.Discarded)
-	assert.Equal(t, 3, batch.Deferred)
-	assert.Equal(t, 3, b.Len())
+		batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible())
+		assert.Equal(t, []int64{25, 28}, keys(batch.Entries), "15 is in flight for its chunk; 25 and 28 are landed and flush")
+		assert.Equal(t, 0, batch.Discarded)
+		assert.Equal(t, 1, batch.Deferred)
+		assert.Equal(t, 1, b.Len())
+	})
 }
 
 // The spread is transitive however long the chain. Four rows each move onto
@@ -419,10 +458,3 @@ func TestDrainAgainstTheZeroSnapshotKeepsEverything(t *testing.T) {
 	assert.Equal(t, 2, batch.Invisible)
 	assert.Equal(t, 2, b.Len())
 }
-
-func sortedPair(a, b int64) []int64 {
-	if a < b {
-		return []int64{a, b}
-	}
-	return []int64{b, a}
-}