From f5bf6056043712e9c3d079b2ff5b0d4f6c69cebc Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Tue, 16 Jun 2026 18:59:38 +0000 Subject: [PATCH] feat(api): /api/perf and /api/healthz expose async migration progress --- cmd/server/async_migrations.go | 208 ++++++++++++++++++++++++++++ cmd/server/async_migrations_test.go | 201 +++++++++++++++++++++++++++ cmd/server/healthz.go | 18 +++ cmd/server/routes.go | 11 ++ cmd/server/types.go | 19 +-- 5 files changed, 448 insertions(+), 9 deletions(-) create mode 100644 cmd/server/async_migrations.go create mode 100644 cmd/server/async_migrations_test.go diff --git a/cmd/server/async_migrations.go b/cmd/server/async_migrations.go new file mode 100644 index 00000000..39f996dc --- /dev/null +++ b/cmd/server/async_migrations.go @@ -0,0 +1,208 @@ +// Read-only surface for async migration status (#1724). +// +// The ingestor writes _async_migrations (status, rows_processed, rows_total, +// last_update_at, error). The server READS this table to surface progress +// via /api/healthz (so the warm-up banner can stay visible while a long +// backfill runs) and /api/perf (so operators see per-migration progress +// + ETA + error message). +// +// Reads go through SetMaxOpenConns(1)? No — cmd/server/db.go uses +// SetMaxOpenConns(4) in mode=ro, but the underlying SQLite file's writer +// is single-threaded (the ingestor). To avoid every /api/healthz request +// hitting the disk while a migration is mid-batch, we cache the result +// for asyncMigrationsTTL. + +package main + +import ( + "database/sql" + "net/http" + "sync" + "time" +) + +const asyncMigrationsTTL = 5 * time.Second + +// AsyncMigrationInfo is the JSON shape returned via /api/perf and embedded +// in /api/healthz. +type AsyncMigrationInfo struct { + Name string `json:"name"` + Status string `json:"status"` // "running" | "done" | "failed" | "unknown" + StartedAt string `json:"startedAt,omitempty"` + EndedAt string `json:"endedAt,omitempty"` + LastUpdateAt string `json:"lastUpdateAt,omitempty"` + RowsProcessed int64 `json:"rowsProcessed"` + RowsTotal int64 `json:"rowsTotal"` + ElapsedSec float64 `json:"elapsedSec"` + EtaSec float64 `json:"etaSec"` // only meaningful when status="running" + RatePerSec float64 `json:"ratePerSec"` // only meaningful when status="running" + ErrorMessage string `json:"errorMessage,omitempty"` +} + +// asyncMigrationsCache caches the latest readAsyncMigrationsRaw result. +var ( + asyncMigrationsCacheMu sync.Mutex + asyncMigrationsCacheAt time.Time + asyncMigrationsCached []AsyncMigrationInfo + asyncMigrationsCacheErr error +) + +// asyncMigrationsNow is overridable for tests. +var asyncMigrationsNow = time.Now + +// readAsyncMigrations returns the current set of async migration info, +// using a short TTL cache to avoid hammering the writer-held DB on hot +// paths like /api/healthz. +func readAsyncMigrations(db *sql.DB) ([]AsyncMigrationInfo, error) { + asyncMigrationsCacheMu.Lock() + defer asyncMigrationsCacheMu.Unlock() + if !asyncMigrationsCacheAt.IsZero() && + asyncMigrationsNow().Sub(asyncMigrationsCacheAt) < asyncMigrationsTTL { + return asyncMigrationsCached, asyncMigrationsCacheErr + } + out, err := readAsyncMigrationsRaw(db) + asyncMigrationsCached = out + asyncMigrationsCacheErr = err + asyncMigrationsCacheAt = asyncMigrationsNow() + return out, err +} + +// readAsyncMigrationsRaw bypasses the cache. +func readAsyncMigrationsRaw(db *sql.DB) ([]AsyncMigrationInfo, error) { + if db == nil { + return []AsyncMigrationInfo{}, nil + } + rows, err := db.Query(` + SELECT name, + status, + COALESCE(started_at, ''), + COALESCE(ended_at, ''), + COALESCE(last_update_at, ''), + COALESCE(rows_processed, 0), + COALESCE(rows_total, 0), + COALESCE(error, '') + FROM _async_migrations + ORDER BY name + `) + if err != nil { + // Table may not exist on freshly-initialized ingestor DBs that + // have not run a single migration yet. Empty result is the + // honest answer there; everything else is a real error and + // MUST propagate (operators should see ANY corruption, not + // silently get an empty banner). + return []AsyncMigrationInfo{}, err + } + defer rows.Close() + + now := asyncMigrationsNow() + out := make([]AsyncMigrationInfo, 0, 4) + for rows.Next() { + var info AsyncMigrationInfo + var rawStatus string + if err := rows.Scan(&info.Name, &rawStatus, &info.StartedAt, &info.EndedAt, + &info.LastUpdateAt, &info.RowsProcessed, &info.RowsTotal, &info.ErrorMessage); err != nil { + return nil, err + } + info.Status = mapAsyncStatus(rawStatus) + + startTs, _ := parseAsyncTime(info.StartedAt) + endTs, _ := parseAsyncTime(info.EndedAt) + switch info.Status { + case "running": + if !startTs.IsZero() { + info.ElapsedSec = now.Sub(startTs).Seconds() + if info.ElapsedSec > 0 && info.RowsProcessed > 0 { + info.RatePerSec = float64(info.RowsProcessed) / info.ElapsedSec + remaining := info.RowsTotal - info.RowsProcessed + if remaining > 0 && info.RatePerSec > 0 { + info.EtaSec = float64(remaining) / info.RatePerSec + } + } + } + case "done", "failed": + if !startTs.IsZero() && !endTs.IsZero() { + info.ElapsedSec = endTs.Sub(startTs).Seconds() + } + } + if info.Status != "failed" { + info.ErrorMessage = "" + } + out = append(out, info) + } + if err := rows.Err(); err != nil { + return nil, err + } + return out, nil +} + +// mapAsyncStatus maps the raw ingestor-side status string to the API enum. +// Unknown values map to "unknown" (NOT "running") so a corrupted row +// cannot pin the warm-up banner in a perpetual loading state. +func mapAsyncStatus(raw string) string { + switch raw { + case "pending_async": + return "running" + case "done": + return "done" + case "failed": + return "failed" + default: + return "unknown" + } +} + +// anyAsyncMigrationRunning returns true iff any migration is in status +// "running". Failed migrations DO NOT count (operator should see +// "warm-up complete + alert", not an endless banner). +func anyAsyncMigrationRunning(infos []AsyncMigrationInfo) bool { + for _, m := range infos { + if m.Status == "running" { + return true + } + } + return false +} + +// parseAsyncTime parses either RFC3339 (last_update_at written by +// recordAsyncMigrationProgress) or "YYYY-MM-DD HH:MM:SS" (SQLite's +// datetime('now') default for started_at/ended_at). +func parseAsyncTime(s string) (time.Time, error) { + if s == "" { + return time.Time{}, nil + } + if t, err := time.Parse(time.RFC3339, s); err == nil { + return t, nil + } + if t, err := time.Parse("2006-01-02 15:04:05", s); err == nil { + return t.UTC(), nil + } + return time.Time{}, errParseAsyncTime{s: s} +} + +type errParseAsyncTime struct{ s string } + +func (e errParseAsyncTime) Error() string { return "parseAsyncTime: cannot parse " + e.s } + +// invalidateAsyncMigrationsCache is exported for tests that want to skip +// the TTL gate. +func invalidateAsyncMigrationsCache() { + asyncMigrationsCacheMu.Lock() + asyncMigrationsCacheAt = time.Time{} + asyncMigrationsCached = nil + asyncMigrationsCacheErr = nil + asyncMigrationsCacheMu.Unlock() +} + +// handlePerfAsyncMigrations exposes the read-only async-migration state at +// /api/perf/async-migrations so dashboards / curl can poll progress +// without fetching the full /api/perf payload. +func (s *Server) handlePerfAsyncMigrations(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + out := []AsyncMigrationInfo{} + if s.db != nil { + if infos, err := readAsyncMigrations(s.db.conn); err == nil && infos != nil { + out = infos + } + } + writeJSON(w, out) +} diff --git a/cmd/server/async_migrations_test.go b/cmd/server/async_migrations_test.go new file mode 100644 index 00000000..0a9c48eb --- /dev/null +++ b/cmd/server/async_migrations_test.go @@ -0,0 +1,201 @@ +// Tests for async migration server-side surface (#1724). + +package main + +import ( + "database/sql" + "strings" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +func openAsyncTestDB(t *testing.T) *sql.DB { + t.Helper() + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + _, err = db.Exec(` + CREATE TABLE _async_migrations ( + name TEXT PRIMARY KEY, + status TEXT NOT NULL, + started_at TEXT, + ended_at TEXT, + error TEXT, + rows_processed INTEGER DEFAULT 0, + rows_total INTEGER DEFAULT 0, + last_update_at TEXT + ) + `) + if err != nil { + t.Fatal(err) + } + return db +} + +func TestMapAsyncStatus(t *testing.T) { + cases := map[string]string{ + "pending_async": "running", + "done": "done", + "failed": "failed", + "": "unknown", + "garbage": "unknown", + } + for in, want := range cases { + if got := mapAsyncStatus(in); got != want { + t.Errorf("mapAsyncStatus(%q)=%q, want %q", in, got, want) + } + } +} + +func TestAnyAsyncMigrationRunning_FalseOnFailed(t *testing.T) { + infos := []AsyncMigrationInfo{ + {Name: "x", Status: "failed"}, + {Name: "y", Status: "done"}, + } + if anyAsyncMigrationRunning(infos) { + t.Errorf("anyAsyncMigrationRunning should be false when no migration is 'running'") + } +} + +func TestAnyAsyncMigrationRunning_TrueOnRunning(t *testing.T) { + infos := []AsyncMigrationInfo{ + {Name: "x", Status: "done"}, + {Name: "y", Status: "running"}, + } + if !anyAsyncMigrationRunning(infos) { + t.Errorf("anyAsyncMigrationRunning should be true when any migration is 'running'") + } +} + +func TestReadAsyncMigrations_EtaAndRateRunning(t *testing.T) { + db := openAsyncTestDB(t) + // Fix "now" to a known value relative to started_at. + fixed := time.Date(2026, 6, 16, 12, 0, 0, 0, time.UTC) + asyncMigrationsNow = func() time.Time { return fixed } + t.Cleanup(func() { asyncMigrationsNow = time.Now }) + invalidateAsyncMigrationsCache() + + // Started 10s ago, processed 100/1000. + _, err := db.Exec(`INSERT INTO _async_migrations + (name, status, started_at, rows_processed, rows_total) + VALUES ('m1','pending_async','2026-06-16 11:59:50',100,1000)`) + if err != nil { + t.Fatal(err) + } + got, err := readAsyncMigrations(db) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 { + t.Fatalf("want 1 row, got %d", len(got)) + } + m := got[0] + if m.Status != "running" { + t.Errorf("status=%q, want running", m.Status) + } + if m.ElapsedSec < 9.5 || m.ElapsedSec > 10.5 { + t.Errorf("ElapsedSec=%v, want ~10", m.ElapsedSec) + } + // rate = 100/10 = 10 rows/sec; remaining = 900 → eta = 90s. + if m.RatePerSec < 9.5 || m.RatePerSec > 10.5 { + t.Errorf("RatePerSec=%v, want ~10", m.RatePerSec) + } + if m.EtaSec < 85 || m.EtaSec > 95 { + t.Errorf("EtaSec=%v, want ~90", m.EtaSec) + } +} + +func TestReadAsyncMigrations_FailedSurfacesErrorMessage(t *testing.T) { + db := openAsyncTestDB(t) + invalidateAsyncMigrationsCache() + asyncMigrationsNow = func() time.Time { return time.Date(2026, 6, 16, 12, 0, 0, 0, time.UTC) } + t.Cleanup(func() { asyncMigrationsNow = time.Now }) + + _, err := db.Exec(`INSERT INTO _async_migrations + (name, status, started_at, ended_at, error, rows_processed, rows_total) + VALUES ('boom','failed','2026-06-16 11:59:00','2026-06-16 11:59:30','disk full',50,100)`) + if err != nil { + t.Fatal(err) + } + got, err := readAsyncMigrations(db) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].Status != "failed" { + t.Fatalf("expected one failed row, got %+v", got) + } + if got[0].ErrorMessage != "disk full" { + t.Errorf("ErrorMessage=%q, want 'disk full'", got[0].ErrorMessage) + } + if got[0].ElapsedSec < 29 || got[0].ElapsedSec > 31 { + t.Errorf("ElapsedSec=%v, want ~30", got[0].ElapsedSec) + } + if anyAsyncMigrationRunning(got) { + t.Errorf("failed migration must not count as running (banner would stick)") + } +} + +func TestReadAsyncMigrations_DoneClearsErrorMessage(t *testing.T) { + db := openAsyncTestDB(t) + invalidateAsyncMigrationsCache() + _, _ = db.Exec(`INSERT INTO _async_migrations + (name, status, started_at, ended_at, error) + VALUES ('ok','done','2026-06-16 11:59:00','2026-06-16 11:59:05','stale')`) + got, err := readAsyncMigrations(db) + if err != nil { + t.Fatal(err) + } + if got[0].ErrorMessage != "" { + t.Errorf("done status must not surface ErrorMessage, got %q", got[0].ErrorMessage) + } +} + +func TestReadAsyncMigrations_PropagatesErrors(t *testing.T) { + db, _ := sql.Open("sqlite", ":memory:") + db.Close() + invalidateAsyncMigrationsCache() + _, err := readAsyncMigrationsRaw(db) + if err == nil { + t.Errorf("closed DB must propagate error, not return nil") + } +} + +func TestParseAsyncTime(t *testing.T) { + if _, err := parseAsyncTime("2026-06-16T11:59:00Z"); err != nil { + t.Errorf("RFC3339 should parse: %v", err) + } + if _, err := parseAsyncTime("2026-06-16 11:59:00"); err != nil { + t.Errorf("SQLite datetime should parse: %v", err) + } + if _, err := parseAsyncTime("not a time"); err == nil { + t.Errorf("bogus value must error") + } + tz, err := parseAsyncTime("") + if err != nil || !tz.IsZero() { + t.Errorf("empty should be zero+nil, got %v / %v", tz, err) + } +} + +func TestReadAsyncMigrations_CachesWithinTTL(t *testing.T) { + db := openAsyncTestDB(t) + invalidateAsyncMigrationsCache() + _, _ = db.Exec(`INSERT INTO _async_migrations(name,status,started_at) VALUES ('a','done','2026-06-16 11:59:00')`) + g1, _ := readAsyncMigrations(db) + // Add another row; cached result must NOT include it. + _, _ = db.Exec(`INSERT INTO _async_migrations(name,status,started_at) VALUES ('b','done','2026-06-16 11:59:00')`) + g2, _ := readAsyncMigrations(db) + if len(g1) != len(g2) { + t.Errorf("cache TTL not honored: g1=%d g2=%d", len(g1), len(g2)) + } +} + +func TestErrParseAsyncTime_Message(t *testing.T) { + e := errParseAsyncTime{s: "x"} + if !strings.Contains(e.Error(), "x") { + t.Errorf("error message missing input") + } +} diff --git a/cmd/server/healthz.go b/cmd/server/healthz.go index 3f554952..26cc63a8 100644 --- a/cmd/server/healthz.go +++ b/cmd/server/healthz.go @@ -41,6 +41,22 @@ func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { // /api/healthz never observes a torn state (e.g. done=true with // processed