From 425dd0d45b73a7effcee6e3ee3f882b1c899eb3a Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 7 Oct 2026 12:56:47 +0100 Subject: [PATCH] Keep same-serial disks on different machines apart findMatch refuses a physical-disk candidate under another parent, but ingestRecord then minted the same hardware-keyed canonical ID from the serial and merged into its holder on the collision. Disks that report one usable serial on several machines (dual-ported SAS shelves, cloned VMs with an explicit serial, fixed-serial USB bridges, TrueNAS systems) collapsed into one resource parented to whichever machine ingested last, and every other machine lost the disk from inventory. A new disk now resolves its parent to a machine (the nearest host ancestor, so pools and Unraid arrays count as their host) and stays apart only on evidence: one reporter placing the serial on two machines, or machines whose hostnames share nothing. Reporters on one box still join across grouping parents, so the TrueNAS API and the agent on that box stay one disk, and an unknown machine never splits a disk. Once a serial spans machines, every copy takes the serial keyed to its own machine and the unscoped holder is re-keyed, so IDs do not depend on machine ingest order and the unscoped ID never passes between disks that coexist. A disk only one machine reports keeps its ID. Operator splits now hold on this path; the old collision merge ignored them. The registry indexes disks by hardware key because the identity matcher keeps one resource per machine ID. --- .../internal/subsystems/unified-resources.md | 42 ++ internal/unifiedresources/registry.go | 273 +++++++++++ .../registry_merge_policy_test.go | 462 ++++++++++++++++++ 3 files changed, 777 insertions(+) diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 6f82da2f6d..957af34194 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -5325,6 +5325,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 6c9a786cad..afc35b3e21 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) @@ -3836,6 +3847,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 { @@ -5260,6 +5272,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) + } +}