Merge reviewed candidate 20260919T180436Z-core-runtime (pulse f569c6a8eb)

Harden pkg/metrics/store.go rollup checkpoints, SQLite busy/locked retry classification, rollup min/max spread preservation, and one-connection auto-vacuum migration (#2079 findings 2-4, 6). Includes the performance-and-scalability contract sections, the regression proof moved into the registry-accepted store_additional_test.go, and the reviewed completion-pair registration for the split proof commit.

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_additional_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_additional_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_additional_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_additional_test.go

@@ -1542,3 +1542,180 @@ func TestConcurrentWritesDuringShutdownDoNotPanic(t *testing.T) {
 	close(stop)
 	wg.Wait()
 }
+
+// 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)
+		}
+	}
+}

scripts/release_control/canonical_completion_history.json

@@ -5,6 +5,11 @@
       "incomplete_commit": "a2dbb27a68463adb02b614e9927174543f5b7799",
       "completion_commit": "5f99f5c13fcc547e5e34286dfe873415a4876cfa",
       "reason": "Preserve the reviewed diagnostic-redaction commit while evaluating its separately reviewed contracts and API proof as one historical completion unit."
+    },
+    {
+      "incomplete_commit": "e1d97b361edfac311dd57b6695ab3ad9bef613bc",
+      "completion_commit": "087bbf8c20db8ffec1c24037e78a1a88ea4ff69e",
+      "reason": "Evaluate the metrics-store rollup, retry, upsert and auto-vacuum repair together with the follow-up commit that moves its regression proof into the registry-accepted pkg/metrics/store_additional_test.go proof file, as one reviewed completion unit."
     }
   ]
 }