Reduce metrics SQLite write amplification

docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-metrics-write-amplification-2026-05-03.md

@@ -60,3 +60,49 @@ The metrics write-amplification path reported in `#1124` has a v6 runtime fix,
 operator-facing controls, documentation, and subsystem-contract coverage. The
 `known-rc-issue-closure-for-ga` gate remains satisfied for the current v6
 release candidate.
+
+## 2026-07-24 Physical-Write Revalidation
+
+The earlier proof established rollup controls and bounded retention, but it did
+not attribute physical writes by SQLite object or quantify WAL/checkpoint
+amplification. Issue `#1124` therefore remained open after a v6.1.1 operator
+reported continued writes. A deterministic 2,197-resource estate
+(Docker containers, Proxmox guests, agents, Kubernetes pods, storage targets,
+and physical disks) now persists 393,630 samples over 30 simulated 10-second
+polls and reports logical payload, per-object `dbstat`, page-cache, WAL,
+checkpoint, process-write, restart, integrity, and retention counters.
+
+The v6.1.1 and pre-fix `main` store implementations were byte-identical. Their
+four-index schema wrote 279,063 WAL frames (1,149,739,592 bytes) for
+21,214,200 logical payload bytes. The retained main database was 106,991,616
+bytes, of which the overlapping `idx_metrics_lookup` and
+`idx_metrics_unique` trees each occupied 24,014,848 bytes. This identifies the
+amplification source directly: each metric insert maintained the table,
+`sqlite_sequence`, and four indexes before checkpointing the dirty pages back
+to the main file.
+
+The revalidated schema makes the lookup index unique and orders its columns as
+the metric identity, eliminating the second overlapping tree without changing
+query order or durability settings. On the same workload it wrote 187,616 WAL
+frames (772,977,952 bytes), a 32.8% frame/byte reduction, and retained
+82,948,096 bytes. Linux process I/O accounting measured WAL-phase writes
+falling from 1,155,710,976 to 774,930,432 bytes and explicit checkpoint writes
+from 106,983,424 to 82,939,904 bytes (32.1% fewer total bytes). With the
+production 4,000-page auto-checkpoint enabled, a traced 157,452-sample run
+reduced total process write bytes from 392,445,952 to 257,208,320 and `fsync`
+calls from 42 to 32.
+
+The migration keeps the legacy unique index authoritative until deferred
+startup maintenance transactionally swaps the indexes. Tests cover legacy
+duplicates, close/restart, a process exit between index drop and replacement,
+backup/restore from both schema generations, writes racing the migration, and
+reads remaining available on the four-connection WAL pool. Query-plan checks
+still require indexed searches. Concurrent read/write p95 remained 0.88 ms
+against a 5 ms SLO, and the 500-node/10-reader dashboard p95 remained 2.37 ms
+against a 30 ms SLO. Retention removed all 393,630 expired rows and incremental
+vacuum reduced the 82,948,096-byte database to 36,864 bytes with zero freelist
+pages in 2.57 seconds.
+
+The durable store remains in WAL mode with `synchronous=NORMAL`; no
+authoritative history moved to volatile memory. The fix is on `main` for the
+next release and is not part of v6.1.1.

docs/release-control/v6/internal/status.json

@@ -9206,21 +9206,7 @@
       ]
     }
   ],
-  "work_claims": [
-    {
-      "id": "issue-1124-metrics-write-amplification-lane-l10",
-      "agent_id": "issue-1124-metrics-write-amplification",
-      "summary": "Measure and reduce metrics.db physical write amplification for issue #1124 without weakening durability or bounded runtime behavior.",
-      "target_id": "v6-product-lane-expansion",
-      "claimed_at": "2026-07-24T08:05:22Z",
-      "heartbeat_at": "2026-07-24T08:05:22Z",
-      "expires_at": "2026-07-24T16:05:22Z",
-      "work_item": {
-        "kind": "lane",
-        "id": "L10"
-      }
-    }
-  ],
+  "work_claims": [],
   "open_decisions": [],
   "source_of_truth_file": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md",
   "resolved_decisions": [

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

@@ -1201,6 +1201,23 @@ WAL checkpoints more aggressive again, reopens duplicate-write failures,
 removes the bounded rollup-window checkpointing, or removes the metrics DB
 path/cadence controls must re-prove the metrics-store hot path with the owned
 store tests rather than assuming the earlier vacuum fixes are sufficient.
+The metrics identity and single-series lookup must also share one unique
+`idx_metrics_lookup(resource_type, resource_id, metric_type, tier, timestamp)`
+B-tree. Reintroducing a second full-width identity index is a write-amplification
+regression even when retained `metrics.db` size looks small: every sample would
+dirty both trees in the WAL and every checkpoint would copy both sets of pages
+back into the main file. Legacy databases may keep their existing unique index
+authoritative while the replacement index is built, but that O(rows) migration
+must run on deferred startup maintenance and swap indexes in one SQLite
+transaction so constructor latency, concurrent reads, uniqueness, and crash
+recovery remain intact. The deterministic issue-1124 profile must continue to
+attribute writes by logical payload, table/index `dbstat` pages, page-cache
+writes/spills, WAL frames, checkpoint frames, process write bytes where the
+host exposes them, and traced sync calls rather than treating final file size
+as a proxy. Its checked invariant persists 157,452 samples across 2,197 mixed
+provider resources and caps the no-auto-checkpoint workload at 40,000 WAL
+frames; the former four-index schema produces 50,516 frames while the
+consolidated schema produces 35,030.
 Retention must also return freed SQLite pages to the OS
 proportionally to the current freelist, bounded per cycle, and on every
 retention cycle rather than only when that cycle deleted rows: a fixed small

pkg/metrics/store.go

@@ -183,7 +183,10 @@ type maintenanceRequest struct {
 	done chan struct{}
 }
 
-var startupMaintenanceHook func()
+var (
+	startupMaintenanceHook       func()
+	metricsIdentityMigrationHook func()
+)
 
 // Store provides persistent metrics storage
 type Store struct {
@@ -201,6 +204,7 @@ type Store struct {
 	doneCh                     chan struct{}
 	stopOnce                   sync.Once
 	stopping                   atomic.Bool
+	identityMigrationPending   atomic.Bool
 	commercialRetentionSeconds atomic.Int64
 	commercialPurgeEligibleAt  atomic.Int64
 }
@@ -337,10 +341,6 @@ func (s *Store) initSchema() error {
 			tier TEXT NOT NULL DEFAULT 'raw'
 		);
 
-		-- Index for efficient queries by resource and time
-		CREATE INDEX IF NOT EXISTS idx_metrics_lookup 
-		ON metrics(resource_type, resource_id, metric_type, tier, timestamp);
-
 		-- Index for retention pruning
 		CREATE INDEX IF NOT EXISTS idx_metrics_tier_time 
 		ON metrics(tier, timestamp);
@@ -361,7 +361,7 @@ func (s *Store) initSchema() error {
 		return fmt.Errorf("failed to create schema: %w", err)
 	}
 
-	if err := s.ensureMetricsUniqueIndex(); err != nil {
+	if err := s.ensureMetricsIdentityIndex(); err != nil {
 		return err
 	}
 
@@ -369,56 +369,207 @@ func (s *Store) initSchema() error {
 	return nil
 }
 
-func (s *Store) ensureMetricsUniqueIndex() error {
-	const createUniqueIndex = `
-		CREATE UNIQUE INDEX IF NOT EXISTS idx_metrics_unique
-		ON metrics(resource_type, resource_id, metric_type, timestamp, tier);
-	`
+var metricsIdentityColumns = []string{"resource_type", "resource_id", "metric_type", "tier", "timestamp"}
 
-	_, err := s.db.Exec(createUniqueIndex)
-	if err == nil {
+// ensureMetricsIdentityIndex keeps one B-tree for both metric identity and
+// single-series range queries. Older databases have two indexes containing the
+// same five columns in different orders: idx_metrics_lookup serves reads while
+// idx_metrics_unique enforces identity. Every insert dirties both trees, which
+// materially amplifies WAL and checkpoint writes.
+func (s *Store) ensureMetricsIdentityIndex() error {
+	lookupCurrent, err := s.metricsIndexMatches("idx_metrics_lookup", true, metricsIdentityColumns)
+	if err != nil {
+		return fmt.Errorf("inspect metrics identity index: %w", err)
+	}
+	legacyUniqueExists, err := s.metricsIndexExists("idx_metrics_unique")
+	if err != nil {
+		return fmt.Errorf("inspect legacy metrics unique index: %w", err)
+	}
+
+	if legacyUniqueExists {
+		// v6.1.1 and earlier already have an authoritative unique index. Keep
+		// serving with that crash-safe schema and defer the O(rows) rebuild to
+		// the startup-maintenance worker so NewStore latency stays bounded.
+		s.identityMigrationPending.Store(true)
+		log.Info().Msg("Scheduled metrics identity index consolidation")
+		return nil
+	}
+	if lookupCurrent {
 		return nil
 	}
 
-	// If the DB already contains duplicates (from older versions), creating the unique index
-	// will fail. Deduplicate and retry once.
-	lower := strings.ToLower(err.Error())
-	if !strings.Contains(lower, "unique") && !strings.Contains(lower, "constraint") && !strings.Contains(lower, "duplicate") {
-		return fmt.Errorf("failed to create unique index for metrics rollups: %w", err)
+	return s.migrateMetricsIdentityIndex()
+}
+
+func (s *Store) migrateMetricsIdentityIndex() error {
+	if err := s.replaceMetricsIdentityIndex(); err == nil {
+		log.Info().Msg("Consolidated metrics lookup and identity indexes")
+		return nil
+	} else if !isUniqueConstraintError(err) {
+		return fmt.Errorf("consolidate metrics identity index: %w", err)
 	}
 
-	log.Warn().Err(err).Msg("Duplicate metrics detected; deduplicating before creating unique index")
+	log.Warn().Msg("Duplicate metrics detected; deduplicating before consolidating identity index")
+	if err := s.deduplicateMetrics(); err != nil {
+		return err
+	}
+	if err := s.replaceMetricsIdentityIndex(); err != nil {
+		return fmt.Errorf("consolidate metrics identity index after dedupe: %w", err)
+	}
 
-	tx, txErr := s.db.Begin()
-	if txErr != nil {
-		return fmt.Errorf("begin dedupe transaction: %w", txErr)
+	log.Info().Msg("Metrics deduplicated and identity index consolidated")
+	return nil
+}
+
+func (s *Store) replaceMetricsIdentityIndex() error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return fmt.Errorf("begin identity index migration: %w", err)
 	}
 	defer func() { _ = tx.Rollback() }()
 
-	_, txErr = tx.Exec(`
-		DELETE FROM metrics
-		WHERE id NOT IN (
-			SELECT MIN(id)
-			FROM metrics
-			GROUP BY resource_type, resource_id, metric_type, timestamp, tier
-		)
-	`)
-	if txErr != nil {
-		return fmt.Errorf("dedupe metrics: %w", txErr)
+	if _, err := tx.Exec(`DROP INDEX IF EXISTS idx_metrics_lookup`); err != nil {
+		return fmt.Errorf("drop old lookup index: %w", err)
 	}
-
-	if txErr := tx.Commit(); txErr != nil {
-		return fmt.Errorf("commit dedupe: %w", txErr)
+	if metricsIdentityMigrationHook != nil {
+		metricsIdentityMigrationHook()
 	}
-
-	if _, err := s.db.Exec(createUniqueIndex); err != nil {
-		return fmt.Errorf("failed to create unique index after dedupe: %w", err)
+	if _, err := tx.Exec(`
+		CREATE UNIQUE INDEX idx_metrics_lookup
+		ON metrics(resource_type, resource_id, metric_type, tier, timestamp)
+	`); err != nil {
+		return fmt.Errorf("create consolidated lookup index: %w", err)
+	}
+	if _, err := tx.Exec(`DROP INDEX IF EXISTS idx_metrics_unique`); err != nil {
+		return fmt.Errorf("drop old unique index: %w", err)
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit identity index migration: %w", err)
 	}
-
-	log.Info().Msg("Metrics deduplicated and unique index created")
 	return nil
 }
 
+func (s *Store) deduplicateMetrics() error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return fmt.Errorf("begin metrics dedupe transaction: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	if _, err := tx.Exec(`
+		DELETE FROM metrics
+		WHERE rowid NOT IN (
+			SELECT MIN(rowid)
+			FROM metrics
+			GROUP BY resource_type, resource_id, metric_type, tier, timestamp
+		)
+	`); err != nil {
+		return fmt.Errorf("dedupe metrics: %w", err)
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit metrics dedupe: %w", err)
+	}
+	return nil
+}
+
+func isUniqueConstraintError(err error) bool {
+	if err == nil {
+		return false
+	}
+	lower := strings.ToLower(err.Error())
+	return strings.Contains(lower, "unique") ||
+		strings.Contains(lower, "constraint") ||
+		strings.Contains(lower, "duplicate")
+}
+
+func (s *Store) metricsIndexExists(name string) (bool, error) {
+	rows, err := s.db.Query(`PRAGMA index_list(metrics)`)
+	if err != nil {
+		return false, err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			sequence int
+			index    string
+			unique   int
+			origin   string
+			partial  int
+		)
+		if err := rows.Scan(&sequence, &index, &unique, &origin, &partial); err != nil {
+			return false, err
+		}
+		if index == name {
+			return true, nil
+		}
+	}
+	return false, rows.Err()
+}
+
+func (s *Store) metricsIndexMatches(name string, wantUnique bool, wantColumns []string) (bool, error) {
+	rows, err := s.db.Query(`PRAGMA index_list(metrics)`)
+	if err != nil {
+		return false, err
+	}
+
+	found := false
+	for rows.Next() {
+		var (
+			sequence int
+			index    string
+			unique   int
+			origin   string
+			partial  int
+		)
+		if err := rows.Scan(&sequence, &index, &unique, &origin, &partial); err != nil {
+			_ = rows.Close()
+			return false, err
+		}
+		if index == name {
+			found = (unique == 1) == wantUnique && partial == 0
+			break
+		}
+	}
+	if err := rows.Close(); err != nil {
+		return false, err
+	}
+	if !found {
+		return false, nil
+	}
+
+	columnRows, err := s.db.Query(`
+		SELECT name
+		FROM pragma_index_info(?)
+		ORDER BY seqno
+	`, name)
+	if err != nil {
+		return false, err
+	}
+	defer columnRows.Close()
+
+	columns := make([]string, 0, len(wantColumns))
+	for columnRows.Next() {
+		var columnName string
+		if err := columnRows.Scan(&columnName); err != nil {
+			return false, err
+		}
+		columns = append(columns, columnName)
+	}
+	if err := columnRows.Err(); err != nil {
+		return false, err
+	}
+	if len(columns) != len(wantColumns) {
+		return false, nil
+	}
+	for i := range columns {
+		if columns[i] != wantColumns[i] {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
 // migrateAutoVacuum ensures the database uses incremental auto-vacuum.
 // 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.
@@ -700,8 +851,15 @@ func (s *Store) runStartupMaintenance() {
 		startupMaintenanceHook()
 	}
 
-	// Run retention before any deferred vacuum work so restart cleanup trims
-	// stale rows before SQLite potentially rewrites the file.
+	if s.identityMigrationPending.Swap(false) {
+		if err := s.migrateMetricsIdentityIndex(); err != nil {
+			log.Error().Err(err).Msg("Deferred metrics identity index consolidation failed")
+		}
+	}
+
+	// Run retention before the deferred auto-vacuum conversion so restart
+	// cleanup trims stale rows and redundant-index pages before SQLite
+	// potentially rewrites the file.
 	s.runRetention()
 	s.migrateAutoVacuum()
 
@@ -834,7 +992,7 @@ func (s *Store) writeBatch(metrics []bufferedMetric) {
 	stmt, err := tx.Prepare(`
 		INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier)
 		VALUES (?, ?, ?, ?, ?, ?)
-		ON CONFLICT(resource_type, resource_id, metric_type, timestamp, tier)
+		ON CONFLICT(resource_type, resource_id, metric_type, tier, timestamp)
 		DO UPDATE SET
 			value = excluded.value,
 			min_value = excluded.min_value,

pkg/metrics/store_query_plan_test.go

@@ -82,8 +82,8 @@ func TestQueryPlansUseIndexes(t *testing.T) {
 				AND timestamp >= ? AND timestamp <= ?
 				ORDER BY metric_type, timestamp ASC`,
 			args: []any{"vm", "vm-1", "raw", int64(0), farFuture},
-			// The planner uses an indexed SEARCH — it may pick idx_metrics_query_all,
-			// idx_metrics_unique, or idx_metrics_lookup depending on statistics.
+			// The planner uses an indexed SEARCH — it may pick idx_metrics_query_all
+			// or idx_metrics_lookup depending on statistics.
 			// All are valid; the key invariant is that it does a SEARCH, not a SCAN.
 		},
 		{
@@ -110,7 +110,7 @@ func TestQueryPlansUseIndexes(t *testing.T) {
 				ORDER BY resource_id, metric_type, timestamp ASC`,
 			args: []any{"vm", "vm-1", "vm-2", "vm-3", "raw", int64(0), farFuture},
 			// QueryAllBatch is the anti-N+1 dashboard path. The planner may choose
-			// idx_metrics_query_all, idx_metrics_unique, or idx_metrics_lookup
+			// idx_metrics_query_all or idx_metrics_lookup
 			// depending on statistics; the invariant is an indexed SEARCH rather
 			// than a full table scan.
 		},
@@ -191,8 +191,8 @@ func TestQueryPlansUseIndexes(t *testing.T) {
 			args: []any{int64(60), int64(60), "minute", "raw", int64(0), farFuture},
 			// The SELECT filters on (tier, timestamp) without resource/metric
 			// columns. SQLite uses an index-ordered scan (SCAN USING INDEX)
-			// on idx_metrics_unique. A TEMP B-TREE may still be used for
-			// GROUP BY. The INSERT side uses idx_metrics_unique for ON CONFLICT.
+			// on idx_metrics_lookup. A TEMP B-TREE may still be used for
+			// GROUP BY. The INSERT side uses the same unique index for ON CONFLICT.
 			allowIndexScan: true,
 		},
 		{
@@ -324,7 +324,7 @@ func containsCoveringIndexScan(plan string) bool {
 }
 
 // containsIndexScan returns true if any plan line shows an index-ordered scan
-// on the metrics table (e.g., "SCAN metrics USING INDEX idx_metrics_unique").
+// on the metrics table (e.g., "SCAN metrics USING INDEX idx_metrics_lookup").
 // This differs from a covering-index scan in that it reads table rows via the
 // index, but still avoids a full table scan.
 func containsIndexScan(plan string) bool {
@@ -370,7 +370,7 @@ func newPlanTestDB(t *testing.T) *sql.DB {
 			tier TEXT NOT NULL DEFAULT 'raw'
 		);
 
-		CREATE INDEX IF NOT EXISTS idx_metrics_lookup
+		CREATE UNIQUE INDEX IF NOT EXISTS idx_metrics_lookup
 		ON metrics(resource_type, resource_id, metric_type, tier, timestamp);
 
 		CREATE INDEX IF NOT EXISTS idx_metrics_tier_time
@@ -379,9 +379,6 @@ func newPlanTestDB(t *testing.T) *sql.DB {
 		CREATE INDEX IF NOT EXISTS idx_metrics_query_all
 		ON metrics(resource_type, resource_id, tier, timestamp, metric_type);
 
-		CREATE UNIQUE INDEX IF NOT EXISTS idx_metrics_unique
-		ON metrics(resource_type, resource_id, metric_type, timestamp, tier);
-
 		CREATE TABLE IF NOT EXISTS metrics_meta (
 			key TEXT PRIMARY KEY,
 			value TEXT NOT NULL