fix(monitoring): batch unified metric sync writes into one transaction

The five unified syncs ran serially and each waited for its own SQLite commit, so a registry rebuild that carried new data for several resource types opened one transaction per type and re-committed the same WAL. Collect the agent, VM, storage and app-container writes through a batch sink and commit them once; the per-series replay guard still runs first, so the series, values and source observation times are unchanged. Standalone callers keep the immediate write. Focused proof in monitor_host_agents_test.go; contract sections recorded in monitoring.md and agent-lifecycle.md.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/agent-lifecycle.md

@@ -8094,6 +8094,10 @@ sampling does not alter agent admission either. Focused proof lives in
 (`TestDedupeUnifiedMetricWritesDropsExactReplays`,
 `TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`,
 `TestSyncUnifiedPhysicalDiskMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`).
+The batched sync path (`syncAllUnifiedMetrics`) collects the agent, VM, storage
+and app-container writes into one SQLite transaction without changing the
+series, values or observation times, so agent admission, token binding and
+removal-block behaviour are unchanged.
 
 ### Docker update preflight accepts any local RepoDigest for the planned image
 

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -4116,6 +4116,22 @@ and `internal/monitoring/monitor_polling_test.go`
 (`TestSyncUnifiedAppContainerMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`,
 `TestSyncUnifiedVMMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`).
 
+### Unified metric sync commits one batch per pass
+
+The five unified syncs run serially and each previously waited for its own
+SQLite commit, so one registry rebuild that carried new data for several
+resource types opened one transaction per type and committed the same WAL
+repeatedly. `syncAllUnifiedMetrics` now collects the surviving writes of the
+agent, VM, storage and app-container syncs through a batch sink and commits them
+in a single `WriteBatchBounded` transaction. Each sync still runs its own
+per-series replay guard first, so the change only moves the transaction
+boundary: the same series, values and source observation times are written.
+Standalone callers and tests still pass no sink and keep the immediate,
+read-your-writes write. The physical-disk SMART path keeps its own transaction
+because it writes directly. Focused proof lives in
+`internal/monitoring/monitor_polling_test.go`
+(`TestSyncUnifiedStorageMetricsDefersWritesToBatchSink`).
+
 ### Docker image update comparison accepts every local RepoDigest
 
 Docker records every RepoDigest an image is known by, and the digest for a

internal/monitoring/monitor.go

@@ -5097,11 +5097,7 @@ func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot) {
 		return
 	}
 	recordSupplementalResourceChanges(store, m.collectSupplementalChanges())
-	m.syncUnifiedAgentMetrics(store)
-	m.syncUnifiedVMMetrics(store)
-	m.syncUnifiedStorageMetrics(store)
-	m.syncUnifiedPhysicalDiskMetrics(store)
-	m.syncUnifiedAppContainerMetrics(store)
+	m.syncAllUnifiedMetrics(store)
 	m.syncUnifiedResourceAlertsToState(store.GetAll())
 }
 
@@ -5148,11 +5144,7 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) {
 	if atomicStore, ok := store.(AtomicSnapshotResourceStore); ok {
 		atomicStore.PopulateSnapshotAndSupplemental(snapshotForStore, recordsBySource)
 		recordSupplementalResourceChanges(store, supplementalChanges)
-		m.syncUnifiedAgentMetrics(store)
-		m.syncUnifiedVMMetrics(store)
-		m.syncUnifiedStorageMetrics(store)
-		m.syncUnifiedPhysicalDiskMetrics(store)
-		m.syncUnifiedAppContainerMetrics(store)
+		m.syncAllUnifiedMetrics(store)
 		for source, records := range recordsBySource {
 			if len(records) == 0 {
 				continue
@@ -5183,11 +5175,7 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) {
 	}
 
 	recordSupplementalResourceChanges(store, supplementalChanges)
-	m.syncUnifiedAgentMetrics(store)
-	m.syncUnifiedVMMetrics(store)
-	m.syncUnifiedStorageMetrics(store)
-	m.syncUnifiedPhysicalDiskMetrics(store)
-	m.syncUnifiedAppContainerMetrics(store)
+	m.syncAllUnifiedMetrics(store)
 	m.syncUnifiedResourceAlertsToState(store.GetAll())
 }
 
@@ -5282,7 +5270,41 @@ func (m *Monitor) dedupeUnifiedMetricWrites(writes []metrics.WriteMetric) []metr
 	return out
 }
 
-func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
+// syncAllUnifiedMetrics runs the per-resource-type unified syncs and commits
+// the surviving store writes of the batched syncs as one transaction. Each sync
+// still runs the per-series replay guard, so this only changes the transaction
+// boundary, not the series written, their values or their observation times.
+// The syncs run serially and each previously waited for its own commit, so a
+// rebuild that carried new data for several resource types opened one SQLite
+// transaction per type; batching collapses that to one (#1966). The physical
+// disk sync keeps its own transaction because its SMART path writes directly.
+func (m *Monitor) syncAllUnifiedMetrics(store ResourceStoreInterface) {
+	if m == nil {
+		return
+	}
+	if m.metricsStore == nil {
+		// The in-memory history path still needs to run; there is no store
+		// batch to collect.
+		m.syncUnifiedAgentMetrics(store)
+		m.syncUnifiedVMMetrics(store)
+		m.syncUnifiedStorageMetrics(store)
+		m.syncUnifiedPhysicalDiskMetrics(store)
+		m.syncUnifiedAppContainerMetrics(store)
+		return
+	}
+	var batch []metrics.WriteMetric
+	sink := &batch
+	m.syncUnifiedAgentMetrics(store, sink)
+	m.syncUnifiedVMMetrics(store, sink)
+	m.syncUnifiedStorageMetrics(store, sink)
+	m.syncUnifiedAppContainerMetrics(store, sink)
+	m.syncUnifiedPhysicalDiskMetrics(store)
+	if len(batch) > 0 {
+		m.metricsStore.WriteBatchBounded(batch)
+	}
+}
+
+func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) {
 	if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
 		return
 	}
@@ -5389,12 +5411,14 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
-	if len(storeWrites) > 0 {
+	if len(sinks) > 0 && sinks[0] != nil {
+		*sinks[0] = append(*sinks[0], storeWrites...)
+	} else if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
 }
 
-func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
+func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) {
 	if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
 		return
 	}
@@ -5509,12 +5533,14 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
-	if len(storeWrites) > 0 {
+	if len(sinks) > 0 && sinks[0] != nil {
+		*sinks[0] = append(*sinks[0], storeWrites...)
+	} else if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
 }
 
-func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
+func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) {
 	if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
 		return
 	}
@@ -5609,7 +5635,9 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
-	if len(storeWrites) > 0 {
+	if len(sinks) > 0 && sinks[0] != nil {
+		*sinks[0] = append(*sinks[0], storeWrites...)
+	} else if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
 }
@@ -5731,7 +5759,7 @@ func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) {
 	}
 }
 
-func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
+func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) {
 	if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
 		return
 	}
@@ -5843,7 +5871,9 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
-	if len(storeWrites) > 0 {
+	if len(sinks) > 0 && sinks[0] != nil {
+		*sinks[0] = append(*sinks[0], storeWrites...)
+	} else if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
 }

internal/monitoring/monitor_host_agents_test.go

@@ -5931,6 +5931,84 @@ func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) {
 	}
 }
 
+func TestSyncUnifiedStorageMetricsDefersWritesToBatchSink(t *testing.T) {
+	observedAt := time.Now().UTC().Add(-time.Minute).Truncate(time.Second)
+	resourceStore := unifiedresources.NewMonitorAdapter(nil)
+	resourceStore.PopulateFromSnapshot(models.StateSnapshot{
+		PBSInstances: []models.PBSInstance{{
+			ID:       "pbs-main",
+			Name:     "pbs-main",
+			Status:   "online",
+			LastSeen: observedAt,
+			Datastores: []models.PBSDatastore{{
+				Name:   "backups",
+				Status: "available",
+				Total:  1000,
+				Used:   400,
+				Free:   600,
+				Usage:  40,
+			}},
+		}},
+	})
+
+	var targetID string
+	for _, resource := range resourceStore.GetAll() {
+		if resource.Type != unifiedresources.ResourceTypeStorage || resource.Storage == nil || resource.Storage.Platform != "pbs" {
+			continue
+		}
+		target := resourceStore.MetricsTargetForResource(resource.ID)
+		if target == nil || target.ResourceType != "storage" {
+			t.Fatalf("unexpected PBS datastore metrics target: %+v", target)
+		}
+		targetID = target.ResourceID
+		break
+	}
+	if targetID == "" {
+		t.Fatal("expected a PBS datastore in the unified resource store")
+	}
+
+	cfg := metrics.DefaultConfig(t.TempDir())
+	persistentStore, err := metrics.NewStore(cfg)
+	if err != nil {
+		t.Fatalf("metrics.NewStore() error = %v", err)
+	}
+	defer func() { _ = persistentStore.Close() }()
+
+	monitor := &Monitor{
+		metricsHistory: NewMetricsHistory(1024, 24*time.Hour),
+		metricsStore:   persistentStore,
+	}
+
+	// A sink collects the batch so the unified syncs can share one transaction.
+	// The sink path must not write through to the store itself.
+	var sink []metrics.WriteMetric
+	monitor.syncUnifiedStorageMetrics(resourceStore, &sink)
+	if len(sink) == 0 {
+		t.Fatal("expected storage writes to be collected in the batch sink")
+	}
+
+	before, err := persistentStore.Query("storage", targetID, "usage", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0)
+	if err != nil {
+		t.Fatalf("query persisted PBS usage: %v", err)
+	}
+	if len(before) != 0 {
+		t.Fatalf("sink path must not write to the store directly, got %+v", before)
+	}
+
+	persistentStore.WriteBatchBounded(sink)
+
+	after, err := persistentStore.Query("storage", targetID, "usage", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0)
+	if err != nil {
+		t.Fatalf("query persisted PBS usage: %v", err)
+	}
+	if len(after) != 1 {
+		t.Fatalf("persisted PBS usage points = %d, want one after the batch flush: %+v", len(after), after)
+	}
+	if !after[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("persisted PBS usage timestamp = %s, want source observation %s", after[0].Timestamp, observedAt)
+	}
+}
+
 func TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) {
 	previous := truenas.IsFeatureEnabled()
 	truenas.SetFeatureEnabled(true)