Keep same-serial disks on different machines apart (#2660)

## Problem `ingestRecord` refuses a physical-disk match under another parent (`findMatch` skips scope-incompatible candidates), then `chooseNewID` mints the same hardware-keyed canonical ID (`machine:<serial>`) and the ingest merges into whatever disk already holds it. Disks that report one usable serial on several machines collapse into a single resource parented to whichever machine ingested last, and every other machine loses the disk from inventory. Real triggers: dual-ported SAS shelves seen by two nodes, cloned VMs with an explicit identical serial, fixed-serial USB bridges, TrueNAS systems. Repro: `IngestSnapshot` with three PVE nodes in one instance, each with a `/dev/sda` of the same serial, gave one disk. ## Fix - A new disk resolves its parent to a machine: the nearest host ancestor, so a TrueNAS pool or Unraid array counts as its system or host. - Two disks with one identity stay apart only on evidence: one reporter placing the serial on two machines (cloned VMs keep both serial and hostname), or machines whose hostnames share nothing. Reporters on one box still join across grouping parents and unmerged host resources, so the TrueNAS API and the agent on that box stay one disk. An unknown machine never splits a disk, so a PVE row whose node row is missing still joins its host's agent disk. - Once a serial spans machines, every copy takes the serial keyed to its own machine, and the holder of the unscoped ID is re-keyed with its source mappings, retired-ID claims and relationship endpoints. IDs don't depend on machine ingest order, and the unscoped ID never passes between disks that coexist. A disk only one machine reports keeps its current ID. - Operator splits ("Split merged resource" exclusions) now hold on this path; the old collision merge ignored them. A split recorded against any ID the disk can hold applies, so it survives a re-key. - The registry indexes disks by hardware key (`physicalDisksByHardware`) because `IdentityMatcher` keeps one resource per machine ID. ## ID stability decision Machine-scoped IDs only for identities that actually span machines, rather than a pre-pass. A pre-pass needs final parent scopes before ingest, but host/node merging happens during ingest. Giving the first-ingested machine the unscoped ID would hand one machine's ID, and its operator state, to another machine when an earlier-sorting machine appears. With the chosen design a disk's ID changes when a same-serial disk first appears on another machine. That orphans ID-keyed state rather than misapplying it. ## Residuals (documented in unified-resources.md) - Like any serial-keyed ID, the unscoped ID still passes to a same-serial disk that appears once the previous holder is gone. - Alert-history rows owned under the unscoped ID don't follow a re-key. The PVE disk alert owner (#2646) already leaves shared-serial rows unbound unless a WWN tells them apart. - Agent SMART and Unraid disk source IDs are the bare serial. Same-serial agent disks on different hosts share one `SourceAgent` mapping, and only the last ingested keeps the agent source target. This is tracked as a separate task, because the fix re-keys persisted mappings. - Same-serial disks share one metrics history key. - Across reporters, hostnames that normalize alike still join. ## Verification - New tests in `registry_merge_policy_test.go`: per-machine disks for PVE-only and linked-agent clusters, including reversed ingest order; rehydration with a machine joining; agent + TrueNAS on one box versus a cloned TrueNAS with the same hostname; a disk moving into a pool; a parentless PVE row on a shared serial; exclusions honoured, surviving a re-key, with per-machine fallback; serialized-JSON rehydration; retired-ID claims carried across a re-key. - Every new test fails against the unfixed registry or with its fix removed (`-overlay` negative controls). - `go test ./internal/unifiedresources` passes, as do the disk/SMART/Unraid/TrueNAS tests in `internal/monitoring`, `internal/api` and `internal/truenas`. - Five rounds of independent review; the last round found no P1/P2 issues.
2026-10-24 04:33:17 +00:00 · 2026-10-07 14:50:15 +01:00
parent 49d4100385 425dd0d45b
commit 156d2e274d
3 changed files with 777 additions and 0 deletions
@@ -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
@@ -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()
@@ -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)
+	}
+}