From 89a05ed7357c983df6003c22b38ad491ceaaed4c Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 7 Oct 2026 23:06:06 +0100 Subject: [PATCH] Pin each manual-link side from its own identity A manual link merges two resources for display, and the durable identity pin was written from that merged identity. A link primary was pinned with the folded resource's machine key, and the store's one-owner-per-key rule then deleted the folded resource's own pin. A Proxmox node's pin records its qualified endpoint as the hostname, so on the next rebuild the node completed the linked agent's machine key from that pin, minted the agent's machine-derived ID and merged with the agent by ID before the link applied. With the node as primary, succession then rewrote the link into a self-link on the agent's ID; in either direction, removing the link left the pair merged. The registry now captures each link side's own pin when the link first joins it in a rebuild, and PersistIdentityPins writes those instead of the merged projection, keeping a pinnable folded side under its own canonical ID. Proxmox auto-links merge by identity before any manual link applies, so their boot-window pins are unchanged. A link side rewrites its whole row through a new ReplaceResourceIdentityPins. Like an upsert it keeps stored fields its own pin leaves empty, but it drops a strong key another resource holds, since an earlier release borrowed that key from the merged identity and the store's one-owner rule would otherwise delete one row or the other. A folded side drops a strong key a live resource or an earlier folded side already holds, so the two never trade it on every rebuild, and writes no pin when that would leave it without its machine evidence. A node row an earlier release filled with the agent's machine key is handed back only when the node cannot complete from it. With a qualified endpoint or a cluster name, completion still supplies the borrowed key before the link applies, so such a pair stays merged. --- .../internal/subsystems/unified-resources.md | 80 ++- .../unifiedresources/canonical_id_pins.go | 204 +++++++- internal/unifiedresources/registry.go | 26 + .../registry_merge_policy_test.go | 457 ++++++++++++++++++ internal/unifiedresources/store.go | 54 ++- internal/unifiedresources/store_test.go | 60 +++ 6 files changed, 846 insertions(+), 35 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 95805c44a6..4d3db9fa90 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -2752,7 +2752,7 @@ application resource-provider or WebSocket lifecycle. (`canonical_id` ↔ machine ID / DMI UUID / cluster / hostname) in the `resource_identities` table (`ResourceStore.UpsertResourceIdentityPins` / `ListResourceIdentityPins`, written diff-aware after monitor-adapter - rebuilds) and completes weak incoming identities from those pins + rebuilds; manual-link sides write through `ReplaceResourceIdentityPins`) and completes weak incoming identities from those pins before matching and ID derivation (`internal/unifiedresources/canonical_id_pins.go`). Persisted primary hostnames preserve their normalized dotted form. The pin index resolves @@ -2768,7 +2768,15 @@ application resource-provider or WebSocket lifecycle. must not mint host canonical IDs from snapshot-content-dependent identity subsets without consulting the pins, and pin writes stay on the durable store-backed registry (ephemeral per-request registries - consult, never write). Pinned hostnames preserve the full dotted name + consult, never write). A pin records the identity a resource derives + from its own sources rather than a manual link's merged projection: + each side of a link writes the pin captured before the link merged + them, keeping its row's other fields unless another resource holds that + key, and a pinnable folded side keeps its pin under its own canonical + ID, less any strong key a live resource or an earlier folded side + holds. Capture follows pin completion, so a key an earlier release + borrowed into a side's row can still reach the capture (see "Operator + links reach record-ingested resources"). Pinned hostnames preserve the full dotted name (`NormalizeFullHostname`): distinct machines that share a short hostname (`cloud.rnd-lax1` vs `cloud.gce-or1`) must keep distinct pins, distinct pin-index buckets, and distinct presentation host rows. @@ -6567,11 +6575,50 @@ separate. system's facet, and the system's pools re-parent to the merged row. - The registry remembers each resource a manual link folded into its primary. A folded resource is still observed, so neither pin-driven nor - record-declared succession may succeed its ID. An agent-type primary - carries the folded resource's machine keys and is pinned with them, and - succeeding the folded ID re-keyed the link onto the primary itself, which - split the pair on the next rebuild. That already broke snapshot links whose - primary is a machine-keyless agent-type row, such as a Proxmox node. + record-declared succession may succeed its ID: that re-keyed the link onto + the primary itself, which split the pair on the next rebuild, and already + broke snapshot links whose primary is a machine-keyless agent-type row, + such as a Proxmox node. +- Each side of a link is pinned only from its own identity, captured when + the link first joins it in a rebuild (`recordLinkOwnPin`). A side with no + pinnable identity (not agent-shaped, or no strong key) writes no pin, and + a pinnable folded side keeps its pin under its own canonical ID. The merged + primary used to be pinned with both sides' keys, and the store's + one-owner-per-key rule then deleted the folded side's pin. A Proxmox node's + pin records its qualified endpoint as the hostname, so the node's next + rebuild completed the linked agent's machine key from that merged pin, + minted the agent's machine-derived ID and merged with the agent by ID + before the link applied. With the node as primary, succession then + rewrote the link into a self-link on the agent's ID; in either direction, + removing the link left the pair merged. +- A link side rewrites its whole row (`ReplaceResourceIdentityPins`) + instead of upserting it. Like an upsert it keeps the stored fields its + own pin leaves empty, so evidence from an earlier observation survives a + weaker one, but it never keeps a machine, DMI or cluster-and-hostname key + another resource holds this rebuild, including a cluster key a kept + hostname would complete. That is how a key an earlier release borrowed + into the row from the merged identity leaves it; a key a resource reports + now also beats one a link side's row only kept, as the store already + hands a moved key to the new pin. The store gives each strong key one + owner and deletes any other row holding a key a written pin carries, so a + folded side drops each own key held by a live non-link resource's + upserted row (an upsert cannot drop a field), a live link side's own pin, + or a folded side earlier in canonical-ID order. It writes no pin when no + strong key is left, or when it would lose the machine or DMI key it + reports: a cluster-only row reads as a machine-keyless host's, which + cluster-slot succession lets a later machine in that slot absorb. A row kept for a side that writes no pin goes + when another written pin carries one of its keys, such as the folded + agent's own pin carrying the machine key an earlier release borrowed into + a Proxmox node's row. That hand-back only reaches a row the node cannot + complete from: while the node's qualified endpoint, or its cluster and + name, still resolve to that row, completion supplies the borrowed key + before any link applies, so the pair merges by ID and succession rewrites + the link into a self-link, as it did before this change. A link side whose + row loses a borrowed machine key becomes a machine-keyless row, open to + the cluster-slot succession any machine-keyless host gets. Two live resources that share a strong key + still take it from each other on alternate rebuilds, as before. Evidence + a later record pass of the same rebuild merges into an already-joined row + reaches neither side's pin. - Fold history is registry-local: resource seeding and overlay clones do not carry it. A reference to the folded resource's canonical ID, such as an availability check configured against an agent later linked into its @@ -6586,6 +6633,25 @@ resources API seeds, lists the same rows pins the node case (`TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence`), record-declared succession (`TestRecordDeclaredSuccessionSparesALinkFoldedResource`) and both payloads (`TestManualLinkKeepsProviderPayloadsThePrimaryLacks`). +It also pins each side's own pin for a qualified-endpoint node linked to an +agent, standalone and clustered, in both directions, through three rebuilds +with no steady-state pin writes and a separate pair once the link is removed +(`TestManualLinkPinsEachSideFromItsOwnIdentity`), an unclaimed stored field +kept without a rewrite every rebuild +(`TestManualLinkSideKeepsAnUnclaimedStoredField`), a shared +cluster-and-hostname key kept by the live primary while the folded side keeps +its machine key, and handed to the folded side when the primary's row only +kept it (`TestManualLinkFoldedSideNeverTakesALiveResourcesKey`), a kept +hostname refused where it would complete another resource's cluster key +(`TestManualLinkSideKeepsNoHostnameThatCompletesAHeldClusterKey`), no pin for a +folded side whose machine key another host holds +(`TestManualLinkFoldedSideWritesNoPinWithoutItsMachineKey`), a folded side +re-pinned after the primary's new pin deletes its widened row +(`TestManualLinkFoldedSideKeepsItsPinWhenAPrimaryClaimsItsWiderRow`), and the +agent's own pin taking back the machine key an earlier release borrowed into +a node's row (`TestManualLinkReturnsABorrowedMachineKeyToItsAgent`). +`store_test.go` (`TestResourceIdentityPinReplaceClearsFieldsUpsertKeeps`) pins +replace against upsert on both stores. `monitor_adapter_read_state_test.go` (`TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds`) pins the thresholds on the rebuild, supplemental and overlay paths, and diff --git a/internal/unifiedresources/canonical_id_pins.go b/internal/unifiedresources/canonical_id_pins.go index c48aa0cee3..af5a52e444 100644 --- a/internal/unifiedresources/canonical_id_pins.go +++ b/internal/unifiedresources/canonical_id_pins.go @@ -3,6 +3,7 @@ package unifiedresources import ( "log" "net" + "slices" "strings" ) @@ -326,12 +327,106 @@ func (index *identityPinIndex) successionsFor(pin ResourceIdentityPin) []Canonic return successions } +// upsertedOnto returns the row UpsertResourceIdentityPins stores when pin is +// written over existing: an empty field keeps the stored value. +func (pin ResourceIdentityPin) upsertedOnto(existing ResourceIdentityPin) ResourceIdentityPin { + if pin.MachineID == "" { + pin.MachineID = existing.MachineID + } + if pin.DMIUUID == "" { + pin.DMIUUID = existing.DMIUUID + } + if pin.ClusterName == "" { + pin.ClusterName = existing.ClusterName + } + if pin.Hostname == "" { + pin.Hostname = existing.Hostname + } + return pin +} + +// keptOver returns the row a manual-link side writes over existing: like an +// upsert it keeps the stored fields its own pin leaves empty, so evidence +// from earlier observations survives a weaker one, but it never keeps a +// strong key another resource holds. Earlier releases pinned a link's +// primary with the merged identity, so such a key may be the other side's, +// and a key a resource reports now beats one a row only kept, as the store +// already hands a moved key to the new pin. +func (pin ResourceIdentityPin) keptOver(existing ResourceIdentityPin, held map[string]string) ResourceIdentityPin { + if pin.Hostname == "" && (pin.ClusterName == "" || !heldByOther(held, clusterPinKey(pin.ClusterName, existing.Hostname), pin.CanonicalID)) { + pin.Hostname = existing.Hostname + } + if pin.MachineID == "" && existing.MachineID != "" && !heldByOther(held, machinePinKey(existing.MachineID), pin.CanonicalID) { + pin.MachineID = existing.MachineID + } + if pin.DMIUUID == "" && existing.DMIUUID != "" && !heldByOther(held, dmiPinKey(existing.DMIUUID), pin.CanonicalID) { + pin.DMIUUID = existing.DMIUUID + } + if pin.ClusterName == "" && existing.ClusterName != "" && !heldByOther(held, clusterPinKey(existing.ClusterName, pin.Hostname), pin.CanonicalID) { + pin.ClusterName = existing.ClusterName + } + return pin +} + +// withoutKeysHeldElsewhere drops the strong keys another resource holds. +func (pin ResourceIdentityPin) withoutKeysHeldElsewhere(held map[string]string) ResourceIdentityPin { + if pin.MachineID != "" && heldByOther(held, machinePinKey(pin.MachineID), pin.CanonicalID) { + pin.MachineID = "" + } + if pin.DMIUUID != "" && heldByOther(held, dmiPinKey(pin.DMIUUID), pin.CanonicalID) { + pin.DMIUUID = "" + } + if pin.ClusterName != "" && heldByOther(held, clusterPinKey(pin.ClusterName, pin.Hostname), pin.CanonicalID) { + pin.ClusterName = "" + } + return pin +} + +func heldByOther(held map[string]string, key, canonicalID string) bool { + owner, taken := held[key] + return taken && owner != canonicalID +} + +// ownerKeys lists the strong keys the store lets only one pin hold, matching +// the conflict rule in UpsertResourceIdentityPins. +func (pin ResourceIdentityPin) ownerKeys() []string { + var keys []string + if pin.MachineID != "" { + keys = append(keys, machinePinKey(pin.MachineID)) + } + if pin.DMIUUID != "" { + keys = append(keys, dmiPinKey(pin.DMIUUID)) + } + if pin.ClusterName != "" { + keys = append(keys, clusterPinKey(pin.ClusterName, pin.Hostname)) + } + return keys +} + +func machinePinKey(machineID string) string { return "machine:" + machineID } + +func dmiPinKey(dmiUUID string) string { return "dmi:" + dmiUUID } + +func clusterPinKey(clusterName, hostname string) string { + return "cluster:" + clusterName + "\x00" + hostname +} + // PersistIdentityPins writes the identity pins for the registry's current // host resources into the resource store. Only new or changed pins are // written, so steady-state rebuild ticks cost no writes. Call this after a // rebuild on the durable store-backed registry; ephemeral per-request // registries consult pins but do not write them. // +// Each resource is pinned from its own sources. Both sides of a manual link +// are pinned from the identity captured before the link merged them +// (recordLinkOwnPin), and a pinnable folded side keeps its pin under its own +// canonical ID: a link is display intent, so the merged identity must not +// lend one machine's keys to the other's pin, or a later rebuild would +// complete those keys, mint the same canonical ID for both sides and keep +// them merged after the link is gone. A link side's row is rewritten whole +// (see keptOver), because an upsert would keep a key an earlier release +// borrowed into it from the merged identity. +// // When a new pin supersedes an earlier era's pin for the same physical host // (see successionsFor), the store re-keys the host's operator-owned rows to // the new canonical ID before the pin write, so operator intent @@ -347,31 +442,93 @@ func (rr *ResourceRegistry) PersistIdentityPins() { } rr.mu.RLock() - var pins []ResourceIdentityPin - var successions []CanonicalIDSuccession - for _, resource := range rr.resources { + // held maps each strong key to the resource that holds it this rebuild: + // a live resource's row (an upsert only ever widens it), each link + // side's own pin, and then the keys link sides keep from their rows. + held := make(map[string]string) + hold := func(pin ResourceIdentityPin) { + for _, key := range pin.ownerKeys() { + if _, taken := held[key]; !taken { + held[key] = pin.CanonicalID + } + } + } + var pins, linkPins []ResourceIdentityPin + for id, resource := range rr.resources { + if own, linked := rr.linkOwnPins[id]; linked { + if own != nil { + hold(*own) + linkPins = append(linkPins, *own) + } + continue + } pin, ok := identityPinForResource(resource) if !ok { continue } - if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known && existing == pin { - continue + effective := pin + if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known { + effective = pin.upsertedOnto(existing) } + hold(effective) pins = append(pins, pin) - for _, succession := range rr.identityPins.successionsFor(pin) { - // A primary that absorbed a linked resource also carries its - // machine keys. That resource is still observed, and succeeding - // it would re-key the link onto the primary itself, splitting - // the pair again on the next rebuild. - if rr.canonicalIDObservedLocked(succession.OldCanonicalID) { - continue - } - successions = append(successions, succession) + } + folded := make([]string, 0, len(rr.linkOwnPins)) + for id, own := range rr.linkOwnPins { + if _, live := rr.resources[id]; !live && own != nil { + folded = append(folded, id) } } + slices.Sort(folded) + for _, id := range folded { + // The store gives each strong key one owner. A folded side drops a + // key a live resource, or an earlier folded side, already holds + // rather than taking it back on every other rebuild. A live non-link + // row is upserted, which cannot drop a field, so its kept fields + // hold their keys here. A side that would lose its machine evidence + // writes no pin: a cluster-only row reads as a machine-keyless + // host's, which cluster-slot succession lets a later machine absorb. + own := *rr.linkOwnPins[id] + pin := own.withoutKeysHeldElsewhere(held) + if !pin.hasStrongKey() || (pin.MachineID == "" && pin.DMIUUID == "" && (own.MachineID != "" || own.DMIUUID != "")) { + continue + } + hold(pin) + linkPins = append(linkPins, pin) + } + slices.SortFunc(linkPins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) }) + for i, pin := range linkPins { + if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known { + linkPins[i] = pin.keptOver(existing, held) + hold(linkPins[i]) + } + } + + var successions []CanonicalIDSuccession + changed := func(batch []ResourceIdentityPin) []ResourceIdentityPin { + var out []ResourceIdentityPin + for _, pin := range batch { + if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known && existing == pin { + continue + } + out = append(out, pin) + for _, succession := range rr.identityPins.successionsFor(pin) { + // A resource a manual link folded into its primary is still + // observed. Succeeding it would re-key the link onto the + // primary itself, splitting the pair on the next rebuild. + if rr.canonicalIDObservedLocked(succession.OldCanonicalID) { + continue + } + successions = append(successions, succession) + } + } + return out + } + pins = changed(pins) + linkPins = changed(linkPins) rr.mu.RUnlock() - if len(pins) == 0 { + if len(pins) == 0 && len(linkPins) == 0 { return } if len(successions) > 0 { @@ -381,9 +538,20 @@ func (rr *ResourceRegistry) PersistIdentityPins() { } } } - if err := rr.store.UpsertResourceIdentityPins(pins); err != nil { - log.Printf("unifiedresources: failed to persist identity pins: %v", err) - return + // Upserts go first: one that deletes a link side's old row, which still + // held a key the side has now dropped, is then followed by the side's + // replace, which writes the row again. + if len(pins) > 0 { + if err := rr.store.UpsertResourceIdentityPins(pins); err != nil { + log.Printf("unifiedresources: failed to persist identity pins: %v", err) + return + } + } + if len(linkPins) > 0 { + if err := rr.store.ReplaceResourceIdentityPins(linkPins); err != nil { + log.Printf("unifiedresources: failed to persist manual-link identity pins: %v", err) + return + } } refreshed, err := rr.store.ListResourceIdentityPins() if err != nil { diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index c1b036344f..c850d02563 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -99,6 +99,12 @@ type ResourceRegistry struct { // primary. They are still observed, so identity succession treats them // as live. linkMergedIDs map[string]struct{} + // linkOwnPins holds the identity pin each side of a manual link derives + // from its own sources, captured before the link merged them. A nil + // value records a side with no pinnable identity. PersistIdentityPins + // writes these instead of the merged projection, so a link never lends + // one machine's identity keys to another resource's durable pin. + linkOwnPins map[string]*ResourceIdentityPin // supersededIndex maps record-declared retired canonical IDs to the live // resource that superseded them, so references persisted under a retired // ID (availability links, API reads) keep resolving. An empty value marks @@ -4842,6 +4848,8 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura primaryID, otherID = otherID, primaryID } + rr.recordLinkOwnPin(primaryID, primary) + rr.recordLinkOwnPin(otherID, other) rr.mergeResourceData(primary, other, thresholds) delete(rr.resources, otherID) if rr.linkMergedIDs == nil { @@ -4852,6 +4860,24 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura } } +// recordLinkOwnPin captures the identity pin a manual-link side derives from +// its own sources, before the link merge projects the other side's identity +// onto the primary. The first capture in a registry generation wins, so a +// primary that absorbs several links keeps the pin it had before any of them. +func (rr *ResourceRegistry) recordLinkOwnPin(id string, resource *Resource) { + if _, recorded := rr.linkOwnPins[id]; recorded { + return + } + if rr.linkOwnPins == nil { + rr.linkOwnPins = make(map[string]*ResourceIdentityPin) + } + if pin, ok := identityPinForResource(resource); ok { + rr.linkOwnPins[id] = &pin + return + } + rr.linkOwnPins[id] = nil +} + func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource, thresholds map[DataSource]time.Duration) { if other == nil || primary == nil { return diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index c133a0e533..fa8f6fd5c1 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -655,6 +655,463 @@ func TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence(t *testing.T) } } +// pinCountingStore counts identity pin writes, so tests can tell a +// steady-state rebuild (no writes) from one that rewrites a row every tick. +type pinCountingStore struct { + *MemoryStore + pinWrites int +} + +func (s *pinCountingStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error { + s.pinWrites++ + return s.MemoryStore.UpsertResourceIdentityPins(pins) +} + +func (s *pinCountingStore) ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error { + s.pinWrites++ + return s.MemoryStore.ReplaceResourceIdentityPins(pins) +} + +// manualLinkPinSnapshot is a Proxmox node and an unrelated host agent. With a +// cluster name the node is a cluster member whose instance names the cluster, +// so its identity carries the cluster. +func manualLinkPinSnapshot(now time.Time, clusterName string) models.StateSnapshot { + instance := "lab" + if clusterName != "" { + instance = clusterName + } + return models.StateSnapshot{ + Nodes: []models.Node{{ + ID: instance + "-pve1", Name: "pve1", Instance: instance, ClusterName: clusterName, IsClusterMember: clusterName != "", + Host: "https://pve1.example.test:8006", Status: "online", LastSeen: now, + }}, + Hosts: []models.Host{{ + ID: "host-box", Hostname: "box-x", MachineID: "machine-box", Status: "online", LastSeen: now, + }}, + LastUpdate: now, + } +} + +// A manual link joins two resources for display; it does not make them one +// machine, so each side keeps pinning its own identity. A Proxmox node whose +// configured endpoint is a qualified name pins that name. Pinned with the +// linked agent's machine key instead, the node's next rebuild completed that +// key, minted the agent's machine-derived ID and merged with the agent before +// the link applied, and removing the link could not separate them. +func TestManualLinkPinsEachSideFromItsOwnIdentity(t *testing.T) { + for _, tc := range []struct { + name string + clusterName string + nodePrimary bool + }{ + {name: "standalone node primary", nodePrimary: true}, + {name: "standalone agent primary"}, + {name: "cluster node primary", clusterName: "lab-cluster", nodePrimary: true}, + {name: "cluster agent primary", clusterName: "lab-cluster"}, + } { + t.Run(tc.name, func(t *testing.T) { + now := time.Now().UTC() + snapshot := manualLinkPinSnapshot(now, tc.clusterName) + store := &pinCountingStore{MemoryStore: NewMemoryStore()} + adapter := NewMonitorAdapter(NewRegistry(store)) + rebuild := func(step int) { + next := snapshot + next.LastUpdate = now.Add(time.Duration(step) * time.Second) + adapter.PopulateSnapshotAndSupplemental(next, nil) + } + rebuild(0) + var nodeID, agentID string + for _, resource := range adapter.GetAll() { + if resource.Proxmox != nil { + nodeID = resource.ID + } else if resource.Agent != nil { + agentID = resource.ID + } + } + if nodeID == "" || agentID != MachineIdentityCanonicalID(ResourceTypeAgent, "machine-box") || nodeID == agentID { + t.Fatalf("unlinked estate = %v, want a node and the machine-keyed agent", resourceIDs(adapter.GetAll())) + } + assertOwnPins := func(step string) { + t.Helper() + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("%s: list identity pins: %v", step, err) + } + agentPinned, nodePinned := false, false + for _, pin := range pins { + switch pin.CanonicalID { + case agentID: + if pin.MachineID != "machine-box" || pin.Hostname != "box-x" || pin.ClusterName != "" { + t.Fatalf("%s: agent pin = %+v, want its own machine key and hostname", step, pin) + } + agentPinned = true + case nodeID: + if pin.MachineID != "" || pin.ClusterName != tc.clusterName || pin.Hostname != "pve1.example.test" { + t.Fatalf("%s: node pin = %+v, want only its own cluster and endpoint", step, pin) + } + nodePinned = true + default: + t.Fatalf("%s: unexpected pin %+v", step, pin) + } + } + if !agentPinned { + t.Fatalf("%s: pins = %+v, want the agent's own machine-keyed pin", step, pins) + } + if tc.clusterName != "" && !nodePinned { + t.Fatalf("%s: pins = %+v, want the cluster node's own pin", step, pins) + } + } + assertOwnPins("unlinked") + + primaryID, otherID := agentID, nodeID + if tc.nodePrimary { + primaryID, otherID = nodeID, agentID + } + if err := store.AddLink(ResourceLink{ResourceA: primaryID, ResourceB: otherID, PrimaryID: primaryID}); err != nil { + t.Fatalf("add link: %v", err) + } + for step := 1; step <= 3; step++ { + writesBefore := store.pinWrites + rebuild(step) + if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{primaryID}) { + t.Fatalf("rebuild %d: resources = %v, want only the linked primary %s", step, got, primaryID) + } + links, err := store.GetLinks() + if err != nil { + t.Fatalf("get links: %v", err) + } + if len(links) != 1 || links[0].ResourceA != primaryID || links[0].ResourceB != otherID { + t.Fatalf("rebuild %d: links = %+v, want %s -> %s kept", step, links, primaryID, otherID) + } + assertOwnPins("linked rebuild") + if step > 1 && store.pinWrites != writesBefore { + t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", step, store.pinWrites-writesBefore) + } + } + + // Removing the link separates the pair on the next rebuild: + // nothing the link lent either side may outlive it. + store.mu.Lock() + store.links = nil + store.mu.Unlock() + rebuild(4) + if got, want := resourceIDs(adapter.GetAll()), resourceIDs([]Resource{{ID: nodeID}, {ID: agentID}}); !slices.Equal(got, want) { + t.Fatalf("after unlink: resources = %v, want %v", got, want) + } + for _, resource := range adapter.GetAll() { + if resource.ID == nodeID && (resource.Agent != nil || resource.Identity.MachineID != "") { + t.Fatalf("after unlink: node = %+v, want no agent payload or borrowed machine key", resource) + } + } + assertOwnPins("after unlink") + }) + } +} + +// Earlier releases pinned a link's primary with the merged identity, so an +// agent primary's row can still hold the linked node's cluster. No resource +// holds that cluster with the agent's hostname, so the row keeps it, as an +// upsert keeps any stored field, and is not rewritten on every rebuild. +func TestManualLinkSideKeepsAnUnclaimedStoredField(t *testing.T) { + now := time.Now().UTC() + snapshot := manualLinkPinSnapshot(now, "lab-cluster") + snapshot.Nodes[0].Host = "https://192.0.2.10:8006" + agentID := MachineIdentityCanonicalID(ResourceTypeAgent, "machine-box") + store := &pinCountingStore{MemoryStore: NewMemoryStore()} + widened := ResourceIdentityPin{ + CanonicalID: agentID, ResourceType: ResourceTypeAgent, + MachineID: "machine-box", ClusterName: "lab-cluster", Hostname: "box-x", + } + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{widened}); err != nil { + t.Fatalf("seed widened pin: %v", err) + } + unlinked := NewMonitorAdapter(NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental(snapshot, nil) + var nodeID string + for _, resource := range unlinked.GetAll() { + if resource.Proxmox != nil { + nodeID = resource.ID + } + } + if nodeID == "" || nodeID == agentID { + t.Fatalf("unlinked estate = %v, want a node beside agent %s", resourceIDs(unlinked.GetAll()), agentID) + } + if err := store.AddLink(ResourceLink{ResourceA: agentID, ResourceB: nodeID, PrimaryID: agentID}); err != nil { + t.Fatalf("add link: %v", err) + } + + adapter := NewMonitorAdapter(NewRegistry(store)) + for step := 0; step < 3; step++ { + writesBefore := store.pinWrites + next := snapshot + next.LastUpdate = now.Add(time.Duration(step) * time.Second) + adapter.PopulateSnapshotAndSupplemental(next, nil) + if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{agentID}) { + t.Fatalf("rebuild %d: resources = %v, want only the linked agent %s", step, got, agentID) + } + if step > 0 && store.pinWrites != writesBefore { + t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", step, store.pinWrites-writesBefore) + } + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + for _, pin := range pins { + if pin.CanonicalID == agentID && pin != widened { + t.Fatalf("rebuild %d: agent pin = %+v, want the stored row kept %+v", step, pin, widened) + } + } + } +} + +// The store gives each strong key one owner. Two hosts can keep separate +// machine IDs yet share a pool name and hostname (an old and a replacement +// enrollment); linked, the live primary keeps that key and the folded side +// keeps only its machine key, rather than the two taking the pool from each +// other every rebuild. A key the primary's row only kept from an earlier +// release's merged pin goes to the folded side that reports it. +func TestManualLinkFoldedSideNeverTakesALiveResourcesKey(t *testing.T) { + newPin := ResourceIdentityPin{CanonicalID: "agent-new", ResourceType: ResourceTypeAgent, MachineID: "machine-new", ClusterName: "pool-a", Hostname: "xcp1"} + oldPin := ResourceIdentityPin{CanonicalID: "agent-old", ResourceType: ResourceTypeAgent, MachineID: "machine-old", ClusterName: "pool-a", Hostname: "xcp1"} + ownNewPin, ownOldPin := newPin, oldPin + ownNewPin.ClusterName, ownOldPin.ClusterName = "", "" + for _, tc := range []struct { + name string + primaryPool string + seed []ResourceIdentityPin + want []ResourceIdentityPin + }{ + {name: "both report the pool", primaryPool: "pool-a", want: []ResourceIdentityPin{newPin, ownOldPin}}, + {name: "primary row borrowed the pool", seed: []ResourceIdentityPin{newPin}, want: []ResourceIdentityPin{ownNewPin, oldPin}}, + } { + t.Run(tc.name, func(t *testing.T) { + now := time.Now().UTC() + host := func(id, machineID, pool string) Resource { + return Resource{ + ID: id, Type: ResourceTypeAgent, Name: "xcp1", Status: StatusOnline, LastSeen: now, + Sources: []DataSource{SourceAgent}, + Agent: &AgentData{AgentID: id, Hostname: "xcp1"}, + Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: []string{"xcp1"}}, + } + } + store := &pinCountingStore{MemoryStore: NewMemoryStore()} + if err := store.UpsertResourceIdentityPins(tc.seed); err != nil { + t.Fatalf("seed pins: %v", err) + } + if err := store.AddLink(ResourceLink{ResourceA: "agent-new", ResourceB: "agent-old", PrimaryID: "agent-new"}); err != nil { + t.Fatalf("add link: %v", err) + } + for rebuild := 1; rebuild <= 3; rebuild++ { + writesBefore := store.pinWrites + registry := NewRegistry(store) + registry.IngestResources([]Resource{host("agent-new", "machine-new", tc.primaryPool), host("agent-old", "machine-old", "pool-a")}) + registry.PersistIdentityPins() + if rebuild > 1 && store.pinWrites != writesBefore { + t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", rebuild, store.pinWrites-writesBefore) + } + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + slices.SortFunc(pins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) }) + if !slices.Equal(pins, tc.want) { + t.Fatalf("rebuild %d: pins = %+v, want %+v", rebuild, pins, tc.want) + } + } + }) + } +} + +// Two hosts can share a cloned machine ID while an exclusion keeps them +// apart. A folded side whose machine key the other host holds writes no pin +// rather than a cluster-only one: cluster-slot succession treats a +// machine-keyless row as a host a later machine in that slot may absorb. +func TestManualLinkFoldedSideWritesNoPinWithoutItsMachineKey(t *testing.T) { + now := time.Now().UTC() + host := func(id, machineID, pool, hostname string) Resource { + return Resource{ + ID: id, Type: ResourceTypeAgent, Name: hostname, Status: StatusOnline, LastSeen: now, + Sources: []DataSource{SourceAgent}, + Agent: &AgentData{AgentID: id, Hostname: hostname}, + Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: []string{hostname}}, + } + } + store := NewMemoryStore() + if err := store.AddLink(ResourceLink{ResourceA: "agent-primary", ResourceB: "agent-clone", PrimaryID: "agent-primary"}); err != nil { + t.Fatalf("add link: %v", err) + } + registry := NewRegistry(store) + registry.IngestResources([]Resource{ + host("agent-original", "machine-cloned", "", "swarm-a"), + host("agent-clone", "machine-cloned", "swarm", "swarm-b"), + host("agent-primary", "machine-primary", "", "primary"), + }) + registry.PersistIdentityPins() + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + for _, pin := range pins { + if pin.CanonicalID == "agent-clone" { + t.Fatalf("folded clone pinned as %+v, want no pin without its machine key", pin) + } + } + if len(pins) != 2 { + t.Fatalf("pins = %+v, want the original's and the primary's", pins) + } +} + +// A link side's row keeps a stored hostname its own pin lacks, but never +// where that hostname would complete a cluster key another resource holds: +// the store would delete that resource's whole row. +func TestManualLinkSideKeepsNoHostnameThatCompletesAHeldClusterKey(t *testing.T) { + now := time.Now().UTC() + host := func(id, machineID, pool string, hostnames ...string) Resource { + return Resource{ + ID: id, Type: ResourceTypeAgent, Name: id, Status: StatusOnline, LastSeen: now, + Sources: []DataSource{SourceAgent}, + Agent: &AgentData{AgentID: id}, + Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: hostnames}, + } + } + store := &pinCountingStore{MemoryStore: NewMemoryStore()} + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{ + CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", ClusterName: "pool-d", Hostname: "h1", + }}); err != nil { + t.Fatalf("seed pin: %v", err) + } + if err := store.AddLink(ResourceLink{ResourceA: "agent-a", ResourceB: "agent-f", PrimaryID: "agent-a"}); err != nil { + t.Fatalf("add link: %v", err) + } + want := []ResourceIdentityPin{ + {CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", ClusterName: "pool-c"}, + {CanonicalID: "agent-b", ResourceType: ResourceTypeAgent, MachineID: "machine-b", ClusterName: "pool-c", Hostname: "h1"}, + {CanonicalID: "agent-f", ResourceType: ResourceTypeAgent, MachineID: "machine-f", Hostname: "f1"}, + } + for rebuild := 1; rebuild <= 3; rebuild++ { + writesBefore := store.pinWrites + registry := NewRegistry(store) + registry.IngestResources([]Resource{ + host("agent-a", "machine-a", "pool-c"), + host("agent-b", "machine-b", "pool-c", "h1"), + host("agent-f", "machine-f", "", "f1"), + }) + registry.PersistIdentityPins() + if rebuild > 1 && store.pinWrites != writesBefore { + t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", rebuild, store.pinWrites-writesBefore) + } + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + slices.SortFunc(pins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) }) + if !slices.Equal(pins, want) { + t.Fatalf("rebuild %d: pins = %+v, want %+v", rebuild, pins, want) + } + } +} + +// A folded side's own pin can be narrower than a row an earlier release +// widened with a pool its primary now reports. The primary's new pin claims +// that pool and hostname, so the store deletes the folded side's whole row; +// the folded side's own pin is written after it, so the side stays pinned. +func TestManualLinkFoldedSideKeepsItsPinWhenAPrimaryClaimsItsWiderRow(t *testing.T) { + now := time.Now().UTC() + host := func(id, machineID, pool string) Resource { + return Resource{ + ID: id, Type: ResourceTypeAgent, Name: "xcp1", Status: StatusOnline, LastSeen: now, + Sources: []DataSource{SourceAgent}, + Agent: &AgentData{AgentID: id, Hostname: "xcp1"}, + Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: []string{"xcp1"}}, + } + } + store := &pinCountingStore{MemoryStore: NewMemoryStore()} + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{ + CanonicalID: "agent-old", ResourceType: ResourceTypeAgent, + MachineID: "machine-old", ClusterName: "pool-a", Hostname: "xcp1", + }}); err != nil { + t.Fatalf("seed widened pin: %v", err) + } + if err := store.AddLink(ResourceLink{ResourceA: "agent-new", ResourceB: "agent-old", PrimaryID: "agent-new"}); err != nil { + t.Fatalf("add link: %v", err) + } + want := []ResourceIdentityPin{ + {CanonicalID: "agent-new", ResourceType: ResourceTypeAgent, MachineID: "machine-new", ClusterName: "pool-a", Hostname: "xcp1"}, + {CanonicalID: "agent-old", ResourceType: ResourceTypeAgent, MachineID: "machine-old", Hostname: "xcp1"}, + } + for rebuild := 1; rebuild <= 3; rebuild++ { + writesBefore := store.pinWrites + registry := NewRegistry(store) + registry.IngestResources([]Resource{host("agent-new", "machine-new", "pool-a"), host("agent-old", "machine-old", "")}) + registry.PersistIdentityPins() + if rebuild > 1 && store.pinWrites != writesBefore { + t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", rebuild, store.pinWrites-writesBefore) + } + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + slices.SortFunc(pins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) }) + if !slices.Equal(pins, want) { + t.Fatalf("rebuild %d: pins = %+v, want %+v", rebuild, pins, want) + } + } +} + +// An earlier release pinned a linked node primary with the agent's machine +// key and dropped the agent's own pin. The node has nothing of its own to +// pin, so the agent's own pin takes the key back on the next rebuild. +func TestManualLinkReturnsABorrowedMachineKeyToItsAgent(t *testing.T) { + now := time.Now().UTC() + snapshot := manualLinkPinSnapshot(now, "") + snapshot.Nodes[0].Host = "https://192.0.2.10:8006" + unlinked := NewMonitorAdapter(NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental(snapshot, nil) + var nodeID string + for _, resource := range unlinked.GetAll() { + if resource.Proxmox != nil { + nodeID = resource.ID + } + } + agentID := MachineIdentityCanonicalID(ResourceTypeAgent, "machine-box") + if nodeID == "" || nodeID == agentID { + t.Fatalf("unlinked estate = %v, want a node beside agent %s", resourceIDs(unlinked.GetAll()), agentID) + } + + store := &pinCountingStore{MemoryStore: NewMemoryStore()} + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{ + CanonicalID: nodeID, ResourceType: ResourceTypeAgent, MachineID: "machine-box", Hostname: "pve1", + }}); err != nil { + t.Fatalf("seed borrowed pin: %v", err) + } + if err := store.AddLink(ResourceLink{ResourceA: nodeID, ResourceB: agentID, PrimaryID: nodeID}); err != nil { + t.Fatalf("add link: %v", err) + } + adapter := NewMonitorAdapter(NewRegistry(store)) + for step := 0; step < 2; step++ { + next := snapshot + next.LastUpdate = now.Add(time.Duration(step) * time.Second) + adapter.PopulateSnapshotAndSupplemental(next, nil) + if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{nodeID}) { + t.Fatalf("rebuild %d: resources = %v, want only the linked node %s", step, got, nodeID) + } + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + if len(pins) != 1 || pins[0].CanonicalID != agentID || pins[0].MachineID != "machine-box" || pins[0].Hostname != "box-x" { + t.Fatalf("rebuild %d: pins = %+v, want only the agent's own pin", step, pins) + } + } + links, err := store.GetLinks() + if err != nil { + t.Fatalf("get links: %v", err) + } + if len(links) != 1 || links[0].ResourceA != nodeID || links[0].ResourceB != agentID { + t.Fatalf("links = %+v, want %s -> %s kept", links, nodeID, agentID) + } +} + // A record may name an older canonical ID as superseded on every ingest. When // that ID belongs to a resource a manual link folded into the record's own // resource, it is still observed: succeeding it on a later refresh would diff --git a/internal/unifiedresources/store.go b/internal/unifiedresources/store.go index 06b03e9332..b61a6e0833 100644 --- a/internal/unifiedresources/store.go +++ b/internal/unifiedresources/store.go @@ -29,6 +29,10 @@ type ResourceStore interface { // Identity pins keep canonical IDs for merged-source hosts stable across // restarts. See ResourceIdentityPin. UpsertResourceIdentityPins(pins []ResourceIdentityPin) error + // ReplaceResourceIdentityPins writes each pin as its row's whole + // content, clearing fields the pin leaves empty, where an upsert keeps + // them. Strong-key conflicts resolve as for an upsert. + ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error ListResourceIdentityPins() ([]ResourceIdentityPin, error) RecordChange(change ResourceChange) error ResourceHistoryIDs(resourceID string) ([]string, error) @@ -1521,9 +1525,39 @@ func (s *SQLiteResourceStore) Close() error { } func (s *SQLiteResourceStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error { + return s.writeResourceIdentityPins(pins, false) +} + +func (s *SQLiteResourceStore) ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error { + return s.writeResourceIdentityPins(pins, true) +} + +// writeResourceIdentityPins upserts pins. An upsert keeps a stored field the +// pin leaves empty; a replace writes the pin as the whole row. +func (s *SQLiteResourceStore) writeResourceIdentityPins(pins []ResourceIdentityPin, replace bool) error { if len(pins) == 0 { return nil } + write := `INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(canonical_id) DO UPDATE SET + resource_type = excluded.resource_type, + machine_id = COALESCE(excluded.machine_id, resource_identities.machine_id), + dmi_uuid = COALESCE(excluded.dmi_uuid, resource_identities.dmi_uuid), + cluster_name = COALESCE(NULLIF(excluded.cluster_name, ''), resource_identities.cluster_name), + primary_hostname = COALESCE(NULLIF(excluded.primary_hostname, ''), resource_identities.primary_hostname), + updated_at = CURRENT_TIMESTAMP` + if replace { + write = `INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(canonical_id) DO UPDATE SET + resource_type = excluded.resource_type, + machine_id = excluded.machine_id, + dmi_uuid = excluded.dmi_uuid, + cluster_name = excluded.cluster_name, + primary_hostname = excluded.primary_hostname, + updated_at = CURRENT_TIMESTAMP` + } s.mu.Lock() defer s.mu.Unlock() @@ -1560,15 +1594,7 @@ func (s *SQLiteResourceStore) UpsertResourceIdentityPins(pins []ResourceIdentity ); err != nil { return fmt.Errorf("clear conflicting identity pins for %q: %w", pin.CanonicalID, err) } - if _, err := tx.Exec(`INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname) - VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT(canonical_id) DO UPDATE SET - resource_type = excluded.resource_type, - machine_id = COALESCE(excluded.machine_id, resource_identities.machine_id), - dmi_uuid = COALESCE(excluded.dmi_uuid, resource_identities.dmi_uuid), - cluster_name = COALESCE(NULLIF(excluded.cluster_name, ''), resource_identities.cluster_name), - primary_hostname = COALESCE(NULLIF(excluded.primary_hostname, ''), resource_identities.primary_hostname), - updated_at = CURRENT_TIMESTAMP`, + if _, err := tx.Exec(write, pin.CanonicalID, string(pin.ResourceType), nullIfEmptyArg(pin.MachineID), @@ -3413,6 +3439,14 @@ func NewMemoryStore() *MemoryStore { } func (m *MemoryStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error { + return m.writeResourceIdentityPins(pins, false) +} + +func (m *MemoryStore) ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error { + return m.writeResourceIdentityPins(pins, true) +} + +func (m *MemoryStore) writeResourceIdentityPins(pins []ResourceIdentityPin, replace bool) error { m.mu.Lock() defer m.mu.Unlock() for _, pin := range pins { @@ -3432,7 +3466,7 @@ func (m *MemoryStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) err delete(m.identityPins, canonicalID) } } - if existing, ok := m.identityPins[pin.CanonicalID]; ok { + if existing, ok := m.identityPins[pin.CanonicalID]; ok && !replace { if pin.MachineID == "" { pin.MachineID = existing.MachineID } diff --git a/internal/unifiedresources/store_test.go b/internal/unifiedresources/store_test.go index 92ec4d333c..fdf9fb2be3 100644 --- a/internal/unifiedresources/store_test.go +++ b/internal/unifiedresources/store_test.go @@ -2863,6 +2863,66 @@ func TestSQLiteResourceStore_ResourceOperatorState_RoundTrips(t *testing.T) { } } +// An upsert keeps a stored field the pin leaves empty; a replace writes the +// pin as the whole row. Both give a strong key to the pin that carries it. +func TestResourceIdentityPinReplaceClearsFieldsUpsertKeeps(t *testing.T) { + sqliteStore, err := NewSQLiteResourceStore(t.TempDir(), "") + if err != nil { + t.Fatalf("open sqlite store: %v", err) + } + defer sqliteStore.Close() + + wide := ResourceIdentityPin{CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", DMIUUID: "dmi-a", ClusterName: "pool-a", Hostname: "host-a"} + narrow := ResourceIdentityPin{CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", Hostname: "host-a.lan"} + rival := ResourceIdentityPin{CanonicalID: "agent-b", ResourceType: ResourceTypeAgent, MachineID: "machine-b", ClusterName: "pool-a", Hostname: "host-a"} + for name, store := range map[string]ResourceStore{"memory": NewMemoryStore(), "sqlite": sqliteStore} { + t.Run(name, func(t *testing.T) { + pinFor := func(id string) (ResourceIdentityPin, bool) { + t.Helper() + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list pins: %v", err) + } + for _, pin := range pins { + if pin.CanonicalID == id { + return pin, true + } + } + return ResourceIdentityPin{}, false + } + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{wide}); err != nil { + t.Fatalf("upsert wide pin: %v", err) + } + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{narrow}); err != nil { + t.Fatalf("upsert narrow pin: %v", err) + } + kept := wide + kept.Hostname = narrow.Hostname + if got, _ := pinFor("agent-a"); got != kept { + t.Fatalf("after upsert: pin = %+v, want stored fields kept %+v", got, kept) + } + if err := store.ReplaceResourceIdentityPins([]ResourceIdentityPin{narrow}); err != nil { + t.Fatalf("replace narrow pin: %v", err) + } + if got, _ := pinFor("agent-a"); got != narrow { + t.Fatalf("after replace: pin = %+v, want exactly %+v", got, narrow) + } + if err := store.ReplaceResourceIdentityPins([]ResourceIdentityPin{wide}); err != nil { + t.Fatalf("replace wide pin: %v", err) + } + if err := store.ReplaceResourceIdentityPins([]ResourceIdentityPin{rival}); err != nil { + t.Fatalf("replace rival pin: %v", err) + } + if _, ok := pinFor("agent-a"); ok { + t.Fatalf("a replace claiming pool-a/host-a must delete the row holding that key") + } + if got, _ := pinFor("agent-b"); got != rival { + t.Fatalf("rival pin = %+v, want %+v", got, rival) + } + }) + } +} + func TestResourceChangeReadsMergeCanonicalIDEras(t *testing.T) { const machineID = "7d465a78-test-machine-id" steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID)