fix(#1735): surface async-migration errors on healthz/perf + propagate progress write failures

Group A from PR #1735 round-1 review (must-fix #1, #5, #6, #7).

- cmd/server/healthz.go: on readAsyncMigrations error, include the
  message in the JSON body as async_migrations_error AND keep
  async_migrations_running=true. Fail closed for warm-up: if we can't
  read the bookkeeping table, treat the system as possibly still
  warming up rather than declaring 'all clear'.
- cmd/server/async_migrations.go handlePerfAsyncMigrations: return
  HTTP 500 with the error body on readAsyncMigrations failure instead
  of silently returning an empty list. (Empty list is a meaningful
  operator signal; a query failure must be visible.)
- cmd/server/routes.go /api/perf: log the readAsyncMigrations error
  and surface it via X-Async-Migrations-Error response header so the
  rest of the perf payload still flows.
- cmd/server/async_migrations.go: delete the unread
  asyncMigrationsCacheErr field (finding #5).
- cmd/server/async_migrations.go parseAsyncTime: propagate parse
  errors to the caller; readAsyncMigrationsRaw now appends them to
  ErrorMessage so unparseable timestamps don't silently produce 0s.
- cmd/ingestor/async_migration_progress.go recordAsyncMigrationProgressEx:
  check RowsAffected(); 0 rows updated -> error (bookkeeping row
  missing). cmd/ingestor/db.go: track in-loop progress write failures,
  log them, and treat a failed TERMINAL progress write as a failed
  migration (counts are no longer trustworthy).

cmd/ingestor/async_migration_progress.go

@@ -99,7 +99,19 @@ func recordAsyncMigrationProgressEx(db *sql.DB, name string, processed, total in
 		})
 		return err
 	}
-	_ = res
+	// #1735 finding #7: a UPDATE that affects 0 rows means the migration
+	// bookkeeping row is missing — every caller of this function expects
+	// RunAsyncMigration to have inserted the row already. Silently
+	// returning nil would let backfills "succeed" while their progress
+	// surface stays at 0/0 forever. Treat as a hard error so the caller
+	// can mark the migration failed.
+	n, raErr := res.RowsAffected()
+	if raErr != nil {
+		return fmt.Errorf("recordAsyncMigrationProgress(%s) RowsAffected: %w", name, raErr)
+	}
+	if n == 0 {
+		return fmt.Errorf("recordAsyncMigrationProgress(%s): no row updated (bookkeeping row missing)", name)
+	}
 	return nil
 }
 

cmd/ingestor/db.go

@@ -176,18 +176,39 @@ func OpenStoreWithInterval(dbPath string, sampleIntervalSec int) (*Store, error)
 	if err := s.RunAsyncMigration(context.Background(), "tx_last_seen_backfill_v1",
 		func(ctx context.Context, d *sql.DB) error {
 			log.Println("[migration/async] Backfilling transmissions.last_seen (chunked, reader-yielding)...")
+			// #1735 finding #7 (Group A): track progress-write failures
+			// so a persistent bookkeeping error (missing row, schema
+			// drift) marks the migration failed instead of silently
+			// running to completion with no visible progress.
+			var progressErrs int
 			processed, total, err := chunkedTxLastSeenBackfill(ctx, d, 5000, 100*time.Millisecond,
 				func(p, t int64) {
-					_ = recordAsyncMigrationProgress(d, "tx_last_seen_backfill_v1", p, t)
+					if perr := recordAsyncMigrationProgress(d, "tx_last_seen_backfill_v1", p, t); perr != nil {
+						progressErrs++
+						if progressErrs == 1 {
+							log.Printf("[migration/async] progress write failed (will fail migration if persistent): %v", perr)
+						}
+					}
 				})
 			if err != nil {
 				// Force-write whatever counts we have so the surfaced
 				// progress reflects the failure point, not stale data.
-				_ = recordAsyncMigrationProgressTerminal(d, "tx_last_seen_backfill_v1", processed, total)
+				if perr := recordAsyncMigrationProgressTerminal(d, "tx_last_seen_backfill_v1", processed, total); perr != nil {
+					log.Printf("[migration/async] terminal progress write failed: %v", perr)
+				}
 				return err
 			}
 			// Force-write the terminal stable counts past the rate limiter.
-			_ = recordAsyncMigrationProgressTerminal(d, "tx_last_seen_backfill_v1", processed, total)
+			if perr := recordAsyncMigrationProgressTerminal(d, "tx_last_seen_backfill_v1", processed, total); perr != nil {
+				// Terminal write failure is itself a failed migration:
+				// the surface counts are now untrustworthy.
+				return fmt.Errorf("terminal progress write: %w", perr)
+			}
+			if progressErrs > 0 {
+				// In-loop progress writes failed but terminal succeeded —
+				// log but do not fail. The terminal write is authoritative.
+				log.Printf("[migration/async] %d in-loop progress writes failed (terminal write OK)", progressErrs)
+			}
 			log.Printf("[migration/async] transmissions.last_seen backfill complete: %d / %d rows", processed, total)
 			return nil
 		}); err != nil {

cmd/server/async_migrations.go

@@ -16,13 +16,21 @@ package main
 
 import (
 	"database/sql"
+	"encoding/json"
 	"net/http"
 	"sync"
 	"time"
+
+	"golang.org/x/sync/singleflight"
 )
 
 const asyncMigrationsTTL = 5 * time.Second
 
+// asyncMigrationsSF collapses concurrent /api/healthz + /api/perf calls
+// during a cache miss into a single DB read. Errors are not cached and
+// each caller gets the same error on a shared in-flight read.
+var asyncMigrationsSF singleflight.Group
+
 // AsyncMigrationInfo is the JSON shape returned via /api/perf and embedded
 // in /api/healthz.
 type AsyncMigrationInfo struct {
@@ -39,12 +47,14 @@ type AsyncMigrationInfo struct {
 	ErrorMessage   string  `json:"errorMessage,omitempty"`
 }
 
-// asyncMigrationsCache caches the latest readAsyncMigrationsRaw result.
+// asyncMigrationsCache caches the latest successful readAsyncMigrationsRaw
+// result. Errors are NOT cached (#1735 finding #4 / Group C): every error
+// path retries on the next call so transient I/O failures don't get
+// pinned for asyncMigrationsTTL.
 var (
 	asyncMigrationsCacheMu sync.Mutex
 	asyncMigrationsCacheAt time.Time
 	asyncMigrationsCached  []AsyncMigrationInfo
-	asyncMigrationsCacheErr error
 )
 
 // asyncMigrationsNow is overridable for tests.
@@ -53,18 +63,41 @@ 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.
+//
+// Concurrency contract (#1735 finding #3 / Group C):
+//   - Cache mutex is NEVER held across db.Query — only across the
+//     check/populate steps. The actual I/O runs through singleflight so
+//     concurrent callers during a cache miss share one DB read.
+//   - Errors are NOT cached (#1735 finding #4): a transient query failure
+//     does not pin healthz/perf at "empty" for asyncMigrationsTTL.
 func readAsyncMigrations(db *sql.DB) ([]AsyncMigrationInfo, error) {
+	// Step 1: cache hit under lock, release before any I/O.
 	asyncMigrationsCacheMu.Lock()
-	defer asyncMigrationsCacheMu.Unlock()
 	if !asyncMigrationsCacheAt.IsZero() &&
 		asyncMigrationsNow().Sub(asyncMigrationsCacheAt) < asyncMigrationsTTL {
-		return asyncMigrationsCached, asyncMigrationsCacheErr
+		cached := asyncMigrationsCached
+		asyncMigrationsCacheMu.Unlock()
+		return cached, nil
 	}
-	out, err := readAsyncMigrationsRaw(db)
+	asyncMigrationsCacheMu.Unlock()
+
+	// Step 2: do the I/O through singleflight so a thundering herd of
+	// /api/healthz polls collapses into one query.
+	v, err, _ := asyncMigrationsSF.Do("read", func() (interface{}, error) {
+		return readAsyncMigrationsRaw(db)
+	})
+	if err != nil {
+		// Do NOT cache the error — let the next caller retry.
+		return nil, err
+	}
+	out, _ := v.([]AsyncMigrationInfo)
+
+	// Step 3: re-acquire to populate cache.
+	asyncMigrationsCacheMu.Lock()
 	asyncMigrationsCached = out
-	asyncMigrationsCacheErr = err
 	asyncMigrationsCacheAt = asyncMigrationsNow()
-	return out, err
+	asyncMigrationsCacheMu.Unlock()
+	return out, nil
 }
 
 // readAsyncMigrationsRaw bypasses the cache.
@@ -105,8 +138,21 @@ func readAsyncMigrationsRaw(db *sql.DB) ([]AsyncMigrationInfo, error) {
 		}
 		info.Status = mapAsyncStatus(rawStatus)
 
-		startTs, _ := parseAsyncTime(info.StartedAt)
-		endTs, _ := parseAsyncTime(info.EndedAt)
+		startTs, startErr := parseAsyncTime(info.StartedAt)
+		endTs, endErr := parseAsyncTime(info.EndedAt)
+		// #1735 finding #6: do not silently discard parse errors. Build
+		// the parseMsg now; append it AFTER the status-driven
+		// ErrorMessage wipe below so it survives non-failed statuses too.
+		parseMsg := ""
+		if startErr != nil {
+			parseMsg = "startedAt: " + startErr.Error()
+		}
+		if endErr != nil {
+			if parseMsg != "" {
+				parseMsg += "; "
+			}
+			parseMsg += "endedAt: " + endErr.Error()
+		}
 		switch info.Status {
 		case "running":
 			if !startTs.IsZero() {
@@ -127,6 +173,14 @@ func readAsyncMigrationsRaw(db *sql.DB) ([]AsyncMigrationInfo, error) {
 		if info.Status != "failed" {
 			info.ErrorMessage = ""
 		}
+		// Append parse errors after the wipe so they always surface.
+		if parseMsg != "" {
+			if info.ErrorMessage == "" {
+				info.ErrorMessage = parseMsg
+			} else {
+				info.ErrorMessage = info.ErrorMessage + " | " + parseMsg
+			}
+		}
 		out = append(out, info)
 	}
 	if err := rows.Err(); err != nil {
@@ -189,20 +243,33 @@ 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.
+//
+// #1735 finding #1 (Group A): on readAsyncMigrations error, return
+// HTTP 500 with the error body instead of silently returning an empty
+// list. An empty list is a meaningful operator signal (no migrations
+// pending); a query failure must be visible, not disguised.
 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
-		}
+	if s.db == nil {
+		writeJSON(w, []AsyncMigrationInfo{})
+		return
 	}
-	writeJSON(w, out)
+	infos, err := readAsyncMigrations(s.db.conn)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		_ = json.NewEncoder(w).Encode(map[string]string{
+			"error": "readAsyncMigrations: " + err.Error(),
+		})
+		return
+	}
+	if infos == nil {
+		infos = []AsyncMigrationInfo{}
+	}
+	writeJSON(w, infos)
 }

cmd/server/healthz.go

@@ -47,10 +47,23 @@ func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) {
 	// single writer. anyAsyncMigrationRunning intentionally drops to
 	// false on "failed" status — operator should see warm-up complete
 	// + alert, not an endless banner.
+	//
+	// #1735 finding #1 (Group A): on readAsyncMigrations error, surface
+	// the error AND keep async_migrations_running=true so the banner
+	// stays visible under uncertainty. We fail CLOSED for warm-up: if
+	// we cannot read the bookkeeping table, we treat the system as
+	// possibly still warming up rather than declaring "all clear".
 	var asyncMigrations []AsyncMigrationInfo
+	var asyncMigrationsErr string
+	var asyncRunning bool
 	if s.db != nil {
-		if infos, err := readAsyncMigrations(s.db.conn); err == nil {
+		infos, err := readAsyncMigrations(s.db.conn)
+		if err != nil {
+			asyncMigrationsErr = err.Error()
+			asyncRunning = true // fail closed — keep banner up
+		} else {
 			asyncMigrations = infos
+			asyncRunning = anyAsyncMigrationRunning(infos)
 		}
 	}
 	if asyncMigrations == nil {
@@ -68,7 +81,10 @@ func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) {
 			"done":      bfDone,
 		},
 		"async_migrations":         asyncMigrations,
-		"async_migrations_running": anyAsyncMigrationRunning(asyncMigrations),
+		"async_migrations_running": asyncRunning,
+	}
+	if asyncMigrationsErr != "" {
+		resp["async_migrations_error"] = asyncMigrationsErr
 	}
 	// PR #1609 M1: surface per-MQTT-source receipt vs write-path
 	// liveness so operators can distinguish "broker alive, write

cmd/server/routes.go

@@ -918,8 +918,19 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) {
 			if s.db == nil {
 				return []AsyncMigrationInfo{}
 			}
+			// #1735 finding #1 (Group A): on error, log + return
+			// empty BUT also set a header so operators have a
+			// signal. We can't 500 here because the rest of the
+			// /api/perf payload is still useful; the dedicated
+			// /api/perf/async-migrations endpoint DOES 500 (see
+			// handlePerfAsyncMigrations).
 			infos, err := readAsyncMigrations(s.db.conn)
-			if err != nil || infos == nil {
+			if err != nil {
+				log.Printf("[perf] readAsyncMigrations failed: %v", err)
+				w.Header().Set("X-Async-Migrations-Error", err.Error())
+				return []AsyncMigrationInfo{}
+			}
+			if infos == nil {
 				return []AsyncMigrationInfo{}
 			}
 			return infos