From e1d97b361edfac311dd57b6695ab3ad9bef613bc Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 19 Sep 2026 19:19:13 +0100 Subject: [PATCH] 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 --- .../subsystems/performance-and-scalability.md | 39 ++++ pkg/metrics/store.go | 97 ++++++++-- pkg/metrics/store_test.go | 178 ++++++++++++++++++ 3 files changed, 301 insertions(+), 13 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index c40cb46b1b..29fb7354e5 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/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`. diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index e998c8629d..cea4491c93 100644 --- a/pkg/metrics/store.go +++ b/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 diff --git a/pkg/metrics/store_test.go b/pkg/metrics/store_test.go index 552f16e0ff..7bb6a8e9d1 100644 --- a/pkg/metrics/store_test.go +++ b/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) + } + } +}