Merge reviewed candidate 20260919T190427Z-core-runtime (pulse e2635c3810)

Change-source: pulse-maintainer
2026-10-24 04:33:17 +00:00 · 2026-09-19 20:24:13 +01:00
parent 2bd6266b67 e2635c3810
commit dfb6c5a275
3 changed files with 277 additions and 43 deletions
@@ -3218,3 +3218,36 @@ 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`.
+
+### Metrics-store writes survive the one-time startup write-lock hold
+
+The deferred identity-index rebuild and the auto-vacuum VACUUM can hold the
+SQLite write lock for minutes on a large legacy database. `BEGIN` is deferred,
+so the lock is contended at the first `INSERT` or at `COMMIT`, not at `Begin`;
+a retry loop around `Begin` alone skipped rows and committed an empty
+transaction. `writeBatch` now retries the whole transaction on a retryable lock
+error, and while startup maintenance is active it uses an extended budget
+(`startupMaintenanceBeginAttempts`) instead of the steady-state
+`writeBatchBeginAttempts`, so an upgrade does not silently discard history. The
+`startupMaintenanceActive` signal is set for the duration of
+`runStartupMaintenance` and cleared on completion; the `stopping` check still
+bounds shutdown. `pkg/metrics/store_additional_test.go` pins the signal in
+`TestStoreStartupMaintenanceSignalsWriteRetryBudget` and the extended retry in
+`TestStoreWriteBatchExtendsRetryDuringStartupMaintenance`.
+
+### Metrics-store legacy host migration is probed before it writes
+
+`migrateLegacyHostResourceType` must not open a write transaction on every boot
+when there are no legacy rows. It probes with an indexed
+`SELECT EXISTS(... WHERE resource_type = 'host' LIMIT 1)` and returns without
+writing when the table has no legacy rows, so an already-migrated store does
+not dirty the WAL on startup.
+
+### Metrics-store QueryAll preallocation is capped
+
+`QueryAll` must not reserve a `MetricPoint` slice sized by the caller's
+requested step when the selected tier returns far fewer rows. A 90-day range at
+a 5-second step previously reserved over a million slots per series; the
+preallocation is now capped at `maxQueryAllSeriesCapacity` and append still
+grows the slice for genuinely dense series. `pkg/metrics/store_additional_test.go`
+pins the cap in `TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation`.
@@ -109,6 +109,28 @@ const (
 	defaultRollupInterval = 15 * time.Minute
 	maxRollupChunkWindow  = 5 * time.Minute
 	maxRollupChunksPerRun = 128
+
+	// writeBatchBeginAttempts bounds how many times a metrics write batch
+	// retries a busy transaction before it is dropped. Five attempts is enough
+	// for the steady-state WAL writer.
+	writeBatchBeginAttempts = 5
+	// startupMaintenanceBeginAttempts extends that budget while the one-time
+	// startup maintenance holds the SQLite write lock. The deferred
+	// identity-index rebuild and the auto-vacuum VACUUM can hold the lock for
+	// minutes on a large legacy database; the steady-state budget would drop
+	// every batch enqueued during an upgrade. Each attempt may itself wait
+	// busy_timeout (30s), so this covers roughly half an hour, matching
+	// migrateAutoVacuum's context deadline.
+	startupMaintenanceBeginAttempts = 70
+	// maxWriteRetryBackoff caps the sleep between write attempts so a fast
+	// non-lock failure cannot spin.
+	maxWriteRetryBackoff = 2 * time.Second
+
+	// maxQueryAllSeriesCapacity caps the per-series preallocation in QueryAll.
+	// A caller can request a fine step over a long range (e.g. 5s over 90 days)
+	// while the tier fallback actually yields far fewer rows; without a cap
+	// every series reserves hundreds of thousands of MetricPoint slots.
+	maxQueryAllSeriesCapacity = 4096
 )

 // DefaultConfig returns sensible defaults for metrics storage
@@ -272,6 +294,7 @@ type Store struct {
 	stopping                   atomic.Bool
 	syncWaitWarnNano           atomic.Int64
 	identityMigrationPending   atomic.Bool
+	startupMaintenanceActive   atomic.Bool
 	commercialRetentionSeconds atomic.Int64
 	commercialPurgeEligibleAt  atomic.Int64

@@ -724,6 +747,21 @@ func (s *Store) migrateAutoVacuum() {
 // canonical v6 `resource_type=agent`. This keeps reads/writes agent-only while
 // preserving historical data.
 func (s *Store) migrateLegacyHostResourceType() {
+	// Probe with the resource_type prefix of idx_metrics_query_all before
+	// opening a write transaction. The migration is a no-op on every store
+	// that has already run it, and an unconditional write transaction dirties
+	// the WAL on every boot for nothing.
+	var legacyHostExists bool
+	if err := s.db.QueryRow(
+		`SELECT EXISTS(SELECT 1 FROM metrics WHERE resource_type = 'host' LIMIT 1)`,
+	).Scan(&legacyHostExists); err != nil {
+		log.Warn().Err(err).Msg("Failed to probe legacy host metrics rows")
+		return
+	}
+	if !legacyHostExists {
+		return
+	}
+
 	tx, err := s.db.Begin()
 	if err != nil {
 		log.Warn().Err(err).Msg("Failed to start legacy host->agent metrics migration")
@@ -992,6 +1030,14 @@ func (s *Store) WaitForMaintenance(timeout time.Duration) error {

 func (s *Store) runStartupMaintenance() {
 	start := time.Now()
+
+	// Signal the ingestion worker that the write lock may be held for a long
+	// time by the deferred identity-index rebuild and the auto-vacuum VACUUM.
+	// Without this a large legacy database drops every metric batch enqueued
+	// during the one-time upgrade (#2079 finding 5).
+	s.startupMaintenanceActive.Store(true)
+	defer s.startupMaintenanceActive.Store(false)
+
 	if s.startupHook != nil {
 		s.startupHook()
 	}
@@ -1196,26 +1242,49 @@ func (s *Store) writeBatch(metrics []bufferedMetric) {
 			Msg("Coalesced duplicate metrics before write")
 	}

-	var tx *pdb.InstrumentedTx
-	var err error
-
-	// Retry on SQLITE_BUSY with exponential backoff
-	for i := 0; i < 5; i++ {
-		tx, err = s.db.Begin()
-		if err == nil {
-			break
-		}
-		if i < 4 && isRetryableWriteError(err) {
-			time.Sleep(time.Duration(100*(i+1)) * time.Millisecond)
-			continue
-		}
-		log.Error().Err(err).
-			Str("component", "metrics_store").
-			Str("action", "begin_write_tx").
-			Int("batch_size", len(metrics)).
-			Msg("Failed to begin metrics transaction")
-		return
+	// Retry the whole transaction on SQLITE_BUSY with exponential backoff.
+	// BEGIN is deferred, so the write lock is actually contended at the first
+	// INSERT or at COMMIT; retrying only Begin() would skip rows and commit an
+	// empty transaction. While the one-time startup maintenance holds the
+	// write lock, extend the budget rather than dropping durable history
+	// during an upgrade (#2079 finding 5). The stopping check keeps shutdown
+	// from waiting out the extended budget.
+	maxAttempts := writeBatchBeginAttempts
+	if s.startupMaintenanceActive.Load() {
+		maxAttempts = startupMaintenanceBeginAttempts
 	}
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		retryable, err := s.writeBatchOnce(metrics)
+		if err == nil {
+			log.Debug().Int("input_count", inputCount).Int("count", len(metrics)).Msg("Wrote metrics batch")
+			return
+		}
+		if !retryable || attempt == maxAttempts-1 || s.stopping.Load() {
+			log.Error().Err(err).
+				Str("component", "metrics_store").
+				Str("action", "write_batch").
+				Int("batch_size", len(metrics)).
+				Msg("Failed to write metrics batch")
+			return
+		}
+		backoff := time.Duration(100*(attempt+1)) * time.Millisecond
+		if backoff > maxWriteRetryBackoff {
+			backoff = maxWriteRetryBackoff
+		}
+		time.Sleep(backoff)
+	}
+}
+
+// writeBatchOnce commits one metrics transaction. It reports retryable=true
+// when the failure was a transient lock so the caller can retry the whole
+// batch instead of dropping it. A non-retryable per-row insert error is logged
+// and skipped, preserving the batch's other rows.
+func (s *Store) writeBatchOnce(metrics []bufferedMetric) (bool, error) {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return isRetryableWriteError(err), fmt.Errorf("begin metrics transaction: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()

 	stmt, err := tx.Prepare(`
 		INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier)
@@ -1227,40 +1296,32 @@ func (s *Store) writeBatch(metrics []bufferedMetric) {
 			max_value = COALESCE(excluded.max_value, max_value)
 	`)
 	if err != nil {
-		_ = tx.Rollback()
-		log.Error().Err(err).
-			Str("component", "metrics_store").
-			Str("action", "prepare_write_stmt").
-			Int("batch_size", len(metrics)).
-			Msg("Failed to prepare metrics insert")
-		return
+		return false, fmt.Errorf("prepare metrics insert: %w", err)
 	}
 	defer stmt.Close()

 	for _, m := range metrics {
 		_, err := stmt.Exec(m.resourceType, m.resourceID, m.metricType, m.value, m.timestamp.Unix(), string(m.tier))
-		if err != nil {
-			log.Warn().Err(err).
-				Str("component", "metrics_store").
-				Str("action", "insert_metric").
-				Str("resource_type", m.resourceType).
-				Str("resource_id", m.resourceID).
-				Str("metric_type", m.metricType).
-				Str("tier", string(m.tier)).
-				Msg("Failed to insert metric")
+		if err == nil {
+			continue
 		}
+		if isRetryableWriteError(err) {
+			return true, fmt.Errorf("insert metric: %w", err)
+		}
+		log.Warn().Err(err).
+			Str("component", "metrics_store").
+			Str("action", "insert_metric").
+			Str("resource_type", m.resourceType).
+			Str("resource_id", m.resourceID).
+			Str("metric_type", m.metricType).
+			Str("tier", string(m.tier)).
+			Msg("Failed to insert metric")
 	}

 	if err := tx.Commit(); err != nil {
-		log.Error().Err(err).
-			Str("component", "metrics_store").
-			Str("action", "commit_write_tx").
-			Int("batch_size", len(metrics)).
-			Msg("Failed to commit metrics batch")
-		return
+		return isRetryableWriteError(err), fmt.Errorf("commit metrics batch: %w", err)
 	}
-
-	log.Debug().Int("input_count", inputCount).Int("count", len(metrics)).Msg("Wrote metrics batch")
+	return false, nil
 }

 func coalesceMetricBatch(metrics []bufferedMetric) []bufferedMetric {
@@ -1846,6 +1907,13 @@ func estimateQueryAllBatchSeriesCapacity(start, end time.Time, stepSecs int64) i
 	if buckets < 1 {
 		return 1
 	}
+	// The selected tier's actual resolution is usually far coarser than the
+	// requested step, so the row count is much smaller than the bucket count.
+	// Cap the preallocation; append still grows the slice for genuinely dense
+	// series.
+	if buckets > maxQueryAllSeriesCapacity {
+		return maxQueryAllSeriesCapacity
+	}
 	return buckets
 }

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"testing"
 	"time"
@@ -1719,3 +1720,135 @@ func TestStoreAutoVacuumPersistsAcrossRestart(t *testing.T) {
 		}
 	}
 }
+
+// TestStoreStartupMaintenanceSignalsWriteRetryBudget verifies the ingestion
+// worker can tell when the one-time startup maintenance holds the write lock,
+// and that the signal is cleared once maintenance completes.
+func TestStoreStartupMaintenanceSignalsWriteRetryBudget(t *testing.T) {
+	previousHook := startupMaintenanceHook
+	defer func() { startupMaintenanceHook = previousHook }()
+
+	// The automatic startup run may begin before NewStore returns, so the hook
+	// reads the store through an atomic pointer rather than a captured local.
+	var storePtr atomic.Pointer[Store]
+	var observedDuringHook atomic.Bool
+	startupMaintenanceHook = func() {
+		if s := storePtr.Load(); s != nil {
+			observedDuringHook.Store(s.startupMaintenanceActive.Load())
+		}
+	}
+
+	dir := t.TempDir()
+	cfg := DefaultConfig(dir)
+	cfg.DBPath = filepath.Join(dir, "startup-signal.db")
+	cfg.FlushInterval = time.Hour
+
+	store, err := NewStore(cfg)
+	if err != nil {
+		t.Fatalf("NewStore returned error: %v", err)
+	}
+	defer store.Close()
+	storePtr.Store(store)
+
+	if err := store.WaitForMaintenance(10 * time.Second); err != nil {
+		t.Fatalf("maintenance did not complete: %v", err)
+	}
+
+	// Re-run synchronously so the hook is guaranteed to observe the flag while
+	// it is active.
+	observedDuringHook.Store(false)
+	store.runStartupMaintenance()
+
+	if !observedDuringHook.Load() {
+		t.Fatalf("startup maintenance did not signal an active write-retry budget")
+	}
+	if store.startupMaintenanceActive.Load() {
+		t.Fatalf("startup maintenance left the extended write-retry budget active")
+	}
+}
+
+// TestStoreWriteBatchExtendsRetryDuringStartupMaintenance proves that a batch
+// enqueued while the startup maintenance holds the write lock survives past
+// the steady-state retry budget. Without the extension it is dropped after
+// five fast attempts (#2079 finding 5).
+func TestStoreWriteBatchExtendsRetryDuringStartupMaintenance(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "retry.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)
+	schema := `
+		CREATE TABLE metrics (
+			id INTEGER PRIMARY KEY AUTOINCREMENT,
+			resource_type TEXT NOT NULL,
+			resource_id TEXT NOT NULL,
+			metric_type TEXT NOT NULL,
+			value REAL NOT NULL,
+			min_value REAL,
+			max_value REAL,
+			timestamp INTEGER NOT NULL,
+			tier TEXT NOT NULL DEFAULT 'raw'
+		);
+		CREATE UNIQUE INDEX idx_metrics_lookup
+		ON metrics(resource_type, resource_id, metric_type, tier, timestamp);
+	`
+	if _, err := holder.Exec(schema); err != nil {
+		t.Fatalf("create schema: %v", err)
+	}
+	if _, err := holder.Exec("BEGIN IMMEDIATE"); err != nil {
+		t.Fatalf("begin immediate: %v", err)
+	}
+
+	rawDB, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(0)")
+	if err != nil {
+		t.Fatalf("open writer: %v", err)
+	}
+	defer rawDB.Close()
+	store := &Store{db: pdb.Wrap(rawDB, "metrics")}
+	store.startupMaintenanceActive.Store(true)
+
+	// Hold the write lock for longer than the steady-state budget
+	// (5 attempts, ~1s of backoff) but within the extended budget.
+	released := make(chan struct{})
+	go func() {
+		time.Sleep(2 * time.Second)
+		_, _ = holder.Exec("ROLLBACK")
+		close(released)
+	}()
+
+	store.writeBatch([]bufferedMetric{{
+		resourceType: "vm",
+		resourceID:   "vm-1",
+		metricType:   "cpu",
+		value:        1,
+		timestamp:    time.Now().UTC().Truncate(time.Second),
+		tier:         TierRaw,
+	}})
+	<-released
+
+	var count int
+	if err := store.db.QueryRow("SELECT COUNT(*) FROM metrics").Scan(&count); err != nil {
+		t.Fatalf("count metrics: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("expected the batch to survive the startup lock window, got %d rows", count)
+	}
+}
+
+// TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation verifies a fine
+// requested step over a long range cannot reserve hundreds of thousands of
+// MetricPoint slots per series when the tier fallback returns far fewer rows.
+func TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation(t *testing.T) {
+	start := time.Unix(0, 0)
+	end := start.Add(90 * 24 * time.Hour)
+	if got := estimateQueryAllBatchSeriesCapacity(start, end, 5); got != maxQueryAllSeriesCapacity {
+		t.Fatalf("90 days at 5s step: capacity = %d, want cap %d", got, maxQueryAllSeriesCapacity)
+	}
+	if got := estimateQueryAllBatchSeriesCapacity(start, start.Add(10*time.Minute), 5); got != 122 {
+		t.Fatalf("10 minutes at 5s step: capacity = %d, want 122", got)
+	}
+}