From 45737a4bbdfafe52e8afca21e367ebae84a73e10 Mon Sep 17 00:00:00 2001 From: thesyncim Date: Fri, 28 Aug 2026 11:00:42 +0100 Subject: [PATCH 1/4] fix(verify): compare source rows to target only --- README.md | 36 +++++---- internal/app/verify.go | 6 +- internal/verify/cdc.go | 14 ++-- internal/verify/cdc_test.go | 16 ++-- internal/verify/localize.go | 20 ++--- internal/verify/verify.go | 9 +-- internal/verify/verify_integration_test.go | 86 ++++++++++++++-------- internal/verify/verify_test.go | 20 ++++- 8 files changed, 118 insertions(+), 89 deletions(-) diff --git a/README.md b/README.md index 9ee1f71..72696f3 100644 --- a/README.md +++ b/README.md @@ -173,7 +173,7 @@ verify e2e.orders: checking cdc, 119 of 119 applied rows verify e2e.orders: done 1115 of 1115 rows (100.0%), 12 pages read, measuring rate verify e2e.metrics: done 0 of 0 rows (0.0%), 0 pages read, measuring rate warning: "e2e"."metrics" was not compared, because it has no primary key and no usable unique index, and a sampled row can only be found on the target by key -verified 11 tables: 5236 of 5236 rows sampled, 978 of 978 applied rows checked (101 deletions), 0 divergent, 1 not compared for want of a key +verified 11 tables: 5236 of 5236 rows sampled, 978 of 978 applied rows checked (101 recorded delete keys; target-only rows ignored), 0 divergent, 1 not compared for want of a key ``` The progress lines and the summary go to standard error; the full result goes to @@ -675,10 +675,10 @@ The heap sample cannot reach the rows the applier wrote. It samples by physical position, and on a bloated heap position says nothing about write time: measured on a production shard, none of the sample's 128 windows intersected the band holding nine hours of applied changes, so the sample was validating `pg_restore` -and skipping the applier, which is the new code. Deletions make the case -strongest, because a row the source deleted and the target kept is absent from -every read of the source: on that shard, 8,201 of 8,984 recorded changes on the -busiest table were deletions from a retention job. +and skipping the applier, which is the new code. The recorded keys let verification +check those rows directly when they still exist on the source. Keys deleted from +the source do not require absence on the target: verification only checks source +rows against the target. So the applier records the identity of the rows it writes, as a reservoir sample capped at `--cdc-sample-rows` per relation, kept uniform over the whole change @@ -795,8 +795,10 @@ with a collatable partition key, where rows can route to a different partition. ## Verification -`verify` checks each selected table two ways, and reports them separately -because they answer different questions. +`verify` checks **source → target only**: each checked source row must be present +on the target with matching column values. Extra target rows do not cause a +mismatch. It uses two samples, reported separately because they cover different +rows. It **samples the heap**: it reads about a million rows from the source and looks those exact rows up on the target by key. Both sides return a hash of the whole @@ -814,13 +816,15 @@ opposite things. A relation whose recorded key does not cover the columns `verify` keys rows on — the applier keys a change on the replica identity, which may differ from the primary key — is skipped with that reason. +The same one-way rule applies to CDC keys and rechecks. A key absent from the +source is ignored even if the target still holds it, including recorded deletes. +If the key was reinserted on the source, its current row must match on the target. +The reported delete count describes recorded operations, not verified removals. + **Read this for what a pass does and does not mean.** It is a smoke test, not a -proof: it finds divergence and never proves its absence. The heap sample is also -blind in one direction, because it walks the source, so a row the target holds and -the source does not — an unapplied delete, or a duplicate — is never looked at. A -*recorded* delete is asked about on both sides and so is caught; a target-only row -nobody recorded a change for is still invisible, and finding it would need a -target-side scan. +proof: it finds divergence and never proves its absence. It does not enforce +equal table counts or target → source inclusion, and does not detect unapplied +deletes. A clean result means the source rows checked matched the target. The row budget is spread over `--verify-sample-windows` evenly spaced places in the heap, with the last window pinned to the end of it, because that is where @@ -830,7 +834,7 @@ of the same production table, so a budget spent in one place is a sample of that place rather than of the table. Each window is bounded twice, by its page interval and by a row limit, so a dense region stops early and a sparse one returns less. A table small enough to fit inside the budget is read whole, so `verify` on a small -database compares everything it holds. +database compares every row in its keyed source tables. A table with no primary key and no `NOT NULL` unique index **cannot be checked at all**, because there is nothing to look its rows up on the target by. It is @@ -998,8 +1002,8 @@ controlled by `PGMIGRATE_CDC_BENCH_TRANSACTIONS` and - The delivered e2e bed is PostgreSQL 17 to 17. Cross-major compatibility has focused integration probes but no full cross-major Compose migration. - Verification samples, and reports 64-bit server-side hashes rather than a - cryptographic proof. It cannot see a target-only row nobody recorded a change - for, cannot check a table with no primary key and no `NOT NULL` unique index at + cryptographic proof. It ignores all target-only rows, including unapplied + deletes, cannot check a table with no primary key and no `NOT NULL` unique index at all, and checks the rows replication wrote as a capped sample of what the applier reported rather than exhaustively from the decoded stream. - Cutover enforces nothing. It does not check that application writes stopped and diff --git a/internal/app/verify.go b/internal/app/verify.go index c6a3bc1..2dd414c 100644 --- a/internal/app/verify.go +++ b/internal/app/verify.go @@ -434,7 +434,7 @@ func cdcSummary(result verify.Result) string { line := fmt.Sprintf("%s of %s applied rows checked", compactCount(keys), compactCount(observed)) if deletes := result.CDCDeletes(); deletes > 0 { - line += fmt.Sprintf(" (%s deletions)", compactCount(deletes)) + line += fmt.Sprintf(" (%s recorded delete keys; target-only rows ignored)", compactCount(deletes)) } return line } @@ -472,13 +472,13 @@ func verificationDivergence(result verify.Result) string { // direction it is points at the cause: a missing row is an apply that did not // happen, a differing one an apply that happened wrongly. func diffKinds(rows []verify.RowDiff) string { - counts := make(map[verify.DiffKind]int, 3) + counts := make(map[verify.DiffKind]int, 2) for _, row := range rows { counts[row.Kind]++ } var parts []string for _, kind := range []verify.DiffKind{ - verify.DiffSourceOnly, verify.DiffTargetOnly, verify.DiffDifferent, + verify.DiffSourceOnly, verify.DiffDifferent, } { if counts[kind] > 0 { parts = append(parts, fmt.Sprintf("%d %s", counts[kind], kind)) diff --git a/internal/verify/cdc.go b/internal/verify/cdc.go index b68c7f1..5150b5b 100644 --- a/internal/verify/cdc.go +++ b/internal/verify/cdc.go @@ -49,9 +49,9 @@ type CDCResult struct { // applier saw. The ratio is this stratum's coverage. Keys int64 `json:"keys"` Observed int64 `json:"observed"` - // Deletes is how many of the checked keys were deletes. They are the only - // check either stratum makes that can catch a row the target holds and the - // source does not. + // Deletes is how many of the checked keys were recorded as deletes. These + // still check a source row if the key was later reinserted, but do not require + // the target to remove rows that are absent from the source. Deletes int64 `json:"deletes"` // Dropped is how many changes the applier could not name, and so never // offered. Without it, a relation whose every key is unrenderable looks the @@ -70,10 +70,10 @@ type CDCResult struct { // budget runs out. Comparing the two sides once beforehand would only re-find the // rows that are legitimately in flight. // -// A correctly applied delete is absent on both sides and produces nothing. An -// unapplied one is absent on the source and present on the target, which -// compareRows reports as DiffTargetOnly — the direction the heap sample cannot -// see at all, because it only ever asks about keys the source still has. +// Like the heap sample, this is a source-to-target check. A recorded key absent +// from the source produces no difference, regardless of whether the target still +// holds it or which operation was recorded. A key reinserted on the source is +// checked against its current contents, even if it was recorded as a delete. func (w *worker) verifyCDC( ctx context.Context, table Table, leaves []relation, out *TableResult, ) ([]RowDiff, error) { diff --git a/internal/verify/cdc_test.go b/internal/verify/cdc_test.go index 5a0da41..dc3c0e0 100644 --- a/internal/verify/cdc_test.go +++ b/internal/verify/cdc_test.go @@ -41,22 +41,16 @@ func TestProjectKeyRefusesAKeyItCannotCover(t *testing.T) { } } -// TestCompareRowsNamesAnUnappliedDelete is the whole reason the CDC stratum -// carries delete keys. The heap sample walks the source, so a row the target -// holds and the source does not is invisible to it by construction; asking both -// sides about the same key is what makes it visible. -func TestCompareRowsNamesAnUnappliedDelete(t *testing.T) { +// A recorded key that is absent on the source is not required to be absent on +// the target, even when the CDC check asks both sides about it. +func TestCompareRowsIgnoresAnUnappliedDelete(t *testing.T) { t.Parallel() target := rowSet{ identity([]string{"1", "gone"}): {key: []string{"1", "gone"}, hash: 7}, } diffs := compareRows(rowSet{}, target) - if len(diffs) != 1 { - t.Fatalf("compareRows() found %d differences, want 1", len(diffs)) - } - if diffs[0].Kind != DiffTargetOnly { - t.Errorf("compareRows() = %s, want %s: a row only the target holds is a delete that did not apply", - diffs[0].Kind, DiffTargetOnly) + if len(diffs) != 0 { + t.Fatalf("compareRows() found %v for a row absent on the source, want none", diffs) } } diff --git a/internal/verify/localize.go b/internal/verify/localize.go index 16732ca..3fb3697 100644 --- a/internal/verify/localize.go +++ b/internal/verify/localize.go @@ -18,14 +18,12 @@ type RowDiff struct { Kind DiffKind `json:"kind"` } -// DiffKind says which way a row disagrees, which is what points at the cause: a -// missing row is an apply that did not happen, a differing row is an apply that -// happened wrongly, and an extra row is a delete that did not. +// DiffKind says whether a source row is missing from the target or has different +// contents there. Rows present only on the target are not verification failures. type DiffKind string const ( DiffSourceOnly DiffKind = "source_only" - DiffTargetOnly DiffKind = "target_only" DiffDifferent DiffKind = "different" ) @@ -205,12 +203,9 @@ func scanRows(ctx context.Context, db querier, query string, table Table, into r return rows.Err() } -// compareRows names the rows two sets disagree about. -// -// Sampling cannot produce DiffTargetOnly: the target is only ever asked about keys -// the source supplied, so a row only the target holds is never in either set. That -// is the blind spot the design accepts. A recheck can produce it, because there the -// two sides are asked about the same keys and the source row may have gone. +// compareRows checks that every source row exists on the target with matching +// contents. Extra target rows are ignored, including CDC keys and sampled rows +// that have since been deleted from the source before a recheck. func compareRows(source, target rowSet) []RowDiff { var diffs []RowDiff for key, entry := range source { @@ -222,11 +217,6 @@ func compareRows(source, target rowSet) []RowDiff { diffs = append(diffs, RowDiff{Key: entry.key, Kind: DiffDifferent}) } } - for key, entry := range target { - if _, present := source[key]; !present { - diffs = append(diffs, RowDiff{Key: entry.key, Kind: DiffTargetOnly}) - } - } slices.SortFunc(diffs, func(a, b RowDiff) int { if order := slices.Compare(a.Key, b.Key); order != 0 { return order diff --git a/internal/verify/verify.go b/internal/verify/verify.go index c1fc2a3..17691c0 100644 --- a/internal/verify/verify.go +++ b/internal/verify/verify.go @@ -12,11 +12,10 @@ // 330M row hashes for a 66-minute answer on one production shard, 57 minutes of it // a single table. See docs/design-verify-sampled.md. // -// What a sample cannot do is prove that two tables match. It finds divergence and -// it never proves its absence, and it is blind in one direction: it walks the -// source, so a row the target holds and the source does not — an unapplied delete, -// or a duplicate — is never looked at. A clean result means the rows that were -// compared agreed. +// Verification is one-way: source rows must exist on the target with matching +// contents. Extra target rows are ignored, including during CDC checks and +// rechecks. A sample cannot prove that every source row matches; a clean result +// means only that the source rows compared agreed. package verify import ( diff --git a/internal/verify/verify_integration_test.go b/internal/verify/verify_integration_test.go index 1ef52b2..1321a22 100644 --- a/internal/verify/verify_integration_test.go +++ b/internal/verify/verify_integration_test.go @@ -423,18 +423,26 @@ func TestPostgres17RecheckAttributesAnInFlightRowToReplicationLatency(t *testing if elapsed := time.Since(began); elapsed < cfg.ConvergeTimeout { t.Fatalf("the row was given up on after %s, before its %s budget", elapsed, cfg.ConvergeTimeout) } + + // If a candidate disappears from the source while rechecking, retaining it + // on the target must not turn it into a reverse-direction mismatch. + cfg.WaitApplied = func(ctx context.Context, _ string) error { + _, err := source.Exec(ctx, `DELETE FROM public.items WHERE id=11`) + return err + } + deleted, err := Run(ctx, cfg) + if err != nil { + t.Fatal(err) + } + if !deleted.Complete || !deleted.Converged || deleted.Tables[0].Candidates != 1 || + len(deleted.Tables[0].Unresolved) != 0 { + t.Fatalf("a candidate deleted from the source should converge: %#v", deleted.Tables[0]) + } } -// TestCDCStratumFindsTheDeleteTheHeapSampleCannotSee is the whole reason this -// stratum exists. -// -// The heap sample walks the source, so it only ever asks the target about keys -// the source still has. A row the source deleted and the target kept is therefore -// invisible to it by construction — the one direction an unapplied delete shows -// up in. Asking both sides about a key the applier recorded is what makes it -// visible, and this test asserts both halves: the sample stays clean and the -// stratum reports the row. -func TestCDCStratumFindsTheDeleteTheHeapSampleCannotSee(t *testing.T) { +// CDC checks obey the same source-to-target rule as the heap sample, regardless +// of the recorded operation. They still reject missing or changed source rows. +func TestCDCStratumRequiresOnlyCurrentSourceRows(t *testing.T) { sourceInstance := pgtest.Start(t, 17) targetInstance := pgtest.Start(t, 17) ctx := context.Background() @@ -447,9 +455,9 @@ func TestCDCStratumFindsTheDeleteTheHeapSampleCannotSee(t *testing.T) { exec(t, source, ddl, insert, "ANALYZE public.notes") exec(t, target, ddl, insert, "ANALYZE public.notes") - // The source drops a row and the target does not, which is what an applier - // that failed to replay a delete leaves behind. - exec(t, source, `DELETE FROM public.notes WHERE app_pk=1 AND id='n42'`) + // Rows retained only by the target are allowed, including keys recorded as + // inserts or updates that were subsequently deleted on the source. + exec(t, source, `DELETE FROM public.notes WHERE app_pk=1 AND id IN ('n42','n43','n44')`) tables := inventoryOf(t, source, "public", "notes") base := Config{ @@ -473,9 +481,11 @@ func TestCDCStratumFindsTheDeleteTheHeapSampleCannotSee(t *testing.T) { withCDC.CDCRows = 100 withCDC.CDCKeys = func(context.Context, string, string) (CDCRecorded, error) { return CDCRecorded{ - Observed: 3, + Observed: 5, Keys: []CDCKey{ {Key: map[string]string{"app_pk": "1", "id": "n42"}, Kind: "delete"}, + {Key: map[string]string{"app_pk": "1", "id": "n43"}, Kind: "insert"}, + {Key: map[string]string{"app_pk": "1", "id": "n44"}, Kind: "update"}, {Key: map[string]string{"app_pk": "1", "id": "n7"}, Kind: "insert"}, {Key: map[string]string{"app_pk": "1", "id": "n99"}, Kind: "update"}, }, @@ -485,24 +495,14 @@ func TestCDCStratumFindsTheDeleteTheHeapSampleCannotSee(t *testing.T) { if err != nil { t.Fatal(err) } - if seen.Converged { - t.Fatal("the CDC stratum missed a delete the target never applied") - } - unresolved := seen.Tables[0].Unresolved - if len(unresolved) != 1 { - t.Fatalf("unresolved = %#v, want only the unapplied delete", unresolved) - } - if unresolved[0].Kind != DiffTargetOnly { - t.Errorf("unapplied delete reported as %q, want %q", unresolved[0].Kind, DiffTargetOnly) + if !seen.Converged || !seen.Complete || len(seen.Tables[0].Unresolved) != 0 { + t.Fatalf("target-only rows should not fail CDC verification: %#v", seen.Tables[0]) } - if got := unresolved[0].Key; len(got) != 2 || got[0] != "1" || got[1] != "n42" { - t.Errorf("unapplied delete named %v, want [1 n42]", got) + if cdc := seen.Tables[0].CDC; cdc.Keys != 5 || cdc.Observed != 5 || cdc.Deletes != 1 || cdc.Candidates != 0 { + t.Errorf("CDC result = %+v, want 5 keys of 5 observed with 1 delete and no candidates", cdc) } - if cdc := seen.Tables[0].CDC; cdc.Keys != 3 || cdc.Observed != 3 || cdc.Deletes != 1 { - t.Errorf("CDC result = %+v, want 3 keys of 3 observed with 1 delete", cdc) - } - if seen.CDCKeys() != 3 || seen.CDCObserved() != 3 { - t.Errorf("run totals = %d of %d, want 3 of 3", seen.CDCKeys(), seen.CDCObserved()) + if seen.CDCKeys() != 5 || seen.CDCObserved() != 5 { + t.Errorf("run totals = %d of %d, want 5 of 5", seen.CDCKeys(), seen.CDCObserved()) } // A delete both sides applied is absent on both, and absent on both is @@ -515,4 +515,30 @@ func TestCDCStratumFindsTheDeleteTheHeapSampleCannotSee(t *testing.T) { if !clean.Converged { t.Fatalf("a correctly applied delete was reported: %#v", clean.Tables[0].Unresolved) } + + // A recorded delete key can now exist again. Check its current value along + // with ordinary insert/update keys; the operation must not bypass comparison. + exec(t, source, `INSERT INTO public.notes VALUES (1,'n42','reinserted')`) + exec(t, target, + `INSERT INTO public.notes VALUES (1,'n42','stale')`, + `DELETE FROM public.notes WHERE app_pk=1 AND id='n7'`, + `UPDATE public.notes SET body='wrong' WHERE app_pk=1 AND id='n99'`) + diverged, err := Run(ctx, withCDC) + if err != nil { + t.Fatal(err) + } + if diverged.Converged || !diverged.Complete || diverged.Tables[0].CDC.Candidates != 3 { + t.Fatalf("CDC should reject all three mismatching source rows: %#v", diverged.Tables[0]) + } + want := map[string]DiffKind{"n42": DiffDifferent, "n7": DiffSourceOnly, "n99": DiffDifferent} + found := make(map[string]bool) + for _, diff := range diverged.Tables[0].Unresolved { + if len(diff.Key) != 2 || diff.Key[0] != "1" || want[diff.Key[1]] != diff.Kind { + t.Fatalf("unexpected divergence: %#v", diff) + } + found[diff.Key[1]] = true + } + if len(found) != len(want) { + t.Fatalf("divergent keys = %v, want %v", found, want) + } } diff --git a/internal/verify/verify_test.go b/internal/verify/verify_test.go index 4f2065a..121498a 100644 --- a/internal/verify/verify_test.go +++ b/internal/verify/verify_test.go @@ -171,7 +171,7 @@ func TestPageChunkWhereIsHalfOpenAndInlinesPages(t *testing.T) { } } -func TestCompareRowsNamesEveryDirection(t *testing.T) { +func TestCompareRowsRequiresSourceRowsAndIgnoresExtraTargetRows(t *testing.T) { t.Parallel() source := rowSet{ identity([]string{"1"}): {key: []string{"1"}, hash: 10}, @@ -187,7 +187,6 @@ func TestCompareRowsNamesEveryDirection(t *testing.T) { want := []RowDiff{ {Key: []string{"2"}, Kind: DiffDifferent}, {Key: []string{"3"}, Kind: DiffSourceOnly}, - {Key: []string{"4"}, Kind: DiffTargetOnly}, } if len(got) != len(want) { t.Fatalf("compareRows() = %v", got) @@ -199,6 +198,23 @@ func TestCompareRowsNamesEveryDirection(t *testing.T) { } } +func TestCompareRowsAcceptsATargetSuperset(t *testing.T) { + t.Parallel() + target := rowSet{ + identity([]string{"1"}): {key: []string{"1"}, hash: 10}, + identity([]string{"2"}): {key: []string{"2"}, hash: 20}, + } + for _, source := range []rowSet{ + nil, + {}, + {identity([]string{"1"}): {key: []string{"1"}, hash: 10}}, + } { + if diffs := compareRows(source, target); len(diffs) != 0 { + t.Fatalf("compareRows(%v, %v) = %v, want no differences", source, target, diffs) + } + } +} + func TestCandidateKeysBoundsWhatOneTableReports(t *testing.T) { t.Parallel() diffs := make([]RowDiff, 10) From 3ab27ca73005359ef63e10598fb048d03fbd24fe Mon Sep 17 00:00:00 2001 From: thesyncim Date: Fri, 28 Aug 2026 11:37:25 +0100 Subject: [PATCH 2/4] fix(verify): defer CDC mismatches once and audit target progress --- .github/workflows/ci.yml | 2 + README.md | 85 ++- internal/app/app.go | 10 +- internal/app/verify.go | 29 +- internal/app/verify_audit.go | 73 +++ internal/app/verify_audit_test.go | 151 ++++++ internal/app/verify_test.go | 50 ++ internal/cli/cli.go | 2 +- internal/controller/controller_test.go | 1 + internal/controller/ui.html | 2 +- internal/controller/ui_test.mjs | 63 +++ internal/verify/audit.go | 99 ++++ internal/verify/cdc.go | 51 +- internal/verify/deferred.go | 132 +++++ internal/verify/deferred_integration_test.go | 533 +++++++++++++++++++ internal/verify/deferred_test.go | 61 +++ internal/verify/localize.go | 47 +- internal/verify/progress.go | 11 +- internal/verify/verify.go | 80 ++- internal/verify/verify_integration_test.go | 1 + 20 files changed, 1419 insertions(+), 64 deletions(-) create mode 100644 internal/app/verify_audit.go create mode 100644 internal/app/verify_audit_test.go create mode 100644 internal/app/verify_test.go create mode 100644 internal/controller/ui_test.mjs create mode 100644 internal/verify/audit.go create mode 100644 internal/verify/deferred.go create mode 100644 internal/verify/deferred_integration_test.go create mode 100644 internal/verify/deferred_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b7dc337..0bed3ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,6 +27,8 @@ jobs: echo "$unformatted" >&2 exit 1 fi + - name: controller verification UI + run: node --test internal/controller/ui_test.mjs - run: make vet # Reachability, not Dependabot's module list: the released binary, not the # integration-tagged testcontainers graph. diff --git a/README.md b/README.md index 72696f3..fdd71d9 100644 --- a/README.md +++ b/README.md @@ -453,13 +453,13 @@ standard error. A named divergence, or a table stopped early, exits non-zero. | `--dir ` | required | migration state directory holding the table inventory | | `--source ` | `PGMIGRATE_SOURCE` | source connection string | | `--target ` | `PGMIGRATE_TARGET` | target connection string | -| `--verify-workers ` | `1` | tables checked in parallel. Each one reads the live source, which is why this is not `--workers` | +| `--verify-workers ` | `1` | initial table scans in parallel, plus one worker for deferred CDC key lookups. Each reads the live source, which is why this is not `--workers` | | `--verify-sample-rows ` | `1000000` | rows per table read from the source and looked up on the target. A table smaller than this is read whole. `0` is rejected: there is no exhaustive mode | | `--verify-sample-windows ` | `128` | page intervals those rows are drawn from, spread across the heap with the last pinned to its end | | `--verify-batch-rows ` | `5000` | keys per target lookup statement, clamped down for a very wide key to stay under the bind-parameter limit | | `--verify-duty-cycle ` | `1` | fraction of the wall clock verification may spend querying, sleeping between windows to stay under it. Must be greater than 0 and at most 1 | -| `--verify-table-timeout ` | `20m` | time one table's check may take. `0` disables it. A table stopped here reports incomplete and cannot report convergence | -| `--verify-converge-timeout ` | `1m` | how long a row that appears to differ is given to settle against a fixed WAL position before it is reported | +| `--verify-table-timeout ` | `20m` | active time one table's check may take, including the single CDC recheck, excluding its delay and queue time. `0` disables it. A table stopped here reports incomplete and cannot report convergence | +| `--verify-converge-timeout ` | `1m` | how long a heap-sample row that appears to differ is given to settle against a fixed WAL position before it is reported; CDC mismatches use the separate one-minute deferred check | | `--verify-cdc-rows ` | `100000` | applier-recorded keys per table checked alongside the heap sample. `0` falls back to the default | ### pgmigrate sequences @@ -692,8 +692,8 @@ names a row. ### Reading a live source -A row read from a live source, on a target that is still applying, is *expected* -to differ, and waiting for it to settle would never terminate for a row written +A heap-sample row read from a live source, on a target that is still applying, +is *expected* to differ, and waiting for it to settle would never terminate for a row written to constantly. What settles it is fixing a position rather than a moment: the source rows are re-read, a decodable marker names a WAL position at or after that read, and the target rows are read once apply has passed it. A row that still @@ -708,6 +708,11 @@ nothing until it reaches disk, so on PostgreSQL 17 and later it is written with `pg_logical_emit_message(flush => true)`, and on 16, which has no such argument, a small committed message immediately after forces the same flush. +CDC-key mismatches instead use the one-minute deferred check described under +[Verification](#verification): unchanged source rows require a match or target +advancement, and changed source rows require target advancement. The heap +WAL-marker retry above is not used for CDC-key candidates. + ### The target is tuned for a bulk load, and put back Stock checkpoint settings are the dominant cost of a large load: at the default @@ -795,10 +800,11 @@ with a collatable partition key, where rows can route to a different partition. ## Verification -`verify` checks **source → target only**: each checked source row must be present -on the target with matching column values. Extra target rows do not cause a -mismatch. It uses two samples, reported separately because they cover different -rows. +`verify` checks **source → target only**: extra target rows do not cause a +mismatch. Heap-sample rows must match their target counterparts. The live CDC +check also accepts target rows that advanced during its one-minute observation, +as described below. These two samples are reported separately because they +cover different rows. It **samples the heap**: it reads about a million rows from the source and looks those exact rows up on the target by key. Both sides return a hash of the whole @@ -807,24 +813,73 @@ type handling cannot cancel itself out across the comparison, and a missing row and a wrongly applied column value are the same finding. It also **checks the rows replication wrote**, by key, from what the applier -recorded as it wrote them, using the same recheck rule against a fixed WAL -position. `--verify-cdc-rows` bounds how many of those keys one table's check -looks at, and the check reports what it looked at against what the applier saw, +recorded as it wrote them. `--verify-cdc-rows` bounds how many of those keys one +table's check looks at, and the check reports what it looked at against what the applier saw, so a truncated check says so rather than reading as complete. An empty reservoir reports "no applied rows recorded", never "0 checked", because those mean opposite things. A relation whose recorded key does not cover the columns `verify` keys rows on — the applier keys a change on the replica identity, which may differ from the primary key — is skipped with that reason. -The same one-way rule applies to CDC keys and rechecks. A key absent from the -source is ignored even if the target still holds it, including recorded deletes. +A CDC key absent from the source at the initial read is ignored even if the +target still holds it, including recorded deletes. If the key was reinserted on the source, its current row must match on the target. The reported delete count describes recorded operations, not verified removals. +**CDC mismatches are deferred for at least one minute.** The initial observation +captures each candidate's source hash and row version (`xmin`), and audits both +source and target hashes/presence. A dedicated worker rechecks those exact keys +after the delay while other tables continue scanning. It reads the source, then +the target, then the source again. If the source changed or disappeared since +the initial observation, or changed during the target read, **the target must +have advanced**; the source change alone does not clear the mismatch. Checking +`xmin` also catches no-op updates and changes reverted to the same contents. +Source transaction IDs are never compared to target IDs. + +For source rows that stayed untouched, matching target hashes count as converged. +If the target still differs but a row appeared or its hash/`xmin` changed since +the initial target snapshot, it **advanced**: accept it as progressing and log +that outcome separately. Advancement is accepted by this live CDC check even +though row equality has not been established. A target that neither matches nor +advances is unresolved and fails verification. If the source changed, an +unchanged target fails as `target_stalled`, even if the source changed to the +target's old value. A target deletion counts as advancement only if the source +also disappeared; deleting a target row still required by the source is not +advancement. + +There is **exactly one deferred recheck**, with no retry/defer loop. Advancing +rows are not queued again. Source changes are counted and flagged in the audit; +they never grant an automatic skip. Advancing targets are counted separately +from matched rows. +The result finishes once those outcomes are recorded. The one-minute delay and +queue time do not consume `--verify-table-timeout`, but the reads do. Cancellation +or timeout reports incomplete. `--verify-converge-timeout` only controls the +separate heap-sample retries. + +**Every observed mismatch is audited**, including ones that later converge or +are accepted as advancing, in `/log/verify-audit.jsonl`. This append-only JSONL file is +separate from the controller's bounded output buffer and retained across runs. +Records carry a run ID, UTC timestamp, table, key, stratum, mismatch kind, row +presence, hashes, and row versions where applicable. CDC outcome records also +carry the original source metadata, initial target metadata, and both later +source reads, a `source_changed` flag, and outcomes `converged`, `advanced`, +`unresolved`, or `incomplete`. Incomplete outcomes retain the original source +metadata; earlier records retain the last observed target. Run start/end records +distinguish clean, divergent, and incomplete runs; an interrupted process may +leave a start without an end. All initial heap mismatches are logged before the +heap recheck threshold is applied, and heap recheck observations are logged too. + +Audit batches are synced before verification proceeds; a write failure aborts +the run. The file is created with mode `0600` and contains keys and comparison +metadata, not full row contents. Keep it on durable storage and treat it as +sensitive. It is not automatically truncated or rotated. + **Read this for what a pass does and does not mean.** It is a smoke test, not a proof: it finds divergence and never proves its absence. It does not enforce equal table counts or target → source inclusion, and does not detect unapplied -deletes. A clean result means the source rows checked matched the target. +deletes. A passing CDC check may include advancing targets that do not yet match. +These are reported separately from rows verified equal; source changes do not +excuse a stalled target. The row budget is spread over `--verify-sample-windows` evenly spaced places in the heap, with the last window pinned to the end of it, because that is where diff --git a/internal/app/app.go b/internal/app/app.go index 52bb96a..02c19fb 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -2348,7 +2348,7 @@ func (a App) Status(ctx context.Context, cfg config.Config) error { } } -func verification(ctx context.Context, cfg config.Config, store *state.Store, progressOut io.Writer) (verify.Result, error) { +func verification(ctx context.Context, cfg config.Config, store *state.Store, progressOut io.Writer) (result verify.Result, runErr error) { tables, err := store.ListTables(ctx) if err != nil { return verify.Result{}, err @@ -2369,13 +2369,19 @@ func verification(ctx context.Context, cfg config.Config, store *state.Store, pr } boundary := newMarker(cfg, "verify:", capabilities) defer boundary.close() - // While the migration is following, a row that differs may simply be in flight. + // While the migration is following, a heap-sample row may simply be in flight. // These two hooks are what tell that apart from a defect: mark a source // position, wait for apply to pass it, look again. mark, wait := recheckHooks(cfg, boundary, migration.SlotName, migration.Phase) + audit, err := newVerificationAudit(cfg.Dir) + if err != nil { + return verify.Result{}, err + } + defer func() { runErr = errors.Join(runErr, audit.finish(result.Complete, result.Converged, runErr)) }() return verify.Run(ctx, verify.Config{ Source: connector(cfg.Source), Target: connector(cfg.Target), Tables: verifyTables, Progress: progress, + Audit: audit.write, Workers: cfg.VerifyWorkers, SampleRows: cfg.VerifySampleRows, SampleWindows: cfg.VerifySampleWindows, diff --git a/internal/app/verify.go b/internal/app/verify.go index 2dd414c..8727d26 100644 --- a/internal/app/verify.go +++ b/internal/app/verify.go @@ -229,7 +229,8 @@ func (p *verifyProgress) Update(update verify.Progress) { "rows_per_second": update.Rate, "eta": update.ETA.String(), "coverage": update.Coverage, "candidate_rows": update.Candidates, "cdc_keys": update.CDCKeys, "cdc_observed": update.CDCObserved, - "unresolved": update.Unresolved, + "cdc_pending_rows": update.CDCPending, + "unresolved": update.Unresolved, }) if p.store == nil || record == nil { return @@ -285,9 +286,7 @@ func (p *verifyProgress) record(update verify.Progress) *state.VerifyTable { into.CDCObserved = update.CDCObserved } into.Unresolved = int64(update.Unresolved) - if update.Stage == verify.StageDone { - into.Converged, into.Complete = update.Converged, update.Complete - } + into.Converged, into.Complete = update.Converged, update.Complete copied := *into return &copied } @@ -296,6 +295,11 @@ func (p *verifyProgress) render(update verify.Progress) { if p.out == nil { return } + if update.Stage == verify.StageCDCDeferred || update.Stage == verify.StageCDCRechecking { + p.line(fmt.Sprintf("verify %s: %s, %d CDC rows pending (not final divergence)", + update.Table, update.Stage, update.CDCPending)) + return + } // The CDC stratum checks rows the heap sample cannot reach, so it reports its // own two numbers. Falling through to the line below would repeat the sample's // counts and read as progress through a sample that has already finished. @@ -436,6 +440,21 @@ func cdcSummary(result verify.Result) string { if deletes := result.CDCDeletes(); deletes > 0 { line += fmt.Sprintf(" (%s recorded delete keys; target-only rows ignored)", compactCount(deletes)) } + var changed, pending, advanced int + for _, table := range result.Tables { + changed += table.CDC.SourceChanged + pending += table.CDC.Pending + advanced += table.CDC.Advanced + } + if changed > 0 { + line += fmt.Sprintf("; %d source-changed CDC rows required target advancement", changed) + } + if advanced > 0 { + line += fmt.Sprintf("; %d CDC target rows advanced without matching; accepted as progressing", advanced) + } + if pending > 0 { + line += fmt.Sprintf("; %d CDC rows still pending", pending) + } return line } @@ -478,7 +497,7 @@ func diffKinds(rows []verify.RowDiff) string { } var parts []string for _, kind := range []verify.DiffKind{ - verify.DiffSourceOnly, verify.DiffDifferent, + verify.DiffSourceOnly, verify.DiffDifferent, verify.DiffTargetStalled, } { if counts[kind] > 0 { parts = append(parts, fmt.Sprintf("%d %s", counts[kind], kind)) diff --git a/internal/app/verify_audit.go b/internal/app/verify_audit.go new file mode 100644 index 0000000..fd21e16 --- /dev/null +++ b/internal/app/verify_audit.go @@ -0,0 +1,73 @@ +package app + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sync" + "time" + + "github.com/GetStream/pgmigrate/internal/verify" +) + +// verificationAudit is independent of the controller's truncated output buffer. +// Each batch is synced before verification continues; an audit failure is fatal. +type verificationAudit struct { + mu sync.Mutex + file *os.File + runID string +} + +func newVerificationAudit(dir string) (*verificationAudit, error) { + path := filepath.Join(dir, "log", "verify-audit.jsonl") + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return nil, err + } + file, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600) + if err != nil { + return nil, fmt.Errorf("open verification audit: %w", err) + } + var id [16]byte + if _, err := rand.Read(id[:]); err != nil { + file.Close() + return nil, err + } + out := &verificationAudit{file: file, runID: hex.EncodeToString(id[:])} + if err := out.write([]verify.AuditEvent{{Time: time.Now().UTC(), Outcome: "run_started"}}); err != nil { + file.Close() + return nil, err + } + return out, nil +} + +func (a *verificationAudit) write(events []verify.AuditEvent) error { + a.mu.Lock() + defer a.mu.Unlock() + enc := json.NewEncoder(a.file) + for _, event := range events { + record := struct { + RunID string `json:"run_id"` + verify.AuditEvent + }{a.runID, event} + if err := enc.Encode(record); err != nil { + return err + } + } + return a.file.Sync() +} + +func (a *verificationAudit) finish(complete, converged bool, runErr error) error { + outcome := "run_incomplete" + if runErr == nil && complete { + outcome = "run_diverged" + if converged { + outcome = "run_converged" + } + } + err := a.write([]verify.AuditEvent{{Time: time.Now().UTC(), Outcome: outcome}}) + return errors.Join(err, a.file.Close()) +} diff --git a/internal/app/verify_audit_test.go b/internal/app/verify_audit_test.go new file mode 100644 index 0000000..ec0470a --- /dev/null +++ b/internal/app/verify_audit_test.go @@ -0,0 +1,151 @@ +package app + +import ( + "bufio" + "encoding/json" + "errors" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/GetStream/pgmigrate/internal/verify" +) + +func TestVerificationAuditAppendsEveryConcurrentObservation(t *testing.T) { + dir := t.TempDir() + var ids []string + for run := 0; run < 2; run++ { + audit, err := newVerificationAudit(dir) + if err != nil { + t.Fatal(err) + } + ids = append(ids, audit.runID) + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + events := []verify.AuditEvent{ + {Time: time.Now().UTC(), Table: "public.items", Key: []string{"quoted\"key\n"}, Stratum: "cdc", Outcome: "deferred", Kind: verify.DiffDifferent, Source: &verify.RowSnapshot{Present: true, Hash: "9223372036854775807", Version: "123"}, Target: &verify.RowSnapshot{Present: true, Hash: "0"}}, + {Time: time.Now().UTC(), Table: "public.items", Key: []string{"quoted\"key\n"}, Stratum: "cdc", Outcome: "converged"}, + } + if err := audit.write(events); err != nil { + t.Error(err) + } + }() + } + wg.Wait() + if err := audit.finish(true, true, nil); err != nil { + t.Fatal(err) + } + } + if ids[0] == ids[1] || ids[0] == "" { + t.Fatalf("run IDs not unique: %v", ids) + } + path := filepath.Join(dir, "log", "verify-audit.jsonl") + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0600 { + t.Fatalf("audit exposes keys with permissions %v", info.Mode()) + } + f, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + defer f.Close() + scanner := bufio.NewScanner(f) + counts := make(map[string]map[string]int) + for scanner.Scan() { + var event struct { + RunID string `json:"run_id"` + verify.AuditEvent + } + if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { + t.Fatalf("interleaved/invalid JSON: %v", err) + } + if event.Time.IsZero() { + t.Fatal("missing timestamp") + } + if counts[event.RunID] == nil { + counts[event.RunID] = make(map[string]int) + } + counts[event.RunID][event.Outcome]++ + if event.Outcome == "deferred" && (event.Source.Hash != "9223372036854775807" || event.Source.Version != "123" || event.Key[0] != "quoted\"key\n") { + t.Fatalf("metadata corrupted: %+v", event) + } + } + if err := scanner.Err(); err != nil { + t.Fatal(err) + } + for _, id := range ids { + c := counts[id] + if c["run_started"] != 1 || c["run_converged"] != 1 || c["deferred"] != 8 || c["converged"] != 8 { + t.Fatalf("lost observations for %s: %v", id, c) + } + } +} + +func TestVerificationAuditFinalOutcomesAndWriteFailures(t *testing.T) { + for _, tc := range []struct { + name string + complete, converged bool + err error + want string + }{ + {name: "converged", complete: true, converged: true, want: "run_converged"}, + {name: "diverged", complete: true, want: "run_diverged"}, + {name: "incomplete", want: "run_incomplete"}, + {name: "error", complete: true, converged: true, err: errors.New("stopped"), want: "run_incomplete"}, + } { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + audit, err := newVerificationAudit(dir) + if err != nil { + t.Fatal(err) + } + if err := audit.finish(tc.complete, tc.converged, tc.err); err != nil { + t.Fatal(err) + } + f, err := os.Open(filepath.Join(dir, "log", "verify-audit.jsonl")) + if err != nil { + t.Fatal(err) + } + defer f.Close() + dec := json.NewDecoder(f) + var e verify.AuditEvent + if err := dec.Decode(&e); err != nil { + t.Fatal(err) + } + if err := dec.Decode(&e); err != nil { + t.Fatal(err) + } + if e.Outcome != tc.want { + t.Fatalf("footer=%s want %s", e.Outcome, tc.want) + } + if err := audit.write([]verify.AuditEvent{{Outcome: "deferred"}}); err == nil { + t.Fatal("closed audit write succeeded") + } + }) + } + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "log"), []byte("not a directory"), 0600); err != nil { + t.Fatal(err) + } + if _, err := newVerificationAudit(dir); err == nil { + t.Fatal("invalid audit path silently ignored") + } + audit, err := newVerificationAudit(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := audit.file.Close(); err != nil { + t.Fatal(err) + } + if err := audit.finish(true, true, nil); err == nil { + t.Fatal("footer failure ignored") + } +} diff --git a/internal/app/verify_test.go b/internal/app/verify_test.go new file mode 100644 index 0000000..eaa4aa2 --- /dev/null +++ b/internal/app/verify_test.go @@ -0,0 +1,50 @@ +package app + +import ( + "bytes" + "strings" + "testing" + + "github.com/GetStream/pgmigrate/internal/config" + "github.com/GetStream/pgmigrate/internal/state" + "github.com/GetStream/pgmigrate/internal/verify" +) + +func TestDeferredCDCProgressIsNotCompleteOrDivergent(t *testing.T) { + t.Parallel() + var output bytes.Buffer + p := newVerifyProgress(config.Config{Dir: t.TempDir()}, nil, &output, []state.Table{{OID: 1, Schema: "public", Name: "items"}}) + p.Update(verify.Progress{Table: "public.items", Stage: verify.StageDone, Complete: true, Converged: true}) + p.Update(verify.Progress{Table: "public.items", Stage: verify.StageCDCDeferred, CDCPending: 41, CDCKeys: 100}) + got := p.merged["public.items"] + if got.Complete || got.Converged || got.Unresolved != 0 || got.Stage != string(verify.StageCDCDeferred) { + t.Fatalf("pending progress = %+v", got) + } + if !strings.Contains(output.String(), "41 CDC rows pending (not final divergence)") { + t.Fatalf("missing provisional explanation: %s", output.String()) + } + p.Update(verify.Progress{Table: "public.items", Stage: verify.StageCDCRechecking, CDCPending: 41}) + if got.Complete || got.Converged { + t.Fatalf("rechecking marked complete: %+v", got) + } + p.Update(verify.Progress{Table: "public.items", Stage: verify.StageDone, Complete: true, Converged: true}) + if !got.Complete || !got.Converged || got.Unresolved != 0 { + t.Fatalf("final progress = %+v", got) + } +} + +func TestCDCSummaryDistinguishesChangedAdvancedAndPending(t *testing.T) { + result := verify.Result{Tables: []verify.TableResult{{CDC: verify.CDCResult{Keys: 10, Observed: 20, SourceChanged: 2, Pending: 3, Advanced: 4}}}} + got := cdcSummary(result) + for _, want := range []string{"10 of 20 applied rows checked", "2 source-changed CDC rows required target advancement", "3 CDC rows still pending", "4 CDC target rows advanced without matching; accepted as progressing"} { + if !strings.Contains(got, want) { + t.Errorf("summary %q missing %q", got, want) + } + } +} + +func TestTargetStalledFailureIsReported(t *testing.T) { + if got := diffKinds([]verify.RowDiff{{Kind: verify.DiffTargetStalled}}); got != "1 target_stalled" { + t.Fatalf("stalled-target failure hidden: %q", got) + } +} diff --git a/internal/cli/cli.go b/internal/cli/cli.go index b2039ce..30a44e5 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -78,7 +78,7 @@ func NewRootCommand() *cobra.Command { flags.Int64Var(&cfg.VerifyBatchRows, "verify-batch-rows", cfg.VerifyBatchRows, "keys per target lookup statement") flags.Float64Var(&cfg.VerifyDutyCycle, "verify-duty-cycle", cfg.VerifyDutyCycle, "fraction of the time verification may spend querying, sleeping between windows to stay under it") flags.DurationVar(&cfg.VerifyTableTimeout, "verify-table-timeout", cfg.VerifyTableTimeout, "time one table's check may take (0 disables)") - flags.DurationVar(&cfg.VerifyConvergeTimeout, "verify-converge-timeout", cfg.VerifyConvergeTimeout, "how long a differing row is given to settle before it is reported") + flags.DurationVar(&cfg.VerifyConvergeTimeout, "verify-converge-timeout", cfg.VerifyConvergeTimeout, "how long a differing heap-sample row is given to settle before it is reported") flags.Int64Var(&cfg.VerifyCDCRows, "verify-cdc-rows", cfg.VerifyCDCRows, "applier-recorded keys per table checked alongside the heap sample") flags.Int64Var(&cfg.CDCSampleRows, "cdc-sample-rows", cfg.CDCSampleRows, "applied keys kept per relation for verification to check the replication path (0 records none)") diff --git a/internal/controller/controller_test.go b/internal/controller/controller_test.go index b75ef89..60d4960 100644 --- a/internal/controller/controller_test.go +++ b/internal/controller/controller_test.go @@ -773,6 +773,7 @@ func TestIndexContainsControllerProgressUI(t *testing.T) { for _, want := range []string{ "pgmigrate controller", "Object completion", "lifecycleBar", "Stop migration", "confirmDialog", "data-action=\"run\" disabled", "no rows compared", + "pending CDC recheck", "pending cdc recheck", "rechecking cdc", "latestCDCRecovery", "CDC files checked", "CDC validation read throughput", "Validating durable CDC segments before reconnecting source capture and target replay.", "replayTrendWarmupSeconds=15", "trend and ETA after", "resetReplaySamplesForOperation", diff --git a/internal/controller/ui.html b/internal/controller/ui.html index 88e580d..50cc0fd 100644 --- a/internal/controller/ui.html +++ b/internal/controller/ui.html @@ -202,7 +202,7 @@

Migration configuration

function phaseDetail(phase,snap){const objects=snap?.objects||{},count=name=>objects[name]||{done:0,total:0};switch(phase){case'preflight':return count('tables').total?`${fmtCount(count('tables').total)} tables inventoried`:'Checking source and target readiness';case'setup':return'Creating durable replication state';case'schema':return'Restoring the selected schema';case'copy':return`Copying parts · ${fmtCount(count('parts').done)} / ${fmtCount(count('parts').total)} (${pct(count('parts').done,count('parts').total).toFixed(1)}%)`;case'indexes':return`Indexes ${fmtCount(count('indexes').done)} / ${fmtCount(count('indexes').total)} · constraints ${fmtCount(count('constraints').done)} / ${fmtCount(count('constraints').total)}`;case'catchup':return`Catching up to the source · ${fmtBytes(snap.apply.lag_bytes)} behind`;case'follow':return`Following live writes · ${fmtBytes(snap.apply.lag_bytes)} behind`;case'drained':return'Replication drained through the cutover boundary';case'cutover':return'Finalizing sequences and cleanup';case'complete':return'Migration complete';default:return'Waiting for preflight.'}} function renderObjects(objects={}){const names=['tables','parts','indexes','constraints','verify'];el('objectCards').replaceChildren(...names.map(name=>{const v=objects[name]||{done:0,total:0},card=document.createElement('div');card.className='card';const head=document.createElement('div');head.className='card-head';const title=document.createElement('strong');title.textContent=name;const count=document.createElement('small');count.textContent=`${fmtCount(v.done)} / ${fmtCount(v.total)}`;head.append(title,count);const bar=document.createElement('div');bar.className='bar';bar.setAttribute('role','progressbar');bar.setAttribute('aria-label',`${name} completion`);bar.setAttribute('aria-valuemin','0');bar.setAttribute('aria-valuemax',String(Math.max(1,v.total)));bar.setAttribute('aria-valuenow',String(v.done));const fill=document.createElement('span');fill.style.width=`${pct(v.done,v.total)}%`;bar.append(fill);card.append(head,bar);return card}))} function renderReplayClaim(claim){const panel=el('replayClaimProgress');panel.hidden=!claim;if(!claim)return;const total=Number(claim.changes_total||claim.work_total||0),done=Number(claim.changes_total?claim.changes_done:claim.work_done||0),percent=pct(done,total);el('replayClaimBar').setAttribute('aria-valuemax',String(Math.max(1,total)));el('replayClaimBar').setAttribute('aria-valuenow',String(done));el('replayClaimFill').style.width=`${percent}%`;setText('replayClaimLabel',`${percent.toFixed(1)}% · ${fmtCount(claim.changes_done)} / ${fmtCount(claim.changes_total)} changes · ${fmtCount(claim.transactions_done)} / ${fmtCount(claim.transactions_total)} tx · ${fmtCount(claim.work_done)} / ${fmtCount(claim.work_total)} receipts`)} -function renderVerification(rows=[]){if(!rows.length){const empty=document.createElement('div');empty.className='empty';empty.textContent='No verification data yet.';el('verification').replaceChildren(empty);return}const table=document.createElement('table'),head=document.createElement('thead');head.innerHTML='TableStageSample coverageRateCDC rowsETAResult';const body=document.createElement('tbody');rows.forEach(v=>{const tr=document.createElement('tr'),coverage=Math.max(0,Math.min(1,v.coverage||0));for(const value of [v.table,v.stage||'waiting']){const td=document.createElement('td');td.textContent=value;tr.append(td)}const progress=document.createElement('td'),bar=document.createElement('div');bar.className='bar';bar.setAttribute('role','progressbar');bar.setAttribute('aria-label',`${v.table} sample coverage`);bar.setAttribute('aria-valuemin','0');bar.setAttribute('aria-valuemax','100');bar.setAttribute('aria-valuenow',String(Math.round(coverage*100)));const fill=document.createElement('span');fill.style.width=`${coverage*100}%`;bar.append(fill);const label=document.createElement('small');label.textContent=` ${(coverage*100).toFixed(2)}% · ${fmtCount(v.sampled_rows)} sampled${v.estimated_rows?` · ${fmtCount(v.estimated_rows)} estimated`:''}`;progress.append(bar,label);tr.append(progress);const rate=document.createElement('td');rate.textContent=v.rows_per_second?`${fmtCount(Math.round(v.rows_per_second))}/s`:'—';tr.append(rate);const cdc=document.createElement('td');cdc.textContent=v.cdc_observed?`${fmtCount(v.cdc_keys)} / ${fmtCount(v.cdc_observed)}`:'—';tr.append(cdc);const eta=document.createElement('td');eta.textContent=v.complete?'done':fmtDuration(v.eta);tr.append(eta);const result=document.createElement('td');if(v.unresolved_rows){result.textContent=`${fmtCount(v.unresolved_rows)} divergent`;result.style.color='var(--red)'}else if(v.complete&&coverage===0&&!v.cdc_observed){result.textContent='no rows compared';result.style.color='var(--amber)'}else if(v.complete&&v.converged){result.textContent=`converged${v.candidate_rows?` · ${fmtCount(v.candidate_rows)} rechecked`:''}`;result.style.color='var(--green)'}else if(v.complete){result.textContent='incomplete';result.style.color='var(--amber)'}else{result.textContent='—'}tr.append(result);body.append(tr)});table.append(head,body);el('verification').replaceChildren(table)} +function renderVerification(rows=[]){if(!rows.length){const empty=document.createElement('div');empty.className='empty';empty.textContent='No verification data yet.';el('verification').replaceChildren(empty);return}const table=document.createElement('table'),head=document.createElement('thead');head.innerHTML='TableStageSample coverageRateCDC rowsETAResult';const body=document.createElement('tbody');rows.forEach(v=>{const tr=document.createElement('tr'),coverage=Math.max(0,Math.min(1,v.coverage||0));for(const value of [v.table,v.stage||'waiting']){const td=document.createElement('td');td.textContent=value;tr.append(td)}const progress=document.createElement('td'),bar=document.createElement('div');bar.className='bar';bar.setAttribute('role','progressbar');bar.setAttribute('aria-label',`${v.table} sample coverage`);bar.setAttribute('aria-valuemin','0');bar.setAttribute('aria-valuemax','100');bar.setAttribute('aria-valuenow',String(Math.round(coverage*100)));const fill=document.createElement('span');fill.style.width=`${coverage*100}%`;bar.append(fill);const label=document.createElement('small');label.textContent=` ${(coverage*100).toFixed(2)}% · ${fmtCount(v.sampled_rows)} sampled${v.estimated_rows?` · ${fmtCount(v.estimated_rows)} estimated`:''}`;progress.append(bar,label);tr.append(progress);const rate=document.createElement('td');rate.textContent=v.rows_per_second?`${fmtCount(Math.round(v.rows_per_second))}/s`:'—';tr.append(rate);const cdc=document.createElement('td');cdc.textContent=v.cdc_observed?`${fmtCount(v.cdc_keys)} / ${fmtCount(v.cdc_observed)}`:'—';tr.append(cdc);const eta=document.createElement('td');eta.textContent=v.complete?'done':fmtDuration(v.eta);tr.append(eta);const result=document.createElement('td');if(!v.complete&&['pending cdc recheck','rechecking cdc'].includes(v.stage)){result.textContent=`${v.unresolved_rows?`${fmtCount(v.unresolved_rows)} divergent · `:''}pending CDC recheck`;result.style.color=v.unresolved_rows?'var(--red)':'var(--amber)'}else if(v.unresolved_rows){result.textContent=`${fmtCount(v.unresolved_rows)} divergent`;result.style.color='var(--red)'}else if(v.complete&&coverage===0&&!v.cdc_observed){result.textContent='no rows compared';result.style.color='var(--amber)'}else if(v.complete&&v.converged){result.textContent=`converged${v.candidate_rows?` · ${fmtCount(v.candidate_rows)} rechecked`:''}`;result.style.color='var(--green)'}else if(v.complete||v.stage==='done'){result.textContent='incomplete';result.style.color='var(--amber)'}else{result.textContent='—'}tr.append(result);body.append(tr)});table.append(head,body);el('verification').replaceChildren(table)} function findingCategory(f){const id=f.id||'';if(f.severity==='error')return['blocker','blocker'];if(id==='collation-version'||id==='wal-retention-unbounded')return['risk','accepted risk'];if(id.startsWith('target-tuning'))return['performance','performance only'];if(id.includes('timeout')||id.includes('replica-identity'))return['managed','managed automatically'];if(f.severity==='info')return['managed','information'];return['risk','review']} function renderFindings(data){const root=el('findings'),items=[],counts={blocker:0,managed:0,performance:0,risk:0};(data.findings||[]).forEach(f=>{const[category,label]=findingCategory(f),details=document.createElement('details'),summary=document.createElement('summary'),title=document.createElement('strong'),kind=document.createElement('span'),text=document.createElement('div');counts[category]++;details.className=`finding ${category}`;title.textContent=`${f.id}`;kind.textContent=label;summary.append(title,kind);text.textContent=f.message;details.append(summary,text);items.push(details)});if(data.failure){const f=data.failure,details=document.createElement('details'),summary=document.createElement('summary'),title=document.createElement('strong'),kind=document.createElement('span'),text=document.createElement('div'),category='blocker';counts[category]++;details.className=`finding ${category}`;details.open=true;title.textContent=`Last run failed in ${f.phase} (${f.consecutive}×)`;kind.textContent='blocker';summary.append(title,kind);text.textContent=f.detail||f.signature;details.append(summary,text);items.unshift(details)}const chips=[['blocker',`${counts.blocker} blockers`],['risk',`${counts.risk} accepted risks`],['performance',`${counts.performance} performance notes`],['managed',`${counts.managed} managed / info`]].map(([kind,label])=>{const chip=document.createElement('span');chip.className=`finding-chip ${kind}`;chip.textContent=label;return chip});el('findingSummary').replaceChildren(...chips);if(!items.length){const empty=document.createElement('div');empty.className='empty';empty.textContent='No open findings or recorded failure.';items.push(empty)}root.replaceChildren(...items)} function active(op){return ['running','stopping'].includes(op?.state)} diff --git a/internal/controller/ui_test.mjs b/internal/controller/ui_test.mjs new file mode 100644 index 0000000..46de48f --- /dev/null +++ b/internal/controller/ui_test.mjs @@ -0,0 +1,63 @@ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; +import vm from 'node:vm'; + +const html = readFileSync(new URL('./ui.html', import.meta.url), 'utf8'); +const render = html.match(/^function renderVerification\(.*$/m)[0]; + +function element() { + return { + children: [], style: {}, textContent: '', + append(...children) { this.children.push(...children); }, + replaceChildren(...children) { this.children = children; }, + setAttribute() {}, + }; +} + +function resultCell(row) { + const root = element(); + const context = vm.createContext({ + document: { createElement: element }, el: () => root, + fmtCount: value => String(value ?? 0), fmtDuration: () => '—', + }); + vm.runInContext(render, context); + context.renderVerification([{ table: 'public.items', coverage: 1, ...row }]); + return root.children[0].children[1].children[0].children[6]; +} + +test('controller script parses', () => { + const script = html.match(/