diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b7dc337..af587a7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -35,6 +35,8 @@ jobs:
# Compiles the integration-tagged files without running them; those and the
# Compose end-to-end suites need PostgreSQL and are run by hand.
- run: make test
+ - name: Controller JavaScript regression tests
+ run: make ui-test
# Apply, the copy workers, and the CDC handoff are concurrent, and a race
# between them is not something anyone reproduces by hand twice.
- run: make race
diff --git a/Makefile b/Makefile
index 47d2a45..c28cb77 100644
--- a/Makefile
+++ b/Makefile
@@ -1,7 +1,8 @@
GO ?= go
GOFLAGS ?=
+NODE ?= node
-.PHONY: fmt vet test race integration bench cdc-bench e2e controller-e2e restart-e2e crash-e2e
+.PHONY: fmt vet test race ui-test integration bench cdc-bench e2e controller-e2e restart-e2e crash-e2e
fmt:
$(GO) $(GOFLAGS) fmt ./...
@@ -15,6 +16,9 @@ test:
race:
$(GO) $(GOFLAGS) test -race ./...
+ui-test:
+ $(NODE) --test internal/controller/progress_test.cjs
+
integration:
$(GO) $(GOFLAGS) test -tags=integration ./...
diff --git a/README.md b/README.md
index 9c0afaf..8c8fd7b 100644
--- a/README.md
+++ b/README.md
@@ -349,6 +349,40 @@ first durable part completion. The lifecycle bar is stage progress, not an
elapsed-time estimate; the object and verification bars use the recorded
completed and total work.
+The **WAL positions and gaps** panel compares current primary WAL with the
+recorded durable capture and apply checkpoints. It shows source minus captured
+(not yet captured), captured minus applied (staged replay), and source minus
+applied (end-to-end gap). LSNs and decimal byte differences preserve all 64 bits.
+These are WAL-byte distances, not row counts or CDC file sizes. Source WAL is
+sampled after the local checkpoints, so live gaps are conservative and include
+sampling delay. A zero staged gap or `follow` phase does **not** establish that
+the source has been caught up. Even zero end-to-end gap is not data validation
+or authorization to cut over.
+
+Source generation, capture, and replay rates are measured separately over a
+rolling window; end-to-end ETA uses replay minus actual source generation and
+is shown only when that net rate is positive. Missing, inconsistent, or stale
+samples show unavailable, not zero. The source identity must match the durable
+migration before its LSN is compared.
+
+The **VACUUM and index-build progress** panel reads PostgreSQL's progress views
+in each configured database, including autovacuum, `CREATE INDEX`/`REINDEX`
+and their concurrent variants, and `VACUUM FULL`. It shows table/index, current
+phase, PID, elapsed time, waits and phase-specific block/tuple/locker counters.
+A phase bar reaching 100% is not overall completion; some phases have no
+measurable total and vacuum phases may repeat. Completed jobs disappear from
+the live views. See [PostgreSQL progress reporting](https://www.postgresql.org/docs/current/progress-reporting.html).
+
+These authenticated, read-only diagnostics use `GET /api/diagnostics`, separate
+from durable status and replay. All tabs share a five-second sample; collection
+has a three-second deadline, at most two database connections per sample, 1.5-second SQL
+timeouts, and at most 100 maintenance jobs per database. No table data or query
+text is read, and no maintenance commands are issued. `pg_read_all_stats` is
+needed to see other users' job details; source identity checking also needs
+permission to execute `pg_control_system()`. Insufficient permissions remain
+visible as unavailable/restricted results; the controller never grants them.
+Source and target monitoring failures are independent and do not stop replay.
+
The lifecycle is also rendered as an ordered, numbered ten-step path from
preflight through completion, with cutover marked CLI-only. Findings are
collapsed by default and classified as blockers, accepted risks,
diff --git a/docs/images/controller-index-progress.png b/docs/images/controller-index-progress.png
new file mode 100644
index 0000000..09a770a
Binary files /dev/null and b/docs/images/controller-index-progress.png differ
diff --git a/docs/images/controller-vacuum-progress.png b/docs/images/controller-vacuum-progress.png
new file mode 100644
index 0000000..a789551
Binary files /dev/null and b/docs/images/controller-vacuum-progress.png differ
diff --git a/docs/images/controller-wal-progress.png b/docs/images/controller-wal-progress.png
new file mode 100644
index 0000000..b6b00a4
Binary files /dev/null and b/docs/images/controller-wal-progress.png differ
diff --git a/internal/controller/controller.go b/internal/controller/controller.go
index 8f41d89..16ad396 100644
--- a/internal/controller/controller.go
+++ b/internal/controller/controller.go
@@ -36,7 +36,7 @@ const (
outputLimit = 64 << 10
)
-//go:embed ui.html
+//go:embed ui.html progress.js
var assets embed.FS
// Action is one operation the controller may supervise.
@@ -78,6 +78,7 @@ type Server struct {
configRevision uint64
configurationPath string
copySample copySample
+ diagnostics diagnosticsCache
}
type copySample struct {
@@ -318,6 +319,8 @@ func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /", s.index)
mux.HandleFunc("GET /api/status", s.status)
+ mux.HandleFunc("GET /api/diagnostics", s.serveDiagnostics)
+ mux.HandleFunc("GET /progress.js", s.progressScript)
mux.HandleFunc("GET /api/config", s.getConfiguration)
mux.HandleFunc("PUT /api/config", s.putConfiguration)
mux.HandleFunc("POST /api/actions/{action}", s.action)
diff --git a/internal/controller/diagnostics.go b/internal/controller/diagnostics.go
new file mode 100644
index 0000000..e1db822
--- /dev/null
+++ b/internal/controller/diagnostics.go
@@ -0,0 +1,377 @@
+package controller
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/GetStream/pgmigrate/internal/config"
+ "github.com/GetStream/pgmigrate/internal/setup"
+ "github.com/GetStream/pgmigrate/internal/state"
+ "github.com/jackc/pglogrepl"
+ "github.com/jackc/pgx/v5"
+ "golang.org/x/sync/singleflight"
+)
+
+const diagnosticsInterval = 5 * time.Second
+
+type diagnosticsView struct {
+ Revision string `json:"revision"`
+ SampledAt time.Time `json:"sampled_at"`
+ SampleAge int64 `json:"sample_age_ms"`
+ WAL walView `json:"wal"`
+ Source maintenanceView `json:"source"`
+ Target maintenanceView `json:"target"`
+}
+
+type walView struct {
+ SourceLSN string `json:"source_lsn,omitempty"`
+ StagedLSN string `json:"staged_lsn,omitempty"`
+ AppliedLSN string `json:"applied_lsn,omitempty"`
+ SourceAt *time.Time `json:"source_sampled_at,omitempty"`
+ CheckpointAt *time.Time `json:"checkpoint_sampled_at,omitempty"`
+ ApplyUpdatedAt *time.Time `json:"apply_updated_at,omitempty"`
+ // Decimal strings preserve the full pg_lsn range in JavaScript and JSON.
+ UncapturedBytes *string `json:"uncaptured_bytes"`
+ ReplayBytes *string `json:"replay_bytes"`
+ TotalBytes *string `json:"total_bytes"`
+ Error string `json:"error,omitempty"`
+}
+
+type maintenanceView struct {
+ Jobs []maintenanceJob `json:"jobs"`
+ Error string `json:"error,omitempty"`
+ Restricted bool `json:"restricted"`
+ Truncated bool `json:"truncated"`
+}
+
+type maintenanceJob struct {
+ PID int `json:"pid"`
+ Kind string `json:"kind"`
+ Command string `json:"command"`
+ Phase string `json:"phase"`
+ Table string `json:"table"`
+ Index string `json:"index,omitempty"`
+ ElapsedSeconds float64 `json:"elapsed_seconds"`
+ WaitEvent string `json:"wait_event,omitempty"`
+ LockerPID int64 `json:"locker_pid,omitempty"`
+ IndexCycles int64 `json:"index_cycles,omitempty"`
+ Done int64 `json:"done,string"`
+ Total int64 `json:"total,string"`
+ Unit string `json:"unit,omitempty"`
+ Percent *float64 `json:"percent,omitempty"`
+}
+
+// Diagnostics never participate in replay, acknowledgement or durable state.
+// One in-flight sample and one five-second cache are shared by all browser tabs.
+type diagnosticsCache struct {
+ mu sync.Mutex
+ value diagnosticsView
+ flight singleflight.Group
+}
+
+func (c *diagnosticsCache) sample(ctx context.Context, revision string, collect func() diagnosticsView) (diagnosticsView, error) {
+ result := c.flight.DoChan(revision, func() (any, error) {
+ c.mu.Lock()
+ cached := c.value
+ c.mu.Unlock()
+ if cached.Revision == revision && time.Since(cached.SampledAt) < diagnosticsInterval {
+ return cached, nil
+ }
+ value := collect()
+ value.Revision = revision
+ c.mu.Lock()
+ c.value = value
+ c.mu.Unlock()
+ return value, nil
+ })
+ select {
+ case <-ctx.Done():
+ return diagnosticsView{}, ctx.Err()
+ case result := <-result:
+ return result.Val.(diagnosticsView), result.Err
+ }
+}
+
+func (s *Server) serveDiagnostics(w http.ResponseWriter, r *http.Request) {
+ if !s.authorized(r) {
+ writeError(w, http.StatusUnauthorized, "controller token is missing or invalid")
+ return
+ }
+ w.Header().Set("Cache-Control", "no-store")
+ s.mu.Lock()
+ cfg, revision := s.cfg, s.configurationRevisionLocked()
+ s.mu.Unlock()
+ view, err := s.diagnostics.sample(r.Context(), revision, func() diagnosticsView {
+ ctx, cancel := context.WithTimeout(s.ctx, 3*time.Second)
+ defer cancel()
+ return collectDiagnostics(ctx, cfg)
+ })
+ if err != nil {
+ writeError(w, http.StatusServiceUnavailable, "diagnostics request canceled")
+ return
+ }
+ view.SampleAge = max(0, time.Since(view.SampledAt).Milliseconds())
+ writeJSON(w, http.StatusOK, view)
+}
+
+func collectDiagnostics(ctx context.Context, cfg config.Config) diagnosticsView {
+ view := diagnosticsView{SampledAt: time.Now().UTC()}
+ var fingerprint string
+ store, err := state.OpenReadOnly(ctx, cfg.Dir)
+ if err == nil {
+ snapshot, readErr := store.Snapshot(ctx)
+ store.Close()
+ if readErr == nil {
+ at := time.Now().UTC()
+ view.WAL.StagedLSN, view.WAL.AppliedLSN = snapshot.Apply.StagedLSN, snapshot.Apply.AppliedLSN
+ view.WAL.CheckpointAt = &at
+ if !snapshot.Apply.UpdatedAt.IsZero() {
+ view.WAL.ApplyUpdatedAt = &snapshot.Apply.UpdatedAt
+ }
+ fingerprint = snapshot.Migration.SourceFingerprint
+ } else {
+ view.WAL.Error = "Durable checkpoints are unavailable"
+ }
+ } else if !errors.Is(err, state.ErrStateNotFound) {
+ view.WAL.Error = "Durable checkpoints are unavailable"
+ }
+
+ // Each database gets its own bounded read-only connection; a slow source
+ // must not hide target maintenance or delay the normal status endpoint.
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ conn, err := diagnosticsConnect(ctx, cfg.Source)
+ if err != nil {
+ view.Source.Error = "Source diagnostics unavailable (connection or permission)"
+ view.WAL.Error = view.Source.Error
+ return
+ }
+ defer conn.Close(ctx)
+ var position *string
+ var systemID, database string
+ var at time.Time
+ err = conn.QueryRow(ctx, `SELECT
+ CASE WHEN pg_catalog.pg_is_in_recovery() THEN NULL ELSE pg_catalog.pg_current_wal_lsn()::text END,
+ pg_catalog.clock_timestamp(), system_identifier::text, pg_catalog.current_database()
+ FROM pg_catalog.pg_control_system()`).Scan(&position, &at, &systemID, &database)
+ switch {
+ case err != nil:
+ view.WAL.Error = "Source WAL unavailable (connection or permission)"
+ case position == nil:
+ view.WAL.Error = "Configured source is in recovery; primary WAL is unavailable"
+ case fingerprint != "" && setup.SourceFingerprint(systemID, database) != fingerprint:
+ view.WAL.Error = "Configured source does not match the migration's source identity"
+ default:
+ view.WAL.SourceLSN, view.WAL.SourceAt = *position, &at
+ }
+ view.Source = readMaintenance(ctx, conn)
+ }()
+ go func() {
+ defer wg.Done()
+ conn, err := diagnosticsConnect(ctx, cfg.Target)
+ if err != nil {
+ view.Target.Error = "Target diagnostics unavailable (connection or permission)"
+ return
+ }
+ defer conn.Close(ctx)
+ view.Target = readMaintenance(ctx, conn)
+ }()
+ wg.Wait()
+ view.WAL.compare()
+ return view
+}
+
+func diagnosticsConnect(ctx context.Context, dsn string) (*pgx.Conn, error) {
+ if strings.TrimSpace(dsn) == "" {
+ return nil, errors.New("database is not configured")
+ }
+ cfg, err := pgx.ParseConfig(dsn)
+ if err != nil {
+ return nil, err
+ }
+ cfg.ConnectTimeout = 2 * time.Second
+ cfg.RuntimeParams["application_name"] = "pgmigrate-diagnostics"
+ cfg.RuntimeParams["default_transaction_read_only"] = "on"
+ cfg.RuntimeParams["statement_timeout"] = "1500"
+ cfg.RuntimeParams["lock_timeout"] = "500"
+ return pgx.ConnectConfig(ctx, cfg)
+}
+
+func (w *walView) compare() {
+ w.UncapturedBytes, w.ReplayBytes, w.TotalBytes = nil, nil, nil
+ staged, stagedErr := diagnosticLSN(w.StagedLSN)
+ applied, appliedErr := diagnosticLSN(w.AppliedLSN)
+ if stagedErr != nil || staged == 0 {
+ return // Not initialized is unknown, not zero backlog.
+ }
+ applyKnown := appliedErr == nil && applied != 0
+ if applyKnown && applied > staged {
+ w.Error = "Checkpoints are out of order; lag is unavailable"
+ return
+ }
+ decimal := func(value pglogrepl.LSN) *string { s := strconv.FormatUint(uint64(value), 10); return &s }
+ if applyKnown {
+ w.ReplayBytes = decimal(staged - applied)
+ }
+ source, err := diagnosticLSN(w.SourceLSN)
+ if err != nil || source == 0 || w.Error != "" {
+ return
+ }
+ if source < staged {
+ w.Error = "Source WAL is behind the recorded capture checkpoint; comparison is unavailable"
+ return
+ }
+ w.UncapturedBytes = decimal(source - staged)
+ if applyKnown {
+ w.TotalBytes = decimal(source - applied)
+ }
+}
+
+func diagnosticLSN(value string) (pglogrepl.LSN, error) {
+ high, low, ok := strings.Cut(value, "/")
+ if !ok || len(high) < 1 || len(high) > 8 || len(low) < 1 || len(low) > 8 {
+ return 0, errors.New("invalid LSN")
+ }
+ hi, hiErr := strconv.ParseUint(high, 16, 32)
+ lo, loErr := strconv.ParseUint(low, 16, 32)
+ if hiErr != nil || loErr != nil || strings.ContainsAny(high+low, "+-") {
+ return 0, errors.New("invalid LSN")
+ }
+ return pglogrepl.LSN(hi<<32 | lo), nil
+}
+
+// JSON extraction allows PG16/17 to omit the PG18-only vacuum index counters.
+// OIDs are resolved only in the connected database, never against another DB's
+// catalog, and neither query text nor user data is returned.
+const maintenanceSQL = `WITH jobs AS (
+ SELECT pid, relid, index_relid, 'index' AS kind, pg_catalog.to_jsonb(p) AS progress
+ FROM pg_catalog.pg_stat_progress_create_index p WHERE datname = pg_catalog.current_database()
+ UNION ALL
+ SELECT pid, relid, 0::oid, 'vacuum', pg_catalog.to_jsonb(p)
+ FROM pg_catalog.pg_stat_progress_vacuum p WHERE datname = pg_catalog.current_database()
+ UNION ALL
+ SELECT pid, relid, 0::oid, 'vacuum_full', pg_catalog.to_jsonb(p)
+ FROM pg_catalog.pg_stat_progress_cluster p
+ WHERE datname = pg_catalog.current_database() AND command = 'VACUUM FULL'
+)
+SELECT j.pid, j.kind, j.progress,
+ coalesce(pg_catalog.quote_ident(n.nspname)||'.'||pg_catalog.quote_ident(t.relname), ''),
+ coalesce(pg_catalog.quote_ident(ni.nspname)||'.'||pg_catalog.quote_ident(i.relname), ''),
+ coalesce(extract(epoch FROM pg_catalog.clock_timestamp()-a.query_start), 0)::float8,
+ coalesce(a.wait_event_type||': '||a.wait_event, '')
+FROM jobs j
+LEFT JOIN pg_catalog.pg_class t ON t.oid=j.relid
+LEFT JOIN pg_catalog.pg_namespace n ON n.oid=t.relnamespace
+LEFT JOIN pg_catalog.pg_class i ON i.oid=j.index_relid
+LEFT JOIN pg_catalog.pg_namespace ni ON ni.oid=i.relnamespace
+LEFT JOIN pg_catalog.pg_stat_activity a ON a.pid=j.pid
+ORDER BY j.pid LIMIT 101`
+
+type maintenanceCounters struct {
+ Command string `json:"command"`
+ Phase *string `json:"phase"`
+ BlocksDone int64 `json:"blocks_done"`
+ BlocksTotal int64 `json:"blocks_total"`
+ TuplesDone int64 `json:"tuples_done"`
+ TuplesTotal int64 `json:"tuples_total"`
+ LockersDone int64 `json:"lockers_done"`
+ LockersTotal int64 `json:"lockers_total"`
+ CurrentLockerPID int64 `json:"current_locker_pid"`
+ HeapScanned int64 `json:"heap_blks_scanned"`
+ HeapVacuumed int64 `json:"heap_blks_vacuumed"`
+ HeapTotal int64 `json:"heap_blks_total"`
+ IndexCycles int64 `json:"index_vacuum_count"`
+ IndexesDone int64 `json:"indexes_processed"`
+ IndexesTotal int64 `json:"indexes_total"`
+}
+
+func readMaintenance(ctx context.Context, conn *pgx.Conn) maintenanceView {
+ view := maintenanceView{Jobs: []maintenanceJob{}}
+ rows, err := conn.Query(ctx, maintenanceSQL)
+ if err != nil {
+ view.Error = "Maintenance progress unavailable (connection or permission)"
+ return view
+ }
+ defer rows.Close()
+ for count := 0; rows.Next(); count++ {
+ if count == 100 {
+ view.Truncated = true
+ break
+ }
+ var job maintenanceJob
+ var raw []byte
+ if err = rows.Scan(&job.PID, &job.Kind, &raw, &job.Table, &job.Index, &job.ElapsedSeconds, &job.WaitEvent); err != nil {
+ break
+ }
+ var counters maintenanceCounters
+ if err = json.Unmarshal(raw, &counters); err != nil {
+ break
+ }
+ if counters.Phase == nil {
+ view.Restricted = true
+ continue
+ }
+ job.setProgress(counters)
+ view.Jobs = append(view.Jobs, job)
+ }
+ if err != nil || rows.Err() != nil {
+ view.Jobs = nil
+ view.Error = "Maintenance progress unavailable (connection or permission)"
+ }
+ return view
+}
+
+func (j *maintenanceJob) setProgress(p maintenanceCounters) {
+ j.Phase, j.Command, j.LockerPID, j.IndexCycles = *p.Phase, p.Command, p.CurrentLockerPID, p.IndexCycles
+ switch j.Kind {
+ case "index":
+ switch {
+ case strings.HasPrefix(j.Phase, "waiting for"):
+ j.Done, j.Total, j.Unit = p.LockersDone, p.LockersTotal, "lockers"
+ case j.Phase == "building index: loading tuples in tree":
+ j.Done, j.Total, j.Unit = p.TuplesDone, p.TuplesTotal, "tuples"
+ case j.Phase == "building index", j.Phase == "building index: scanning table", j.Phase == "index validation: scanning index", j.Phase == "index validation: scanning table":
+ j.Done, j.Total, j.Unit = p.BlocksDone, p.BlocksTotal, "blocks"
+ if j.Total == 0 && j.Phase == "building index" {
+ j.Done, j.Total, j.Unit = p.TuplesDone, p.TuplesTotal, "tuples"
+ }
+ }
+ case "vacuum":
+ j.Command = "VACUUM / autovacuum"
+ switch j.Phase {
+ case "scanning heap":
+ j.Done, j.Total, j.Unit = p.HeapScanned, p.HeapTotal, "heap blocks scanned"
+ case "vacuuming heap":
+ j.Done, j.Total, j.Unit = p.HeapVacuumed, p.HeapTotal, "heap blocks vacuumed"
+ case "vacuuming indexes", "cleaning up indexes":
+ j.Done, j.Total, j.Unit = p.IndexesDone, p.IndexesTotal, "indexes"
+ }
+ case "vacuum_full":
+ if j.Phase == "seq scanning heap" {
+ j.Done, j.Total, j.Unit = p.HeapScanned, p.HeapTotal, "heap blocks scanned"
+ }
+ }
+ if j.Total > 0 && j.Done >= 0 && j.Done <= j.Total {
+ percent := 100 * float64(j.Done) / float64(j.Total)
+ j.Percent = &percent
+ }
+}
+
+func (s *Server) progressScript(w http.ResponseWriter, _ *http.Request) {
+ data, err := assets.ReadFile("progress.js")
+ if err != nil {
+ http.Error(w, "progress UI unavailable", http.StatusInternalServerError)
+ return
+ }
+ w.Header().Set("Content-Type", "text/javascript; charset=utf-8")
+ w.Header().Set("Cache-Control", "no-store")
+ _, _ = w.Write(data)
+}
diff --git a/internal/controller/diagnostics_integration_test.go b/internal/controller/diagnostics_integration_test.go
new file mode 100644
index 0000000..2dca19d
--- /dev/null
+++ b/internal/controller/diagnostics_integration_test.go
@@ -0,0 +1,256 @@
+//go:build integration
+
+package controller
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/GetStream/pgmigrate/internal/config"
+ "github.com/GetStream/pgmigrate/internal/pgtest"
+ "github.com/GetStream/pgmigrate/internal/setup"
+ "github.com/GetStream/pgmigrate/internal/state"
+ "github.com/jackc/pgx/v5"
+ "github.com/jackc/pgx/v5/pgconn"
+)
+
+func TestPostgresDiagnostics(t *testing.T) {
+ for _, major := range pgtest.Majors(t) {
+ t.Run(fmt.Sprint(major), func(t *testing.T) {
+ instance := pgtest.Start(t, major)
+ ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+ defer cancel()
+ admin := instance.Connect(t)
+ var systemID, database, lsn string
+ if err := admin.QueryRow(ctx, `SELECT system_identifier::text, current_database(), pg_current_wal_lsn()::text FROM pg_control_system()`).Scan(&systemID, &database, &lsn); err != nil {
+ t.Fatal(err)
+ }
+ cfg := config.Config{Source: instance.URI, Target: instance.URI, Dir: t.TempDir()}
+ writer, err := state.Open(ctx, cfg.Dir, state.Fingerprints{Source: setup.SourceFingerprint(systemID, database), Filter: "test"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer writer.Close()
+ if err := writer.UpdateApplyProgress(ctx, state.ApplyProgress{StagedLSN: lsn, AppliedLSN: "0/1", UpdatedAt: time.Now()}); err != nil {
+ t.Fatal(err)
+ }
+ server := newTestServer(t, cfg, "test-token", noOpActions())
+ response := request(t, server, http.MethodGet, "/api/diagnostics", "", "test-token")
+ if response.Code != http.StatusOK {
+ t.Fatalf("status = %d: %s", response.Code, response.Body.String())
+ }
+ var view diagnosticsView
+ decode(t, response, &view)
+ if view.WAL.Error != "" || view.WAL.TotalBytes == nil || view.Source.Error != "" || view.Target.Error != "" {
+ t.Fatalf("diagnostics = %+v", view)
+ }
+ var total, captured string
+ if err := admin.QueryRow(ctx, `SELECT pg_wal_lsn_diff($1::pg_lsn,$2::pg_lsn)::text, pg_wal_lsn_diff($1::pg_lsn,$3::pg_lsn)::text`, view.WAL.SourceLSN, view.WAL.AppliedLSN, view.WAL.StagedLSN).Scan(&total, &captured); err != nil {
+ t.Fatal(err)
+ }
+ if *view.WAL.TotalBytes != total || *view.WAL.UncapturedBytes != captured {
+ t.Fatalf("Go gaps disagree with PostgreSQL: %+v, SQL %s/%s", view.WAL, total, captured)
+ }
+ if view.WAL.SourceAt == nil || view.WAL.CheckpointAt == nil {
+ t.Fatal("sample timestamps are missing")
+ }
+
+ t.Run("read-only and timeout", func(t *testing.T) {
+ conn, err := diagnosticsConnect(ctx, instance.URI)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close(context.Background())
+ _, err = conn.Exec(ctx, "CREATE TABLE must_not_be_created(id int)")
+ var pgErr *pgconn.PgError
+ if !errors.As(err, &pgErr) || pgErr.Code != "25006" {
+ t.Fatalf("write error = %v, want read-only transaction", err)
+ }
+ started := time.Now()
+ _, err = conn.Exec(ctx, "SELECT pg_sleep(5)")
+ if !errors.As(err, &pgErr) || pgErr.Code != "57014" || time.Since(started) > 3*time.Second {
+ t.Fatalf("unbounded diagnostics query: elapsed=%s err=%v", time.Since(started), err)
+ }
+ })
+
+ t.Run("wrong source identity", func(t *testing.T) {
+ wrong := cfg
+ wrong.Dir = t.TempDir()
+ store, err := state.Open(ctx, wrong.Dir, state.Fingerprints{Source: "another-cluster", Filter: "test"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := store.UpdateApplyProgress(ctx, state.ApplyProgress{StagedLSN: lsn, AppliedLSN: "0/1"}); err != nil {
+ t.Fatal(err)
+ }
+ store.Close()
+ got := collectDiagnostics(ctx, wrong)
+ if got.WAL.TotalBytes != nil || !strings.Contains(got.WAL.Error, "source identity") {
+ t.Fatalf("unrelated source was compared: %+v", got.WAL)
+ }
+ })
+
+ if _, err := admin.Exec(ctx, `CREATE TABLE public.progress_fixture(id int PRIMARY KEY, payload text) WITH (autovacuum_enabled=false);
+ INSERT INTO public.progress_fixture SELECT i, repeat(md5(i::text),16) FROM generate_series(1,5000) i`); err != nil {
+ t.Fatal(err)
+ }
+ monitor, err := diagnosticsConnect(ctx, instance.URI)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer monitor.Close(context.Background())
+ t.Run("concurrent index and restricted observer", func(t *testing.T) {
+ blocker, err := admin.Begin(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer blocker.Rollback(context.Background())
+ if _, err := blocker.Exec(ctx, "INSERT INTO public.progress_fixture VALUES (6000,'blocker')"); err != nil {
+ t.Fatal(err)
+ }
+ builder := instance.Connect(t)
+ buildCtx, stopBuild := context.WithCancel(ctx)
+ defer stopBuild()
+ finished := make(chan error, 1)
+ go func() {
+ _, err := builder.Exec(buildCtx, "CREATE INDEX CONCURRENTLY progress_payload_idx ON public.progress_fixture(payload)")
+ finished <- err
+ }()
+ job := waitMaintenance(t, ctx, monitor, func(job maintenanceJob) bool {
+ return job.Kind == "index" && strings.HasPrefix(job.Phase, "waiting for")
+ })
+ if job.Table != "public.progress_fixture" || job.Command != "CREATE INDEX CONCURRENTLY" || job.Unit != "lockers" || job.LockerPID == 0 {
+ t.Fatalf("index progress = %+v", job)
+ }
+ otherCfg, err := pgx.ParseConfig(instance.URI)
+ if err != nil {
+ t.Fatal(err)
+ }
+ otherCfg.Database = "postgres"
+ other, err := pgx.ConnectConfig(ctx, otherCfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer other.Close(context.Background())
+ if got := readMaintenance(ctx, other); got.Error != "" || len(got.Jobs) != 0 || got.Restricted {
+ t.Fatalf("another database's job leaked into results: %+v", got)
+ }
+ roleAdmin := instance.Connect(t)
+ if _, err := roleAdmin.Exec(ctx, "CREATE ROLE progress_observer LOGIN PASSWORD 'test'"); err != nil {
+ t.Fatal(err)
+ }
+ observerCfg, err := pgx.ParseConfig(instance.URI)
+ if err != nil {
+ t.Fatal(err)
+ }
+ observerCfg.User, observerCfg.Password = "progress_observer", "test"
+ observer, err := pgx.ConnectConfig(ctx, observerCfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer observer.Close(context.Background())
+ restricted := readMaintenance(ctx, observer)
+ if restricted.Error != "" || !restricted.Restricted || len(restricted.Jobs) != 0 {
+ t.Fatalf("restricted progress = %+v", restricted)
+ }
+ if _, err := roleAdmin.Exec(ctx, "GRANT pg_read_all_stats TO progress_observer"); err != nil {
+ t.Fatal(err)
+ }
+ visible := readMaintenance(ctx, observer)
+ if visible.Restricted || len(visible.Jobs) == 0 {
+ t.Fatalf("granted progress = %+v", visible)
+ }
+ if err := blocker.Rollback(ctx); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case err := <-finished:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-ctx.Done():
+ t.Fatal(ctx.Err())
+ }
+ if after := readMaintenance(ctx, monitor); after.Error != "" || len(after.Jobs) != 0 {
+ t.Fatalf("completed index remains active: %+v", after)
+ }
+ var ready, valid bool
+ if err := monitor.QueryRow(ctx, "SELECT indisready, indisvalid FROM pg_index WHERE indexrelid='public.progress_payload_idx'::regclass").Scan(&ready, &valid); err != nil || !ready || !valid {
+ t.Fatalf("index completion flags = %v/%v, %v", ready, valid, err)
+ }
+ })
+
+ t.Run("live vacuum", func(t *testing.T) {
+ worker := instance.Connect(t)
+ if _, err := worker.Exec(ctx, "SET vacuum_cost_delay='20ms'; SET vacuum_cost_limit=1"); err != nil {
+ t.Fatal(err)
+ }
+ vacuumCtx, stopVacuum := context.WithCancel(ctx)
+ finished := make(chan error, 1)
+ defer func() { stopVacuum(); <-finished }()
+ go func() { _, err := worker.Exec(vacuumCtx, "VACUUM public.progress_fixture"); finished <- err }()
+ job := waitMaintenance(t, ctx, monitor, func(job maintenanceJob) bool {
+ return job.Kind == "vacuum" && job.Phase == "scanning heap" && job.Total > 0
+ })
+ if job.Table != "public.progress_fixture" || job.Percent == nil || job.Unit != "heap blocks scanned" {
+ t.Fatalf("vacuum progress = %+v", job)
+ }
+ })
+ t.Run("vacuum full index rebuild", func(t *testing.T) {
+ // Hold the rebuild inside a test-only index expression so this
+ // otherwise brief phase can be observed deterministically.
+ if _, err := admin.Exec(ctx, `CREATE FUNCTION public.progress_gate(i int) RETURNS int LANGUAGE plpgsql IMMUTABLE AS $$
+ BEGIN PERFORM pg_advisory_xact_lock(987654); RETURN i; END$$;
+ CREATE INDEX progress_gate_idx ON public.progress_fixture(public.progress_gate(id))`); err != nil {
+ t.Fatal(err)
+ }
+ gate := instance.Connect(t)
+ if _, err := gate.Exec(ctx, "SELECT pg_advisory_lock(987654)"); err != nil {
+ t.Fatal(err)
+ }
+ worker := instance.Connect(t)
+ vacuumCtx, stopVacuum := context.WithCancel(ctx)
+ finished := make(chan error, 1)
+ defer func() { stopVacuum(); <-finished }()
+ go func() { _, err := worker.Exec(vacuumCtx, "VACUUM FULL public.progress_fixture"); finished <- err }()
+ job := waitMaintenance(t, ctx, monitor, func(job maintenanceJob) bool {
+ return job.Kind == "vacuum_full" && job.Phase == "rebuilding index"
+ })
+ if job.Table != "public.progress_fixture" || job.Command != "VACUUM FULL" || job.Percent != nil {
+ t.Fatalf("VACUUM FULL rebuild must not show a finished heap scan as 100%%: %+v", job)
+ }
+ })
+ })
+ }
+}
+
+func waitMaintenance(t *testing.T, ctx context.Context, conn *pgx.Conn, match func(maintenanceJob) bool) maintenanceJob {
+ t.Helper()
+ deadline := time.NewTimer(10 * time.Second)
+ defer deadline.Stop()
+ ticker := time.NewTicker(20 * time.Millisecond)
+ defer ticker.Stop()
+ for {
+ view := readMaintenance(ctx, conn)
+ if view.Error != "" {
+ t.Fatal(view.Error)
+ }
+ for _, job := range view.Jobs {
+ if match(job) {
+ return job
+ }
+ }
+ select {
+ case <-ticker.C:
+ case <-deadline.C:
+ t.Fatalf("maintenance did not reach expected phase: %+v", view)
+ case <-ctx.Done():
+ t.Fatal(ctx.Err())
+ }
+ }
+}
diff --git a/internal/controller/diagnostics_test.go b/internal/controller/diagnostics_test.go
new file mode 100644
index 0000000..3929f96
--- /dev/null
+++ b/internal/controller/diagnostics_test.go
@@ -0,0 +1,196 @@
+package controller
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/GetStream/pgmigrate/internal/config"
+)
+
+func TestWALComparison(t *testing.T) {
+ for _, test := range []struct {
+ name, source, staged, applied string
+ uncaptured, replay, total string
+ failed bool
+ }{
+ {name: "split backlog", source: "1/30", staged: "1/20", applied: "1/10", uncaptured: "16", replay: "16", total: "32"},
+ {name: "staged drained but capture behind", source: "1/30", staged: "1/10", applied: "1/10", uncaptured: "32", replay: "0", total: "32"},
+ {name: "sampled equality", source: "1/30", staged: "1/30", applied: "1/30", uncaptured: "0", replay: "0", total: "0"},
+ {name: "above JS safe integer", source: "FFFFFFFF/FFFFFFFF", staged: "FFFFFFFF/FFFFFFFE", applied: "0/1", uncaptured: "1", replay: "18446744073709551613", total: "18446744073709551614"},
+ {name: "word rollover", source: "2/0", staged: "1/FFFFFFFF", applied: "1/FFFFFFFE", uncaptured: "1", replay: "1", total: "2"},
+ {name: "source unavailable", staged: "1/20", applied: "1/10", replay: "16"},
+ {name: "copy before apply", source: "1/30", staged: "1/20", applied: "0/0", uncaptured: "16"},
+ {name: "no state", source: "1/30"},
+ {name: "uninitialized", source: "1/30", staged: "0/0", applied: "0/0"},
+ {name: "applied ahead", source: "1/30", staged: "1/10", applied: "1/20", failed: true},
+ {name: "source behind", source: "1/10", staged: "1/30", applied: "1/20", replay: "16", failed: true},
+ {name: "invalid trailing text", source: "1/30junk", staged: "1/20", applied: "1/10", replay: "16"},
+ {name: "invalid checkpoint", source: "1/30", staged: "1/+20", applied: "1/10"},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ view := walView{SourceLSN: test.source, StagedLSN: test.staged, AppliedLSN: test.applied}
+ view.compare()
+ for _, check := range []struct {
+ got *string
+ want string
+ }{{view.UncapturedBytes, test.uncaptured}, {view.ReplayBytes, test.replay}, {view.TotalBytes, test.total}} {
+ if check.want == "" {
+ if check.got != nil {
+ t.Fatalf("unknown gap = %q, want null", *check.got)
+ }
+ } else if check.got == nil || *check.got != check.want {
+ t.Fatalf("gap = %v, want %s", check.got, check.want)
+ }
+ }
+ if (view.Error != "") != test.failed {
+ t.Fatalf("error = %q", view.Error)
+ }
+ encoded, err := json.Marshal(view)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if test.total != "" && !strings.Contains(string(encoded), `"total_bytes":"`+test.total+`"`) {
+ t.Fatalf("JSON lost decimal string: %s", encoded)
+ }
+ })
+ }
+ view := walView{SourceLSN: "1/30", StagedLSN: "1/20", AppliedLSN: "1/10", Error: "source identity mismatch"}
+ view.compare()
+ if view.TotalBytes != nil || view.UncapturedBytes != nil || view.ReplayBytes == nil {
+ t.Fatalf("identity failure did not suppress source comparison: %+v", view)
+ }
+}
+
+func TestMaintenanceProgressIsPhaseSpecific(t *testing.T) {
+ for _, test := range []struct {
+ name, kind, data, unit string
+ done, total int64
+ measurable bool
+ }{
+ {"index blocks", "index", `{"phase":"building index: scanning table","blocks_done":20,"blocks_total":100}`, "blocks", 20, 100, true},
+ {"index tuples", "index", `{"phase":"building index: loading tuples in tree","blocks_done":100,"blocks_total":100,"tuples_done":2,"tuples_total":8}`, "tuples", 2, 8, true},
+ {"build sort stale blocks", "index", `{"phase":"building index: sorting live tuples","blocks_done":100,"blocks_total":100}`, "", 0, 0, false},
+ {"waiting", "index", `{"phase":"waiting for old snapshots","lockers_done":1,"lockers_total":4,"current_locker_pid":123}`, "lockers", 1, 4, true},
+ {"sort stale blocks", "index", `{"phase":"index validation: sorting tuples","blocks_done":100,"blocks_total":100}`, "", 0, 0, false},
+ {"scan", "vacuum", `{"phase":"scanning heap","heap_blks_scanned":7,"heap_blks_total":10}`, "heap blocks scanned", 7, 10, true},
+ {"vacuum heap", "vacuum", `{"phase":"vacuuming heap","heap_blks_scanned":10,"heap_blks_vacuumed":3,"heap_blks_total":10}`, "heap blocks vacuumed", 3, 10, true},
+ {"PG17 indexes", "vacuum", `{"phase":"vacuuming indexes","heap_blks_scanned":10,"heap_blks_total":10,"index_vacuum_count":1}`, "indexes", 0, 0, false},
+ {"PG18 indexes", "vacuum", `{"phase":"cleaning up indexes","indexes_processed":1,"indexes_total":3}`, "indexes", 1, 3, true},
+ {"cleanup not done", "vacuum", `{"phase":"performing final cleanup","heap_blks_scanned":10,"heap_blks_total":10}`, "", 0, 0, false},
+ {"vacuum full scan", "vacuum_full", `{"command":"VACUUM FULL","phase":"seq scanning heap","heap_blks_scanned":3,"heap_blks_total":10}`, "heap blocks scanned", 3, 10, true},
+ {"vacuum full rebuild", "vacuum_full", `{"command":"VACUUM FULL","phase":"rebuilding index","heap_blks_scanned":10,"heap_blks_total":10}`, "", 0, 0, false},
+ {"unknown total", "index", `{"phase":"building index","blocks_done":50}`, "tuples", 0, 0, false},
+ {"inconsistent counters", "index", `{"phase":"building index","blocks_done":101,"blocks_total":100}`, "blocks", 101, 100, false},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ var p maintenanceCounters
+ if err := json.Unmarshal([]byte(test.data), &p); err != nil {
+ t.Fatal(err)
+ }
+ job := maintenanceJob{Kind: test.kind}
+ job.setProgress(p)
+ if job.Done != test.done || job.Total != test.total || job.Unit != test.unit || (job.Percent != nil) != test.measurable {
+ t.Fatalf("progress = %+v", job)
+ }
+ if job.Percent != nil && *job.Percent != 100*float64(test.done)/float64(test.total) {
+ t.Fatalf("percent = %v", *job.Percent)
+ }
+ })
+ }
+}
+
+func TestDiagnosticsAuthAndUnavailableStateAreReadOnly(t *testing.T) {
+ dir := filepath.Join(t.TempDir(), "not-created")
+ server := newTestServer(t, config.Config{Dir: dir, Source: "postgres://secret:credential%zz@source/db", Target: "postgres://secret:credential%zz@target/db"}, "token", noOpActions())
+ if got := request(t, server, http.MethodGet, "/api/diagnostics", "", ""); got.Code != http.StatusUnauthorized {
+ t.Fatalf("unauthenticated status = %d", got.Code)
+ }
+ got := request(t, server, http.MethodGet, "/api/diagnostics", "", "token")
+ if got.Code != http.StatusOK {
+ t.Fatalf("status = %d, %s", got.Code, got.Body.String())
+ }
+ var view diagnosticsView
+ decode(t, got, &view)
+ if view.WAL.TotalBytes != nil || view.WAL.Error == "" || view.Source.Error == "" || view.Target.Error == "" {
+ t.Fatalf("unavailable diagnostics = %+v", view)
+ }
+ for _, secret := range []string{"credential", "postgres://", "secret", "%zz"} {
+ if strings.Contains(got.Body.String(), secret) {
+ t.Fatalf("diagnostics exposed %q", secret)
+ }
+ }
+ if _, err := os.Stat(dir); !os.IsNotExist(err) {
+ t.Fatalf("diagnostics created state directory: %v", err)
+ }
+ if status := request(t, server, http.MethodGet, "/api/status", "", "token"); status.Code != http.StatusOK {
+ t.Fatal("diagnostics failure affected durable status")
+ }
+ if script := request(t, server, http.MethodGet, "/progress.js", "", ""); script.Code != http.StatusOK || !strings.HasPrefix(script.Header().Get("Content-Type"), "text/javascript") {
+ t.Fatal("progress script is not served")
+ }
+}
+
+func TestDiagnosticsCacheSharesSamplesAndInvalidatesConfiguration(t *testing.T) {
+ var cache diagnosticsCache
+ var calls atomic.Int64
+ collect := func() diagnosticsView { calls.Add(1); return diagnosticsView{SampledAt: time.Now().UTC()} }
+ var wg sync.WaitGroup
+ for range 20 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ if _, err := cache.sample(context.Background(), "one", collect); err != nil {
+ t.Error(err)
+ }
+ }()
+ }
+ wg.Wait()
+ if calls.Load() != 1 {
+ t.Fatalf("20 browser requests collected %d samples", calls.Load())
+ }
+ if _, err := cache.sample(context.Background(), "two", collect); err != nil {
+ t.Fatal(err)
+ }
+ if calls.Load() != 2 {
+ t.Fatal("configuration change did not invalidate cache")
+ }
+ cache.mu.Lock()
+ cache.value.SampledAt = time.Now().Add(-2 * diagnosticsInterval)
+ cache.mu.Unlock()
+ if _, err := cache.sample(context.Background(), "two", collect); err != nil {
+ t.Fatal(err)
+ }
+ if calls.Load() != 3 {
+ t.Fatal("expired sample was reused")
+ }
+}
+
+func TestDiagnosticsWaitingRequestCanCancel(t *testing.T) {
+ var cache diagnosticsCache
+ entered, release := make(chan struct{}), make(chan struct{})
+ defer close(release)
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() {
+ _, err := cache.sample(ctx, "one", func() diagnosticsView { close(entered); <-release; return diagnosticsView{SampledAt: time.Now()} })
+ done <- err
+ }()
+ <-entered
+ cancel()
+ select {
+ case err := <-done:
+ if err != context.Canceled {
+ t.Fatalf("cancel error = %v", err)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("request cancellation blocked on diagnostics")
+ }
+}
diff --git a/internal/controller/progress.js b/internal/controller/progress.js
new file mode 100644
index 0000000..5c948c0
--- /dev/null
+++ b/internal/controller/progress.js
@@ -0,0 +1,162 @@
+// Diagnostics are display-only. No maintenance or migration actions live here.
+const pgProgress = (() => {
+ const maxAge = 15000;
+ let latest = null, status = null, pending = false, samples = [];
+
+ function lsnBytes(value) {
+ if (!/^[0-9a-f]{1,8}\/[0-9a-f]{1,8}$/i.test(String(value || ''))) return null;
+ const [high, low] = value.split('/');
+ return (BigInt('0x' + high) << 32n) + BigInt('0x' + low);
+ }
+
+ function walRates(points) {
+ if (points.length < 2) return null;
+ const first = points[0], last = points[points.length - 1];
+ const seconds = (last.at - first.at) / 1000;
+ if (seconds < 15) return null;
+ const source = Number(last.source - first.source) / seconds;
+ const capture = Number(last.capture - first.capture) / seconds;
+ const replay = Number(last.replay - first.replay) / seconds;
+ const drain = Number((last.replay - first.replay) - (last.source - first.source)) / seconds;
+ return {seconds, source, capture, replay, drain};
+ }
+
+ function addSample(points, data) {
+ const w = data.wal, at = Date.parse(data.sampled_at);
+ const source = lsnBytes(w.source_lsn), capture = lsnBytes(w.staged_lsn), replay = lsnBytes(w.applied_lsn);
+ if (w.error || !Number.isFinite(at) || w.total_bytes == null || [source, capture, replay].some(x => x === null) || source < capture || capture < replay) return [];
+ const last = points[points.length - 1];
+ if (last && (data.revision !== last.revision || at < last.at || source < last.source || capture < last.capture || replay < last.replay)) points = [];
+ else if (last && at === last.at) return points;
+ points.push({at, source, capture, replay, revision: data.revision});
+ while (points.length > 2 && points[1].at <= at - 300000) points.shift();
+ return points;
+ }
+
+ const byID = id => document.getElementById(id);
+ const text = (id, value) => { byID(id).textContent = value; };
+ const bytes = value => value == null ? 'unavailable' : fmtBytes(Number(value));
+ // Server age plus a local monotonic clock avoids comparing clocks on two hosts.
+ const fresh = (data, now = performance.now()) => data && Number.isFinite(data.sample_age_ms) && data.sample_age_ms >= 0 && now >= data.receivedAt && data.sample_age_ms + now - data.receivedAt < maxAge;
+
+ function summary() {
+ const w = latest?.wal, usable = fresh(latest) && !w.error && w.total_bytes != null;
+ const running = ['running', 'stopping'].includes(status?.operations?.migration?.state);
+ const replaying = running && ['catchup', 'follow', 'drained'].includes(status?.snapshot?.phase);
+ text('lag', usable ? bytes(w.total_bytes) : '—');
+ text('lagTrend', '—'); text('lagTrendLabel', 'end-to-end WAL trend unavailable');
+ text('replayIO', '—'); text('replayIOLabel', 'WAL apply / actual source generation');
+ if (!usable) return;
+ const rate = replaying ? walRates(samples) : null;
+ if (!rate) {
+ text('lagTrendLabel', replaying ? 'collecting 15s of end-to-end samples' : 'replay is not running');
+ return;
+ }
+ text('replayIO', `${fmtBytes(rate.replay)}/s`);
+ text('replayIOLabel', `WAL apply · source ${fmtBytes(rate.source)}/s · ${Math.round(rate.seconds)}s avg`);
+ text('lagTrend', `${rate.drain < 0 ? '+' : rate.drain > 0 ? '−' : ''}${fmtBytes(Math.abs(rate.drain))}/s`);
+ const eta = rate.drain > 0 && BigInt(w.total_bytes) > 0n ? ` · ETA ${fmtDuration(Number(w.total_bytes) / rate.drain * 1e9)}` : '';
+ text('lagTrendLabel', BigInt(w.total_bytes) === 0n ? 'no WAL gap at sampled checkpoints' : rate.drain > 0 ? `end-to-end drain${eta}` : rate.drain < 0 ? 'end-to-end backlog growing · no convergent ETA' : 'end-to-end backlog unchanged · no convergent ETA');
+ }
+
+ function renderMaintenance(id, view) {
+ const root = byID(id);
+ root.replaceChildren();
+ if (view.error) { root.textContent = view.error; return; }
+ if (view.restricted) {
+ const warning = document.createElement('p');
+ warning.textContent = 'Some sessions are hidden: pg_read_all_stats is required for full visibility.';
+ root.append(warning);
+ }
+ if (!view.jobs?.length) {
+ const empty = document.createElement('p');
+ empty.className = 'muted'; empty.textContent = 'No visible active VACUUM or index builds at this sample.';
+ root.append(empty); return;
+ }
+ for (const job of view.jobs) {
+ const card = document.createElement('div'); card.className = 'maintenance-job';
+ const name = document.createElement('strong');
+ name.textContent = `${job.command} · ${job.table || 'relation unavailable'}${job.index ? ` · ${job.index}` : ''}`;
+ const phase = document.createElement('p');
+ phase.textContent = `${job.phase} · PID ${job.pid} · ${Math.max(0, Math.floor(job.elapsed_seconds))}s${job.wait_event ? ` · ${job.wait_event}` : ''}${job.locker_pid ? ` · waiting for PID ${job.locker_pid}` : ''}`;
+ card.append(name, phase);
+ if (Number.isFinite(job.percent)) {
+ const progress = document.createElement('progress');
+ progress.max = 100; progress.value = job.percent;
+ progress.setAttribute('aria-label', `${job.table} ${job.phase} phase progress`);
+ const label = document.createElement('p');
+ label.textContent = `${job.percent.toFixed(1)}% of this phase · ${job.done} / ${job.total} ${job.unit}`;
+ card.append(progress, label);
+ } else {
+ const unknown = document.createElement('p'); unknown.className = 'muted';
+ unknown.textContent = 'PostgreSQL does not report a measurable total for this phase.';
+ card.append(unknown);
+ }
+ if (job.index_cycles) {
+ const cycles = document.createElement('p'); cycles.textContent = `${job.index_cycles} index vacuum cycles completed`; card.append(cycles);
+ }
+ root.append(card);
+ }
+ if (view.truncated) {
+ const note = document.createElement('p'); note.textContent = 'Results limited to the first 100 sessions.'; root.append(note);
+ }
+ }
+
+ function render(data) {
+ const w = data.wal;
+ text('diagnosticsTime', `Sampled ${new Date(data.sampled_at).toISOString()} · refresh every 5s`);
+ for (const [id, value] of [['sourceLSN', w.source_lsn], ['capturedLSN', w.staged_lsn], ['appliedLSN', w.applied_lsn]]) text(id, value || 'unavailable');
+ for (const [id, value] of [['uncapturedGap', w.uncaptured_bytes], ['replayGap', w.replay_bytes], ['totalGap', w.total_bytes]]) {
+ text(id, bytes(value)); byID(id).title = value == null ? 'unavailable' : `${value} bytes`;
+ }
+ text('walMessage', w.error || (w.total_bytes == null ? 'Waiting for source WAL and initialized durable checkpoints.' : 'Source sampled after the recorded checkpoints; gaps are conservative WAL-byte distances, not rows, queue-file sizes, or data validation. Zero is not cutover authorization.'));
+ text('walTimes', `Checkpoints sampled: ${w.checkpoint_sampled_at || 'unavailable'} · source sampled: ${w.source_sampled_at || 'unavailable'} · last durable apply: ${w.apply_updated_at || 'unavailable'}`);
+ const rate = walRates(samples);
+ text('walRates', rate ? `Source generation ${fmtBytes(rate.source)}/s · capture ${fmtBytes(rate.capture)}/s · replay ${fmtBytes(rate.replay)}/s · ${Math.round(rate.seconds)}s average` : 'Rates need at least 15s of valid, fresh samples.');
+ renderMaintenance('sourceMaintenance', data.source);
+ renderMaintenance('targetMaintenance', data.target);
+ summary();
+ }
+
+ function unavailable(message) {
+ latest = null; samples = [];
+ text('diagnosticsTime', message);
+ for (const id of ['sourceLSN', 'capturedLSN', 'appliedLSN', 'uncapturedGap', 'replayGap', 'totalGap']) { text(id, 'unavailable'); byID(id).removeAttribute('title'); }
+ text('walMessage', 'Live diagnostics unavailable; durable migration status is independent.');
+ text('walTimes', ''); text('walRates', '');
+ text('sourceMaintenance', 'Live progress unavailable.'); text('targetMaintenance', 'Live progress unavailable.');
+ summary();
+ }
+
+ async function refresh() {
+ if (pending || !configurationLoaded || configurationToken !== token.value) return;
+ pending = true;
+ const revision = configurationRevision, auth = token.value;
+ try {
+ const response = await fetch('/api/diagnostics', {headers: {'X-PGMigrate-Token': auth}, signal: AbortSignal.timeout(4500)});
+ if (!response.ok) throw new Error('Diagnostics unavailable; authentication or connection failed.');
+ const data = await response.json();
+ data.receivedAt = performance.now();
+ if (auth !== token.value || revision !== configurationRevision) return;
+ if (data.revision !== revision) throw new Error('Configuration changed; waiting for matching diagnostics.');
+ if (!fresh(data)) throw new Error('Diagnostics sample is stale; waiting for a fresh sample.');
+ samples = addSample(samples, data); latest = data; render(data);
+ } catch (error) {
+ if (auth === token.value && revision === configurationRevision) unavailable(error.message);
+ } finally { pending = false; }
+ }
+
+ function updateStatus(data) {
+ status = data;
+ if (latest && (!fresh(latest) || latest.revision !== configurationRevision)) unavailable('Diagnostics sample expired or configuration changed.');
+ summary();
+ }
+
+ function start() {
+ refresh();
+ setInterval(refresh, 5000);
+ setInterval(() => { if (latest && !fresh(latest)) unavailable('Diagnostics sample expired.'); }, 1000);
+ }
+
+ return {lsnBytes, walRates, addSample, fresh, updateStatus, unavailable, start};
+})();
diff --git a/internal/controller/progress_test.cjs b/internal/controller/progress_test.cjs
new file mode 100644
index 0000000..b2fee5e
--- /dev/null
+++ b/internal/controller/progress_test.cjs
@@ -0,0 +1,81 @@
+const {test} = require('node:test');
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const vm = require('node:vm');
+const path = require('node:path');
+
+const context = vm.createContext({});
+vm.runInContext(fs.readFileSync(path.join(__dirname, 'progress.js'), 'utf8'), context);
+const progress = vm.runInContext('pgProgress', context);
+const sample = (at, source, capture, replay, revision = 'one') => ({
+ sampled_at: new Date(at).toISOString(), revision,
+ wal: {source_lsn: source, staged_lsn: capture, applied_lsn: replay, total_bytes: '16'},
+});
+
+test('sample expiry uses server age and local elapsed time, not clock synchronization', () => {
+ const data = {sampled_at: '1999-01-01T00:00:00Z', sample_age_ms: 3000, receivedAt: 100};
+ assert.equal(progress.fresh(data, 100), true);
+ assert.equal(progress.fresh(data, 12099), true);
+ assert.equal(progress.fresh(data, 12100), false);
+ assert.equal(progress.fresh({...data, sample_age_ms: 16000}, 100), false);
+ assert.equal(progress.fresh({...data, sample_age_ms: -1}, 100), false);
+ assert.equal(progress.fresh({...data, sample_age_ms: undefined}, 100), false);
+});
+
+test('LSNs retain every bit, including above the JS integer limit', () => {
+ assert.equal(progress.lsnBytes('FFFFFFFF/FFFFFFFF'), 18446744073709551615n);
+ assert.equal(progress.lsnBytes('200000/1') - progress.lsnBytes('200000/0'), 1n);
+ assert.equal(progress.lsnBytes('2/0') - progress.lsnBytes('1/FFFFFFFF'), 1n);
+ for (const value of ['', '1/', '/1', '1/2junk', '1/+2', '100000000/0', '0/100000000', '1/2/3', 'NaN/1']) assert.equal(progress.lsnBytes(value), null, value);
+});
+
+test('real source generation, capture and replay are separate rates', () => {
+ let points = progress.addSample([], sample(0, '2/100', '2/80', '2/40'));
+ points = progress.addSample(points, sample(20000, '2/200', '2/C0', '2/80'));
+ const rates = progress.walRates(points);
+ assert.equal(rates.source, 256 / 20);
+ assert.equal(rates.capture, 64 / 20);
+ assert.equal(rates.replay, 64 / 20);
+ assert.equal(rates.drain, -192 / 20);
+});
+
+test('unchanged positions produce zero throughput, not a frozen old rate', () => {
+ let points = progress.addSample([], sample(0, '200000/3', '200000/2', '200000/1'));
+ points = progress.addSample(points, sample(20000, '200000/4', '200000/3', '200000/2'));
+ assert.equal(progress.walRates(points).replay, 1 / 20);
+ points = progress.addSample(points, sample(330000, '200000/4', '200000/3', '200000/2'));
+ assert.equal(progress.walRates(points).replay, 0);
+ assert.equal(progress.walRates(points).source, 0);
+});
+
+test('duplicate cached samples, reset positions and configuration changes', () => {
+ let points = progress.addSample([], sample(1000, '2/30', '2/20', '2/10'));
+ assert.equal(progress.addSample(points, sample(1000, '2/30', '2/20', '2/10')).length, 1);
+ points = progress.addSample(points, sample(5000, '2/30', '2/20', '2/10'));
+ assert.equal(progress.walRates(points), null);
+ assert.equal(progress.addSample(points, sample(6000, '2/30', '2/20', '2/10', 'two')).length, 1);
+ assert.equal(progress.addSample(points, sample(6000, '1/30', '1/20', '1/10')).length, 1);
+ assert.equal(progress.addSample(points, sample(0, '2/30', '2/20', '2/10')).length, 1);
+});
+
+test('unknown, failed and reversed comparisons never become a zero backlog', () => {
+ const missing = sample(0, '2/30', '2/20', '2/10');
+ missing.wal.total_bytes = null;
+ assert.equal(progress.addSample([], missing).length, 0);
+ const failed = sample(0, '2/30', '2/20', '2/10');
+ failed.wal.error = 'source identity mismatch';
+ assert.equal(progress.addSample([], failed).length, 0);
+ assert.equal(progress.addSample([], sample(0, '2/10', '2/20', '2/10')).length, 0);
+});
+
+test('the shipped replay sampler uses exact LSN deltas and samples idle time', () => {
+ const html = fs.readFileSync(path.join(__dirname, 'ui.html'), 'utf8');
+ const functions = html.split('\n').filter(line => line.startsWith('function lsnBytes(') || line.startsWith('function sampleReplay(')).join('\n');
+ vm.runInContext('let replaySamples=[];\n' + functions, context);
+ const replay = vm.runInContext('sampleReplay', context);
+ const state = {transactions: 0, rows: 0, applied_lsn: '200000/0', staged_lsn: '200000/1', lag_bytes: 1};
+ assert.equal(replay(state, true, 1000), null);
+ const next = {...state, applied_lsn: '200000/1', staged_lsn: '200000/2'};
+ assert.equal(replay(next, true, 2000).appliedBytes, 1);
+ assert.equal(replay(next, true, 303000).appliedBytes, 0);
+});
diff --git a/internal/controller/ui.html b/internal/controller/ui.html
index 88e580d..3cc42fa 100644
--- a/internal/controller/ui.html
+++ b/internal/controller/ui.html
@@ -18,6 +18,7 @@
.grid { display:grid; grid-template-columns:repeat(12,1fr); gap:16px; }
.panel { background:linear-gradient(180deg,rgba(22,31,52,.96),rgba(15,23,40,.96)); border:1px solid var(--line); border-radius:14px; padding:18px; box-shadow:0 14px 35px rgba(0,0,0,.2); }
.configuration,.objects,.verify,.findings,.operation { grid-column:span 12; } .overview { grid-column:span 8; } .controls { grid-column:span 4; }
+ .diagnostics { grid-column:span 12; } .wal-grid { display:grid; grid-template-columns:repeat(3,minmax(0,1fr)); gap:12px; margin:14px 0; } .wal-grid .mono,.wal-times { overflow-wrap:anywhere; } .maintenance-job { padding:12px 0; border-bottom:1px solid var(--line); } .maintenance-job progress { width:100%; accent-color:var(--cyan); } @media(max-width:560px) { .wal-grid { grid-template-columns:1fr; } }
.phase-row { display:flex; justify-content:space-between; align-items:baseline; gap:12px; margin-bottom:10px; }
.phase { font-size:22px; font-weight:750; text-transform:capitalize; } .phase-count { color:var(--muted); }
.bar { width:100%; height:10px; overflow:hidden; border-radius:999px; background:#08101f; border:1px solid #23304a; }
@@ -156,9 +157,27 @@
Migration configuration
Lifecycle phase
not started
0 / 10
Steps run in order. Cutover and sequence advancement are CLI-only; the dashboard follows their durable state.
Waiting for preflight.
Current durable replay claimwaiting for receipts
-
—apply lag
—since last durable commit
—replay rate
—WAL apply throughput
—net lag trend
0changes applied
—copy rate
0 Bdata streamed
0rows streamed
0review items
+
—end-to-end WAL lag
—since last durable commit
—replay rate
—WAL apply throughput
—net lag trend
0changes applied
—copy rate
0 Bdata streamed
0rows streamed
0review items
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.
+
+
WAL positions and gaps
Waiting for live diagnostics.
+
+
1 · Source WAL
unavailable
Current primary WAL position
+
2 · Captured
unavailable
Recorded durable CDC checkpoint
+
3 · Applied
unavailable
Recorded committed replay checkpoint
+
unavailable
Source − captured
Not yet captured at the recorded checkpoint
+
unavailable
Captured − applied
Staged replay backlog
+
unavailable
Source − applied
End-to-end WAL gap
+
+
+
+
+
VACUUM and index-build progress
+
Read-only activity in each configured database. Bars measure the current phase, not overall completion. Phases can repeat; completed jobs disappear. VACUUM FULL is tracked separately.