fix(monitoring): anchor and guard unified physical-disk SMART writes

syncUnifiedPhysicalDiskMetrics stamped every SMART sample with the registry rebuild wall clock and wrote through the unguarded buffered path, so each read-side rebuild re-issued the same physical-disk observation as fresh rows (#1966). writeSMARTMetrics now anchors both the in-memory history and the persisted batch to the source observation time and drops exact replays through the existing unified guard; the unified sync passes unifiedResourceObservedAt. Adds a two-rebuild focused proof and updates the TrueNAS disk fixture to a current observation time.

Change-source: pulse-maintainer

internal/monitoring/canonical_guardrails_test.go

@@ -2197,7 +2197,7 @@ func TestUnifiedPhysicalDiskMetricsUseCanonicalDiskHistoryPath(t *testing.T) {
 		"m.syncUnifiedPhysicalDiskMetrics(store)",
 		`if target == nil || target.ResourceType != "disk" || strings.TrimSpace(target.ResourceID) == "" {`,
 		`if source == unifiedresources.SourceProxmox || source == unifiedresources.SourceAgent {`,
-		`m.writeSMARTMetrics(disk, now)`,
+		`m.writeSMARTMetrics(disk, unifiedResourceObservedAt(resource, now))`,
 	}
 	for _, snippet := range requiredSnippets {
 		if !strings.Contains(source, snippet) {

internal/monitoring/monitor.go

@@ -891,6 +891,11 @@ func hostsFromReadState(readState unifiedresources.ReadState) []models.Host {
 
 // writeSMARTMetrics writes SMART temperature history to the in-memory chart
 // buffer and persists SMART attributes when the metrics store is enabled.
+//
+// now is the source observation time, not the rebuild wall clock: a read-side
+// registry rebuild that re-issues one unchanged observation must map to a
+// single sample. The store writes go through the unified replay guard so the
+// same observation is not re-committed on every rebuild (#1966).
 func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) {
 	if shouldSkipNativeMockStateMetricWrites() {
 		return
@@ -901,56 +906,88 @@ func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) {
 		return
 	}
 
+	// Temperature (always write if > 0)
+	if disk.Temperature > 0 && m.metricsHistory != nil {
+		m.metricsHistory.AddDiskMetric(resourceID, "smart_temp", float64(disk.Temperature), now)
+	}
+
+	writes := m.smartMetricStoreWrites(disk, resourceID, now)
+	if len(writes) == 0 {
+		return
+	}
+	writes = m.dedupeUnifiedMetricWrites(writes)
+	if len(writes) > 0 {
+		m.metricsStore.WriteBatchBounded(writes)
+	}
+}
+
+// smartMetricStoreWrites builds the persisted SMART writes for one physical
+// disk observation. The caller owns batching so the unified sync can anchor the
+// observation time and drop exact replays (#1966).
+func (m *Monitor) smartMetricStoreWrites(disk models.PhysicalDisk, resourceID string, observedAt time.Time) []metrics.WriteMetric {
+	if m.metricsStore == nil {
+		return nil
+	}
+
+	writes := make([]metrics.WriteMetric, 0, 12)
+	appendWrite := func(metricType string, value float64) {
+		writes = append(writes, metrics.WriteMetric{
+			ResourceType: "disk",
+			ResourceID:   resourceID,
+			MetricType:   metricType,
+			Value:        value,
+			Timestamp:    observedAt,
+			Tier:         metrics.TierRaw,
+		})
+	}
+
 	// Temperature (always write if > 0)
 	if disk.Temperature > 0 {
-		if m.metricsHistory != nil {
-			m.metricsHistory.AddDiskMetric(resourceID, "smart_temp", float64(disk.Temperature), now)
-		}
-		if m.metricsStore != nil {
-			m.metricsStore.Write("disk", resourceID, "smart_temp", float64(disk.Temperature), now)
-		}
+		appendWrite("smart_temp", float64(disk.Temperature))
 	}
 
 	attrs := disk.SmartAttributes
-	if attrs == nil || m.metricsStore == nil {
-		return
+	if attrs == nil {
+		return writes
 	}
 
 	// Common
 	if attrs.PowerOnHours != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_power_on_hours", float64(*attrs.PowerOnHours), now)
+		appendWrite("smart_power_on_hours", float64(*attrs.PowerOnHours))
 	}
 	if attrs.PowerCycles != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_power_cycles", float64(*attrs.PowerCycles), now)
+		appendWrite("smart_power_cycles", float64(*attrs.PowerCycles))
 	}
 
 	// SATA-specific
 	if attrs.ReallocatedSectors != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_reallocated_sectors", float64(*attrs.ReallocatedSectors), now)
+		appendWrite("smart_reallocated_sectors", float64(*attrs.ReallocatedSectors))
 	}
 	if attrs.PendingSectors != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_pending_sectors", float64(*attrs.PendingSectors), now)
+		appendWrite("smart_pending_sectors", float64(*attrs.PendingSectors))
 	}
 	if attrs.OfflineUncorrectable != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_offline_uncorrectable", float64(*attrs.OfflineUncorrectable), now)
+		appendWrite("smart_offline_uncorrectable", float64(*attrs.OfflineUncorrectable))
 	}
 	if attrs.UDMACRCErrors != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_crc_errors", float64(*attrs.UDMACRCErrors), now)
+		appendWrite("smart_crc_errors", float64(*attrs.UDMACRCErrors))
 	}
 
 	// NVMe-specific
 	if attrs.PercentageUsed != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_percentage_used", float64(*attrs.PercentageUsed), now)
+		appendWrite("smart_percentage_used", float64(*attrs.PercentageUsed))
 	}
 	if attrs.AvailableSpare != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_available_spare", float64(*attrs.AvailableSpare), now)
+		appendWrite("smart_available_spare", float64(*attrs.AvailableSpare))
 	}
 	if attrs.MediaErrors != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_media_errors", float64(*attrs.MediaErrors), now)
+		appendWrite("smart_media_errors", float64(*attrs.MediaErrors))
 	}
 	if attrs.UnsafeShutdowns != nil {
-		m.metricsStore.Write("disk", resourceID, "smart_unsafe_shutdowns", float64(*attrs.UnsafeShutdowns), now)
+		appendWrite("smart_unsafe_shutdowns", float64(*attrs.UnsafeShutdowns))
 	}
+
+	return writes
 }
 
 // PollExecutor defines the contract for executing polling tasks.
@@ -5687,7 +5724,10 @@ func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) {
 		if disk.Serial == "" {
 			disk.ID = targetID
 		}
-		m.writeSMARTMetrics(disk, now)
+		// Anchor to the source observation time so a read-side registry rebuild
+		// re-issuing the same snapshot maps to one sample instead of a fresh
+		// wall-clock row (#1966).
+		m.writeSMARTMetrics(disk, unifiedResourceObservedAt(resource, now))
 	}
 }
 

internal/monitoring/monitor_polling_test.go

@@ -1138,7 +1138,10 @@ func TestSyncUnifiedPhysicalDiskMetricsRecordsTrueNASDiskHistory(t *testing.T) {
 	defer func() { _ = store.Close() }()
 
 	resourceStore := unifiedresources.NewMonitorAdapter(nil)
-	records := truenas.NewProvider(truenas.DefaultFixtures()).Records()
+	fixtures := truenas.DefaultFixtures()
+	fixtures.CollectedAt = time.Now().UTC().Truncate(time.Second)
+	fixtures.System.CollectedAt = fixtures.CollectedAt
+	records := truenas.NewProvider(fixtures).Records()
 	resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{
 		unifiedresources.SourceTrueNAS: records,
 	})
@@ -1186,6 +1189,79 @@ func TestSyncUnifiedPhysicalDiskMetricsRecordsTrueNASDiskHistory(t *testing.T) {
 	}
 }
 
+func TestSyncUnifiedPhysicalDiskMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) {
+	previous := truenas.IsFeatureEnabled()
+	truenas.SetFeatureEnabled(true)
+	t.Cleanup(func() {
+		truenas.SetFeatureEnabled(previous)
+	})
+
+	fixtures := truenas.DefaultFixtures()
+	fixtures.CollectedAt = time.Now().UTC().Truncate(time.Second)
+	fixtures.System.CollectedAt = fixtures.CollectedAt
+	observedAt := fixtures.CollectedAt
+
+	resourceStore := unifiedresources.NewMonitorAdapter(nil)
+	resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{
+		unifiedresources.SourceTrueNAS: truenas.NewProvider(fixtures).Records(),
+	})
+
+	var diskResourceID string
+	for _, resource := range resourceStore.GetAll() {
+		if resource.Type != unifiedresources.ResourceTypePhysicalDisk || resource.PhysicalDisk == nil {
+			continue
+		}
+		if strings.TrimSpace(resource.PhysicalDisk.DevPath) == "/dev/sdc" {
+			diskResourceID = resource.ID
+			break
+		}
+	}
+	if diskResourceID == "" {
+		t.Fatal("expected TrueNAS sdc disk resource in unified store")
+	}
+
+	target := resourceStore.MetricsTargetForResource(diskResourceID)
+	if target == nil || target.ResourceType != "disk" || target.ResourceID != "WD-WX12A3456" {
+		t.Fatalf("unexpected physical-disk metrics target %+v", target)
+	}
+
+	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,
+	}
+
+	// Read-side registry rebuilds must not invent a fresh sample for the same
+	// source observation (the write amplification in #1966).
+	monitor.syncUnifiedPhysicalDiskMetrics(resourceStore)
+	monitor.syncUnifiedPhysicalDiskMetrics(resourceStore)
+
+	inMemory := monitor.GetDiskMetrics(target.ResourceID, "smart_temp", time.Hour)
+	if len(inMemory) != 1 {
+		t.Fatalf("in-memory disk smart_temp points = %d, want one per source observation: %+v", len(inMemory), inMemory)
+	}
+	if !inMemory[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("in-memory disk smart_temp timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt)
+	}
+
+	persisted, err := persistentStore.Query("disk", target.ResourceID, "smart_temp", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0)
+	if err != nil {
+		t.Fatalf("query persisted disk smart_temp: %v", err)
+	}
+	if len(persisted) != 1 {
+		t.Fatalf("persisted disk smart_temp points = %d, want one per source observation: %+v", len(persisted), persisted)
+	}
+	if !persisted[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("persisted disk smart_temp timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt)
+	}
+}
+
 func TestSyncUnifiedStorageAndDiskMetricsSkipMockOwnedTrueNASHistoryWhenMockEnabled(t *testing.T) {
 	previousFeature := truenas.IsFeatureEnabled()
 	truenas.SetFeatureEnabled(true)