feat(api): /api/perf and /api/healthz expose async migration progress

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)
+}

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")
+	}
+}

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<total).
 	bfTotal, bfProcessed, bfDone := fromPubkeyBackfillSnapshot()
+
+	// #1724: surface async migration progress so the warm-up banner can
+	// stay visible while a long-running backfill is still consuming the
+	// single writer. anyAsyncMigrationRunning intentionally drops to
+	// false on "failed" status — operator should see warm-up complete
+	// + alert, not an endless banner.
+	var asyncMigrations []AsyncMigrationInfo
+	if s.db != nil {
+		if infos, err := readAsyncMigrations(s.db.conn); err == nil {
+			asyncMigrations = infos
+		}
+	}
+	if asyncMigrations == nil {
+		asyncMigrations = []AsyncMigrationInfo{}
+	}
+
 	w.WriteHeader(http.StatusOK)
 	resp := map[string]interface{}{
 		"ready":     true,
@@ -51,6 +67,8 @@ func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) {
 			"processed": bfProcessed,
 			"done":      bfDone,
 		},
+		"async_migrations":         asyncMigrations,
+		"async_migrations_running": anyAsyncMigrationRunning(asyncMigrations),
 	}
 	// PR #1609 M1: surface per-MQTT-source receipt vs write-path
 	// liveness so operators can distinguish "broker alive, write

cmd/server/routes.go

@@ -230,6 +230,7 @@ func (s *Server) RegisterRoutes(r *mux.Router) {
 	r.HandleFunc("/api/perf/io", s.handlePerfIO).Methods("GET")
 	r.HandleFunc("/api/perf/sqlite", s.handlePerfSqlite).Methods("GET")
 	r.HandleFunc("/api/perf/write-sources", s.handlePerfWriteSources).Methods("GET")
+	r.HandleFunc("/api/perf/async-migrations", s.handlePerfAsyncMigrations).Methods("GET")
 	r.HandleFunc("/api/mqtt/status", s.handleMqttStatus).Methods("GET")
 	r.Handle("/api/perf/reset", s.requireAPIKey(http.HandlerFunc(s.handlePerfReset))).Methods("POST")
 	// /api/admin/prune removed in #1283 — pruning is owned by the
@@ -913,6 +914,16 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) {
 		Cache:         perfCS,
 		PacketStore:   pktStoreStats,
 		Sqlite:        sqliteStats,
+		AsyncMigrations: func() []AsyncMigrationInfo {
+			if s.db == nil {
+				return []AsyncMigrationInfo{}
+			}
+			infos, err := readAsyncMigrations(s.db.conn)
+			if err != nil || infos == nil {
+				return []AsyncMigrationInfo{}
+			}
+			return infos
+		}(),
 		GoRuntime: func() *GoRuntimeStats {
 			ms := s.getMemStats()
 			return &GoRuntimeStats{

cmd/server/types.go

@@ -261,15 +261,16 @@ type SqliteStats struct {
 }
 
 type PerfResponse struct {
-	Uptime        int                           `json:"uptime"`
-	TotalRequests int64                         `json:"totalRequests"`
-	AvgMs         float64                       `json:"avgMs"`
-	Endpoints     map[string]*EndpointStatsResp `json:"endpoints"`
-	SlowQueries   []SlowQuery                   `json:"slowQueries"`
-	Cache         PerfCacheStats                `json:"cache"`
-	PacketStore   *PerfPacketStoreStats         `json:"packetStore"`
-	Sqlite        *SqliteStats                  `json:"sqlite"`
-	GoRuntime     *GoRuntimeStats               `json:"goRuntime,omitempty"`
+	Uptime          int                           `json:"uptime"`
+	TotalRequests   int64                         `json:"totalRequests"`
+	AvgMs           float64                       `json:"avgMs"`
+	Endpoints       map[string]*EndpointStatsResp `json:"endpoints"`
+	SlowQueries     []SlowQuery                   `json:"slowQueries"`
+	Cache           PerfCacheStats                `json:"cache"`
+	PacketStore     *PerfPacketStoreStats         `json:"packetStore"`
+	Sqlite          *SqliteStats                  `json:"sqlite"`
+	GoRuntime       *GoRuntimeStats               `json:"goRuntime,omitempty"`
+	AsyncMigrations []AsyncMigrationInfo          `json:"asyncMigrations"`
 }
 
 // GoRuntimeStats holds Go runtime metrics for the perf endpoint.