From df442387972c07d6c36aef4cdcc59a11aee61796 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:15:31 +0100 Subject: [PATCH] fix(monitoring): drop replayed unified metric samples Read-side registry rebuilds re-run the unified metric sync as often as every two seconds, so one poll observation was re-written many times between polls. The metrics store upserts on (resource, metric, tier, timestamp), so each replay UPDATEs the same row and commits a fresh transaction for no new data, churning the SQLite WAL and driving the disk-write amplification reported in #1966. Track the last sample handed to the store per series and skip exact timestamp+value repeats before the batch is enqueued. A corrected value at the same observation time is still written, so the guard cannot mask a real change. Record the behaviour in the monitoring and agent-lifecycle contracts and move the focused regression into the accepted monitor.go proof file. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 9 +++ .../v6/internal/subsystems/monitoring.md | 15 ++++ internal/monitoring/monitor.go | 69 +++++++++++++++++++ .../monitoring/monitor_host_agents_test.go | 45 ++++++++++++ 4 files changed, 138 insertions(+) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index d70c06430f..b0acd1ed24 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8071,3 +8071,12 @@ This does not change token scopes, agent admission, or existing agent cleanup. Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and `TestInitializeSystemSettingsPreservesExistingBytes`, `TestInitializeSystemSettingsMissing`, `TestInitializeSystemSettingsReadError`. + +### Unified metric replay guard does not change agent admission + +The unified metric sync runs on host-agent ingest boundaries and on read-side +registry rebuilds. A per-series replay guard now drops exact timestamp+value +repeats before the metrics batch is enqueued, reducing WAL churn without +changing host admission, token binding or removal-block behaviour. Focused proof +lives in `internal/monitoring/monitor_host_agents_test.go` +(`TestDedupeUnifiedMetricWritesDropsExactReplays`). diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index e3897ec5e2..9790120617 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4065,3 +4065,18 @@ provider keys, canonical IDs, JSON projection, temperature/size/cadence, separat nodes/controller members, and confirmed inventory removal. `TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata. This is synthetic runtime evidence, not USB hardware or reporter acceptance. + +### Unified metric replay guard bounds registry-rebuild writes + +Read-side registry rebuilds (/api/state, websocket hydrate) re-run the unified +metric sync as often as every two seconds, and storage observations are anchored +to the provider's LastSeen. Before this change each rebuild re-issued the same +rows as SQLite UPDATEs and committed a fresh transaction for no new data, +churning the WAL and driving the disk-write amplification in #1966. Monitoring +now tracks the last sample handed to the store per (resource type, resource id, +metric type) series and drops exact timestamp+value repeats before the batch is +enqueued; a corrected value at the same observation time is still written, so a +real change is never masked. The guard is bounded so a long-lived monitor with +churning resource IDs cannot pin memory. Focused proof lives in +`internal/monitoring/monitor_host_agents_test.go` +(`TestDedupeUnifiedMetricWritesDropsExactReplays`). diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 49d0cf35bb..4be2bf5093 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1247,6 +1247,14 @@ type Monitor struct { // that already hold m.mu can consult it without lock re-entrancy. licenseCheckerMu sync.RWMutex licenseChecker func(feature string) bool + + // Unified metric sync replay guard. The sync runs on both ingest + // boundaries and read-side registry rebuilds, so one observation is often + // replayed many times between polls. Without this every replay re-writes + // the same metrics row (an UPDATE in SQLite) and churns the WAL for no new + // data, which is the dominant write amplification in #1966. + unifiedMetricSyncMu sync.Mutex + unifiedMetricSyncLast map[unifiedMetricSampleKey]unifiedMetricSample } func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) { @@ -5171,6 +5179,63 @@ func recordSupplementalResourceChanges(store ResourceStoreInterface, changes []u } } +// unifiedMetricSampleKey identifies one persisted unified-metric series. The +// metrics store keys rows by (resource type, resource id, metric type, tier, +// timestamp), so the replay guard must match at least those coordinates. +type unifiedMetricSampleKey struct { + resourceType string + resourceID string + metricType string +} + +// unifiedMetricSample is the last sample handed to the metrics store for a +// series. Timestamps are compared at the store's second resolution so a +// sub-second replay of the same observation is still recognised as redundant. +type unifiedMetricSample struct { + timestampUnix int64 + value float64 +} + +// unifiedMetricSyncMaxSeries bounds the replay guard so long-lived monitors +// with churning resource IDs cannot pin memory. The map is reset wholesale +// when the cap is crossed; the worst case is one extra write per series. +const unifiedMetricSyncMaxSeries = 100000 + +// dedupeUnifiedMetricWrites drops exact repeats of a sample already handed to +// the metrics store. The unified sync runs on ingest boundaries and on +// read-side registry rebuilds (which fire as often as every two seconds), so +// the same observation is replayed many times between polls. Each replay would +// otherwise UPDATE the same row and commit a fresh transaction, churning the +// SQLite WAL with no new data (#1966). A changed value at the same timestamp is +// still written so corrections are not lost. +func (m *Monitor) dedupeUnifiedMetricWrites(writes []metrics.WriteMetric) []metrics.WriteMetric { + if len(writes) == 0 { + return writes + } + + m.unifiedMetricSyncMu.Lock() + defer m.unifiedMetricSyncMu.Unlock() + if m.unifiedMetricSyncLast == nil || len(m.unifiedMetricSyncLast) > unifiedMetricSyncMaxSeries { + m.unifiedMetricSyncLast = make(map[unifiedMetricSampleKey]unifiedMetricSample, len(writes)) + } + + out := writes[:0] + for _, write := range writes { + key := unifiedMetricSampleKey{ + resourceType: write.ResourceType, + resourceID: write.ResourceID, + metricType: write.MetricType, + } + sample := unifiedMetricSample{timestampUnix: write.Timestamp.Unix(), value: write.Value} + if previous, ok := m.unifiedMetricSyncLast[key]; ok && previous == sample { + continue + } + m.unifiedMetricSyncLast[key] = sample + out = append(out, write) + } + return out +} + func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) { return @@ -5276,6 +5341,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { appendStoreWrite("agent", targetID, "diskwrite", metric.Value) } } + storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } @@ -5391,6 +5457,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { appendStoreWrite("vm", targetID, "diskwrite", metric.Value) } } + storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } @@ -5490,6 +5557,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) { appendStoreWrite("storage", targetID, "avail", float64(free), observedAt) } } + storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } @@ -5697,6 +5765,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value) } } + storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index f8d32ea0e8..97211ac167 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -5884,3 +5884,48 @@ func TestAgentReportsPreserveExplicitDiskIncludes(t *testing.T) { check(t, host.Disks) }) } + +// Issue #1966: read-side registry rebuilds re-run the unified metric sync as +// often as every two seconds, so one poll observation was re-written many +// times between polls. The store upserts on (resource, metric, tier, +// timestamp), so each replay committed a fresh transaction for no new data and +// churned the SQLite WAL. The replay guard drops exact timestamp+value repeats +// before the batch is enqueued while still writing a corrected value. +func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) { + monitor := &Monitor{} + observedAt := time.Date(2026, time.September, 19, 12, 0, 0, 0, time.UTC) + write := metrics.WriteMetric{ + ResourceType: "storage", + ResourceID: "pool:tank", + MetricType: "usage", + Value: 62, + Timestamp: observedAt, + Tier: metrics.TierRaw, + } + + first := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write}) + if len(first) != 1 { + t.Fatalf("first write count = %d, want 1", len(first)) + } + // The same observation replayed by a read-side registry rebuild must not + // reach the store again: that UPDATE is the write amplification in #1966. + replay := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write}) + if len(replay) != 0 { + t.Fatalf("replayed write count = %d, want 0", len(replay)) + } + + // A corrected value at the same observation time is still written. + corrected := write + corrected.Value = 63 + if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{corrected}); len(got) != 1 { + t.Fatalf("corrected write count = %d, want 1", len(got)) + } + + // A newer observation for the same series is written. + advanced := write + advanced.Value = 63 + advanced.Timestamp = observedAt.Add(time.Minute) + if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{advanced}); len(got) != 1 { + t.Fatalf("advanced write count = %d, want 1", len(got)) + } +}