applier: catch-up loop with convergence tests and progress counters

[Kiran01bm]

Why

pkg/applier has a Buffer that merges the stream into one entry per key and drains it against the copier's Position, and a Flusher that applies one batch under the table lock — but nothing drives them. pkg/decode has a Stream with no consumer. This PR lands the catch-up loop CO-4 and ST-4 describe: the one consumer of the stream, which gathers, drains against a copier position read after the last add, flushes, requeues or holds, and confirms a position the slot can safely resume from. It is also the first time the buffer and the flusher run end to end against a live copy under concurrent writes, so the convergence tests the design promised for CO-4, CO-5, CO-6, and CO-8 land here, one per race, each comparing the source with its shadow row for row.

Stacked on kiran01bm/cs7-pgoutput (#151); this PR's diff is the loop, its tests, and the progress counters.

Before / after

Source app.workload, schema change ALTER TABLE … DROP COLUMN label, copier chunking 50 rows at a time, workload writing throughout.

Before                                           After
──────                                           ─────
decode.Stream ──► Delivery{Change, Delivered}    applier.NewCatchup(stream, flusher, copier, opts)
   (no consumer)                                   └─ Run(ctx, pool)            ◄── runs once; tracker WorkSource while it runs

applier.Buffer.Add / Drain(pos) / Requeue / Hold   cycle, every 250 ms or 10 000 changes:
applier.Flusher.Flush(ctx, pool, batch)              gather    stream.Next → buffer.Add(change)
   (caller-driven; no caller)                        release   buffer.Release(stream.Delivered)        ◄── held images whose read the stream passed
                                                     drain     buffer.Drain(copier.Position())        ◄── read AFTER the last Add (CO-4)
                                                                 key uncut      → discarded (copy's own read sees it)
                                                                 key in flight  → stays buffered (Deferred)
                                                                 key landed     → in the Batch
                                                     flush     flusher.Flush(batch)
                                                                 ok             → Applied += images + deletes
                                                                 fallback ran   → Fallbacks++ (CO-6)
                                                                 Result.Held    → buffer.Hold (D13)
                                                                 ErrBatchDeferred → buffer.Requeue; BatchesDeferred++
                                                     confirm   stream.Confirm(min(Delivered, buffer.OldestPending))   (ST-4)

progress: copy and checksum steps carry Work      progress: catch-up step carries Work too (format_version 6)
                                                     changes_applied · changes_buffered · lag_bytes

Why the position is read after the last add, with chunk [1496, 1545] in flight:

copier:  Claim [1496,1545] ─── SELECT … FOR KEY SHARE (reads source) ─── INSERT shadow ─── Land
stream:         UPDATE 1500 arrives ─────┐
catch-up:                                 Add(1500) ──► Position() ──► Drain: 1500 is in flight → wait for the chunk

Read the position before the add and 1500 can be judged uncut while the copier's SELECT has already run: the change is discarded, the chunk lands the stale row, and nothing replays it. Read after, and the key is in flight, so it waits and is flushed once the chunk lands.

What

• pkg/applier/catchup.go — PositionSource, CatchupOptions{Interval, MaxChanges, Clock, Tracker} with defaults and refusals (ErrInvalidCatchupOptions), NewCatchup (a nil stream, flusher, or position source is an ErrInvariantViolation), Status{Delivered, Confirmed, ServerWALEnd, Buffered, Deferred, Held, Applied, Discarded, Fallbacks, BatchesDeferred} and Status.Lag, Run (ErrCatchupAlreadyRun on a second call), the cycle (gather → Release → Drain → flush → confirm), confirmBound, and finish (lock loss first, then the context's cause wrapped with the table, else the stream's error).
• pkg/applier/catchup_work.go — Work, the progress.WorkSource the catch-up registers for the lifetime of Run; all three counters read from memory.
• pkg/progress — OperationCatchUp, Work.ChangesApplied / ChangesBuffered / LagBytes, FormatVersion 6; docs/progress-report.md gains the engine-measured catch-up section, the counter table, the operations row, and a fourth JSON example.
• internal/testutil/workload.go — Mix.KeyMove and Summary.KeyMoves: an UPDATE … SET id = nextval(pg_get_serial_sequence(…)) on a random row, so the primary key moves to a fresh identity value.
• Tests — catchup_test.go (defaults, refusals, confirm bound below / equal / above the oldest pending and with nothing pending, Lag, Work mirrors Status); catchup_integration_test.go (fixture: slot under the table lock before the shadow is built, stream from the consistent point, copier and catch-up sharing the lock session, catch-up started before the copier; five convergence tests and TestCatchupRunsOnce); work_source_test.go (TestSnapshotJSONShapeForACatchUpStep).
• Docs — pkg/applier/doc.go; SAFETY.md pkg/applier row (scheduling no longer planned; ST-4 added); design package map; docs/invariants.md CO-4 enforcement and the test-plan row.

Decisions to veto

• Confirm bound is min(Delivered, OldestPending), both keyed on the Delivered a change arrived with. Nothing below the bound is still in the buffer, held, or deferred, so a stream reopened from it misses nothing; keying on arrival position rather than the change's own LSN keeps the bound from falling below what the stream already confirmed (the interleaved-transaction case decode: stream the slot into change events with caller-confirmed feedback #151 documents). The slot therefore stalls while any key waits on an in-flight chunk or a deferred batch; that is the intended ST-4 trade, and lag_bytes makes the stall visible.
• Position read after the last Add, not mutual exclusion between copy and flush. The design's CO-4 text called the enforcement "scheduling that makes a chunk copy and a backlog flush mutually exclusive". Ordering the read is enough: an in-flight key is never discarded, a landed key's flush and a later chunk cannot overlap on the same key because the copier never re-reads a landed chunk, and the flusher already takes the copier's guard inside its transaction. No lock between the two stages, so a slow flush never stalls the copy and a slow chunk never stalls the flush. The CO-4 invariant text is updated to say so.
• Key-move and unique-swap loads start after the copy has landed. A row that moves key (or swaps a unique value) mid copy can straddle a chunk the copier is reading: the chunk's own INSERT of the row under its new key then collides on the unique index with the stale shadow row the flush has not yet deleted (SQLSTATE 23505, reproduced for both primary-key moves and unique-key swaps). That is the copier's race — the catch-up's part of the exchange is already correct — and its resolution is tracked as an internal follow-up on the copier. Mixed load, hot rows, and TOAST updates run across the whole copy.
• The synchronous-commit visibility window stays open. With synchronous_commit waiting on a standby, the walsender can send a transaction's commit while its backend is still in the proc array, so a chunk snapshot taken after the stream delivered the change may not see it: the drain judges the key uncut and discards the change, the chunk copies the pre-change row, and nothing replays it. Reading the position after the last add does not close this — the copier's snapshot comes later still. The CO-4 text names the window as open; closing it (discard an uncut entry only when its transaction's xid is below the copier's snapshot xmin and not in its xip, or hold uncut entries until the chunk that covers the key claims) is tracked as an internal follow-up. The convergence tests run without synchronous replication and do not exercise it.
• Run returns the context's cause, wrapped, when the caller stopped it. The stream reports a cancelled receive as its own error; a caller that ended the context gets catch-up on s.t stopped: <cause> instead, and a lost table lock outranks both, so the orchestrator can tell a stop it asked for from a stream or lock failure. Run runs once because the stream is single-use; a resume opens a new stream and a new catch-up.
• Defaults 250 ms / 10 000 changes. Interval bounds how long a quiet table waits before confirming (so WAL releases on a quiet source) and how long a change can sit before its flush; MaxChanges bounds the buffer and the flush transaction under a burst. Both are options; an interval below one millisecond or a zero MaxChanges is refused rather than defaulted silently. Both are callers' knobs later, not CLI flags now.
• Three counters, not the whole Status, in the progress JSON. changes_applied (exact, monotone), changes_buffered (the backlog an operator watches drain before cutover), and lag_bytes (how far the confirmed position trails the server's write position). Deferred / held / fallback counts stay on Status for the library caller; the JSON contract grows when a consumer needs them. This is a contract change, hence format_version 6.
• An empty cycle still confirms. On a quiet table the delivered position moves on keepalives (between transactions only, per decode: stream the slot into change events with caller-confirmed feedback #151), and confirming it is what lets the slot release WAL; the alternative — confirm only after a flush — would hold WAL for as long as the source is idle.
• Discarded counts entries, not events. The buffer already merged events per key before the drain, so what it drops is one entry per uncut key; this is the figure that says the copy's own read is doing the work mid copy, and the mixed-load test waits on it before starting the copier.

Verification

• make lint 0 issues; gofmt -l clean; go build ./...; make test-unit green (docs guards included).
• go test -race -count=1 ./pkg/applier/ ./pkg/progress/ ./internal/testutil/ ./pkg/decode/ green on PG 16 (applier ~63 s, decode ~67 s).
• scripts/test-flaky.sh 'TestCatchupConverges.*' 10 ./pkg/applier/ — all five convergence tests, 10/10 iterations with -race, ~18 s each.
• Integration, each ending with a row-for-row source/shadow comparison on every column the schema change kept, after waiting for the confirmed position to reach the server's write position: mixed insert/update/delete with 50 % TOAST rewrites across the whole copy (Discarded > 0 before the copy, Applied > 0 after, Buffered and BatchesDeferred zero at the end; the tracker's poll carries changes_applied mid-run and no work after Run returns; Work after Run still reports the applied count); primary-key moves after the copy (KeyMoves > 0); hot-row contention on the ten lowest keys (HotRowFraction 1); unchanged-TOAST updates (ToastRewriteFraction 0); unique-key swaps after the copy (Fallbacks > 0); a second Run → ErrCatchupAlreadyRun; a stop → wrapped context.Canceled naming the table.
• Unit: option defaults and refusals, nil-part refusals, confirm bound with nothing pending / pending below / equal / above delivered, Lag with and without a server position, Work mirroring Status; pkg/progress JSON pins for the catch-up step at format_version 6.

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.

[morgo]

🤖 Automated adversarial review, posted on Morgan Tocker's behalf.

Approving. I went at the confirm bound hardest, since a bound one position too high silently loses a change and the row-for-row oracle would not catch it if the lost change were overwritten later. It holds, and the reason it holds is not the one the code makes obvious. The notes below are a coverage claim in docs/invariants.md that is broader than the tests behind it, a Run signature whose success path is always an error, and a counter that freezes exactly when an operator needs it moving.

What I tried to break and could not

Confirming a position a buffered change still sits at. confirmBound lowers to oldestPending only when it is strictly below delivered, so a pending entry recorded at exactly the current delivered position does not lower the bound. If an entry's position were its own commit LSN, that would confirm an unapplied transaction and lose it on resume.

It is not. decode sets ev.Delivered = s.delivered when it yields the change, and raiseDelivered runs on the commit message, which arrives after every change in its transaction. So a change's recorded position is the end of the previous committed transaction, strictly below its own. Equality can therefore only mean no commit has been processed since the change arrived — in which case everything at or below the bound genuinely is applied or discarded, and the bound is right. The >= is load-bearing and correct, and it is the single most dangerous line in the PR to change later. The unit tests pinning the bound below, equal to, and above the oldest pending are the right three cases.

A keepalive raising Delivered past a buffered entry. This is the other way the bound could outrun the buffer. raiseDelivered(s.serverWALEnd) on a keepalive does move Delivered independently of any transaction, but the buffered entry's own recorded position does not move with it, so oldestPending < delivered and the bound is pulled back down. The stall that follows is the intended ST-4 trade and lag_bytes is the signal for it.

Hold and Requeue landing after the drain but before the confirm. The ordering in cycle is what makes the bound honest: Drain removes entries, Hold(result.Held) and Requeue(batch) put the unapplied ones back, and only then does confirm read OldestPending(). Reading the oldest pending before those two calls would see an emptier buffer and confirm past entries that were about to return to it. Worth a comment saying the confirm must follow them, because the three calls read as independent steps.

Reading the copier position before the last Add. The // INV: CO-4 comment sits on the right line: c.positions.Position() is evaluated as the argument to Drain, after gather has returned and after Release. The before/after diagram in the description is the failure this ordering prevents and it is accurate.

The progress poll racing the cycle. Work goes through c.Status(), which takes c.mu, and every writer (recordFlush, recordDeferral, recordPositions) takes the same lock, so a poll never sees a half-updated Status. report's stop is called by defer before Run returns and is documented to fence an in-flight poll, so no poll completes against a finished catch-up. That is the part a -race run would not necessarily have caught on its own.

Lag underflowing. if s.ServerWALEnd <= s.Confirmed { return 0 } before the subtraction. ServerWALEnd can legitimately sit below Confirmed under commit ordering — decode's own comment says the server's write position can lie below Delivered — so this guard is not defensive, it is reachable, and without it lag_bytes would report a number near 2^64 to an operator deciding whether to cut over.

Option validation being bypassable. withDefaults runs before validate, and the comment says so, so a zero Interval becomes 250 ms rather than failing the sub-millisecond check. A negative interval still fails. The refusal message telling the caller to pass zero for the default is the right shape.

The convergence tests being weaker than they read. Two of the five are, but the test file says so plainly: the key-move and unique-swap loads start after the copy lands, and the comment explains the 23505 collision that makes mid-copy versions fail and attributes it to the copier rather than this stage. That is a level of disclosure most PRs this size do not reach, and it is why the note below is about docs/invariants.md and not about the tests.

Non-blocking

docs/invariants.md now cites coverage the tests do not provide, and names only one of the two open holes. The new CO-4 paragraph says the race is ordered rather than locked out, backed by "convergence tests under mixed load, key moves, hot-row contention, unchanged-TOAST updates, and unique-key swaps", and the test-plan table row for CO-4, CO-5, CO-6 and CO-8 now points at those tests. The *Open:* sentence that follows names the synchronous-commit visibility window — but not the mid-copy key move, which the test file documents as a reproduced failure (SQLSTATE 23505, for both primary-key moves and unique-value swaps) and works around by starting that load only after the copy has landed.

So a reader who goes to the invariants doc to ask "is a key that moves mid copy covered?" is told yes by the evidence list and finds nothing in the *Open:* text to correct it. The limitation lives in the test file and in this description, neither of which that reader is in. In a repo where the invariants doc is the safety contract and where you have already written one *Open:* paragraph for the other hole, the symmetrical fix is a second sentence: a key move or unique-value swap that straddles an in-flight chunk collides with the stale shadow row on the copier's own insert, the catch-up's half of the exchange is correct, the resolution is on the copier, and the convergence tests for those two races run after the copy lands. One sentence, and the doc stops over-claiming.

Run can never return nil. The loop is for { if err := c.cycle(...); err != nil { return c.finish(ctx, err) } }, so every exit is an error — including the expected one, where the orchestrator watches lag_bytes reach zero, cancels, and gets catch-up on s.t stopped: context.Canceled. A method whose only success path is a wrapped context.Canceled is easy to mis-handle, and there is no caller in the tree yet to establish the pattern. The doc comment describes the behaviour but does not say outright that nil is unreachable. Either say so in one line, or give the caller a sentinel to compare against, so the first orchestrator does not have to decide for itself which errors mean "this worked".

Related and smaller: finish attributes any error to the caller whenever ctx.Err() != nil. That is the documented precedence and it is the right default, but it means a genuine stream failure arriving in the same cycle as a cancellation is reported as a clean stop. Since the expected shutdown always takes this branch, a real failure during shutdown is the case most likely to be swallowed. Wrapping the original error alongside the cause — rather than replacing it — would cost nothing and keep that case legible.

changes_buffered and lag_bytes freeze while a cycle is in flight, which is when they matter most. Status only moves in recordFlush, recordDeferral, and recordPositions, all of which run after Flush returns. A flush stuck behind a lock wait, or a very large batch, leaves Status pinned at the previous cycle's numbers for as long as it takes. The field comments are honest — "after the last cycle", "as of the last cycle" — but the operator-facing reading of a lag_bytes that has stopped changing is "caught up and steady", which is the opposite of what a stalled flush means. The counters are the cutover signal, so the failure mode is an operator cutting over on a stale zero.

Recording positions at the top of each cycle as well as the end would keep lag_bytes and ServerWALEnd moving independently of the flush, since neither depends on the flush's outcome. Failing that, a timestamp on the snapshot would let a consumer tell a steady value from a stuck one.

gather drops delivery.Delivered and the position key lives elsewhere. c.buffer.Add(*delivery.Change) passes only the change, so the fact that the entry is keyed on the Delivered it arrived with — the fact the whole confirm bound rests on — is established in decode by ev.Delivered = s.delivered and is invisible at the one place a reader of catchup.go would look for it. confirmBound's comment explains why the buffer's remaining entries bound the confirmation but not what their positions are. One sentence pointing at Change.Delivered would connect the two halves of the argument.