Reduce metrics store transaction churn

Refs #1124

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

@@ -401,6 +401,16 @@ onto the shared backend `resourceType=k8s` transport, but it must preserve the
 canonical frontend target types for clusters, nodes, pods, and deployments so
 the workloads and drawer hot paths do not silently drop deployment history or
 split Kubernetes charts across incompatible cache keys.
+That same protected metrics-store boundary also owns write-path churn on local
+instances. `pkg/metrics/store.go` must prefer fewer, larger SQLite write
+transactions over tiny frequent commits: the default write buffer should stay
+large enough to absorb one poll cycle on modest multi-node installs, queued
+flush batches should coalesce before commit when the worker is already behind,
+and WAL auto-checkpointing should avoid tiny segment rewrite loops that keep
+rewriting `metrics.db` on SSD-backed systems. Any future change that reduces
+that batching headroom or makes WAL checkpoints more aggressive again must
+re-prove the metrics-store hot path with the owned store tests rather than
+assuming the earlier vacuum fixes are sufficient.
 contract instead of inventing an infrastructure-local summary filter branch.
 For shared line charts on that hot path, the shared sparkline primitive may
 isolate the selected series inside the existing render budget, but that

pkg/metrics/store.go

@@ -58,8 +58,11 @@ func DefaultConfig(dataDir string) StoreConfig {
 	}
 
 	return StoreConfig{
-		DBPath:          dbPath,
-		WriteBufferSize: 100,
+		DBPath: dbPath,
+		// Large installs can enqueue hundreds of metric points per poll cycle.
+		// A larger buffer keeps those writes inside a single SQLite transaction
+		// more often, which materially reduces WAL churn on SSD-backed setups.
+		WriteBufferSize: 500,
 		FlushInterval:   5 * time.Second,
 		RetentionRaw:    2 * time.Hour,
 		RetentionMinute: 24 * time.Hour,
@@ -154,7 +157,9 @@ func NewStore(config StoreConfig) (*Store, error) {
 			"journal_mode(WAL)",
 			"synchronous(NORMAL)",
 			"auto_vacuum(INCREMENTAL)",
-			"wal_autocheckpoint(500)",
+			// Checkpoint less aggressively so high-cardinality installs don't
+			// keep rewriting tiny WAL segments back into the main DB file.
+			"wal_autocheckpoint(4000)",
 		},
 	}.Encode()
 	rawDB, err := sql.Open("sqlite", dsn)
@@ -674,6 +679,32 @@ func (s *Store) writeBatch(metrics []bufferedMetric) {
 	log.Debug().Int("count", len(metrics)).Msg("Wrote metrics batch")
 }
 
+// coalesceQueuedBatches drains any already-queued write batches so the worker
+// can commit them in a single SQLite transaction. This reduces WAL write
+// amplification when the in-memory buffer flushes multiple times during one
+// poll cycle.
+func (s *Store) coalesceQueuedBatches(initial []bufferedMetric) []bufferedMetric {
+	if len(initial) == 0 {
+		return nil
+	}
+
+	combined := initial
+	for {
+		select {
+		case next, ok := <-s.writeCh:
+			if !ok {
+				return combined
+			}
+			if len(next) == 0 {
+				continue
+			}
+			combined = append(combined, next...)
+		default:
+			return combined
+		}
+	}
+}
+
 // Query retrieves metrics for a resource within a time range, with optional downsampling
 func (s *Store) Query(resourceType, resourceID, metricType string, start, end time.Time, stepSecs int64) ([]MetricPoint, error) {
 	resourceType = normalizeMetricResourceType(resourceType)
@@ -1279,13 +1310,20 @@ func (s *Store) backgroundWorker() {
 			s.Flush()
 			// Process remaining writes
 			close(s.writeCh)
+			var remaining []bufferedMetric
 			for batch := range s.writeCh {
-				s.writeBatch(batch)
+				if len(batch) == 0 {
+					continue
+				}
+				remaining = append(remaining, batch...)
+			}
+			if len(remaining) > 0 {
+				s.writeBatch(remaining)
 			}
 			return
 
 		case batch := <-s.writeCh:
-			s.writeBatch(batch)
+			s.writeBatch(s.coalesceQueuedBatches(batch))
 
 		case maintenance := <-s.maintenanceCh:
 			if maintenance.run != nil {

pkg/metrics/store_additional_test.go

@@ -12,6 +12,37 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+func TestDefaultConfigUsesLargerWriteBuffer(t *testing.T) {
+	cfg := DefaultConfig("/tmp/pulse")
+	if cfg.WriteBufferSize != 500 {
+		t.Fatalf("WriteBufferSize = %d, want 500", cfg.WriteBufferSize)
+	}
+}
+
+func TestStoreCoalesceQueuedBatches(t *testing.T) {
+	store := &Store{
+		writeCh: make(chan []bufferedMetric, 4),
+	}
+
+	initial := []bufferedMetric{
+		{resourceType: "vm", resourceID: "vm-1", metricType: "cpu", value: 10},
+	}
+	store.writeCh <- []bufferedMetric{
+		{resourceType: "vm", resourceID: "vm-1", metricType: "memory", value: 20},
+	}
+	store.writeCh <- []bufferedMetric{
+		{resourceType: "vm", resourceID: "vm-2", metricType: "cpu", value: 30},
+	}
+
+	combined := store.coalesceQueuedBatches(initial)
+	if len(combined) != 3 {
+		t.Fatalf("expected 3 combined metrics, got %d", len(combined))
+	}
+	if len(store.writeCh) != 0 {
+		t.Fatalf("expected queued batches to be drained, got %d remaining", len(store.writeCh))
+	}
+}
+
 func TestStoreWriteBatchSync(t *testing.T) {
 	dir := t.TempDir()
 	store, err := NewStore(DefaultConfig(dir))