diff --git a/cmd/ingestor/async_migration_progress.go b/cmd/ingestor/async_migration_progress.go index 21721b0d..83789302 100644 --- a/cmd/ingestor/async_migration_progress.go +++ b/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 } diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index 62385a02..b0d1d60a 100644 --- a/cmd/ingestor/db.go +++ b/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 { diff --git a/cmd/server/async_migrations.go b/cmd/server/async_migrations.go index 39f996dc..8f96bdbc 100644 --- a/cmd/server/async_migrations.go +++ b/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) } diff --git a/cmd/server/healthz.go b/cmd/server/healthz.go index 26cc63a8..f5ef9248 100644 --- a/cmd/server/healthz.go +++ b/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 diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 40771d2e..e149b5f2 100644 --- a/cmd/server/routes.go +++ b/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