diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 9d3644cb63..094f0761b5 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -5339,6 +5339,48 @@ serial or WWN only inside a compatible physical-disk parent scope; when those identifiers are absent or placeholders, host/device/topology fallback identity is authoritative. An exact seeded source mapping may be reused after restart only when its resource type and parent scope still agree. +A usable serial or WWN identifies a disk on one machine only, and the canonical +ID a new disk takes must honour that, not just matching. Dual-ported SAS +shelves, cloned VMs with an explicit serial, fixed-serial USB bridges and +TrueNAS systems report one identifier on several machines, and merging them on +a hardware-keyed ID collision dropped every machine but the last from +inventory. `physicalDiskIDForMachineLocked` resolves each disk's parent to its +machine (the nearest host ancestor, so a TrueNAS pool or Unraid array counts as +its system or host) and treats two disks carrying one identity as different +machines only on evidence: one reporter placing the identity on two machines +(for a single-source disk rehydrated without per-source parents, its own +parent), or differing machines whose hostnames share nothing. Reporters on one +machine still join across grouping parents and unmerged host resources (the +TrueNAS API and the agent on that box), and an unknown machine never splits a +disk. A split (operator exclusion) recorded against any ID the disk can hold, +unscoped, current or machine-scoped, keeps the observation on its +source-specific ID, keyed to its machine when another machine's disk already +holds that ID (agent disk source IDs are the bare serial). Once the identity +spans machines the unscoped ID no longer names one disk, so a split recorded +against it applies to every copy, erring towards an extra row rather than a +merge the operator forbade, and a manual link recorded against it stops +applying. More than one joinable disk joins none, except that the holder of the +unscoped ID keeps it. Once an identity spans machines, every copy not split off +onto its source-specific ID takes the identity keyed to the machine of its +canonical parent at that point, and the holder of the unscoped ID is re-keyed +with its source mappings, retired-ID claims and relationship endpoints. +Machines may be ingested in any order, and the unscoped ID never passes between +disks that coexist on different machines. A disk only one machine reports keeps +its unscoped ID, apart from observations an operator split off. Residuals: like +any serial-keyed ID, the unscoped ID still passes to a same-serial disk that +appears once the previous holder is gone; a disk's ID changes when a +same-serial disk first appears on another machine and back when it goes away, +and alert-history rows owned under the unscoped ID do not follow a re-key; a +disk two reporters share is scoped to its canonical parent when the identity +first spans machines, which the fixed ingest order (snapshot sources, then +supplemental records) and source priority decide; agent SMART and Unraid disk +source IDs are the bare serial, so same-serial agent disks on different hosts +share one `SourceAgent` mapping and only the last ingested keeps the agent +source target; same-serial disks share one metrics history key; and across +reporters, hostnames that normalize alike (default TrueNAS names, short names +across domains) still join. `IdentityMatcher` keeps one resource per machine +ID, so the registry indexes disks by hardware key (`physicalDisksByHardware`) +to see every copy. That same canonical physical-disk view must also expose source-independent host context. When a disk is API-backed rather than node-backed, typed views should fall back to canonical host identity such as `identity.hostnames` instead of diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 7d59ac78b4..7b452be5b7 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -104,6 +104,11 @@ type ResourceRegistry struct { // nil during a batch means reference reads use the live scan until the // final identity refresh rebuilds the index. Empty values are ambiguous. canonicalIdentityIndex map[string]string + // physicalDisksByHardware lists every disk resource ever indexed under a + // hardware key (physicalDiskHardwareKey). The IdentityMatcher keeps one + // resource per machine ID, so it cannot see same-serial disks on several + // machines. Entries are validated on read, never removed eagerly. + physicalDisksByHardware map[string]map[string]struct{} // Cached typed view indexes. Invalidated on ingest, rebuilt lazily on // first access. Protected by mu — callers hold RLock to read, and the @@ -835,6 +840,7 @@ func (rr *ResourceRegistry) ingestResources(resources []Resource, thresholds map rr.seedSourceMappingsFromResourceLocked(rr.resources[resourceID]) if resource := rr.resources[resourceID]; resource != nil { rr.indexSupersededCanonicalIDsLocked(resourceID, resource.SupersededCanonicalIDs) + rr.indexPhysicalDiskHardwareLocked(resource) } } rr.applyManualLinks(thresholds) @@ -2929,6 +2935,7 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res } stampPhysicalDiskTemperatureReading(&resource, source, sourceID) rr.resources[resource.ID] = &resource + rr.indexPhysicalDiskHardwareLocked(&resource) rr.bySource[source][sourceID] = resource.ID rr.matcher.Add(resource.ID, identity) return resource.ID @@ -2936,6 +2943,9 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res } resource.ID = rr.chooseNewID(resource.Type, identity, source, sourceID) + if resource.Type == ResourceTypePhysicalDisk { + resource.ID = rr.physicalDiskIDForMachineLocked(resource.ID, candidateID, &resource, source, onlyMissing) + } normalizeResourceRelationships(&resource) if existing := rr.resources[resource.ID]; existing != nil { if onlyMissing { @@ -2948,6 +2958,7 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res } stampPhysicalDiskTemperatureReading(&resource, source, sourceID) rr.resources[resource.ID] = &resource + rr.indexPhysicalDiskHardwareLocked(&resource) rr.bySource[source][sourceID] = resource.ID rr.matcher.Add(resource.ID, identity) @@ -3842,6 +3853,7 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou existing.Name = incoming.Name } } + rr.indexPhysicalDiskHardwareLocked(existing) } func clearUnavailableSourceMemoryMetric(metrics *ResourceMetrics, incoming *Resource, source DataSource) *ResourceMetrics { @@ -5266,6 +5278,267 @@ func (rr *ResourceRegistry) chooseNewID(resourceType ResourceType, identity Reso return rr.sourceSpecificID(resourceType, source, sourceID) } +// physicalDiskIDForMachineLocked decides which resource a new disk +// observation joins once findMatch has refused every candidate. A usable +// serial or WWN identifies a disk only on one machine: dual-ported SAS shelves, +// cloned VMs with an explicit serial, fixed-serial USB bridges and TrueNAS +// systems report one identifier on several machines. id is chooseNewID's +// hardware-keyed ID and candidateID the observation's source-specific ID. +// +// With no other disk carrying the identity, the disk keeps id, so a disk seen +// on one machine never changes ID. Otherwise it joins the one disk it does not +// conflict with (a second reporter on the same machine), unless an operator +// exclusion separates them, which leaves it on candidateID as findMatch's +// excluded branch does. An exclusion recorded against any ID the disk can hold +// (id, its current ID or its machine-scoped ID) applies, so a split survives a +// re-key; once the identity spans machines, id no longer names one disk, so a +// split recorded against it applies to every copy (erring towards an extra +// row, never a merge the operator forbade). Manual links recorded against id +// stop applying for the same reason. When every disk carrying the identity +// sits on another machine, each +// copy takes the identity keyed to the machine of its canonical parent at that +// point, and the copy holding id is re-keyed too. So machines can be ingested +// in any order, and id never passes between disks that coexist on different +// machines. Like any serial-keyed ID it still passes to a same-serial disk +// that appears once the previous holder is gone, and a disk's ID changes when +// a same-serial disk first appears on another machine and back when it goes +// away. More than one joinable disk joins none, except that the holder of id +// keeps it. +func (rr *ResourceRegistry) physicalDiskIDForMachineLocked(id, candidateID string, incoming *Resource, source DataSource, onlyMissing bool) string { + key := physicalDiskHardwareKey(incoming.Identity) + if key == "" { + return id + } + siblings := rr.physicalDisksWithHardwareKeyLocked(key) + if len(siblings) == 0 { + return id + } + var joinable []*Resource + excluded := false + for _, sibling := range siblings { + if rr.physicalDiskMachinesConflictLocked(sibling, incoming, source) { + continue + } + if rr.isExcluded(sibling.ID, candidateID) || rr.isExcluded(id, candidateID) || + rr.isExcluded(rr.physicalDiskScopedIDLocked(sibling), candidateID) { + excluded = true + continue + } + joinable = append(joinable, sibling) + } + if len(joinable) == 1 { + return joinable[0].ID + } + for _, sibling := range joinable { + if sibling.ID == id { + return id + } + } + if excluded && len(joinable) == 0 { + return rr.physicalDiskFallbackIDLocked(candidateID, incoming, source) + } + scopedID := rr.physicalDiskScopedIDLocked(incoming) + if scopedID == "" { + return id + } + if holder := rr.resources[scopedID]; holder != nil && + (holder.Type != ResourceTypePhysicalDisk || rr.physicalDiskMachinesConflictLocked(holder, incoming, source)) { + return rr.physicalDiskFallbackIDLocked(candidateID, incoming, source) + } + if len(joinable) == 0 && !onlyMissing { + for _, sibling := range siblings { + if sibling.ID == id { + rr.rekeyPhysicalDiskLocked(sibling, rr.physicalDiskScopedIDLocked(sibling)) + } + } + } + return scopedID +} + +// physicalDiskMachinesConflictLocked reports whether two disks carrying one +// hardware identity provably sit on different machines. One reporter placing +// the identity on two machines is proof even when their hostnames match, as +// cloned VMs keep both. Different reporters may sit on machine resources the +// registry never merged, such as a TrueNAS system and the agent running on it, +// so there differing machines conflict only when both disks name hosts with no +// hostname in common. An unknown machine is never a conflict. +func (rr *ResourceRegistry) physicalDiskMachinesConflictLocked(existing, incoming *Resource, source DataSource) bool { + incomingMachine := rr.physicalDiskMachineLocked(incoming.ParentID) + reported, ok := existing.parentBySource[source] + if !ok && existing.ParentID != nil && len(existing.Sources) == 1 && existing.Sources[0] == source { + // Per-source parents do not survive serialization; a single-source + // disk's own parent is that source's. + reported, ok = *existing.ParentID, true + } + if ok && incomingMachine != "" { + if reportedMachine := rr.physicalDiskMachineLocked(&reported); reportedMachine != "" && reportedMachine != incomingMachine { + return true + } + } + if existingMachine := rr.physicalDiskMachineLocked(existing.ParentID); existingMachine != "" && existingMachine == incomingMachine { + return false + } + if physicalDiskHostname(existing) == "" || physicalDiskHostname(incoming) == "" { + return false + } + return !identitiesShareHostname(existing.Identity, incoming.Identity) +} + +// physicalDiskMachineLocked resolves a disk's parent to the machine it sits +// on: the nearest host-shaped ancestor (a Proxmox node, agent host or TrueNAS +// system above a pool or Unraid array), else the topmost known ancestor. +func (rr *ResourceRegistry) physicalDiskMachineLocked(parentID *string) string { + if parentID == nil { + return "" + } + id := CanonicalResourceID(strings.TrimSpace(*parentID)) + for depth := 0; id != "" && depth < 8; depth++ { + parent := rr.resources[id] + if parent == nil || parent.Type == ResourceTypeAgent || parent.ParentID == nil { + return id + } + next := CanonicalResourceID(strings.TrimSpace(*parent.ParentID)) + if next == "" || next == id { + return id + } + id = next + } + return id +} + +// physicalDiskScopedIDLocked keys a disk's hardware identity to its machine, +// or to its hostname when the machine is unknown. +func (rr *ResourceRegistry) physicalDiskScopedIDLocked(disk *Resource) string { + key := physicalDiskHardwareKey(disk.Identity) + scope := rr.physicalDiskScopeLocked(disk) + if key == "" || scope == "" { + return "" + } + return buildHashID(ResourceTypePhysicalDisk, key+"@"+scope) +} + +// physicalDiskFallbackIDLocked returns the observation's source-specific ID, +// keyed to its machine when another machine's disk already holds it: agent +// disk source IDs are the bare serial, so one candidate ID serves every host. +func (rr *ResourceRegistry) physicalDiskFallbackIDLocked(candidateID string, incoming *Resource, source DataSource) string { + holder := rr.resources[candidateID] + if holder == nil || (holder.Type == ResourceTypePhysicalDisk && !rr.physicalDiskMachinesConflictLocked(holder, incoming, source)) { + return candidateID + } + if scope := rr.physicalDiskScopeLocked(incoming); scope != "" { + return buildHashID(ResourceTypePhysicalDisk, candidateID+"@"+scope) + } + return candidateID +} + +func (rr *ResourceRegistry) physicalDiskScopeLocked(disk *Resource) string { + if machine := rr.physicalDiskMachineLocked(disk.ParentID); machine != "" { + return "machine:" + machine + } + if hostname := physicalDiskHostname(disk); hostname != "" { + return "host:" + hostname + } + return "" +} + +// physicalDiskHardwareKey is the stable key chooseNewID hashes into a disk's +// hardware-keyed canonical ID, or "" when the disk has no hardware identity. +func physicalDiskHardwareKey(identity ResourceIdentity) string { + if machineID := strings.TrimSpace(identity.MachineID); machineID != "" { + return "machine:" + machineID + } + if dmiUUID := strings.TrimSpace(identity.DMIUUID); dmiUUID != "" { + return "dmi:" + dmiUUID + } + return "" +} + +// physicalDiskHostname is the disk's least full hostname, so a scoped ID does +// not depend on the order merges appended hostnames in. +func physicalDiskHostname(disk *Resource) string { + least := "" + for _, hostname := range disk.Identity.Hostnames { + if NormalizeHostname(hostname) == "" { + continue + } + if full := NormalizeFullHostname(hostname); least == "" || full < least { + least = full + } + } + return least +} + +func (rr *ResourceRegistry) indexPhysicalDiskHardwareLocked(disk *Resource) { + if disk == nil || disk.Type != ResourceTypePhysicalDisk { + return + } + key := physicalDiskHardwareKey(disk.Identity) + if key == "" { + return + } + if rr.physicalDisksByHardware == nil { + rr.physicalDisksByHardware = make(map[string]map[string]struct{}) + } + ids := rr.physicalDisksByHardware[key] + if ids == nil { + ids = make(map[string]struct{}) + rr.physicalDisksByHardware[key] = ids + } + ids[disk.ID] = struct{}{} +} + +// physicalDisksWithHardwareKeyLocked returns the live disks carrying key, +// sorted by ID, and drops index entries that no longer match. +func (rr *ResourceRegistry) physicalDisksWithHardwareKeyLocked(key string) []*Resource { + ids := rr.physicalDisksByHardware[key] + disks := make([]*Resource, 0, len(ids)) + for id := range ids { + disk := rr.resources[id] + if disk == nil || disk.ID != id || disk.Type != ResourceTypePhysicalDisk || physicalDiskHardwareKey(disk.Identity) != key { + delete(ids, id) + continue + } + disks = append(disks, disk) + } + sort.Slice(disks, func(i, j int) bool { return disks[i].ID < disks[j].ID }) + return disks +} + +// rekeyPhysicalDiskLocked moves a disk to newID, carrying its source mappings, +// identity index entries, retired-ID claims, child parent references and +// relationship endpoints. +func (rr *ResourceRegistry) rekeyPhysicalDiskLocked(disk *Resource, newID string) { + oldID := disk.ID + if newID == "" || newID == oldID || rr.resources[newID] != nil { + return + } + delete(rr.resources, oldID) + disk.ID = newID + rr.resources[newID] = disk + rr.updateSourceMappings(oldID, newID) + rr.matcher.Add(newID, disk.Identity) + rr.indexPhysicalDiskHardwareLocked(disk) + for retiredID, claimant := range rr.supersededIndex { + if claimant == oldID { + rr.supersededIndex[retiredID] = newID + } + } + for _, resource := range rr.resources { + if resource == nil { + continue + } + for index := range resource.Relationships { + relationship := &resource.Relationships[index] + if relationship.SourceID == oldID { + relationship.SourceID = newID + } + if relationship.TargetID == oldID { + relationship.TargetID = newID + } + } + } +} + func (rr *ResourceRegistry) sourceResourceID(source DataSource, sourceID string) string { rr.mu.RLock() defer rr.mu.RUnlock() diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index 5b7a92adb6..89fafc70b0 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -2,6 +2,7 @@ package unifiedresources import ( "encoding/json" + "slices" "strconv" "strings" "testing" @@ -1926,3 +1927,464 @@ func TestPhysicalDiskMergeTemperatureReadingsAcrossIngestBoundaries(t *testing.T } }) } + +const sharedDiskSerial = "SHARED-SERIAL-1" + +// sharedSerialDiskSnapshot puts one /dev/sda carrying sharedDiskSerial under +// each named PVE node of one instance, optionally with a linked host agent +// reporting that disk over SMART on every node. pve1 also carries a disk whose +// serial no other machine reports. +func sharedSerialDiskSnapshot(now time.Time, withAgents bool, nodeNames ...string) models.StateSnapshot { + var snapshot models.StateSnapshot + for _, name := range nodeNames { + node := models.Node{ + ID: proxmoxNodeSourceID("pve", name), + Name: name, + Instance: "pve", + Host: "https://" + name + ":8006", + Status: "online", + LastSeen: now, + } + if withAgents { + node.LinkedAgentID = "host-" + name + snapshot.Hosts = append(snapshot.Hosts, models.Host{ + ID: "host-" + name, + Hostname: name, + MachineID: "machine-" + name, + LinkedNodeID: node.ID, + Status: "online", + LastSeen: now, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sda", + Model: "SEAGATE ST4000NM0023", + Serial: sharedDiskSerial, + Type: "sas", + Temperature: 31, + Health: "PASSED", + }}}, + }) + } + snapshot.Nodes = append(snapshot.Nodes, node) + snapshot.PhysicalDisks = append(snapshot.PhysicalDisks, models.PhysicalDisk{ + ID: ProxmoxPhysicalDiskSourceID("pve", name, "/dev/sda", "", ""), + Instance: "pve", + Node: name, + DevPath: "/dev/sda", + Model: "SEAGATE ST4000NM0023", + Serial: sharedDiskSerial, + Type: "sas", + Health: "PASSED", + LastChecked: now, + }) + if name == "pve1" { + snapshot.PhysicalDisks = append(snapshot.PhysicalDisks, models.PhysicalDisk{ + ID: ProxmoxPhysicalDiskSourceID("pve", name, "/dev/sdb", "", ""), + Instance: "pve", + Node: name, + DevPath: "/dev/sdb", + Model: "Samsung SSD 870", + Serial: "UNIQUE-SERIAL-1", + Type: "ssd", + Health: "PASSED", + LastChecked: now, + }) + } + } + return snapshot +} + +// reversedSnapshot reverses every slice sharedSerialDiskSnapshot fills, so a +// test can prove IDs do not depend on ingest order. +func reversedSnapshot(snapshot models.StateSnapshot) models.StateSnapshot { + slices.Reverse(snapshot.Nodes) + slices.Reverse(snapshot.Hosts) + slices.Reverse(snapshot.PhysicalDisks) + return snapshot +} + +// sharedSerialDisksByNode indexes the registry's sharedDiskSerial disks by +// Proxmox node, failing when two claim one node or one lost its node. +func sharedSerialDisksByNode(t *testing.T, rr *ResourceRegistry) map[string]Resource { + t.Helper() + byNode := make(map[string]Resource) + for _, disk := range rr.ListByType(ResourceTypePhysicalDisk) { + if disk.PhysicalDisk == nil || disk.PhysicalDisk.Serial != sharedDiskSerial { + continue + } + if disk.Proxmox == nil || disk.Proxmox.NodeName == "" { + t.Fatalf("disk %s has no Proxmox node: %+v", disk.ID, disk.Proxmox) + } + if _, dup := byNode[disk.Proxmox.NodeName]; dup { + t.Fatalf("two disks claim node %s", disk.Proxmox.NodeName) + } + byNode[disk.Proxmox.NodeName] = disk + } + return byNode +} + +func TestPhysicalDisksSharingASerialStayOnEachMachine(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + for _, tc := range []struct { + name string + withAgents bool + sources []DataSource + }{ + {name: "proxmox only", sources: []DataSource{SourceProxmox}}, + {name: "linked host agents", withAgents: true, sources: []DataSource{SourceProxmox, SourceAgent}}, + } { + t.Run(tc.name, func(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshot(sharedSerialDiskSnapshot(now, tc.withAgents, "pve1", "pve2", "pve3")) + + byNode := sharedSerialDisksByNode(t, rr) + if len(byNode) != 3 { + t.Fatalf("shared-serial disks by node = %d, want one per node", len(byNode)) + } + bareID := rr.canonicalIDFromIdentity(ResourceTypePhysicalDisk, ResourceIdentity{MachineID: sharedDiskSerial}) + for _, name := range []string{"pve1", "pve2", "pve3"} { + disk := byNode[name] + nodeID := rr.bySource[SourceProxmox][proxmoxNodeSourceID("pve", name)] + if disk.ParentID == nil || *disk.ParentID != nodeID { + t.Fatalf("disk on %s parent = %v, want its node %s", name, disk.ParentID, nodeID) + } + for _, source := range tc.sources { + if !containsDataSource(disk.Sources, source) { + t.Fatalf("disk on %s sources = %v, want %s merged in", name, disk.Sources, source) + } + } + if !slices.ContainsFunc(rr.GetChildren(nodeID), func(child Resource) bool { return child.ID == disk.ID }) { + t.Fatalf("node %s does not list its disk %s as a child", name, disk.ID) + } + // No machine keeps the unscoped ID, so it can never pass + // from one machine's disk to another's. + if disk.ID == bareID { + t.Fatalf("disk on %s holds the unscoped ID %s shared by every machine", name, bareID) + } + } + + uniqueID := rr.canonicalIDFromIdentity(ResourceTypePhysicalDisk, ResourceIdentity{MachineID: "UNIQUE-SERIAL-1"}) + if unique, ok := rr.Get(uniqueID); !ok || unique.PhysicalDisk == nil || unique.PhysicalDisk.Serial != "UNIQUE-SERIAL-1" { + t.Fatalf("a disk only one machine reports must keep its unscoped ID %s", uniqueID) + } + + reordered := NewRegistry(nil) + reordered.IngestSnapshot(reversedSnapshot(sharedSerialDiskSnapshot(now, tc.withAgents, "pve1", "pve2", "pve3"))) + for name, disk := range sharedSerialDisksByNode(t, reordered) { + if disk.ID != byNode[name].ID { + t.Fatalf("reversed ingest moved the disk on %s from %s to %s", name, byNode[name].ID, disk.ID) + } + } + }) + } +} + +func TestPhysicalDiskIDsHoldAcrossRehydrationWhenAMachineJoins(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + first := NewRegistry(nil) + first.IngestSnapshot(sharedSerialDiskSnapshot(now, false, "pve2", "pve3")) + before := sharedSerialDisksByNode(t, first) + + rehydrated := NewRegistry(nil) + rehydrated.IngestResources(first.List()) + rehydrated.IngestSnapshot(sharedSerialDiskSnapshot(now.Add(time.Minute), false, "pve1", "pve2", "pve3")) + after := sharedSerialDisksByNode(t, rehydrated) + if len(after) != 3 { + t.Fatalf("shared-serial disks by node = %d, want one per node", len(after)) + } + for name, disk := range before { + if after[name].ID != disk.ID { + t.Fatalf("disk on %s moved from %s to %s when pve1 joined", name, disk.ID, after[name].ID) + } + } + + fresh := NewRegistry(nil) + fresh.IngestSnapshot(sharedSerialDiskSnapshot(now.Add(time.Minute), false, "pve1", "pve2", "pve3")) + for name, disk := range sharedSerialDisksByNode(t, fresh) { + if after[name].ID != disk.ID { + t.Fatalf("disk on %s is %s after rehydration but %s after a fresh rebuild", name, after[name].ID, disk.ID) + } + } +} + +// trueNASDiskRecords describes one TrueNAS system holding one disk, under a +// pool when pool is set. +func trueNASDiskRecords(now time.Time, system, hostname, pool, serial string) []IngestRecord { + records := []IngestRecord{{ + SourceID: "system:" + system, + Resource: Resource{Type: ResourceTypeAgent, Name: hostname, Status: StatusOnline, LastSeen: now, TrueNAS: &TrueNASData{Hostname: hostname}}, + Identity: ResourceIdentity{Hostnames: []string{hostname}}, + }} + diskParent := "system:" + system + if pool != "" { + diskParent = "pool:" + system + ":" + pool + records = append(records, IngestRecord{ + SourceID: diskParent, + ParentSourceID: "system:" + system, + Resource: Resource{Type: ResourceTypeStorage, Name: pool, Status: StatusOnline, LastSeen: now}, + Identity: ResourceIdentity{Hostnames: []string{hostname}}, + }) + } + return append(records, IngestRecord{ + SourceID: "disk:" + system + ":sda", + ParentSourceID: diskParent, + Resource: Resource{ + Type: ResourceTypePhysicalDisk, Name: "sda", Status: StatusOnline, LastSeen: now, + PhysicalDisk: &PhysicalDiskMeta{DevPath: "/dev/sda", Serial: serial, Wearout: WearoutUnreported}, + }, + Identity: ResourceIdentity{MachineID: serial, Hostnames: []string{hostname}}, + }) +} + +func TestPhysicalDiskReportersOnOneMachineJoinAcrossGroupingParents(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + agentOnArchive := models.StateSnapshot{Hosts: []models.Host{{ + ID: "host-archive", Hostname: "archive", MachineID: "machine-archive", Status: "online", LastSeen: now, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sda", Model: "WDC WD80EFAX", Serial: sharedDiskSerial, Type: "sata", Temperature: 30, Health: "PASSED", + }}}, + }}} + + // The TrueNAS API and the agent on the same box report one disk under + // different parents (pool and agent host) that the registry never merged. + rr := NewRegistry(nil) + rr.IngestSnapshot(agentOnArchive) + rr.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "a", "archive", "tank", sharedDiskSerial)) + disks := rr.ListByType(ResourceTypePhysicalDisk) + if len(disks) != 1 || !containsDataSource(disks[0].Sources, SourceAgent) || !containsDataSource(disks[0].Sources, SourceTrueNAS) { + t.Fatalf("disks = %+v, want the agent and TrueNAS rows of one box merged", disks) + } + + // A cloned TrueNAS keeps both the serial and the hostname; one reporter + // placing the serial on two systems still proves two disks. + rr.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "b", "archive", "tank", sharedDiskSerial)) + disks = rr.ListByType(ResourceTypePhysicalDisk) + if len(disks) != 2 { + t.Fatalf("disks = %d, want the cloned system's disk kept apart", len(disks)) + } + bySystem := make(map[string]Resource) + for _, disk := range disks { + for _, source := range []string{"disk:a:sda", "disk:b:sda"} { + if rr.bySource[SourceTrueNAS][source] == disk.ID { + bySystem[source] = disk + } + } + } + if len(bySystem) != 2 { + t.Fatalf("TrueNAS disk mappings = %+v, want one disk per system", rr.bySource[SourceTrueNAS]) + } + if !containsDataSource(bySystem["disk:a:sda"].Sources, SourceAgent) { + t.Fatalf("system a disk sources = %v, want the agent row still merged", bySystem["disk:a:sda"].Sources) + } + + // A disk moving into a pool on the same machine is the same disk. + moved := NewRegistry(nil) + moved.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "a", "archive", "", sharedDiskSerial)) + moved.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now.Add(time.Minute), "a", "archive", "tank", sharedDiskSerial)) + if disks := moved.ListByType(ResourceTypePhysicalDisk); len(disks) != 1 { + t.Fatalf("disks after moving into a pool = %d, want 1", len(disks)) + } +} + +func TestPhysicalDiskWithoutItsNodeRowJoinsItsOwnMachinesDisk(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + snapshot := sharedSerialDiskSnapshot(now, false, "pve1", "pve2") + // pve3 has an agent but no node row this poll, so its PVE disk arrives + // without a parent while the serial already spans other machines. + snapshot.Hosts = append(snapshot.Hosts, models.Host{ + ID: "host-pve3", Hostname: "pve3", MachineID: "machine-pve3", Status: "online", LastSeen: now, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sda", Model: "SEAGATE ST4000NM0023", Serial: sharedDiskSerial, Type: "sas", Temperature: 31, Health: "PASSED", + }}}, + }) + snapshot.PhysicalDisks = append(snapshot.PhysicalDisks, models.PhysicalDisk{ + ID: ProxmoxPhysicalDiskSourceID("pve", "pve3", "/dev/sda", "", ""), Instance: "pve", Node: "pve3", + DevPath: "/dev/sda", Model: "SEAGATE ST4000NM0023", Serial: sharedDiskSerial, Type: "sas", Health: "PASSED", LastChecked: now, + }) + + rr := NewRegistry(nil) + rr.IngestSnapshot(snapshot) + byNode := sharedSerialDisksByNode(t, rr) + if len(byNode) != 3 { + t.Fatalf("shared-serial disks by node = %d, want one per node", len(byNode)) + } + if sources := byNode["pve3"].Sources; !containsDataSource(sources, SourceAgent) || !containsDataSource(sources, SourceProxmox) { + t.Fatalf("pve3 disk sources = %v, want its agent and parentless PVE rows merged", sources) + } +} + +func TestPhysicalDiskSiblingJoinHonoursManualExclusions(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + // An agent on pve1 that the registry never linked to the node reports the + // disk after both nodes' rows; hostname evidence alone joins it to pve1's. + agentOnPVE1 := models.StateSnapshot{Hosts: []models.Host{{ + ID: "host-pve1", Hostname: "pve1", MachineID: "machine-pve1", Status: "online", LastSeen: now, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sda", Model: "SEAGATE ST4000NM0023", Serial: sharedDiskSerial, Type: "sas", Temperature: 31, Health: "PASSED", + }}}, + }}} + ingest := func(store ResourceStore) *ResourceRegistry { + rr := NewRegistry(store) + rr.IngestSnapshot(sharedSerialDiskSnapshot(now, false, "pve1", "pve2")) + rr.IngestSnapshot(agentOnPVE1) + return rr + } + + joined := ingest(nil) + pve1Disk := sharedSerialDisksByNode(t, joined)["pve1"] + if !containsDataSource(pve1Disk.Sources, SourceAgent) { + t.Fatalf("pve1 disk sources = %v, want the agent row joined", pve1Disk.Sources) + } + + store := NewMemoryStore() + agentCandidate := buildHashID(ResourceTypePhysicalDisk, string(SourceAgent)+":"+normalizeSourceID(sharedDiskSerial)) + if err := store.AddExclusion(ResourceExclusion{ResourceA: pve1Disk.ID, ResourceB: agentCandidate}); err != nil { + t.Fatal(err) + } + split := ingest(store) + if disk, ok := split.Get(pve1Disk.ID); !ok || containsDataSource(disk.Sources, SourceAgent) { + t.Fatalf("excluded pve1 disk = %+v, want it kept apart from the agent row", disk) + } + if agentDisk, ok := split.Get(agentCandidate); !ok || !containsDataSource(agentDisk.Sources, SourceAgent) { + t.Fatalf("agent row should keep its source-specific ID %s once excluded", agentCandidate) + } +} + +func TestPhysicalDiskClonesStayApartAfterSerializedRehydration(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + first := NewRegistry(nil) + first.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "a", "archive", "tank", sharedDiskSerial)) + payload, err := json.Marshal(first.List()) + if err != nil { + t.Fatal(err) + } + var persisted []Resource + if err := json.Unmarshal(payload, &persisted); err != nil { + t.Fatal(err) + } + + rehydrated := NewRegistry(nil) + rehydrated.IngestResources(persisted) + // A clone keeps the hostname; the persisted disk lost its per-source + // parents, so only its own parent says which system TrueNAS saw it on. + rehydrated.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "b", "archive", "tank", sharedDiskSerial)) + if disks := rehydrated.ListByType(ResourceTypePhysicalDisk); len(disks) != 2 { + t.Fatalf("disks = %d, want the clone's disk kept apart after rehydration", len(disks)) + } +} + +func TestPhysicalDiskRekeyCarriesRetiredIDClaims(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + records := trueNASDiskRecords(now, "a", "archive-a", "tank", sharedDiskSerial) + records[len(records)-1].SupersededCanonicalIDs = []string{"physical_disk-retired"} + rr := NewRegistry(nil) + rr.IngestRecords(SourceTrueNAS, records) + bareID := rr.canonicalIDFromIdentity(ResourceTypePhysicalDisk, ResourceIdentity{MachineID: sharedDiskSerial}) + if got := rr.supersededResourceIDLocked("physical_disk-retired"); got != bareID { + t.Fatalf("retired ID resolves to %q, want the disk %s", got, bareID) + } + + rr.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "b", "archive-b", "tank", sharedDiskSerial)) + systemADisk := rr.bySource[SourceTrueNAS]["disk:a:sda"] + if systemADisk == bareID || systemADisk == "" { + t.Fatalf("system a disk = %q, want it re-keyed off the unscoped ID", systemADisk) + } + if got := rr.supersededResourceIDLocked("physical_disk-retired"); got != systemADisk { + t.Fatalf("retired ID resolves to %q after the re-key, want %s", got, systemADisk) + } +} + +func TestPhysicalDiskSplitSurvivesARekey(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + agentDisk := func(id, hostname string) models.Host { + return models.Host{ + ID: id, Hostname: hostname, MachineID: "machine-" + hostname, Status: "online", LastSeen: now, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sda", Model: "WDC WD80EFAX", Serial: sharedDiskSerial, Type: "sata", Temperature: 30, Health: "PASSED", + }}}, + } + } + trueNASCandidate := buildHashID(ResourceTypePhysicalDisk, string(SourceTrueNAS)+":"+normalizeSourceID("disk:a:sda")) + bareID := MachineIdentityCanonicalID(ResourceTypePhysicalDisk, sharedDiskSerial) + ingest := func(t *testing.T, excludedID string, hosts ...models.Host) *ResourceRegistry { + t.Helper() + store := NewMemoryStore() + if err := store.AddExclusion(ResourceExclusion{ResourceA: excludedID, ResourceB: trueNASCandidate}); err != nil { + t.Fatal(err) + } + rr := NewRegistry(store) + rr.IngestSnapshot(models.StateSnapshot{Hosts: hosts}) + rr.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "a", "archive", "tank", sharedDiskSerial)) + return rr + } + split := func(t *testing.T, rr *ResourceRegistry) { + t.Helper() + trueNASDisk, ok := rr.Get(trueNASCandidate) + if !ok || containsDataSource(trueNASDisk.Sources, SourceAgent) { + t.Fatalf("TrueNAS row = %+v, want it kept on %s apart from the agent row", trueNASDisk, trueNASCandidate) + } + } + + t.Run("split recorded against the unscoped ID after another machine re-keyed the disk", func(t *testing.T) { + rr := ingest(t, bareID, agentDisk("host-archive", "archive"), agentDisk("host-other", "other")) + if _, ok := rr.Get(bareID); ok { + t.Fatalf("the unscoped ID %s should be re-keyed once two machines report the serial", bareID) + } + split(t, rr) + }) + t.Run("split recorded against the scoped ID before the serial spans machines", func(t *testing.T) { + scoped := ingest(t, bareID, agentDisk("host-archive", "archive"), agentDisk("host-other", "other")) + archiveAgentDisk := "" + for _, disk := range scoped.ListByType(ResourceTypePhysicalDisk) { + if disk.ParentID != nil && *disk.ParentID == scoped.bySource[SourceAgent]["host-archive"] { + archiveAgentDisk = disk.ID + } + } + if archiveAgentDisk == "" || archiveAgentDisk == bareID { + t.Fatalf("archive agent disk = %q, want a machine-scoped ID", archiveAgentDisk) + } + split(t, ingest(t, archiveAgentDisk, agentDisk("host-archive", "archive"))) + }) +} + +func TestPhysicalDiskExclusionFallbackStaysPerMachine(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + agents := models.StateSnapshot{} + for _, name := range []string{"pve1", "pve2"} { + agents.Hosts = append(agents.Hosts, models.Host{ + ID: "host-" + name, Hostname: name, MachineID: "machine-" + name, Status: "online", LastSeen: now, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sda", Model: "SEAGATE ST4000NM0023", Serial: sharedDiskSerial, Type: "sas", Temperature: 31, Health: "PASSED", + }}}, + }) + } + probe := NewRegistry(nil) + probe.IngestSnapshot(sharedSerialDiskSnapshot(now, false, "pve1", "pve2")) + byNode := sharedSerialDisksByNode(t, probe) + + // Both nodes' disks are split from the agent observation, whose + // source-specific ID is the same on every host. + store := NewMemoryStore() + agentCandidate := buildHashID(ResourceTypePhysicalDisk, string(SourceAgent)+":"+normalizeSourceID(sharedDiskSerial)) + for _, name := range []string{"pve1", "pve2"} { + if err := store.AddExclusion(ResourceExclusion{ResourceA: byNode[name].ID, ResourceB: agentCandidate}); err != nil { + t.Fatal(err) + } + } + rr := NewRegistry(store) + rr.IngestSnapshot(sharedSerialDiskSnapshot(now, false, "pve1", "pve2")) + rr.IngestSnapshot(agents) + + agentDisks := 0 + for _, disk := range rr.ListByType(ResourceTypePhysicalDisk) { + if !containsDataSource(disk.Sources, SourceAgent) { + continue + } + agentDisks++ + if containsDataSource(disk.Sources, SourceProxmox) { + t.Fatalf("disk %s merged an excluded agent row into a node's disk", disk.ID) + } + } + if agentDisks != 2 { + t.Fatalf("agent disks = %d, want one per host rather than both hosts on the shared fallback ID", agentDisks) + } +}