fix(metrics): harden store rollups, retries and auto-vacuum

Address the remaining verified findings from the pkg/metrics/store.go review
(#2079 findings 2-4 and 6):

- rollupTier no longer leaps its checkpoint over a source gap. Previously a
  trailing or interspersed gap advanced the checkpoint to the rollup cutoff, so
  a lower tier that backfilled the gap later had those rows stranded below the
  upper checkpoint and purged by retention.
- Retry conditions now classify SQLITE_BUSY/LOCKED by driver result code. The
  modernc driver reports "database is locked (5) (SQLITE_BUSY)", which the
  previous exact-string comparison never matched, so a busy writer dropped its
  batch without retrying.
- The metrics upsert preserves an existing rollup row's min/max spread with
  COALESCE instead of overwriting it with NULL.
- migrateAutoVacuum pins the pragma and VACUUM to one connection and verifies
  the persisted mode, so the one-time conversion cannot silently repeat a
  full-file VACUUM on every restart.

Regressions: TestStoreRollupTierBackfillsInterspersedGap,
TestStoreWriteBatchPreservesRollupSpread, TestIsRetryableWriteError,
TestIsRetryableWriteErrorMatchesDriverBusyCode,
TestStoreAutoVacuumPersistsAcrossRestart. The performance-and-scalability
contract records the invariants.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/performance-and-scalability.md

@@ -3179,3 +3179,42 @@ and panics the process during shutdown. The worker instead drains already-queued
 requests without closing the channel; a late write lands in the buffered channel
 and is discarded with the store. `pkg/metrics/store_additional_test.go` pins the
 post-shutdown enqueue and the concurrent-close race.
+
+### Metrics-store rollup checkpoints never leap source gaps
+
+A tier rollup must not advance its checkpoint past a window that has no source
+rows. The previous implementation jumped the checkpoint to the rollup cutoff
+whenever a source gap followed already-processed windows, so a lower tier that
+backfilled the gap later (for example after a transient raw-to-minute failure)
+had its rows stranded below the upper tier's checkpoint and eventually purged by
+retention. `rollupTier` now stops at the gap, leaving the checkpoint at the gap
+start; the next invocation skips the gap once it is confirmed empty. The
+`nextSourceRollupBucket` prefix skip still runs only before the first source
+window of an invocation, preserving bounded catch-up. `pkg/metrics/store_test.go`
+pins the interspersed-gap backfill in
+`TestStoreRollupTierBackfillsInterspersedGap` and keeps
+`TestStoreRollupTierEmptyWindowPreservesCheckpoint`.
+
+### Metrics-store retries match the SQLite driver error code
+
+Transient write and read retries must classify SQLite lock errors by the
+`modernc.org/sqlite` result code rather than by an exact message. The driver
+reports `database is locked (5) (SQLITE_BUSY)`, which an equality check against
+`database is locked` never matched, so a busy writer dropped its batch without
+retrying. `isRetryableWriteError` matches the BUSY/LOCKED code family with a
+substring fallback for closed-pool errors. A direct aggregate-tier write also
+preserves an existing rollup row's `min_value`/`max_value` via `COALESCE` rather
+than nulling the stored spread. `pkg/metrics/store_test.go` pins both in
+`TestIsRetryableWriteErrorMatchesDriverBusyCode` and
+`TestStoreWriteBatchPreservesRollupSpread`.
+
+### One-time auto-vacuum conversion runs on one connection
+
+`migrateAutoVacuum` must run `PRAGMA auto_vacuum = INCREMENTAL` and the
+follow-up `VACUUM` on the same SQLite connection, because the pragma is only a
+per-connection setting until that connection's `VACUUM` rewrites the file
+header. Running them on separate pool connections could leave the file at NONE
+and repeat the full-file VACUUM on every restart. The migration now pins a
+connection, verifies the persisted mode, and leaves the conversion for a later
+restart if verification fails. `pkg/metrics/store_test.go` pins the persisted
+mode in `TestStoreAutoVacuumPersistsAcrossRestart`.

pkg/metrics/store.go

@@ -17,7 +17,7 @@ import (
 	"time"
 
 	"github.com/rs/zerolog/log"
-	_ "modernc.org/sqlite"
+	sqlite "modernc.org/sqlite"
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
 	pdb "github.com/rcourtman/pulse-go-rewrite/pkg/db"
@@ -28,6 +28,52 @@ const (
 	privateFilePerm = 0o600
 )
 
+// SQLite result codes for transient lock conditions. The modernc driver
+// surfaces these as *sqlite.Error, and its message text is e.g.
+// "database is locked (5) (SQLITE_BUSY)", so callers must match the code
+// rather than a bare "database is locked" string.
+const (
+	sqliteCodeBusy              = 5
+	sqliteCodeLocked            = 6
+	sqliteCodeBusyRecovery      = sqliteCodeBusy | (1 << 8)
+	sqliteCodeBusySnapshot      = sqliteCodeBusy | (2 << 8)
+	sqliteCodeBusyTimeout       = sqliteCodeBusy | (3 << 8)
+	sqliteCodeLockedSharedCache = sqliteCodeLocked | (1 << 8)
+	sqliteCodeLockedVTab        = sqliteCodeLocked | (2 << 8)
+)
+
+// isRetryableWriteError reports whether err is a transient SQLite lock or a
+// closed pool connection that should be retried rather than dropped. The
+// previous exact string comparison against "database is locked" never matched
+// the modernc driver's actual message, so a busy writer dropped its batch
+// immediately instead of backing off.
+func isRetryableWriteError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	var sqliteErr *sqlite.Error
+	if errors.As(err, &sqliteErr) {
+		switch sqliteErr.Code() {
+		case sqliteCodeBusy,
+			sqliteCodeLocked,
+			sqliteCodeBusyRecovery,
+			sqliteCodeBusySnapshot,
+			sqliteCodeBusyTimeout,
+			sqliteCodeLockedSharedCache,
+			sqliteCodeLockedVTab:
+			return true
+		}
+	}
+
+	message := strings.ToLower(err.Error())
+	return strings.Contains(message, "database is locked") ||
+		strings.Contains(message, "sqlite_busy") ||
+		strings.Contains(message, "sqlite_locked") ||
+		strings.Contains(message, "database table is locked") ||
+		message == "sql: database is closed"
+}
+
 // Tier represents the granularity of stored metrics
 type Tier string
 
@@ -630,8 +676,22 @@ func (s *Store) metricsIndexMatches(name string, wantUnique bool, wantColumns []
 // SQLite cannot switch from NONE to INCREMENTAL without a full VACUUM to
 // restructure the file, so we detect and convert on first run after upgrade.
 func (s *Store) migrateAutoVacuum() {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+
+	// auto_vacuum is only a per-connection setting until a VACUUM on that same
+	// connection rewrites the file header. Running the pragma and the VACUUM on
+	// separate pool connections could leave the file at NONE, so pin both to one
+	// connection and verify the mode actually persisted.
+	conn, err := s.db.DB.Conn(ctx)
+	if err != nil {
+		log.Debug().Err(err).Msg("Failed to acquire connection for auto_vacuum check")
+		return
+	}
+	defer conn.Close()
+
 	var mode int
-	if err := s.db.QueryRow("PRAGMA auto_vacuum").Scan(&mode); err != nil {
+	if err := conn.QueryRowContext(ctx, "PRAGMA auto_vacuum").Scan(&mode); err != nil {
 		log.Debug().Err(err).Msg("Failed to check auto_vacuum mode")
 		return
 	}
@@ -643,15 +703,20 @@ func (s *Store) migrateAutoVacuum() {
 	start := time.Now()
 
 	// Set the desired mode then VACUUM to restructure the file.
-	if _, err := s.db.Exec("PRAGMA auto_vacuum = INCREMENTAL"); err != nil {
+	if _, err := conn.ExecContext(ctx, "PRAGMA auto_vacuum = INCREMENTAL"); err != nil {
 		log.Warn().Err(err).Msg("Failed to set auto_vacuum mode")
 		return
 	}
-	if _, err := s.db.Exec("VACUUM"); err != nil {
+	if _, err := conn.ExecContext(ctx, "VACUUM"); err != nil {
 		log.Warn().Err(err).Msg("Auto-vacuum migration VACUUM failed (will retry next restart)")
 		return
 	}
 
+	if err := conn.QueryRowContext(ctx, "PRAGMA auto_vacuum").Scan(&mode); err != nil || mode != 2 {
+		log.Warn().Err(err).Int("mode", mode).Msg("Auto-vacuum migration did not persist incremental mode")
+		return
+	}
+
 	log.Info().Dur("duration", time.Since(start)).Msg("Metrics database auto-vacuum migration complete")
 }
 
@@ -1140,7 +1205,7 @@ func (s *Store) writeBatch(metrics []bufferedMetric) {
 		if err == nil {
 			break
 		}
-		if i < 4 && (err.Error() == "database is locked" || err.Error() == "sql: database is closed") {
+		if i < 4 && isRetryableWriteError(err) {
 			time.Sleep(time.Duration(100*(i+1)) * time.Millisecond)
 			continue
 		}
@@ -1158,8 +1223,8 @@ func (s *Store) writeBatch(metrics []bufferedMetric) {
 		ON CONFLICT(resource_type, resource_id, metric_type, tier, timestamp)
 		DO UPDATE SET
 			value = excluded.value,
-			min_value = excluded.min_value,
-			max_value = excluded.max_value
+			min_value = COALESCE(excluded.min_value, min_value),
+			max_value = COALESCE(excluded.max_value, max_value)
 	`)
 	if err != nil {
 		_ = tx.Rollback()
@@ -1662,7 +1727,7 @@ func (s *Store) queryRetainedChunk(resourceType string, resourceIDs []string, me
 		if err == nil {
 			break
 		}
-		if i < 4 && (err.Error() == "database is locked" || err.Error() == "sql: database is closed") {
+		if i < 4 && isRetryableWriteError(err) {
 			time.Sleep(time.Duration(100*(i+1)) * time.Millisecond)
 			continue
 		}
@@ -1971,15 +2036,21 @@ func (s *Store) rollupTier(fromTier, toTier Tier, bucketSize, minAge time.Durati
 	for windowStart < cutoffBucket {
 		nextBucket, hasSource := s.nextSourceRollupBucket(fromTier, windowStart, cutoffBucket, bucketSecs)
 		if !hasSource {
-			if processedAny {
-				if err := s.setMetaInt(metaKey, cutoffBucket); err != nil {
-					log.Warn().Err(err).Str("tier", string(fromTier)).Msg("Failed to persist rollup checkpoint")
-				}
-			}
+			// No source remains at or after the checkpoint. Leave the checkpoint
+			// where it is so a late backfill into the trailing gap is still rolled
+			// up on a later run; do not leap it to the cutoff.
 			return
 		}
 
 		if nextBucket > windowStart {
+			if processedAny {
+				// A gap sits between already-processed source and the next source.
+				// Stop and leave the checkpoint at the gap start so the gap can
+				// still be rolled up if the lower tier backfills it later. The next
+				// run skips the gap once it is known to be empty.
+				return
+			}
+			// Skip the empty prefix before the first source in this run.
 			windowStart = nextBucket
 		}
 		windowEnd := windowStart + chunkSecs

pkg/metrics/store_test.go

@@ -1,6 +1,7 @@
 package metrics
 
 import (
+	"database/sql"
 	"fmt"
 	"path/filepath"
 	"testing"
@@ -794,3 +795,180 @@ func TestQueryAllBatch(t *testing.T) {
 		}
 	})
 }
+
+// TestStoreRollupTierBackfillsInterspersedGap verifies that a rollup checkpoint
+// does not leap over a gap between already-processed source windows. A late
+// backfill into that gap must still be aggregated on a later run instead of
+// being stranded below the checkpoint and eventually purged.
+func TestStoreRollupTierBackfillsInterspersedGap(t *testing.T) {
+	dir := t.TempDir()
+	cfg := DefaultConfig(dir)
+	cfg.DBPath = filepath.Join(dir, "metrics-rollup-gap.db")
+	cfg.FlushInterval = time.Hour
+
+	store, err := NewStore(cfg)
+	if err != nil {
+		t.Fatalf("NewStore returned error: %v", err)
+	}
+	defer store.Close()
+
+	base := time.Now().UTC().Add(-4 * time.Hour).Truncate(time.Hour)
+	firstHour := base.Unix()
+	thirdHour := base.Add(2 * time.Hour).Unix()
+	gapHour := base.Add(time.Hour).Unix()
+
+	if _, err := store.db.Exec(
+		`INSERT INTO metrics (resource_type, resource_id, metric_type, value, min_value, max_value, timestamp, tier) VALUES
+		('vm','vm-gap','cpu',1.0,1.0,1.0,?, 'minute'),
+		('vm','vm-gap','cpu',3.0,3.0,3.0,?, 'minute')`, firstHour, thirdHour); err != nil {
+		t.Fatalf("insert source rows: %v", err)
+	}
+
+	store.rollupTier(TierMinute, TierHourly, time.Hour, time.Hour)
+
+	// Backfill the interspersed gap after the first rollup ran.
+	if _, err := store.db.Exec(
+		`INSERT INTO metrics (resource_type, resource_id, metric_type, value, min_value, max_value, timestamp, tier) VALUES
+		('vm','vm-gap','cpu',2.0,2.0,2.0,?, 'minute')`, gapHour); err != nil {
+		t.Fatalf("backfill gap row: %v", err)
+	}
+
+	store.rollupTier(TierMinute, TierHourly, time.Hour, time.Hour)
+
+	var count int
+	if err := store.db.QueryRow(
+		`SELECT COUNT(*) FROM metrics WHERE tier='hourly' AND resource_id='vm-gap' AND timestamp=?`, gapHour).Scan(&count); err != nil {
+		t.Fatalf("query hourly gap: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("expected backfilled gap hour %d to roll up to hourly, got %d rows", gapHour, count)
+	}
+}
+
+// TestStoreWriteBatchPreservesRollupSpread verifies that a direct write to an
+// aggregate tier does not erase an existing rollup row's min/max spread. The
+// write statement omits min/max, so excluded.* are NULL and a naive upsert
+// would null the stored spread.
+func TestStoreWriteBatchPreservesRollupSpread(t *testing.T) {
+	dir := t.TempDir()
+	cfg := DefaultConfig(dir)
+	cfg.DBPath = filepath.Join(dir, "metrics-spread.db")
+	cfg.FlushInterval = time.Hour
+
+	store, err := NewStore(cfg)
+	if err != nil {
+		t.Fatalf("NewStore returned error: %v", err)
+	}
+	defer store.Close()
+
+	ts := time.Now().UTC().Add(-2 * time.Hour).Truncate(time.Minute)
+	if _, err := store.db.Exec(
+		`INSERT INTO metrics (resource_type, resource_id, metric_type, value, min_value, max_value, timestamp, tier) VALUES
+		('vm','vm-spread','cpu',5.0,1.0,9.0,?, 'minute')`, ts.Unix()); err != nil {
+		t.Fatalf("insert rollup row: %v", err)
+	}
+
+	store.writeBatch([]bufferedMetric{
+		{resourceType: "vm", resourceID: "vm-spread", metricType: "cpu", value: 7.0, timestamp: ts, tier: TierMinute},
+	})
+
+	var minValue, maxValue sql.NullFloat64
+	if err := store.db.QueryRow(
+		`SELECT min_value, max_value FROM metrics WHERE tier='minute' AND resource_id='vm-spread' AND timestamp=?`, ts.Unix()).Scan(&minValue, &maxValue); err != nil {
+		t.Fatalf("query spread: %v", err)
+	}
+	if !minValue.Valid || !maxValue.Valid {
+		t.Fatalf("direct write erased rollup spread: min=%+v max=%+v", minValue, maxValue)
+	}
+	if minValue.Float64 != 1.0 || maxValue.Float64 != 9.0 {
+		t.Fatalf("unexpected spread after direct write: min=%v max=%v", minValue.Float64, maxValue.Float64)
+	}
+}
+
+func TestIsRetryableWriteError(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"driver busy", fmt.Errorf("database is locked (5) (SQLITE_BUSY)"), true},
+		{"legacy busy", fmt.Errorf("database is locked"), true},
+		{"closed pool", fmt.Errorf("sql: database is closed"), true},
+		{"unrelated", fmt.Errorf("no such table: metrics"), false},
+	}
+	for _, tc := range cases {
+		if got := isRetryableWriteError(tc.err); got != tc.want {
+			t.Fatalf("%s: isRetryableWriteError(%v) = %v, want %v", tc.name, tc.err, got, tc.want)
+		}
+	}
+}
+
+// TestIsRetryableWriteErrorMatchesDriverBusyCode proves the classifier matches
+// a real SQLITE_BUSY error returned by the modernc driver, not just its message.
+func TestIsRetryableWriteErrorMatchesDriverBusyCode(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "busy.db")
+
+	holder, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(0)")
+	if err != nil {
+		t.Fatalf("open holder: %v", err)
+	}
+	defer holder.Close()
+	holder.SetMaxOpenConns(1)
+	if _, err := holder.Exec("CREATE TABLE t (x INTEGER)"); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+	if _, err := holder.Exec("BEGIN IMMEDIATE"); err != nil {
+		t.Fatalf("begin immediate: %v", err)
+	}
+	defer holder.Exec("ROLLBACK")
+
+	waiter, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(0)")
+	if err != nil {
+		t.Fatalf("open waiter: %v", err)
+	}
+	defer waiter.Close()
+
+	_, err = waiter.Exec("BEGIN IMMEDIATE")
+	if err == nil {
+		waiter.Exec("ROLLBACK")
+		t.Skip("driver did not report SQLITE_BUSY on a contended write lock")
+	}
+	if !isRetryableWriteError(err) {
+		t.Fatalf("real driver busy error not classified retryable: %v", err)
+	}
+}
+
+// TestStoreAutoVacuumPersistsAcrossRestart verifies that the one-time
+// auto-vacuum conversion is committed to the database file and remains in
+// effect after reopening the store, rather than being re-run every restart.
+func TestStoreAutoVacuumPersistsAcrossRestart(t *testing.T) {
+	dir := t.TempDir()
+	cfg := DefaultConfig(dir)
+	cfg.DBPath = filepath.Join(dir, "metrics-autovacuum.db")
+	cfg.FlushInterval = time.Hour
+
+	for i := 0; i < 2; i++ {
+		store, err := NewStore(cfg)
+		if err != nil {
+			t.Fatalf("open %d: %v", i, err)
+		}
+		if err := store.WaitForMaintenance(10 * time.Second); err != nil {
+			store.Close()
+			t.Fatalf("maintenance %d: %v", i, err)
+		}
+		var mode int
+		if err := store.db.QueryRow("PRAGMA auto_vacuum").Scan(&mode); err != nil {
+			store.Close()
+			t.Fatalf("pragma %d: %v", i, err)
+		}
+		if mode != 2 {
+			store.Close()
+			t.Fatalf("open %d: auto_vacuum = %d, want 2 (INCREMENTAL)", i, mode)
+		}
+		if err := store.Close(); err != nil {
+			t.Fatalf("close %d: %v", i, err)
+		}
+	}
+}