Bound synchronous metrics writes so a slow disk cannot stall monitoring

WriteBatchSync waited unboundedly for the ingestion worker's commit. The
monitoring pipeline calls it inline from state broadcast, agent ingest,
and poll publish, so a metrics disk slow enough to back up the write
queue froze the monitor after its first cycle: polls stopped being
scheduled, PollStatus.LastSuccess never advanced, and healthy API
sources degraded to stale/agent-only while SQLite ground through
retention maintenance (107s cleanup, multi-second commits on the
reporter's instance).

enqueueAndWait now shares a single 2s budget across enqueue and commit.
A queue that cannot accept the batch within the budget drops it with a
warning, matching enqueueWrite's saturation behavior. A batch that
enqueues but has not committed stays queued and is not lost; the caller
moves on and a rate-limited warning records the backlog. Healthy disks
keep read-your-writes semantics.

Refs #1437

Contract-Neutral: behavioral fix: bound metrics store sync write wait (#1437), no public contract delta

pkg/metrics/store.go

@@ -205,6 +205,7 @@ type Store struct {
 	maintenanceDoneCh          chan struct{}
 	stopOnce                   sync.Once
 	stopping                   atomic.Bool
+	syncWaitWarnNano           atomic.Int64
 	identityMigrationPending   atomic.Bool
 	commercialRetentionSeconds atomic.Int64
 	commercialPurgeEligibleAt  atomic.Int64
@@ -916,13 +917,67 @@ func (s *Store) enqueueWrite(req writeRequest) {
 	}
 }
 
+// syncWriteWaitTimeout bounds how long a synchronous batch write blocks on the
+// ingestion worker. The monitoring pipeline calls WriteBatchSync inline (state
+// broadcast, agent ingest, poll publish), so an unbounded wait lets a slow
+// metrics disk starve polling entirely: on #1437's instance SQLite commits ran
+// for seconds to minutes and the monitor froze after its first cycle while
+// history writes queued behind retention maintenance. Within the budget the
+// call keeps read-your-writes; past it the caller moves on and history lands
+// whenever the worker catches up.
+const syncWriteWaitTimeout = 2 * time.Second
+
+// enqueueAndWait hands the batch to the ingestion worker and waits for its
+// commit, but never longer than syncWriteWaitTimeout in total. If the queue
+// cannot even accept the batch within the budget the batch is dropped, matching
+// enqueueWrite's saturation behavior. If only the commit is outstanding the
+// batch stays queued and is not lost.
 func (s *Store) enqueueAndWait(req writeRequest) {
 	if req.done == nil {
 		req.done = make(chan struct{})
 	}
 
-	s.writeCh <- req
-	<-req.done
+	timer := time.NewTimer(syncWriteWaitTimeout)
+	defer timer.Stop()
+
+	select {
+	case s.writeCh <- req:
+	case <-s.stopCh:
+		return
+	case <-timer.C:
+		log.Warn().
+			Str("component", "metrics_store").
+			Str("action", "drop_sync_write_batch").
+			Int("batch_size", len(req.metrics)).
+			Int("write_queue_depth", len(s.writeCh)).
+			Int("write_queue_capacity", cap(s.writeCh)).
+			Msg("Metrics write queue saturated, dropping synchronous batch to keep monitoring live")
+		return
+	}
+
+	select {
+	case <-req.done:
+	case <-s.stopCh:
+	case <-timer.C:
+		s.warnSyncWriteBacklog(len(req.metrics))
+	}
+}
+
+// warnSyncWriteBacklog reports a lagging ingestion worker at most once per
+// 30-second window. Unlike a dropped batch this is not data loss, so the
+// per-call signal is redundant while the condition persists.
+func (s *Store) warnSyncWriteBacklog(batchSize int) {
+	now := time.Now().UnixNano()
+	last := s.syncWaitWarnNano.Load()
+	if now-last < int64(30*time.Second) || !s.syncWaitWarnNano.CompareAndSwap(last, now) {
+		return
+	}
+	log.Warn().
+		Str("component", "metrics_store").
+		Str("action", "sync_write_backlogged").
+		Int("batch_size", batchSize).
+		Int("write_queue_depth", len(s.writeCh)).
+		Msg("Metrics write worker lagging, batch left queued and monitoring continues")
 }
 
 func (s *Store) drainBuffer() []bufferedMetric {

pkg/metrics/store_write_backpressure_test.go

@@ -0,0 +1,86 @@
+package metrics
+
+import (
+	"testing"
+	"time"
+)
+
+// Regression coverage for #1437: WriteBatchSync sits on the monitoring
+// pipeline (state broadcast, agent ingest, poll publish). When the ingestion
+// worker cannot keep up with the disk, the call must return within its wait
+// budget instead of stalling the monitor until polling stops.
+
+// newUnservicedStore builds a bare store whose ingestion worker never runs,
+// modelling a writer wedged behind slow SQLite maintenance.
+func newUnservicedStore(queueCap int) *Store {
+	return &Store{
+		writeCh: make(chan writeRequest, queueCap),
+		stopCh:  make(chan struct{}),
+	}
+}
+
+func backpressureProbeMetric() WriteMetric {
+	return WriteMetric{
+		ResourceType: "vm",
+		ResourceID:   "backpressure-probe",
+		MetricType:   "cpu",
+		Value:        1,
+		Timestamp:    time.Unix(1_700_000_000, 0),
+		Tier:         TierRaw,
+	}
+}
+
+func requireReturnsWithinWaitBudget(t *testing.T, s *Store, label string) {
+	t.Helper()
+
+	done := make(chan struct{})
+	go func() {
+		s.WriteBatchSync([]WriteMetric{backpressureProbeMetric()})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(syncWriteWaitTimeout + 3*time.Second):
+		t.Fatalf("WriteBatchSync stalled past its wait budget (%s)", label)
+	}
+}
+
+func TestWriteBatchSyncReturnsWhenQueueSaturated(t *testing.T) {
+	s := newUnservicedStore(1)
+	s.writeCh <- writeRequest{metrics: []bufferedMetric{{}}}
+
+	requireReturnsWithinWaitBudget(t, s, "saturated queue")
+
+	if got := len(s.writeCh); got != 1 {
+		t.Fatalf("saturated queue depth = %d, want the pre-existing batch only", got)
+	}
+}
+
+func TestWriteBatchSyncReturnsWhenCommitLags(t *testing.T) {
+	s := newUnservicedStore(4)
+
+	requireReturnsWithinWaitBudget(t, s, "lagging commit")
+
+	if got := len(s.writeCh); got != 1 {
+		t.Fatalf("queued batches = %d, want 1 (batch must stay queued, not be dropped)", got)
+	}
+}
+
+func TestWriteBatchSyncReturnsOnClosedStore(t *testing.T) {
+	s := newUnservicedStore(1)
+	s.writeCh <- writeRequest{metrics: []bufferedMetric{{}}}
+	close(s.stopCh)
+
+	done := make(chan struct{})
+	go func() {
+		s.WriteBatchSync([]WriteMetric{backpressureProbeMetric()})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("WriteBatchSync did not observe store shutdown")
+	}
+}