applier: wait for a delivered commit until the source shows it visible

docs/invariants.md

@@ -237,20 +237,27 @@ after the buffer's last `Add`, so a key the copier cuts in between is in flight
 chunk, a deferred batch stays buffered until the next pass, and the confirmed position never
 passes a change still buffered, held, or deferred (convergence tests under mixed load, key
 moves, hot-row contention, unchanged-TOAST updates, and unique-key swaps; a refused batch
-requeued while its chunk is in flight and flushed once the chunk lands). *Open:* a source
-transaction whose commit the walsender has already sent while its backend is still in a
-synchronous-commit wait is not yet visible to a snapshot taken after the stream delivered it, so
-an uncut key's change can be discarded ahead of a chunk read that does not see it; the drain
-has no rule for that window yet. The same window reaches a moved image whose unchanged-TOAST
-markers the flush completes from the source: an image `CompleteFirst` names whose old key's
-shadow row is absent or was reused, so its completion is read from the source row under `Key`.
-Inside the window that read does not see the delivered move. When it finds no row, the flush
-writes a delete at `Key` (`Result.Skipped`) with no stream gate, and the shadow ends up without
-a row the source holds; when the same invisible commit also removed another row from `Key`, the
-read finds that row and completes the image with its stale marker values. Nothing completes the
-image again. The checksum (CO-1) finds the differing row when the loss lands before the pass
-settles its chunk, and under `abort` refuses the cutover while `repair` recopies the chunk; a
-loss between the pass and the swap has no later comparison. Also open: a primary-key move or a
+requeued while its chunk is in flight and flushed once the chunk lands). The server writes a
+transaction's commit to the WAL — and waits out any synchronous standby — before it marks the
+transaction visible to new snapshots, and the walsender decodes the commit as soon as it is
+flushed, so the stream can deliver a change whose transaction no snapshot shows yet; a chunk
+read or a completion read from the source taken in that window does not see it, and a drain
+that discarded an uncut key's change, or flushed a moved image completed from a source row that
+does not yet show the move, would lose it. The drain judges that window as it judges an
+in-flight chunk: every `ChangeEvent` carries its transaction's xid, each buffered `Entry`
+carries the xid of the latest event merged into it (a key move stamps both halves, a delete its
+marker, the flush's replacement marker the image it replaced), and `Drain` takes the source's
+`pg_current_snapshot()` — read on the pool before the `Position`, and only when the buffer holds
+anything — beside the position: an entry whose xid the snapshot does not show as committed is
+treated as in flight, kept buffered and counted in `Batch.Invisible` (its paired entries
+deferred with it under the three-way rule), and judged again next cycle, when visibility, which
+is monotone, has caught up; the zero `Snapshot` shows nothing visible, so a drain with no
+snapshot keeps everything. The swap's drain (`flushAll`) drains against `AllVisible`, since a
+writer holds its locks until its commit is visible and the swap's `ACCESS EXCLUSIVE` lock waited
+them all out (sync-commit test: a writer under `synchronous_commit = on` against a standby that
+never answers has its UPDATE delivered and left buffered with `Invisible` one and nothing
+applied, the confirmed position below the delivered one, until the backend is cancelled, after
+which the change lands and the catch-up converges). Also open: a primary-key move or a
 unique-value swap that straddles a chunk the copier is reading collides, on the copier's own
 insert of the moved row, with the stale shadow row the flush has not yet deleted (SQLSTATE
 `23505`); the catch-up's half of the exchange is correct and the resolution is the copier's, so
@@ -263,8 +270,10 @@ with the DELETE not yet delivered, is held until the stream passes the flush's c
 catch-up converges, while the same image for a row the source still holds and the shadow lost
 fails closed once the stream has passed it (CO-8); a
 key-moving UPDATE that straddles the watermark (`UPDATE t SET id = 5000 WHERE id = 5`, watermark
-1000) asserts key 5 is deleted from the shadow. *Source:* this doc set (Spirit's model
-translated).
+1000) asserts key 5 is deleted from the shadow; an UPDATE committed under `synchronous_commit =
+on` against a standby that never acknowledges asserts the delivered change stays buffered, with
+nothing applied and the slot unconfirmed past it, until the commit is visible, and lands then.
+*Source:* this doc set (Spirit's model translated).
 
 ### CO-5 — The change buffer is disjoint and current at flush time
 

docs/schemabot-integration.md

@@ -520,7 +520,7 @@ each asks for a different action. Whether the pass wrote the shadow is part of t
 | --- | --- | --- | --- |
 | `Outcome` with `VerifiedShadow` minted (`Live()` true) | Every chunk was equal or settled against the catch-up's confirmed position; the shadow equals the source as of those positions and every change since is in the stream | Hand the proof to `GateCutover`; `Cutover` requires a `DrainFunc` that applies the stream through the final position under the swap lock, and refuses the proof without one (`cutover-undrained`) | No |
 | `*UnresolvedError`, no repairs | A chunk could not be settled: the catch-up did not confirm past the read within the wait (`lagging`), or the chunk changed on every reread (`hot`). Not a finding about the data | Run the pass again — after the catch-up has confirmed further, or when the source is quieter; `Outcome.Report.Unsettled` counts the chunks that did not settle | No |
-| `*DivergenceError` (policy `abort`) | At least one chunk held still on both sides across a confirmed position and still differed; the stream delivered everything and the shadow does not match. `Report.Unsettled` may also be non-zero | Stop and report; a divergence under the stream is a defect, or the known synchronous-commit window recorded in [CO-4's *Open* note](invariants.md#co-4--the-copyapply-ordering-invariants), and either way not a retry: a second pass reads the same shadow and stops the same way, so start the run over (the `*checksum.DivergenceError` row above). A chunk that settled is not compared again, so a loss that lands in that window after the pass has settled its chunk is not caught | No |
+| `*DivergenceError` (policy `abort`) | At least one chunk held still on both sides across a confirmed position and still differed; the stream delivered everything and the shadow does not match. `Report.Unsettled` may also be non-zero | Stop and report; a divergence under the stream is a defect in the copy/apply ordering [CO-4](invariants.md#co-4--the-copyapply-ordering-invariants) enforces, not a retry: a second pass reads the same shadow and stops the same way, so start the run over (the `*checksum.DivergenceError` row above) | No |
 | `*UnresolvedError` with `Outcome.Repairs` listed (policy `repair`) | Every confirmed divergence was recopied and read back clean, but another chunk did not settle | The repairs are committed; run the pass again as above. No proof was minted (CO-2) | Yes — the listed repairs |
 | `*RepairError` (policy `repair`) | A repaired chunk was a confirmed divergence again after its recopy | Stop and report; the recopy committed and is listed in `Outcome.Repairs`, and something other than the stream is writing the shadow | Yes — the listed repairs |
 

pkg/applier/absent_integration_test.go

@@ -109,7 +109,7 @@ func TestHeldAbsentImageWaitsForTheStreamToPassTheCheck(t *testing.T) {
 	assert.GreaterOrEqual(t, result.Held[0].ReadLSN, deleteDone, "the check reads a position at or past the DELETE's commit")
 	assert.Equal(t, 0, buffer.Release(deleteDone-1), "the stream has not delivered the DELETE")
 	assert.Equal(t, 1, buffer.Held())
-	next := buffer.Drain(everyKeyLanded())
+	next := buffer.Drain(everyKeyLanded(), applier.AllVisible())
 	assert.Empty(t, next.Entries, "the held image is not flushed before the stream passes the check")
 	assert.Equal(t, 1, next.Held)
 }
@@ -159,7 +159,7 @@ func TestFlushRefusesAnAbsentRowOnceTheStreamHasPassedTheCheck(t *testing.T) {
 
 	require.Len(t, result.Held, 1)
 	assert.Equal(t, 1, buffer.Release(passed(result.Held)), "the stream passed the check and nothing replaced the image")
-	_, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded()))
+	_, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded(), applier.AllVisible()))
 
 	require.ErrorIs(t, err, applier.ErrInvariantViolation)
 	assert.ErrorContains(t, err, "(CO-8): flush key 2")
@@ -187,7 +187,7 @@ func TestFlushRefusesAHeldImageAnUpdateFilledInWhenTheRowIsStillAbsent(t *testin
 	require.Len(t, result.Held, 1)
 	require.NoError(t, buffer.Add(updateEvent(2, 2, col("slot", "Q"), col("doc", "a new document"), col("note", "seat 2"))))
 	assert.Equal(t, 1, buffer.Release(passed(result.Held)), "an UPDATE keeps the held image pending")
-	_, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded()))
+	_, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded(), applier.AllVisible()))
 
 	require.ErrorIs(t, err, applier.ErrInvariantViolation)
 	assert.ErrorContains(t, err, "(CO-8): flush key 2")
@@ -212,12 +212,12 @@ func TestFlushInRefusesALostRowOnTheSecondPassOfTheSwapDrain(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = tx.Rollback(context.WithoutCancel(t.Context())) })
 
-	first, err := p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded()))
+	first, err := p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded(), applier.AllVisible()))
 	require.NoError(t, err)
 	require.Len(t, first.Held, 1)
 	require.NoError(t, buffer.Hold(first.Held))
 	assert.Equal(t, 1, buffer.Release(math.MaxUint64))
-	_, err = p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded()))
+	_, err = p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded(), applier.AllVisible()))
 
 	require.ErrorIs(t, err, applier.ErrInvariantViolation)
 	assert.ErrorContains(t, err, "(CO-8): flush key 2")

pkg/applier/buffer.go

@@ -103,7 +103,7 @@ func (b *Buffer) addInsert(ev decode.ChangeEvent) error {
 		}
 	}
 	b.markOldKeyReused(ev.Key, nil)
-	b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: firstLSN(ev.Delivered, cur)})
+	b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: firstLSN(ev.Delivered, cur), Xid: ev.Xid})
 	return nil
 }
 
@@ -117,13 +117,14 @@ func (b *Buffer) addUpdate(ev decode.ChangeEvent) error {
 	cur, ok := b.entries[ev.Key]
 	if !ok {
 		b.markOldKeyReused(ev.Key, nil)
-		b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: ev.Delivered})
+		b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: ev.Delivered, Xid: ev.Xid})
 		return nil
 	}
 	if cur.Kind == DeleteMarker {
 		return fmt.Errorf("%w (CO-5): buffer key %d: update over a buffered delete", ErrInvariantViolation, ev.Key)
 	}
 	cur.overlay(ev.Columns)
+	cur.Xid = ev.Xid
 	return nil
 }
 
@@ -152,7 +153,7 @@ func (b *Buffer) addKeyMove(ev decode.ChangeEvent) error {
 		return fmt.Errorf("%w (CO-5): buffer key %d: key move from %d, which is buffered deleted", ErrInvariantViolation, ev.Key, oldKey)
 	}
 	b.markOldKeyReused(ev.Key, from)
-	moved := &Entry{Key: ev.Key, Kind: Image, OldKey: &oldKey, FirstLSN: firstLSN(ev.Delivered, cur, from)}
+	moved := &Entry{Key: ev.Key, Kind: Image, OldKey: &oldKey, FirstLSN: firstLSN(ev.Delivered, cur, from), Xid: ev.Xid}
 	if from != nil {
 		moved.Columns = cloneColumns(from.Columns)
 		moved.overlay(ev.Columns)
@@ -165,14 +166,14 @@ func (b *Buffer) addKeyMove(ev decode.ChangeEvent) error {
 		moved.Columns = cloneColumns(ev.Columns)
 	}
 	b.put(moved)
-	b.put(&Entry{Key: oldKey, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, from)})
+	b.put(&Entry{Key: oldKey, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, from), Xid: ev.Xid})
 	return nil
 }
 
 // addDelete replaces whatever the key held with a delete marker: the newest
 // fact about the key is that its row is gone (CO-5).
 func (b *Buffer) addDelete(ev decode.ChangeEvent) {
-	b.put(&Entry{Key: ev.Key, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, b.entries[ev.Key])})
+	b.put(&Entry{Key: ev.Key, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, b.entries[ev.Key]), Xid: ev.Xid})
 }
 
 // markOldKeyReused records that key now holds a row other than the one each

pkg/applier/buffer_test.go

@@ -167,7 +167,7 @@ func TestBufferReusedOldKeyIsNotACompletionSource(t *testing.T) {
 	t.Run("chain, after the old key's marker was discarded", func(t *testing.T) {
 		b := NewBuffer()
 		require.NoError(t, b.Add(keyMove(10, 50, 15, col("label", "r"), marker("doc"))))
-		_ = b.Drain(position(t, 30, [2]int64{11, 20}))
+		_ = b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible())
 		require.NoError(t, b.Add(insert(20, 50, col("label", "s"), col("doc", "S-doc"))))
 		require.NoError(t, b.Add(keyMove(30, 15, 25, col("label", "r"), marker("doc"))))
 		got := entry(t, b, 25)
@@ -178,7 +178,7 @@ func TestBufferReusedOldKeyIsNotACompletionSource(t *testing.T) {
 	t.Run("reuse seen as an update of a key the buffer does not hold", func(t *testing.T) {
 		b := NewBuffer()
 		require.NoError(t, b.Add(keyMove(10, 50, 25, col("label", "r"), marker("doc"))))
-		_ = b.Drain(position(t, 30, [2]int64{21, 30}))
+		_ = b.Drain(position(t, 30, [2]int64{21, 30}), AllVisible())
 		require.NoError(t, b.Add(update(20, 50, col("label", "s2"))))
 		assert.True(t, entry(t, b, 25).OldKeyReused, "a live row at 50 is not R")
 	})
@@ -212,7 +212,7 @@ func TestBufferReuseIndexFollowsTheEntries(t *testing.T) {
 	t.Run("image drained, then the old key reused", func(t *testing.T) {
 		b := NewBuffer()
 		require.NoError(t, b.Add(keyMove(10, 50, 25, col("label", "r"), marker("doc"))))
-		batch := b.Drain(position(t, 6000))
+		batch := b.Drain(position(t, 6000), AllVisible())
 		require.Equal(t, []int64{25, 50}, keys(batch.Entries))
 		require.NoError(t, b.Add(insert(20, 50, col("label", "s"), col("doc", "S-doc"))))
 		assert.Equal(t, 1, b.Len())
@@ -389,6 +389,32 @@ func TestBufferKeysEveryEventKindOnItsDeliveredPosition(t *testing.T) {
 	}
 }
 
+// An entry carries the transaction of the latest event merged into it: an
+// overlaying UPDATE replaces it, a key move stamps both halves with the
+// move's, and a DELETE stamps the marker with its own.
+func TestBufferEntryCarriesItsLatestTransaction(t *testing.T) {
+	t.Run("overlay", func(t *testing.T) {
+		b := NewBuffer()
+		require.NoError(t, b.Add(inTx(update(10, 1, col("label", "a")), 600)))
+		assert.Equal(t, uint32(600), entry(t, b, 1).Xid)
+		require.NoError(t, b.Add(inTx(update(20, 1, col("label", "b")), 700)))
+		assert.Equal(t, uint32(700), entry(t, b, 1).Xid)
+	})
+	t.Run("key move", func(t *testing.T) {
+		b := NewBuffer()
+		require.NoError(t, b.Add(inTx(update(10, 1, col("label", "a")), 600)))
+		require.NoError(t, b.Add(inTx(keyMove(20, 1, 2, col("label", "b")), 700)))
+		assert.Equal(t, uint32(700), entry(t, b, 1).Xid, "the marker at the old key")
+		assert.Equal(t, uint32(700), entry(t, b, 2).Xid, "the moved image")
+	})
+	t.Run("delete", func(t *testing.T) {
+		b := NewBuffer()
+		require.NoError(t, b.Add(inTx(update(10, 1, col("label", "a")), 600)))
+		require.NoError(t, b.Add(inTx(del(20, 1), 700)))
+		assert.Equal(t, uint32(700), entry(t, b, 1).Xid)
+	})
+}
+
 func TestEntryKindStrings(t *testing.T) {
 	assert.Equal(t, "image", Image.String())
 	assert.Equal(t, "delete-marker", DeleteMarker.String())

pkg/applier/catchup.go

@@ -126,6 +126,12 @@ type Status struct {
 	// check that found an unmoved image's shadow row absent. A nonzero value
 	// is normal for the cycle or two a hold lasts.
 	Held int
+	// Invisible is the number of entries the last drain left waiting for
+	// the source to show their transaction visible: a commit the stream
+	// delivered that the server was still flushing or waiting on a
+	// synchronous standby for. A nonzero value is normal for a cycle;
+	// one that persists means the source's commits are not completing.
+	Invisible int
 	// Applied is the number of images and delete markers committed flushes
 	// have written to the shadow.
 	Applied uint64
@@ -224,9 +230,11 @@ func (c *Catchup) Confirmed() decode.LSN {
 
 // Run consumes the stream in cycles until ctx ends or the stream, the buffer,
 // or a flush fails. Each cycle gathers changes until MaxChanges are buffered
-// or Interval has passed, reads the copier's Position after the last Add — so
-// a discarded uncut key's change is one the copier's read will see (CO-4) —
-// drains, flushes the batch, puts a refused batch back (ErrBatchDeferred) or
+// or Interval has passed, takes the source's snapshot and then reads the
+// copier's Position after the last Add — so a discarded uncut key's change
+// is one the copier's read will see, and a change whose commit the source
+// has not yet made visible waits for a later cycle (CO-4) — drains, flushes
+// the batch, puts a refused batch back (ErrBatchDeferred) or
 // holds the images the flush did not write — those it completed and those
 // whose shadow row it found absent — and then confirms the lowest of the
 // stream's delivered position and the buffer's oldest pending position:
@@ -343,9 +351,15 @@ func (c *Catchup) apply(ctx context.Context, pool *pgxpool.Pool) (cycleOutcome,
 		return cycleOutcome{paused: true}, nil
 	}
 	defer c.releaseFlushes()
-	// INV: CO-4 — the position is read after the last Add of this cycle.
+	// INV: CO-4 — the snapshot is taken, and then the position read, after
+	// the last Add of this cycle: a transaction the snapshot shows visible
+	// is visible to every chunk read the copier makes after the position.
+	snap, err := c.snapshot(ctx, pool)
+	if err != nil {
+		return cycleOutcome{}, err
+	}
 	pos := c.positions.Position()
-	batch := c.buffer.Drain(pos)
+	batch := c.buffer.Drain(pos, snap)
 	result, err := c.flusher.Flush(ctx, pool, batch)
 	outcome := cycleOutcome{batch: batch, result: result}
 	switch {
@@ -369,6 +383,20 @@ func (c *Catchup) apply(ctx context.Context, pool *pgxpool.Pool) (cycleOutcome,
 	return outcome, nil
 }
 
+// snapshot takes the source's snapshot on pool for this cycle's drain, and
+// skips the round trip when the buffer holds nothing to judge: the zero
+// Snapshot then judges nothing.
+func (c *Catchup) snapshot(ctx context.Context, pool *pgxpool.Pool) (Snapshot, error) {
+	if c.buffer.Len() == 0 {
+		return Snapshot{}, nil
+	}
+	snap, err := ReadSnapshot(ctx, pool)
+	if err != nil {
+		return Snapshot{}, fmt.Errorf("catch-up on %s.%s: %w", c.flusher.target.Schema(), c.flusher.target.Table(), err)
+	}
+	return snap, nil
+}
+
 // checkSlot reads the stream's slot from the catalog after the cycle has
 // confirmed and judges it against the ceiling: what the slot retains after
 // this cycle's confirmation is what the next cycle will have to live with.
@@ -495,10 +523,12 @@ func (c *Catchup) record(outcome cycleOutcome, walEnd decode.LSN) {
 	case outcome.deferred:
 		c.status.Discarded += uint64(outcome.batch.Discarded)
 		c.status.Deferred = outcome.batch.Deferred + len(outcome.batch.Entries)
+		c.status.Invisible = outcome.batch.Invisible
 		c.status.BatchesDeferred++
 	default:
 		c.status.Discarded += uint64(outcome.batch.Discarded)
 		c.status.Deferred = outcome.batch.Deferred
+		c.status.Invisible = outcome.batch.Invisible
 		c.status.Applied += uint64(outcome.result.Images + outcome.result.Deletes)
 		if outcome.result.Fallback {
 			c.status.Fallbacks++