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 9ee1f71..2e0ad04 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 @@ -453,14 +453,15 @@ 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` | heap convergence budget and maximum replay confirmation wait during the single deferred CDC check; confirmation timeout reports incomplete | | `--verify-cdc-rows ` | `100000` | applier-recorded keys per table checked alongside the heap sample. `0` falls back to the default | +| `--verify-ignore-apps ` | empty | comma-separated positive `app_pk` IDs whose mismatches are audited but excluded from verification verdicts; copy, capture and replay are unchanged | ### pgmigrate sequences @@ -675,10 +676,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 @@ -692,8 +693,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 +709,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): a stable source/target match passes, and rows +that still differ require target advancement even if the source changed. 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,8 +801,11 @@ 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**: 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 @@ -805,22 +814,95 @@ 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. +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, emits +a source WAL marker, waits for replay to pass that marker, then reads the target +and the source again. This confirms replay has seen the fresh source snapshot; +elapsed time alone does not establish that. Without a live applier, the marker +and wait are omitted. **A matching target passes first**, provided +the source hash and version stayed stable across the bracketed reads. This also +passes when the source changed since the initial observation and the target +already held the matching value in its initial snapshot; no further target +write is needed. If the rows still differ and 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. + +Stable matching source/target reads 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 and no +stable match was established, an unchanged target fails as `target_stalled`. +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 reads and replay +confirmation do. Confirmation also has a hard `--verify-converge-timeout` limit, +even with table timeouts disabled. Failure to reach the marker within that budget +reports incomplete, retaining pending keys for audit, not divergence or success. +Cancellation and execution errors also cannot pass. No candidate is requeued. + +**Optional app exclusions affect verification only.** Configure +`--verify-ignore-apps 7,42`, or `verify_ignore_apps` in the controller settings, to +exclude mismatches belonging to those source `app_pk` values. The column can be +part of the key or an ordinary column. Tables without `app_pk` and rows with a +NULL or other source app remain in scope; a target app value cannot exempt them. +No rows are skipped by copy, capture, or replay. Samples are still read and +compared, and every excluded mismatch is recorded as `ignored_app` with `app_id` +and snapshots. `ignored_rows` counts observations, so a key checked by both heap +and CDC can contribute twice. The run audit records `ignored_apps`; the summary +and controller configuration banner disclose the exclusion. A clean result +applies only to the non-excluded scope. Empty settings clear the exclusion. + +**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`, `ignored_app`, or `incomplete`, and the replay confirmation boundary +when one was established. 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. 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 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 that still differs. 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 +912,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 +1080,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/app.go b/internal/app/app.go index 52bb96a..4779eba 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,23 @@ 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) + ignoredApps, err := cfg.IgnoredVerificationApps() + if err != nil { + return verify.Result{}, err + } + audit, err := newVerificationAudit(cfg.Dir, ignoredApps...) + 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, @@ -2385,6 +2395,7 @@ func verification(ctx context.Context, cfg config.Config, store *state.Store, pr ConvergeTimeout: cfg.VerifyConvergeTimeout, CDCKeys: recordedCDCKeys(store), CDCRows: cfg.VerifyCDCRows, + IgnoreApps: ignoredApps, Boundary: mark, WaitApplied: wait, }) diff --git a/internal/app/verify.go b/internal/app/verify.go index c6a3bc1..8261456 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. @@ -384,10 +388,10 @@ func verificationWarnings(result verify.Result) string { // hide it. func verificationSummary(result verify.Result) string { var ( - compared, skipped int - sampled, estimated int64 - leastName string - least = 1.0 + compared, skipped, ignored int + sampled, estimated int64 + leastName string + least = 1.0 ) for _, table := range result.Tables { if len(table.Table.Key.Columns) == 0 { @@ -395,6 +399,7 @@ func verificationSummary(result verify.Result) string { continue } compared++ + ignored += table.IgnoredRows sampled += table.Source.Rows // An unanalyzed table estimates zero rows, and a stale estimate can be // under what was read. Neither is a denominator. @@ -415,6 +420,9 @@ func verificationSummary(result verify.Result) string { if skipped > 0 { line += fmt.Sprintf(", %d not compared for want of a key", skipped) } + if ignored > 0 { + line += fmt.Sprintf("; %d mismatches ignored by application scope (audited)", ignored) + } return line + "\n" } @@ -434,7 +442,22 @@ 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)) + } + 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 checked for convergence or 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 } @@ -472,13 +495,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, 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..22ddb2e --- /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, ignoredApps ...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", IgnoredApps: ignoredApps}}); 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..6f7d939 --- /dev/null +++ b/internal/app/verify_audit_test.go @@ -0,0 +1,155 @@ +package app + +import ( + "bufio" + "encoding/json" + "errors" + "os" + "path/filepath" + "slices" + "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, "7", "42") + 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 event.Outcome == "run_started" && !slices.Equal(event.IgnoredApps, []string{"7", "42"}) { + t.Fatalf("run scope missing from audit: %+v", event) + } + 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..f098e45 --- /dev/null +++ b/internal/app/verify_test.go @@ -0,0 +1,58 @@ +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 checked for convergence or 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 TestVerificationSummaryDisclosesIgnoredMismatches(t *testing.T) { + table := verify.Table{Key: verify.Key{Columns: []verify.KeyColumn{{Name: "id"}}}} + result := verify.Result{Tables: []verify.TableResult{{Table: table, IgnoredRows: 2}, {Table: table, IgnoredRows: 3}}} + if got := verificationSummary(result); !strings.Contains(got, "5 mismatches ignored by application scope (audited)") { + t.Fatalf("missing exclusion summary: %s", got) + } +} + +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..447bcdc 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -78,8 +78,9 @@ 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, "budget for heap convergence and deferred CDC replay confirmation") flags.Int64Var(&cfg.VerifyCDCRows, "verify-cdc-rows", cfg.VerifyCDCRows, "applier-recorded keys per table checked alongside the heap sample") + flags.StringVar(&cfg.VerifyIgnoreApps, "verify-ignore-apps", cfg.VerifyIgnoreApps, "comma-separated app_pk IDs whose mismatches are audited but excluded from verification; replication is unchanged") flags.Int64Var(&cfg.CDCSampleRows, "cdc-sample-rows", cfg.CDCSampleRows, "applied keys kept per relation for verification to check the replication path (0 records none)") application := app.App{Out: root.OutOrStdout()} diff --git a/internal/cli/cli_test.go b/internal/cli/cli_test.go index d9a65e4..9808cd0 100644 --- a/internal/cli/cli_test.go +++ b/internal/cli/cli_test.go @@ -40,6 +40,23 @@ func TestReplayWorkersFlagDefaultsConservatively(t *testing.T) { } } +func TestVerificationIgnoreAppsFlagParsesAndValidates(t *testing.T) { + root := NewRootCommand() + if err := root.ParseFlags([]string{"--verify-ignore-apps=7,42"}); err != nil { + t.Fatal(err) + } + value, err := root.PersistentFlags().GetString("verify-ignore-apps") + if err != nil || value != "7,42" { + t.Fatalf("ignore apps flag = %q, %v", value, err) + } + cfg := config.FromEnvironment() + cfg.Source, cfg.Target, cfg.Dir = "postgres://source/db", "postgres://target/db", t.TempDir() + cfg.VerifyIgnoreApps = "invalid" + if err := validateDatabaseConfig(cfg); err == nil || !strings.Contains(err.Error(), "verify-ignore-apps") { + t.Fatalf("invalid app IDs accepted: %v", err) + } +} + func TestDatabaseConfigurationBoundsReplayWorkers(t *testing.T) { cfg := config.FromEnvironment() cfg.Source = "postgres://source/db" diff --git a/internal/config/config.go b/internal/config/config.go index 8ffd3a0..60b0afa 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,6 +6,8 @@ import ( "fmt" "os" "runtime" + "slices" + "strconv" "strings" "time" @@ -76,6 +78,9 @@ type Config struct { VerifyTableTimeout time.Duration VerifyConvergeTimeout time.Duration VerifyCDCRows int64 + // VerifyIgnoreApps only excludes app_pk mismatches from verification verdicts. + // It never changes base copy, CDC capture, or replay. + VerifyIgnoreApps string // CDCSampleRows bounds the reservoir of applied keys the run keeps per // relation, which is what lets verification check the replication path at @@ -107,7 +112,25 @@ func (c Config) ValidateVerify() error { case c.VerifyCDCRows < 0: return errors.New("verify-cdc-rows must not be negative (0 falls back to the default)") } - return nil + _, err := c.IgnoredVerificationApps() + return err +} + +// IgnoredVerificationApps validates and canonicalizes comma-separated app IDs. +func (c Config) IgnoredVerificationApps() ([]string, error) { + if strings.TrimSpace(c.VerifyIgnoreApps) == "" { + return nil, nil + } + var apps []string + for _, value := range strings.Split(c.VerifyIgnoreApps, ",") { + id, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + if err != nil || id <= 0 { + return nil, errors.New("verify-ignore-apps must contain comma-separated positive bigint app IDs") + } + apps = append(apps, strconv.FormatInt(id, 10)) + } + slices.Sort(apps) + return slices.Compact(apps), nil } // TuningOverrides returns the operator-supplied target tuning values, validated. diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 10a6f61..a4fc4d0 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1,12 +1,40 @@ package config_test import ( + "slices" "strings" "testing" "github.com/GetStream/pgmigrate/internal/config" ) +func TestIgnoredVerificationApps(t *testing.T) { + for _, tc := range []struct { + input string + want []string + }{ + {"", nil}, {" \t", nil}, {"7", []string{"7"}}, + {" 7,42,007 ", []string{"42", "7"}}, + } { + cfg := config.FromEnvironment() + cfg.VerifyIgnoreApps = tc.input + got, err := cfg.IgnoredVerificationApps() + if err != nil || !slices.Equal(got, tc.want) { + t.Errorf("IgnoredVerificationApps(%q) = %v, %v; want %v", tc.input, got, err, tc.want) + } + if err := cfg.ValidateVerify(); err != nil { + t.Errorf("ValidateVerify(%q): %v", tc.input, err) + } + } + for _, input := range []string{"0", "-1", "1,", ",1", "1,,2", "foo", "1.5", "9223372036854775808", "1); DROP TABLE items"} { + cfg := config.FromEnvironment() + cfg.VerifyIgnoreApps = input + if err := cfg.ValidateVerify(); err == nil || !strings.Contains(err.Error(), "verify-ignore-apps") { + t.Errorf("ValidateVerify(%q) = %v, want app ID validation error", input, err) + } + } +} + func TestFromEnvironment(t *testing.T) { t.Setenv(config.SourceEnv, "postgres://source/db") t.Setenv(config.TargetEnv, "postgres://target/db") diff --git a/internal/controller/config_persistence.go b/internal/controller/config_persistence.go index 3a83082..317820b 100644 --- a/internal/controller/config_persistence.go +++ b/internal/controller/config_persistence.go @@ -55,6 +55,7 @@ type persistedConfiguration struct { VerifyTableTimeout time.Duration `json:"verify_table_timeout"` VerifyConvergeTimeout time.Duration `json:"verify_converge_timeout"` VerifyCDCRows int64 `json:"verify_cdc_rows"` + VerifyIgnoreApps string `json:"verify_ignore_apps,omitempty"` CDCSampleRows int64 `json:"cdc_sample_rows"` } @@ -76,7 +77,8 @@ func persistedConfigurationFrom(cfg config.Config) persistedConfiguration { VerifySampleWindows: cfg.VerifySampleWindows, VerifyBatchRows: cfg.VerifyBatchRows, VerifyDutyCycle: cfg.VerifyDutyCycle, VerifyTableTimeout: cfg.VerifyTableTimeout, VerifyConvergeTimeout: cfg.VerifyConvergeTimeout, VerifyCDCRows: cfg.VerifyCDCRows, - CDCSampleRows: cfg.CDCSampleRows, + VerifyIgnoreApps: cfg.VerifyIgnoreApps, + CDCSampleRows: cfg.CDCSampleRows, } } @@ -111,6 +113,7 @@ func (persisted persistedConfiguration) merge(base config.Config) config.Config base.VerifyTableTimeout = persisted.VerifyTableTimeout base.VerifyConvergeTimeout = persisted.VerifyConvergeTimeout base.VerifyCDCRows = persisted.VerifyCDCRows + base.VerifyIgnoreApps = persisted.VerifyIgnoreApps base.CDCSampleRows = persisted.CDCSampleRows return base } diff --git a/internal/controller/controller.go b/internal/controller/controller.go index 8f41d89..6cb644f 100644 --- a/internal/controller/controller.go +++ b/internal/controller/controller.go @@ -188,6 +188,7 @@ type configurationView struct { VerifyTableTimeout string `json:"verify_table_timeout"` VerifyConvergeTimeout string `json:"verify_converge_timeout"` VerifyCDCRows int64 `json:"verify_cdc_rows"` + VerifyIgnoreApps string `json:"verify_ignore_apps"` CDCSampleRows int64 `json:"cdc_sample_rows"` } @@ -227,6 +228,7 @@ type configurationUpdate struct { VerifyTableTimeout *string `json:"verify_table_timeout"` VerifyConvergeTimeout *string `json:"verify_converge_timeout"` VerifyCDCRows *int64 `json:"verify_cdc_rows"` + VerifyIgnoreApps *string `json:"verify_ignore_apps"` CDCSampleRows *int64 `json:"cdc_sample_rows"` } @@ -717,6 +719,7 @@ func applyConfigurationUpdate(candidate *config.Config, update configurationUpda setIfPresent(&candidate.VerifyBatchRows, update.VerifyBatchRows) setIfPresent(&candidate.VerifyDutyCycle, update.VerifyDutyCycle) setIfPresent(&candidate.VerifyCDCRows, update.VerifyCDCRows) + setIfPresent(&candidate.VerifyIgnoreApps, update.VerifyIgnoreApps) setIfPresent(&candidate.CDCSampleRows, update.CDCSampleRows) setIfPresent(&candidate.ReplayWorkers, update.ReplayWorkers) setIfPresent(&candidate.ReplayBatchBytes, update.ReplayBatchBytes) @@ -800,7 +803,8 @@ func viewConfiguration(cfg config.Config, revision string) configurationView { VerifySampleWindows: cfg.VerifySampleWindows, VerifyBatchRows: cfg.VerifyBatchRows, VerifyDutyCycle: cfg.VerifyDutyCycle, VerifyTableTimeout: cfg.VerifyTableTimeout.String(), VerifyConvergeTimeout: cfg.VerifyConvergeTimeout.String(), VerifyCDCRows: cfg.VerifyCDCRows, - CDCSampleRows: cfg.CDCSampleRows, + VerifyIgnoreApps: cfg.VerifyIgnoreApps, + CDCSampleRows: cfg.CDCSampleRows, } } diff --git a/internal/controller/controller_test.go b/internal/controller/controller_test.go index b75ef89..88112e6 100644 --- a/internal/controller/controller_test.go +++ b/internal/controller/controller_test.go @@ -410,6 +410,7 @@ func TestConfigurationUpdateParsesValuesAndPreservesDefaults(t *testing.T) { "verify_table_timeout":"1h30m", "verify_converge_timeout":"90s", "verify_cdc_rows":120, + "verify_ignore_apps":"7,42", "cdc_sample_rows":300 }` got := requestJSON(t, server, http.MethodPut, "/api/config", body, "secret") @@ -424,7 +425,7 @@ func TestConfigurationUpdateParsesValuesAndPreservesDefaults(t *testing.T) { if view.Workers != 7 || view.SplitThreshold != 2048 || view.RestoreJobs != 3 || view.WALSampleDuration != "45s" || view.SegmentPruneInterval != "2m0s" || view.ReplayWorkers != 12 || view.ReplayBatchBytes != 67_108_864 || view.ReplayBatchChanges != 262_144 || - view.VerifyWorkers != 2 || view.VerifyTableTimeout != "1h30m0s" || view.VerifyConvergeTimeout != "1m30s" { + view.VerifyWorkers != 2 || view.VerifyTableTimeout != "1h30m0s" || view.VerifyConvergeTimeout != "1m30s" || view.VerifyIgnoreApps != "7,42" { t.Fatalf("updated view = %#v", view) } expected := cfg @@ -459,6 +460,7 @@ func TestConfigurationUpdateParsesValuesAndPreservesDefaults(t *testing.T) { expected.VerifyTableTimeout = 90 * time.Minute expected.VerifyConvergeTimeout = 90 * time.Second expected.VerifyCDCRows = 120 + expected.VerifyIgnoreApps = "7,42" expected.CDCSampleRows = 300 if updated := server.configurationSnapshot(); updated != expected { t.Fatalf("complete update did not round trip\nwant: %#v\ngot: %#v", expected, updated) @@ -474,12 +476,16 @@ func TestConfigurationUpdateParsesValuesAndPreservesDefaults(t *testing.T) { if updated.Source != "postgres://new-source/database" || updated.Target != "postgres://new-target/database" { t.Fatalf("credentials changed after blank update: source=%q target=%q", updated.Source, updated.Target) } - if updated.Workers != 7 || updated.AckWarnings { + if updated.Workers != 7 || updated.AckWarnings || updated.VerifyIgnoreApps != "7,42" { t.Fatalf("partial update lost values: workers=%d ack=%v", updated.Workers, updated.AckWarnings) } if updated.Dir != cfg.Dir || !updated.NoCleanup || updated.SequenceOffset != 1234 || updated.EndPosition != "0/123" { t.Fatalf("startup/CLI-only configuration changed: %#v", updated) } + got = requestJSON(t, server, http.MethodPut, "/api/config", `{"verify_ignore_apps":""}`, "secret") + if got.Code != http.StatusOK || server.configurationSnapshot().VerifyIgnoreApps != "" { + t.Fatalf("explicitly clearing ignored apps failed: %s", got.Body.String()) + } } func TestControllerConfigurationPersistsNonSecretsAcrossRestart(t *testing.T) { @@ -495,7 +501,8 @@ func TestControllerConfigurationPersistsNonSecretsAcrossRestart(t *testing.T) { "target":"postgres://replacement-target:replacement-password@target/database", "replay_workers":24, "replay_batch_bytes":4194304, - "replay_batch_changes":8192 + "replay_batch_changes":8192, + "verify_ignore_apps":"7,42" }`, "controller-token-secret") if got.Code != http.StatusOK { t.Fatalf("PUT config status = %d, body = %s", got.Code, got.Body.String()) @@ -530,7 +537,7 @@ func TestControllerConfigurationPersistsNonSecretsAcrossRestart(t *testing.T) { restarted.EndPosition = "0/BEEF" second := newTestServer(t, restarted, "new-runtime-token", noOpActions()) loaded := second.configurationSnapshot() - if loaded.ReplayWorkers != 24 || loaded.ReplayBatchBytes != 4_194_304 || loaded.ReplayBatchChanges != 8192 { + if loaded.ReplayWorkers != 24 || loaded.ReplayBatchBytes != 4_194_304 || loaded.ReplayBatchChanges != 8192 || loaded.VerifyIgnoreApps != "7,42" { t.Fatalf("persisted replay configuration was not restored: %#v", loaded) } if loaded.Source != restarted.Source || loaded.Target != restarted.Target || loaded.Dir != restarted.Dir || @@ -584,6 +591,7 @@ func TestInvalidConfigurationDoesNotReplaceCurrentConfiguration(t *testing.T) { `{"replay_batch_changes":0}`, `{"wal_sample_duration":"tomorrow"}`, `{"verify_duty_cycle":2}`, + `{"verify_ignore_apps":"7,,42"}`, `{"unknown_setting":true}`, } { got := requestJSON(t, server, http.MethodPut, "/api/config", body, "") @@ -773,6 +781,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", @@ -868,7 +877,7 @@ func TestIndexContainsCompleteWriteOnlyConfigurationUI(t *testing.T) { "maintenance_work_mem", "max_parallel_maintenance_workers", "max_wal_size", "checkpoint_timeout", "verify_workers", "verify_sample_rows", "verify_sample_windows", "verify_batch_rows", "verify_duty_cycle", - "verify_table_timeout", "verify_converge_timeout", "verify_cdc_rows", + "verify_table_timeout", "verify_converge_timeout", "verify_cdc_rows", "verify_ignore_apps", "cdc_sample_rows", } { if !strings.Contains(body, `data-config="`+field+`"`) { diff --git a/internal/controller/ui.html b/internal/controller/ui.html index 88e580d..b2ec350 100644 --- a/internal/controller/ui.html +++ b/internal/controller/ui.html @@ -146,6 +146,7 @@

Migration configuration

+ @@ -160,7 +161,7 @@

Migration configuration

Controls

Completed work and replay position are durable; an interrupted operation can resume from its recorded phase.

Cutover and sequence advancement are intentionally CLI-only.

Object completion

-

Verification progress

+

Verification progress

Findings and failures

Controller operations

migration · idle
Migration output
No migration action has run.
verification · idle
Verification output
No verification action has run.
@@ -194,7 +195,8 @@

Migration configuration

function setConnectionState(id,configured){const state=el(id);state.textContent=configured?'configured':'not configured';state.className=`connection-state${configured?' configured':''}`} function renderLocked(){lastStatus=null;configurationLoaded=false;configurationSaved=false;configurationToken=null;configurationRevision=null;sourceDsn.value='';targetDsn.value='';setConnectionState('sourceState',false);setConnectionState('targetState',false);el('sourceState').textContent='locked';el('targetState').textContent='locked';setConfigurationEnabled(false);setConfigurationMessage('Enter the controller token above to load configuration.');disableControls();el('connection').textContent='locked';el('connection').className='status-pill locked';showError('Dashboard locked: controller token is missing or invalid.')} function setConfigurationEnabled(enabled){[...configurationInputs,...secretInputs].forEach(input=>{input.disabled=!enabled});saveConfiguration.disabled=!enabled||configurationLoading||configurationSaving} -function populateConfiguration(data,saved=false){configurationInputs.forEach(input=>{const value=data[input.dataset.config];if(input.type==='checkbox')input.checked=Boolean(value);else input.value=value??''});sourceDsn.value='';targetDsn.value='';setConnectionState('sourceState',data.source_configured);setConnectionState('targetState',data.target_configured);const configured=Boolean(data.source_configured&&data.target_configured&&data.revision);configurationLoaded=true;configurationSaved=saved||configured;configurationToken=token.value;configurationRevision=data.revision;if(saved)setConfigurationMessage('Configuration saved. Database URLs were cleared from the form.','success');else if(configured)setConfigurationMessage('Saved controller configuration loaded. Database URLs remain write-only.','success');else setConfigurationMessage('Defaults loaded. Review and save before running an action.');if(lastStatus)render(lastStatus)} +function renderVerificationScope(apps){const scope=el('verificationScope');scope.hidden=!apps;scope.textContent=apps?`Current verification configuration ignores mismatches for app_pk: ${apps}. Ignored mismatches are audited; replication is unchanged.`:''} +function populateConfiguration(data,saved=false){renderVerificationScope(data.verify_ignore_apps||'');configurationInputs.forEach(input=>{const value=data[input.dataset.config];if(input.type==='checkbox')input.checked=Boolean(value);else input.value=value??''});sourceDsn.value='';targetDsn.value='';setConnectionState('sourceState',data.source_configured);setConnectionState('targetState',data.target_configured);const configured=Boolean(data.source_configured&&data.target_configured&&data.revision);configurationLoaded=true;configurationSaved=saved||configured;configurationToken=token.value;configurationRevision=data.revision;if(saved)setConfigurationMessage('Configuration saved. Database URLs were cleared from the form.','success');else if(configured)setConfigurationMessage('Saved controller configuration loaded. Database URLs remain write-only.','success');else setConfigurationMessage('Defaults loaded. Review and save before running an action.');if(lastStatus)render(lastStatus)} async function responseError(response){try{return(await response.json()).error||response.statusText}catch{return response.statusText}} async function loadConfiguration(){if(configurationLoading)return;configurationLoading=true;setConfigurationEnabled(false);setConfigurationMessage('Loading configuration…');try{const response=await fetch('/api/config',{headers:{'X-PGMigrate-Token':token.value}});if(!response.ok)throw new Error(await responseError(response));populateConfiguration(await response.json())}catch(error){configurationLoaded=false;configurationSaved=false;configurationToken=null;setConfigurationMessage(error.message,'error');throw error}finally{configurationLoading=false;if(lastStatus)render(lastStatus)}} function configurationPayload(){const payload={};configurationInputs.forEach(input=>{const key=input.dataset.config;if(input.type==='checkbox')payload[key]=input.checked;else if(input.type==='number')payload[key]=Number(input.value);else payload[key]=input.value});if(sourceDsn.value.trim())payload.source=sourceDsn.value;if(targetDsn.value.trim())payload.target=targetDsn.value;return payload} @@ -202,7 +204,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..d1bdee5 --- /dev/null +++ b/internal/controller/ui_test.mjs @@ -0,0 +1,76 @@ +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(/