Split a manually linked pair when the API unlinks it

POST /api/resources/{id}/unlink recorded only an exclusion. The link row
stayed, and applyManualLinks never consults exclusions, so the monitor's
rebuilds, the resources API and the websocket kept the pair folded while
the API answered "Resources unlinked".

A pair now carries one operator decision. AddExclusion deletes the pair's
links and AddLink its exclusions, in one transaction and in either ID
order. Reads keep each pair's latest decision, so rows written before
this change, or rewritten onto one pair by canonical-ID succession,
resolve by their later write, and a registry resolves its two reads
together so an unlink landing between them cannot load both.

While a Proxmox node was linked to an agent, the merged resource's
identity pin could take the agent's machine key. After unlink the node
completed that key, minted the agent's ID and folded into it by ID, where
no exclusion check runs. A pin no longer completes a record the operator
split from the pinned resource.

The broadcast coalesced host views without exclusions while REST's
ListForPresentation honoured them, so a split host pair was one websocket
row and two REST rows. The broadcast now coalesces through the read state
that listed the rows, with the same exclusion filter.
This commit is contained in:
rcourtman
2026-10-07 23:36:39 +01:00
parent 1a094646d7
commit 58ee64fc1c
15 changed files with 1028 additions and 30 deletions
@@ -8998,3 +8998,15 @@ No agent registration, enrolment, install, update or removal path changed.
for every method. No agent route lives under `/debug/`, so agent registration,
enrolment, install, update, removal, report ingest and `/api/agent/ws`
handshakes are unchanged on a demo instance.
### Broadcast host coalescing honours operator splits
`internal/monitoring/monitor.go` changed only so the websocket broadcast
coalesces host views with the operator's merge exclusions
(`MonitorAdapter.CoalesceForPresentation`), as the resources API's
`ListForPresentation` already did, so an agent host the operator split from
a same-named platform view by unlink or report-merge is coalesced on the
websocket as REST coalesces it (the filter's limits are recorded under
"Unlink replaces the pair's operator link" in the unified-resources
contract). Agent registration, enrolment, install, update, removal, report
identity and continuity are unchanged.
@@ -5826,3 +5826,24 @@ that actually accepted the same occurrence. The connected
HTTP 200/503 split, reopens the persistent queue, and requires the old accepted
destination's recovery plus the new firing, with no recovery to the unannounced
destination. Quiet hours and disabled recovery controls still apply.
### Broadcast host coalescing honours operator splits
The websocket broadcast coalesced top-level host views with
`CoalescePresentationHostResources`, which applies no merge exclusions, while
the resources API's `ResourceRegistry.ListForPresentation` applies the
registry's. A host pair the operator split (unlink or report-merge) could
therefore be one broadcast row and two REST rows.
`Monitor.buildBroadcastFrontendStateFromSnapshotWithClock` now calls
`coalesceResourcesForPresentation`, which asks the view's read state for
`MonitorAdapter.CoalesceForPresentation` when its registry is store-backed
(the resource store's adapter, or a host-continuity overlay that loaded the
store's exclusions afresh), and otherwise the resource store's adapter: the
mock-mode view lists through a store-less registry that carries no operator
decisions. Both surfaces share `presentationExclusionFilter`. Polling, rebuild cadence,
alert evaluation and the broadcast payload shape are unchanged.
`TestBroadcastPresentationCoalesceHonoursMergeExclusions` compares broadcast
and REST row counts with and without the split, and
`TestBroadcastPresentationCoalescePrefersTheListingReadState` pins the
listing read state over an older store generation and the store's exclusions
for a mock-style view.
@@ -6852,3 +6852,90 @@ above the hosts table without its filter toolbar. Nothing renders when
vCenter reports no signals. The Overview's own model, workload snapshot and
navigation facets still come from the Overview query; the Health tab keeps
the filterable table.
### Unlink replaces the pair's operator link
`POST /api/resources/{id}/link` records a `ResourceLink`; unlink and
report-merge record a `ResourceExclusion`. An operator decides each resource
pair one way, and the later decision replaces the earlier: the store's
`AddExclusion` deletes the pair's link rows and `AddLink` deletes its
exclusion rows, in the same transaction as the write, matching both ID
orders because canonical-ID succession can store either. Unlink used to add
only the exclusion. The link row survived, and `applyManualLinks`, which
never consults exclusions, kept the pair folded in the monitor's rebuilds and
in the resources API's registry, so the agent facet stayed on the VM and the
agent row stayed hidden although the API answered "Resources unlinked".
`GetLinks` and `GetExclusions` return only each pair's latest decision
(`effectiveManualPairDecisions` in `manual_link_decisions.go`), so a store
that still holds both rows for a pair, written before this change or
rewritten onto one pair by succession, reads its later write, and a tie
reads as apart. Every registry loads its links and exclusions from its store
when it is constructed, so the monitor's rebuild and the resources API's
store-backed registry apply one decision with no exclusion check in
`applyManualLinks`. Unlink keeps the exclusion, so identity matching and
presentation coalescing also keep a pair apart that would otherwise match
automatically.
Identity pins honour the exclusion too. While a Proxmox node was linked to a
standalone agent, the merged resource's pin could record the node's endpoint
hostname with the agent's machine key. After unlink the node completed that
key in `completeIdentityFromPins`, minted the agent's machine-derived ID and
folded into the agent by ID, a path no exclusion check sees, so the pair
stayed merged with the link row gone. A pin whose canonical ID the operator
excluded from the ID the record takes without the pin (its `chooseNewID` or
source-specific ID) now completes nothing. This covers pins already written
that way, not only new ones. Merges through an agent's own declared node link
(`resolveLinkedResource`) still ignore exclusions.
The resources API seeds its registry from the monitor's listing, which
already has the link applied, so it cannot split the pair itself: an unlink
shows on REST, `/api/state` and the websocket from the monitor's next
registry rebuild, the same generation in which a new link reaches the
broadcast. REST applies a new link at once, because its own registry applies
the store's links on top of the seed. The mock-mode view applies no operator
links, so in mock mode REST splits a manually linked pair as soon as the
unlink is recorded. Mock fixtures that identity matching merged stay merged
on every surface: the fixture graph is built without the store's exclusions.
The websocket broadcast coalesced host views with
`CoalescePresentationHostResources`, which honours no exclusions, while the
resources API's `ListForPresentation` honours the registry's, so a host pair
the operator split could be one broadcast row and two REST rows. The
broadcast now coalesces through `Monitor.coalesceResourcesForPresentation`,
which asks the read state that listed the rows for
`MonitorAdapter.CoalesceForPresentation` when its registry is store-backed
(the resource store's adapter, or a host-continuity overlay that loaded the
store's exclusions afresh). The mock view and other views built from an
already-unified list sit on a store-less registry, which declines, so the
resource store's adapter supplies the exclusions instead. Either applies its
current registry's exclusions through the same `presentationExclusionFilter`
that `ListForPresentation` uses; a rebuild published between listing the
rows and coalescing them is judged by the newer generation's exclusions for
that one broadcast. Both surfaces share one known limit: the filter compares
the IDs of the rows being merged, so a third same-host fragment that merges
first can carry an excluded pair into one row.
A write leaves one row for its pair: `AddLink` and `AddExclusion` also delete
a row of their own kind that succession stored in the other order, and the
in-memory store replaces a pair's rows on every write and stamps `CreatedAt`
as SQLite does. Reads keep the newest link per pair, so an older reversed copy
cannot win with its primary, and drop exclusions older than it; duplicate
exclusions already stored stay, which changes no decision. A registry resolves its `GetLinks` and `GetExclusions` results
together in `loadOverrides`, so an unlink committed between the two reads
cannot load both decisions for one generation. Report-merge still records
exclusions against candidate IDs derived from the merged resource's type,
which do not name the folded side of a cross-type manual link, so it does not
undo such a link; unlink does. `TestManualPairDecisionReplacesThePreviousOne` and
`TestManualPairDecisionReadsResolveStoredConflictsByRecency` pin the store
rule on SQLite and the in-memory store,
`TestUnlinkSplitsHostPairWhosePinTookTheOtherSidesKey` links and unlinks
node and agent pairs (IP and FQDN endpoints, either side primary) through
three monitor rebuilds per step,
`TestRegistryResolvesDecisionsAcrossItsTwoReads` loads a link read before an
unlink and the exclusion read after it,
`TestResourceUnlinkSplitsManuallyLinkedPair` drives link, unlink and relink
through the API handlers against a monitor adapter rebuilt twice per step,
and `TestMonitorAdapterCoalesceForPresentationHonoursExclusions`,
`TestBroadcastPresentationCoalesceHonoursMergeExclusions` and
`TestBroadcastPresentationCoalescePrefersTheListingReadState` pin broadcast
and REST row counts for a split host pair.
@@ -0,0 +1,144 @@
package resourceapi
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
// monitorAdapterSeedProvider seeds the resources API from a monitor adapter's
// current generation, as the monitor does in production: REST receives the
// already-linked list and applies the store's decisions on top.
type monitorAdapterSeedProvider struct {
adapter *unified.MonitorAdapter
snapshot models.StateSnapshot
}
func (p *monitorAdapterSeedProvider) ReadSnapshot() models.StateSnapshot { return p.snapshot }
func (p *monitorAdapterSeedProvider) UnifiedResourceSnapshot() ([]unified.Resource, time.Time) {
return p.adapter.GetAll(), p.adapter.LastRebuiltAt()
}
// Unlinking a manually linked pair splits it on every surface. The link row
// used to survive the unlink, so the agent stayed folded into the VM in the
// monitor's rebuilds and in REST although the API answered "Resources
// unlinked". A later link joins the pair again: each decision replaces the
// previous one for that pair.
func TestResourceUnlinkSplitsManuallyLinkedPair(t *testing.T) {
now := time.Now().UTC()
h := NewQueryService(&config.Config{DataPath: t.TempDir()})
store, err := h.getStore("default")
if err != nil {
t.Fatalf("getStore: %v", err)
}
snapshot := models.StateSnapshot{
LastUpdate: now,
VMs: []models.VM{{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now}},
Hosts: []models.Host{{ID: "host-app", Hostname: "app-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now}},
}
adapter := unified.NewMonitorAdapter(unified.NewRegistry(store))
h.SetStateProvider(&monitorAdapterSeedProvider{adapter: adapter, snapshot: snapshot})
// Identity pins persist on the first rebuild and steer the second, so a
// decision only counts once it survives two generations.
rebuild := func() {
t.Helper()
adapter.PopulateFromSnapshot(snapshot)
adapter.PopulateFromSnapshot(snapshot)
}
rebuild()
var vmID, agentID string
for _, resource := range adapter.GetAll() {
switch resource.Type {
case unified.ResourceTypeVM:
vmID = resource.ID
case unified.ResourceTypeAgent:
agentID = resource.ID
}
}
if vmID == "" || agentID == "" {
t.Fatalf("fixture did not produce a standalone VM and agent: %+v", adapter.GetAll())
}
post := func(action string) {
t.Helper()
body, _ := json.Marshal(map[string]string{"targetId": agentID})
rec := httptest.NewRecorder()
h.HandleResourceRoutes(rec, httptest.NewRequest(http.MethodPost, "/api/resources/"+vmID+"/"+action, bytes.NewReader(body)))
if rec.Code != http.StatusOK {
t.Fatalf("%s status = %d body=%s", action, rec.Code, rec.Body.String())
}
}
get := func(id string) (unified.Resource, bool) {
t.Helper()
rec := httptest.NewRecorder()
h.HandleResourceRoutes(rec, httptest.NewRequest(http.MethodGet, "/api/resources/"+id, nil))
if rec.Code == http.StatusNotFound {
return unified.Resource{}, false
}
if rec.Code != http.StatusOK {
t.Fatalf("get %s status = %d body=%s", id, rec.Code, rec.Body.String())
}
var resource unified.Resource
if err := json.NewDecoder(rec.Body).Decode(&resource); err != nil {
t.Fatalf("decode %s: %v", id, err)
}
return resource, true
}
assertLinked := func(step string, wantLinked bool) {
t.Helper()
vm, ok := get(vmID)
if !ok {
t.Fatalf("%s: VM %s missing from REST", step, vmID)
}
_, agentListed := get(agentID)
if wantLinked && (vm.Agent == nil || agentListed) {
t.Fatalf("%s: want agent folded into the VM, VM agent facet=%v agent row listed=%v", step, vm.Agent != nil, agentListed)
}
if !wantLinked && (vm.Agent != nil || !agentListed) {
t.Fatalf("%s: want VM and agent apart, VM agent facet=%v agent row listed=%v", step, vm.Agent != nil, agentListed)
}
var monitorVM *unified.Resource
monitorAgentListed := false
for _, resource := range adapter.GetAll() {
resource := resource
switch resource.ID {
case vmID:
monitorVM = &resource
case agentID:
monitorAgentListed = true
}
}
if monitorVM == nil || (monitorVM.Agent != nil) != wantLinked || monitorAgentListed == wantLinked {
t.Fatalf("%s: monitor disagrees with REST: VM present=%v agent listed=%v, want linked=%v", step, monitorVM != nil, monitorAgentListed, wantLinked)
}
}
post("link")
rebuild()
assertLinked("after link", true)
post("unlink")
rebuild()
assertLinked("after unlink", false)
links, err := store.GetLinks()
if err != nil || len(links) != 0 {
t.Fatalf("unlink left links %+v (err %v)", links, err)
}
post("link")
rebuild()
assertLinked("after relink", true)
exclusions, err := store.GetExclusions()
if err != nil || len(exclusions) != 0 {
t.Fatalf("relink left exclusions %+v (err %v)", exclusions, err)
}
}
@@ -650,7 +650,8 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) {
for _, snippet := range []string{
"metricsTargetResolver := unifiedView.metricsTargets",
"metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)",
"broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)",
"broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources)",
"if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {",
"broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)",
"attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)",
"frontendState.CapabilityCatalog = broadcastCatalogs.capabilities",
+34 -1
View File
@@ -4478,7 +4478,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot mode
if metricsTargetResolver == nil {
metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)
}
broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)
broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources)
// Coalescing owns the outer slice. Decorate that one projection in place,
// not three full-resource copies; nested store data is still read-only.
healthAlerts := resourceHealthAlerts(frontendState.ActiveAlerts)
@@ -4501,6 +4501,39 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot mode
return frontendState
}
// presentationCoalescingResourceStore is implemented by resource stores whose
// registry can hold the operator's merge exclusions (MonitorAdapter).
type presentationCoalescingResourceStore interface {
CoalesceForPresentation(resources []unifiedresources.Resource) ([]unifiedresources.Resource, bool)
}
// coalesceResourcesForPresentation applies the presentation host coalesce with
// the operator's merge exclusions, which the resources API's
// ListForPresentation honours too: a host pair the operator split must not
// show as one broadcast row while REST lists two. The read state that listed
// the rows supplies them when its registry is store-backed, as the resource
// store's adapter and a host-continuity overlay are; the mock view and other
// views built from an unified list carry none, so the resource store's
// adapter supplies them instead.
func (m *Monitor) coalesceResourcesForPresentation(readState unifiedresources.ReadState, resources []unifiedresources.Resource) []unifiedresources.Resource {
if coalescer, ok := readState.(presentationCoalescingResourceStore); ok {
if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {
return coalesced
}
}
if m != nil {
m.mu.RLock()
store := m.resourceStore
m.mu.RUnlock()
if coalescer, ok := store.(presentationCoalescingResourceStore); ok {
if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {
return coalesced
}
}
}
return unifiedresources.CoalescePresentationHostResources(resources)
}
// GetLiveStateSnapshot returns the underlying monitor state snapshot without
// applying global mock mode overrides.
//
@@ -6988,6 +6988,113 @@ func TestBroadcastProjectionListsRegistryOnceAndKeepsLiveChanges(t *testing.T) {
}
}
// presentationFixtureStore keeps the fixture registry: the broadcast's read
// refresh would otherwise rebuild it from the empty snapshot.
type presentationFixtureStore struct {
*unifiedresources.MonitorAdapter
}
func (presentationFixtureStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool {
return false
}
// A host pair the operator split (unlink or report-merge) must be two
// broadcast rows exactly when the resources API lists two. The broadcast used
// to coalesce without the store's exclusions, so the websocket folded the
// pair the API kept apart.
func TestBroadcastPresentationCoalesceHonoursMergeExclusions(t *testing.T) {
now := time.Date(2026, 10, 7, 21, 0, 0, 0, time.UTC)
resources := []unifiedresources.Resource{
{
ID: "agent-runtime-alpha", Type: unifiedresources.ResourceTypeAgent, Name: "alpha",
Status: unifiedresources.StatusOnline, LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent},
Identity: unifiedresources.ResourceIdentity{Hostnames: []string{"alpha"}},
Agent: &unifiedresources.AgentData{AgentID: "agent-alpha", Hostname: "alpha"},
},
{
ID: "agent-docker-alpha", Type: unifiedresources.ResourceTypeAgent, Name: "alpha",
Status: unifiedresources.StatusOnline, LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceDocker},
Identity: unifiedresources.ResourceIdentity{Hostnames: []string{"alpha"}},
Docker: &unifiedresources.DockerData{Hostname: "alpha"},
},
}
for _, split := range []bool{false, true} {
store := unifiedresources.NewMemoryStore()
if split {
if err := store.AddExclusion(unifiedresources.ResourceExclusion{ResourceA: "agent-runtime-alpha", ResourceB: "agent-docker-alpha"}); err != nil {
t.Fatal(err)
}
}
registry := unifiedresources.NewRegistry(store)
registry.IngestResources(resources)
adapter := unifiedresources.NewMonitorAdapter(registry)
m := &Monitor{resourceStore: presentationFixtureStore{adapter}}
state := m.buildBroadcastFrontendStateFromSnapshotWithClock(models.EmptyStateSnapshot(), m.mockModeFence.begin(), func() time.Time { return now })
broadcastRows := 0
for _, resource := range state.Resources {
if resource.Name == "alpha" {
broadcastRows++
}
}
// The resources API seeds a store-backed registry from the monitor's
// listing and presents it through ListForPresentation.
rest := unifiedresources.NewRegistry(store)
rest.IngestResources(adapter.GetAll())
restRows := len(rest.ListForPresentation())
want := 1
if split {
want = 2
}
if broadcastRows != want || restRows != want {
t.Fatalf("split=%v: broadcast rows=%d REST rows=%d, want %d on both", split, broadcastRows, restRows, want)
}
}
}
// The rows a broadcast coalesces were listed by its view's read state, such as
// a host-continuity overlay that loaded the store's exclusions afresh, so its
// exclusions decide even when the resource store's generation predates them.
func TestBroadcastPresentationCoalescePrefersTheListingReadState(t *testing.T) {
now := time.Date(2026, 10, 7, 21, 0, 0, 0, time.UTC)
resources := []unifiedresources.Resource{
{ID: "agent-runtime-alpha", Type: unifiedresources.ResourceTypeAgent, Name: "alpha", Status: unifiedresources.StatusOnline, LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent}, Identity: unifiedresources.ResourceIdentity{Hostnames: []string{"alpha"}},
Agent: &unifiedresources.AgentData{AgentID: "agent-alpha", Hostname: "alpha"}},
{ID: "agent-docker-alpha", Type: unifiedresources.ResourceTypeAgent, Name: "alpha", Status: unifiedresources.StatusOnline, LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceDocker}, Identity: unifiedresources.ResourceIdentity{Hostnames: []string{"alpha"}},
Docker: &unifiedresources.DockerData{Hostname: "alpha"}},
}
store := unifiedresources.NewMemoryStore()
base := unifiedresources.NewRegistry(store)
base.IngestResources(resources)
m := &Monitor{resourceStore: unifiedresources.NewMonitorAdapter(base)}
if err := store.AddExclusion(unifiedresources.ResourceExclusion{ResourceA: "agent-runtime-alpha", ResourceB: "agent-docker-alpha"}); err != nil {
t.Fatal(err)
}
overlay := unifiedresources.NewRegistry(store)
overlay.IngestResources(resources)
listing := unifiedresources.NewMonitorAdapter(overlay)
if got := len(m.coalesceResourcesForPresentation(listing, listing.GetAll())); got != 2 {
t.Fatalf("coalesced %d rows with the listing's exclusion, want 2", got)
}
if got := len(m.coalesceResourcesForPresentation(nil, listing.GetAll())); got != 1 {
t.Fatalf("without a listing read state the store generation decides: %d rows, want 1", got)
}
// The mock view lists through a store-less registry, which carries no
// operator decisions, so the resource store's adapter supplies them.
mockView := monitorUnifiedStateViewFromResources(resources, now)
withSplit := &Monitor{resourceStore: listing}
if got := len(withSplit.coalesceResourcesForPresentation(mockView.readState, mockView.resources)); got != 2 {
t.Fatalf("mock-style view coalesced %d rows, want the store's split to keep 2", got)
}
}
// A host agent linked to a Proxmox node merges into one read-state row that
// every PVE poll keeps online and fresh. Once the agent stops reporting, its
// retained filesystem summary must stop standing in for the node's disk (and
+12 -1
View File
@@ -239,7 +239,12 @@ func (rr *ResourceRegistry) loadIdentityPins() {
// knows cluster+hostname) still derives the same canonical ID as a steady
// state rebuild that knows the machine ID. Only missing fields are filled; an
// incoming identity that already carries a machine ID is never overridden.
func (rr *ResourceRegistry) completeIdentityFromPins(source DataSource, resource Resource, identity ResourceIdentity) ResourceIdentity {
// A pin of a resource the operator split from this record (unlink or
// report-merge) completes nothing: the record would mint the pinned
// resource's machine-derived ID and fold into it by ID, which no exclusion
// check sees. A node pin that took a linked agent's machine key kept the pair
// merged after unlink that way.
func (rr *ResourceRegistry) completeIdentityFromPins(source DataSource, sourceID string, resource Resource, identity ResourceIdentity) ResourceIdentity {
if CanonicalResourceType(resource.Type) != ResourceTypeAgent {
return identity
}
@@ -256,6 +261,12 @@ func (rr *ResourceRegistry) completeIdentityFromPins(source DataSource, resource
if !ok {
return identity
}
// The exclusion names the IDs the record holds without the pin.
if len(rr.exclusions) > 0 &&
(rr.isExcluded(pin.CanonicalID, rr.chooseNewID(resource.Type, identity, source, sourceID)) ||
rr.isExcluded(pin.CanonicalID, rr.sourceSpecificID(resource.Type, source, sourceID))) {
return identity
}
if strings.TrimSpace(identity.MachineID) == "" {
identity.MachineID = pin.MachineID
}
@@ -1,6 +1,11 @@
package unifiedresources
import "testing"
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
// The #1559 shape: several standalone agents whose hostnames share a short
// name (cloud.rnd-lax1, cloud.gce-or1, cloud.dmi-lax1). The pin index must
@@ -74,3 +79,85 @@ func TestIdentityPinIndexLegacyCollapsedPinStillMatches(t *testing.T) {
t.Fatalf("legacy collapsed pin should match its host's full hostname, got ok=%v pin=%q", ok, pin.CanonicalID)
}
}
// A Proxmox node linked to a standalone agent: while linked, the merged
// resource's identity pin recorded the node's endpoint hostname with the
// agent's machine key. After unlink the node completed that key from the pin,
// minted the agent's machine-derived ID and folded into it by ID before any
// exclusion check ran, so the pair stayed merged. Each step is judged after
// three monitor rebuilds, since pins persist on one rebuild and steer the next.
func TestUnlinkSplitsHostPairWhosePinTookTheOtherSidesKey(t *testing.T) {
now := time.Now().UTC()
for _, tc := range []struct {
name string
nodeEndpoint string
agentPrimary bool
}{
{"agent-primary", "https://10.0.0.5:8006", true},
{"node-primary", "https://10.0.0.5:8006", false},
{"agent-primary-fqdn", "https://pve1.example.lan:8006", true},
{"node-primary-fqdn", "https://pve1.example.lan:8006", false},
} {
t.Run(tc.name, func(t *testing.T) {
store, err := NewSQLiteResourceStore(t.TempDir(), "default")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = store.Close() })
snapshot := models.StateSnapshot{
LastUpdate: now,
Nodes: []models.Node{{ID: "lab-pve1", Name: "pve1", Instance: "lab", Host: tc.nodeEndpoint, Status: "online", LastSeen: now}},
Hosts: []models.Host{{ID: "host-box", Hostname: "box-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now}},
}
adapter := NewMonitorAdapter(NewRegistry(store))
rebuild := func() []Resource {
for i := 0; i < 3; i++ {
adapter.PopulateFromSnapshot(snapshot)
}
return adapter.GetAll()
}
var nodeID, agentID string
for _, resource := range rebuild() {
switch {
case resource.Proxmox != nil && resource.Agent == nil:
nodeID = resource.ID
case resource.Agent != nil && resource.Proxmox == nil:
agentID = resource.ID
}
}
if nodeID == "" || agentID == "" {
t.Fatalf("fixture did not produce a separate node and agent: %+v", adapter.GetAll())
}
primaryID, otherID := nodeID, agentID
if tc.agentPrimary {
primaryID, otherID = agentID, nodeID
}
if err := store.AddLink(ResourceLink{ResourceA: primaryID, ResourceB: otherID, PrimaryID: primaryID, CreatedAt: time.Now().UTC()}); err != nil {
t.Fatal(err)
}
linked := rebuild()
if len(linked) != 1 {
t.Fatalf("link did not merge the pair: %d resources", len(linked))
}
// Unlink as the API is called: on the merged resource, naming the
// side that disappeared into it.
mergedID := linked[0].ID
splitID := agentID
if mergedID == agentID {
splitID = nodeID
}
if err := store.AddExclusion(ResourceExclusion{ResourceA: mergedID, ResourceB: splitID, CreatedAt: time.Now().UTC()}); err != nil {
t.Fatal(err)
}
unlinked := rebuild()
if len(unlinked) != 2 {
t.Fatalf("unlink left %d resources, want the node and the agent apart: %+v", len(unlinked), unlinked)
}
for _, resource := range unlinked {
if resource.Proxmox != nil && resource.Agent != nil {
t.Fatalf("resource %s still carries both facets after unlink", resource.ID)
}
}
})
}
}
@@ -0,0 +1,67 @@
package unifiedresources
import "time"
// An operator decides each resource pair one way: a link folds the pair into
// one resource, an exclusion keeps it apart (the resources API's unlink and
// report-merge record exclusions). The later decision replaces the earlier
// one. Stores drop the opposite row when they write either, and their reads
// pass through effectiveManualPairDecisions, so a pair that still carries both
// (rows written before writes replaced each other, or IDs that canonical-ID
// succession rewrote onto one pair) is decided by its later write. Every
// registry loads its links and exclusions from the store and resolves the two
// reads together, since a decision can land between them, so the monitor's
// rebuilds, the resources API and the views built from the monitor's links
// all apply the same decision.
// effectiveManualPairDecisions keeps, for each resource pair, only its latest
// decision: the newest link (a later row wins a tie, so a reversed-order copy
// that succession left behind cannot override a newer link's primary) unless
// an exclusion is at least as new. A link and an exclusion written at the
// same instant keep the pair apart: two separate rows lose no data and the
// operator can link again.
func effectiveManualPairDecisions(links []ResourceLink, exclusions []ResourceExclusion) ([]ResourceLink, []ResourceExclusion) {
if len(links) == 0 {
return links, exclusions
}
newestLink := make(map[string]int, len(links))
for i, link := range links {
key := exclusionKey(link.ResourceA, link.ResourceB)
if j, ok := newestLink[key]; !ok || !links[j].CreatedAt.After(link.CreatedAt) {
newestLink[key] = i
}
}
latestExclusion := make(map[string]time.Time, len(exclusions))
for _, exclusion := range exclusions {
key := exclusionKey(exclusion.ResourceA, exclusion.ResourceB)
if at, ok := latestExclusion[key]; !ok || exclusion.CreatedAt.After(at) {
latestExclusion[key] = exclusion.CreatedAt
}
}
keptLinks := make([]ResourceLink, 0, len(newestLink))
for i, link := range links {
key := exclusionKey(link.ResourceA, link.ResourceB)
if newestLink[key] != i {
continue
}
if at, ok := latestExclusion[key]; ok && !link.CreatedAt.After(at) {
continue
}
keptLinks = append(keptLinks, link)
}
keptExclusions := make([]ResourceExclusion, 0, len(exclusions))
for _, exclusion := range exclusions {
if i, ok := newestLink[exclusionKey(exclusion.ResourceA, exclusion.ResourceB)]; ok && links[i].CreatedAt.After(exclusion.CreatedAt) {
continue
}
keptExclusions = append(keptExclusions, exclusion)
}
return keptLinks, keptExclusions
}
// sameManualPair reports whether two ID pairs name the same resources in
// either order.
func sameManualPair(a1, b1, a2, b2 string) bool {
return exclusionKey(a1, b1) == exclusionKey(a2, b2)
}
@@ -358,6 +358,56 @@ func TestCoalescePresentationHostResourcesWithExclusionsHonorsManualSplit(t *tes
}
}
// The broadcast coalesces through the adapter, the resources API through
// ListForPresentation. Both must keep a host pair the operator split apart.
func TestMonitorAdapterCoalesceForPresentationHonoursExclusions(t *testing.T) {
now := time.Date(2026, 10, 7, 21, 0, 0, 0, time.UTC)
resources := []Resource{
{
ID: "agent-runtime-alpha", Type: ResourceTypeAgent, Name: "alpha", Status: StatusOnline, LastSeen: now,
Sources: []DataSource{SourceAgent},
Identity: ResourceIdentity{Hostnames: []string{"alpha"}},
Agent: &AgentData{AgentID: "agent-alpha", Hostname: "alpha"},
},
{
ID: "agent-docker-alpha", Type: ResourceTypeAgent, Name: "alpha", Status: StatusOnline, LastSeen: now,
Sources: []DataSource{SourceDocker},
Identity: ResourceIdentity{Hostnames: []string{"alpha"}},
Docker: &DockerData{Hostname: "alpha"},
},
}
for _, split := range []bool{false, true} {
store := NewMemoryStore()
if split {
if err := store.AddExclusion(ResourceExclusion{ResourceA: "agent-docker-alpha", ResourceB: "agent-runtime-alpha"}); err != nil {
t.Fatal(err)
}
}
registry := NewRegistry(store)
registry.IngestResources(resources)
adapter := NewMonitorAdapter(registry)
want := 1
if split {
want = 2
}
rows, ok := adapter.CoalesceForPresentation(adapter.GetAll())
if !ok || len(rows) != want {
t.Fatalf("split=%v: adapter coalesced %d rows (ok=%v), want %d", split, len(rows), ok, want)
}
if got := len(registry.ListForPresentation()); got != want {
t.Fatalf("split=%v: ListForPresentation gave %d rows, want %d", split, got, want)
}
}
// A registry without a store holds no operator decisions, so it declines
// rather than coalescing as if none existed.
storeless := NewRegistry(nil)
storeless.IngestResources(resources)
if _, ok := NewMonitorAdapter(storeless).CoalesceForPresentation(resources); ok {
t.Fatal("a store-less adapter claimed to carry operator decisions")
}
}
// A Proxmox node whose Pulse Agent has gone offline must show the live PVE CPU,
// not the agent's last (0) reading. The agent is the presentation primary, so
// without a freshness gate its stale 0 CPU was kept and the live value dropped.
@@ -0,0 +1,44 @@
package unifiedresources
// presentationExclusionFilter reports whether the presentation host coalesce
// must keep two resources apart because the operator split the pair (unlink
// or report-merge). It reads a copy of the registry's exclusions, so callers
// coalesce without holding the registry lock; nil means no pair is excluded.
func (rr *ResourceRegistry) presentationExclusionFilter() func(left, right Resource) bool {
rr.mu.RLock()
if len(rr.exclusions) == 0 {
rr.mu.RUnlock()
return nil
}
exclusions := make(map[string]struct{}, len(rr.exclusions))
for key := range rr.exclusions {
exclusions[key] = struct{}{}
}
rr.mu.RUnlock()
return func(left, right Resource) bool {
leftID := CanonicalResourceID(left.ID)
rightID := CanonicalResourceID(right.ID)
if leftID == "" || rightID == "" {
return false
}
_, ok := exclusions[exclusionKey(leftID, rightID)]
return ok
}
}
// CoalesceForPresentation applies the presentation host coalesce to resources
// listed from this adapter's current registry generation, honouring the
// operator's merge exclusions through the same filter that
// ResourceRegistry.ListForPresentation uses, so the websocket broadcast keeps
// a pair the operator split apart as the resources API does. It reports
// false, leaving resources untouched, when the registry has no store and so
// carries no operator decisions: the mock view and other read states built
// from an already-unified list.
func (a *MonitorAdapter) CoalesceForPresentation(resources []Resource) ([]Resource, bool) {
registry := a.currentRegistry()
if registry == nil || registry.store == nil {
return resources, false
}
return CoalescePresentationHostResourcesWithExclusions(resources, registry.presentationExclusionFilter()), true
}
+5 -18
View File
@@ -233,6 +233,9 @@ func (rr *ResourceRegistry) loadOverrides() {
}
exclusions, err := rr.store.GetExclusions()
if err == nil {
// An unlink or link landing between the two reads would load both
// decisions for its pair (manual_link_decisions.go).
rr.links, exclusions = effectiveManualPairDecisions(rr.links, exclusions)
for _, exclusion := range exclusions {
key := exclusionKey(exclusion.ResourceA, exclusion.ResourceB)
rr.exclusions[key] = struct{}{}
@@ -1432,23 +1435,7 @@ func (rr *ResourceRegistry) listMaterialized(withTargets bool) ([]Resource, map[
// merge exclusions.
func (rr *ResourceRegistry) ListForPresentation() []Resource {
resources := rr.List()
rr.mu.RLock()
exclusions := make(map[string]struct{}, len(rr.exclusions))
for key := range rr.exclusions {
exclusions[key] = struct{}{}
}
rr.mu.RUnlock()
return CoalescePresentationHostResourcesWithExclusions(resources, func(left, right Resource) bool {
leftID := CanonicalResourceID(left.ID)
rightID := CanonicalResourceID(right.ID)
if leftID == "" || rightID == "" {
return false
}
_, ok := exclusions[exclusionKey(leftID, rightID)]
return ok
})
return CoalescePresentationHostResourcesWithExclusions(resources, rr.presentationExclusionFilter())
}
// ListByType returns all resources of the provided type.
@@ -2923,7 +2910,7 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res
// pulse-agent's pin on a same-named host) completing the identity would
// re-merge systems the connection scoping keeps apart.
if source != SourceTrueNAS {
identity = rr.completeIdentityFromPins(source, resource, identity)
identity = rr.completeIdentityFromPins(source, sourceID, resource, identity)
}
resource.Identity = identity
resource.Sources = []DataSource{source}
+113 -8
View File
@@ -11,6 +11,7 @@ import (
"os"
"path/filepath"
"regexp"
"slices"
"sort"
"strings"
"sync"
@@ -1194,7 +1195,29 @@ func (s *SQLiteResourceStore) AddLink(link ResourceLink) error {
a, b := normalizePair(link.ResourceA, link.ResourceB)
s.mu.Lock()
defer s.mu.Unlock()
_, err := s.db.Exec(`
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("begin resource link %q<->%q: %w", a, b, err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback()
}
}()
// A link replaces the pair's earlier decision (manual_link_decisions.go):
// its exclusion, and a link canonical-ID succession stored in the other
// order, which the upsert below would not replace.
if _, err := tx.Exec(`DELETE FROM resource_exclusions
WHERE (resource_a = ? AND resource_b = ?) OR (resource_a = ? AND resource_b = ?)`, a, b, b, a); err != nil {
return fmt.Errorf("clear resource exclusion %q<->%q replaced by link: %w", a, b, err)
}
if a != b {
if _, err := tx.Exec(`DELETE FROM resource_links WHERE resource_a = ? AND resource_b = ?`, b, a); err != nil {
return fmt.Errorf("clear reversed resource link %q<->%q: %w", b, a, err)
}
}
if _, err := tx.Exec(`
INSERT INTO resource_links (resource_a, resource_b, primary_id, reason, created_by, created_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(resource_a, resource_b) DO UPDATE SET
@@ -1202,10 +1225,13 @@ func (s *SQLiteResourceStore) AddLink(link ResourceLink) error {
reason=excluded.reason,
created_by=excluded.created_by,
created_at=excluded.created_at
`, a, b, link.PrimaryID, link.Reason, link.CreatedBy, link.CreatedAt)
if err != nil {
`, a, b, link.PrimaryID, link.Reason, link.CreatedBy, link.CreatedAt); err != nil {
return fmt.Errorf("upsert resource link %q<->%q: %w", a, b, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit resource link %q<->%q: %w", a, b, err)
}
committed = true
return nil
}
@@ -1221,21 +1247,76 @@ func (s *SQLiteResourceStore) AddExclusion(exclusion ResourceExclusion) error {
a, b := normalizePair(exclusion.ResourceA, exclusion.ResourceB)
s.mu.Lock()
defer s.mu.Unlock()
_, err := s.db.Exec(`
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("begin resource exclusion %q<->%q: %w", a, b, err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback()
}
}()
// An exclusion replaces the pair's earlier decision, so unlinking a linked
// pair splits it (manual_link_decisions.go): its links in either order,
// and an exclusion canonical-ID succession stored in the other order,
// which the upsert below would not replace.
if _, err := tx.Exec(`DELETE FROM resource_links
WHERE (resource_a = ? AND resource_b = ?) OR (resource_a = ? AND resource_b = ?)`, a, b, b, a); err != nil {
return fmt.Errorf("clear resource link %q<->%q replaced by exclusion: %w", a, b, err)
}
if a != b {
if _, err := tx.Exec(`DELETE FROM resource_exclusions WHERE resource_a = ? AND resource_b = ?`, b, a); err != nil {
return fmt.Errorf("clear reversed resource exclusion %q<->%q: %w", b, a, err)
}
}
if _, err := tx.Exec(`
INSERT INTO resource_exclusions (resource_a, resource_b, reason, created_by, created_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(resource_a, resource_b) DO UPDATE SET
reason=excluded.reason,
created_by=excluded.created_by,
created_at=excluded.created_at
`, a, b, exclusion.Reason, exclusion.CreatedBy, exclusion.CreatedAt)
if err != nil {
`, a, b, exclusion.Reason, exclusion.CreatedBy, exclusion.CreatedAt); err != nil {
return fmt.Errorf("upsert resource exclusion %q<->%q: %w", a, b, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit resource exclusion %q<->%q: %w", a, b, err)
}
committed = true
return nil
}
func (s *SQLiteResourceStore) GetLinks() (links []ResourceLink, err error) {
// GetLinks returns the links that are their pair's latest operator decision.
func (s *SQLiteResourceStore) GetLinks() ([]ResourceLink, error) {
links, err := s.storedLinks()
if err != nil || len(links) == 0 {
return links, err
}
exclusions, err := s.storedExclusions()
if err != nil {
return nil, err
}
links, _ = effectiveManualPairDecisions(links, exclusions)
return links, nil
}
// GetExclusions returns the exclusions that are their pair's latest operator
// decision.
func (s *SQLiteResourceStore) GetExclusions() ([]ResourceExclusion, error) {
exclusions, err := s.storedExclusions()
if err != nil || len(exclusions) == 0 {
return exclusions, err
}
links, err := s.storedLinks()
if err != nil {
return nil, err
}
_, exclusions = effectiveManualPairDecisions(links, exclusions)
return exclusions, nil
}
func (s *SQLiteResourceStore) storedLinks() (links []ResourceLink, err error) {
rows, err := s.db.Query(`SELECT resource_a, resource_b, primary_id, reason, created_by, created_at FROM resource_links`)
if err != nil {
return nil, fmt.Errorf("query resource links: %w", err)
@@ -1267,7 +1348,7 @@ func (s *SQLiteResourceStore) GetLinks() (links []ResourceLink, err error) {
return links, nil
}
func (s *SQLiteResourceStore) GetExclusions() (exclusions []ResourceExclusion, err error) {
func (s *SQLiteResourceStore) storedExclusions() (exclusions []ResourceExclusion, err error) {
rows, err := s.db.Query(`SELECT resource_a, resource_b, reason, created_by, created_at FROM resource_exclusions`)
if err != nil {
return nil, fmt.Errorf("query resource exclusions: %w", err)
@@ -3483,6 +3564,17 @@ func (m *MemoryStore) AddLink(link ResourceLink) error {
link.ResourceA = CanonicalResourceID(link.ResourceA)
link.ResourceB = CanonicalResourceID(link.ResourceB)
link.PrimaryID = CanonicalResourceID(link.PrimaryID)
if link.CreatedAt.IsZero() {
link.CreatedAt = time.Now().UTC()
}
// A link replaces the pair's earlier decision, either way round
// (manual_link_decisions.go).
m.exclusions = slices.DeleteFunc(m.exclusions, func(exclusion ResourceExclusion) bool {
return sameManualPair(exclusion.ResourceA, exclusion.ResourceB, link.ResourceA, link.ResourceB)
})
m.links = slices.DeleteFunc(m.links, func(existing ResourceLink) bool {
return sameManualPair(existing.ResourceA, existing.ResourceB, link.ResourceA, link.ResourceB)
})
m.links = append(m.links, link)
return nil
}
@@ -3492,6 +3584,17 @@ func (m *MemoryStore) AddExclusion(exclusion ResourceExclusion) error {
defer m.mu.Unlock()
exclusion.ResourceA = CanonicalResourceID(exclusion.ResourceA)
exclusion.ResourceB = CanonicalResourceID(exclusion.ResourceB)
if exclusion.CreatedAt.IsZero() {
exclusion.CreatedAt = time.Now().UTC()
}
// An exclusion replaces the pair's earlier decision, so unlinking a
// linked pair splits it (manual_link_decisions.go).
m.links = slices.DeleteFunc(m.links, func(link ResourceLink) bool {
return sameManualPair(link.ResourceA, link.ResourceB, exclusion.ResourceA, exclusion.ResourceB)
})
m.exclusions = slices.DeleteFunc(m.exclusions, func(existing ResourceExclusion) bool {
return sameManualPair(existing.ResourceA, existing.ResourceB, exclusion.ResourceA, exclusion.ResourceB)
})
m.exclusions = append(m.exclusions, exclusion)
return nil
}
@@ -3501,6 +3604,7 @@ func (m *MemoryStore) GetLinks() ([]ResourceLink, error) {
defer m.mu.RUnlock()
out := make([]ResourceLink, len(m.links))
copy(out, m.links)
out, _ = effectiveManualPairDecisions(out, m.exclusions)
return out, nil
}
@@ -3509,6 +3613,7 @@ func (m *MemoryStore) GetExclusions() ([]ResourceExclusion, error) {
defer m.mu.RUnlock()
out := make([]ResourceExclusion, len(m.exclusions))
copy(out, m.exclusions)
_, out = effectiveManualPairDecisions(m.links, out)
return out, nil
}
+242
View File
@@ -644,6 +644,248 @@ func TestNewSQLiteResourceStore_MigratesLegacyStore(t *testing.T) {
}
}
func manualDecisionStores(t *testing.T) map[string]ResourceStore {
t.Helper()
sqlite, err := NewSQLiteResourceStore(t.TempDir(), "default")
if err != nil {
t.Fatalf("NewSQLiteResourceStore: %v", err)
}
t.Cleanup(func() { _ = sqlite.Close() })
return map[string]ResourceStore{"sqlite": sqlite, "memory": NewMemoryStore()}
}
// manualDecisionRows returns the rows a store holds before read resolution,
// so the tests see what a write replaced rather than what a read hides.
func manualDecisionRows(t *testing.T, store ResourceStore) ([]ResourceLink, []ResourceExclusion) {
t.Helper()
switch s := store.(type) {
case *SQLiteResourceStore:
links, err := s.storedLinks()
if err != nil {
t.Fatalf("storedLinks: %v", err)
}
exclusions, err := s.storedExclusions()
if err != nil {
t.Fatalf("storedExclusions: %v", err)
}
return links, exclusions
case *MemoryStore:
s.mu.RLock()
defer s.mu.RUnlock()
return append([]ResourceLink(nil), s.links...), append([]ResourceExclusion(nil), s.exclusions...)
}
t.Fatalf("unexpected store %T", store)
return nil, nil
}
// seedReversedManualRows stores a link and an exclusion for a pair in the
// order canonical-ID succession can leave, which writes never produce.
func seedReversedManualRows(t *testing.T, store ResourceStore, a, b string, at time.Time) {
t.Helper()
switch s := store.(type) {
case *SQLiteResourceStore:
if _, err := s.db.Exec(`INSERT INTO resource_links (resource_a, resource_b, primary_id, reason, created_by, created_at) VALUES (?, ?, ?, '', '', ?)`, a, b, a, at); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`INSERT INTO resource_exclusions (resource_a, resource_b, reason, created_by, created_at) VALUES (?, ?, '', '', ?)`, a, b, at); err != nil {
t.Fatal(err)
}
case *MemoryStore:
s.mu.Lock()
s.links = append(s.links, ResourceLink{ResourceA: a, ResourceB: b, PrimaryID: a, CreatedAt: at})
s.exclusions = append(s.exclusions, ResourceExclusion{ResourceA: a, ResourceB: b, CreatedAt: at})
s.mu.Unlock()
default:
t.Fatalf("unexpected store %T", store)
}
}
// Unlink records an exclusion; it must also delete the link, or every
// registry keeps folding the pair. Link must delete an earlier exclusion the
// same way. The IDs arrive in either order, as the API takes either side as
// the path, and a row succession stored in the other order is replaced too.
func TestManualPairDecisionReplacesThePreviousOne(t *testing.T) {
for name, store := range manualDecisionStores(t) {
t.Run(name, func(t *testing.T) {
at := time.Date(2026, 10, 7, 21, 0, 0, 0, time.UTC)
if err := store.AddLink(ResourceLink{ResourceA: "vm-web", ResourceB: "agent-web", PrimaryID: "vm-web", CreatedAt: at}); err != nil {
t.Fatal(err)
}
if err := store.AddLink(ResourceLink{ResourceA: "vm-db", ResourceB: "agent-db", PrimaryID: "vm-db", CreatedAt: at}); err != nil {
t.Fatal(err)
}
if links, exclusions := manualDecisionRows(t, store); len(links) != 2 || len(exclusions) != 0 {
t.Fatalf("after links: links=%+v exclusions=%+v", links, exclusions)
}
seedReversedManualRows(t, store, "vm-web", "agent-web", at.Add(-time.Hour))
if err := store.AddExclusion(ResourceExclusion{ResourceA: "agent-web", ResourceB: "vm-web", CreatedAt: at.Add(time.Minute)}); err != nil {
t.Fatal(err)
}
links, exclusions := manualDecisionRows(t, store)
if len(links) != 1 || !sameManualPair(links[0].ResourceA, links[0].ResourceB, "vm-db", "agent-db") {
t.Fatalf("unlink kept the wrong link rows: %+v", links)
}
if len(exclusions) != 1 || !exclusions[0].CreatedAt.Equal(at.Add(time.Minute)) {
t.Fatalf("unlink left exclusion rows %+v, want only its own", exclusions)
}
seedReversedManualRows(t, store, "vm-web", "agent-web", at.Add(-time.Hour))
if err := store.AddLink(ResourceLink{ResourceA: "agent-web", ResourceB: "vm-web", PrimaryID: "vm-web", CreatedAt: at.Add(2 * time.Minute)}); err != nil {
t.Fatal(err)
}
links, exclusions = manualDecisionRows(t, store)
if len(links) != 2 || len(exclusions) != 0 {
t.Fatalf("relink: links=%+v exclusions=%+v, want 2 links and no exclusions", links, exclusions)
}
// Linking a linked pair again replaces its link, primary included.
if err := store.AddLink(ResourceLink{ResourceA: "agent-db", ResourceB: "vm-db", PrimaryID: "agent-db", CreatedAt: at.Add(3 * time.Minute)}); err != nil {
t.Fatal(err)
}
links, _ = manualDecisionRows(t, store)
primaries := map[string]string{}
for _, link := range links {
primaries[exclusionKey(link.ResourceA, link.ResourceB)] = link.PrimaryID
}
if len(links) != 2 || primaries[exclusionKey("vm-db", "agent-db")] != "agent-db" {
t.Fatalf("relinking kept stale rows: %+v", links)
}
// A decision written without a time is stamped, as SQLite does,
// so it can be ordered against later ones.
if err := store.AddExclusion(ResourceExclusion{ResourceA: "vm-db", ResourceB: "agent-db"}); err != nil {
t.Fatal(err)
}
if _, exclusions = manualDecisionRows(t, store); len(exclusions) != 1 || exclusions[0].CreatedAt.IsZero() {
t.Fatalf("unstamped exclusion: %+v", exclusions)
}
})
}
}
// Stores written before unlink dropped the link still hold both rows for a
// pair, and canonical-ID succession can rewrite two pairs onto one in either
// order. Reads decide such a pair by its later write; a tie keeps it apart.
func TestManualPairDecisionReadsResolveStoredConflictsByRecency(t *testing.T) {
sqlite, err := NewSQLiteResourceStore(t.TempDir(), "default")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sqlite.Close() })
at := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC)
rows := []struct {
table string
a, b string
createdAt time.Time
withPrimID bool
}{
// Unlinked after linking: the pair stays apart.
{"resource_links", "agent-a", "vm-a", at, true},
{"resource_exclusions", "agent-a", "vm-a", at.Add(time.Minute), false},
// Linked again after an unlink: the pair is linked.
{"resource_exclusions", "agent-b", "vm-b", at, false},
{"resource_links", "agent-b", "vm-b", at.Add(time.Minute), true},
// Same instant: apart.
{"resource_links", "agent-c", "vm-c", at, true},
{"resource_exclusions", "agent-c", "vm-c", at, false},
// Succession stored the link in reverse order.
{"resource_links", "vm-d", "agent-d", at, true},
{"resource_exclusions", "agent-d", "vm-d", at.Add(time.Minute), false},
// Undecided elsewhere: untouched.
{"resource_links", "agent-e", "vm-e", at, true},
{"resource_exclusions", "agent-f", "vm-f", at, false},
// Succession left an older link in the other order: the newer
// link, and its primary, decide.
{"resource_links", "vm-g", "agent-g", at.Add(time.Minute), true},
{"resource_links", "agent-g", "vm-g", at, true},
}
for _, row := range rows {
var err error
if row.withPrimID {
_, err = sqlite.db.Exec(`INSERT INTO resource_links (resource_a, resource_b, primary_id, reason, created_by, created_at) VALUES (?, ?, ?, '', '', ?)`, row.a, row.b, row.b, row.createdAt)
} else {
_, err = sqlite.db.Exec(`INSERT INTO resource_exclusions (resource_a, resource_b, reason, created_by, created_at) VALUES (?, ?, '', '', ?)`, row.a, row.b, row.createdAt)
}
if err != nil {
t.Fatalf("seed %s %s<->%s: %v", row.table, row.a, row.b, err)
}
}
links, err := sqlite.GetLinks()
if err != nil {
t.Fatal(err)
}
exclusions, err := sqlite.GetExclusions()
if err != nil {
t.Fatal(err)
}
linked := map[string]bool{}
for _, link := range links {
linked[exclusionKey(link.ResourceA, link.ResourceB)] = true
}
excluded := map[string]bool{}
for _, exclusion := range exclusions {
excluded[exclusionKey(exclusion.ResourceA, exclusion.ResourceB)] = true
}
for _, want := range []struct {
pair string
linked, excluded bool
}{
{"a", false, true},
{"b", true, false},
{"c", false, true},
{"d", false, true},
{"e", true, false},
{"f", false, true},
{"g", true, false},
} {
key := exclusionKey("agent-"+want.pair, "vm-"+want.pair)
if linked[key] != want.linked || excluded[key] != want.excluded {
t.Errorf("pair %s: linked=%v excluded=%v, want %v/%v", want.pair, linked[key], excluded[key], want.linked, want.excluded)
}
}
if len(links) != 3 || len(exclusions) != 4 {
t.Fatalf("links=%+v exclusions=%+v", links, exclusions)
}
for _, link := range links {
if sameManualPair(link.ResourceA, link.ResourceB, "agent-g", "vm-g") && link.PrimaryID != "agent-g" {
t.Fatalf("older reversed link won: %+v", link)
}
}
}
// tornDecisionStore answers GetLinks from before an unlink and GetExclusions
// from after it, as two reads around a concurrent unlink do.
type tornDecisionStore struct {
*MemoryStore
staleLinks []ResourceLink
}
func (s tornDecisionStore) GetLinks() ([]ResourceLink, error) { return s.staleLinks, nil }
// A registry must not load both decisions for a pair when an unlink lands
// between its two reads: applyManualLinks does not consult exclusions, so the
// generation would stay merged.
func TestRegistryResolvesDecisionsAcrossItsTwoReads(t *testing.T) {
at := time.Date(2026, 10, 7, 21, 0, 0, 0, time.UTC)
store := tornDecisionStore{
MemoryStore: NewMemoryStore(),
staleLinks: []ResourceLink{{ResourceA: "vm-web", ResourceB: "agent-web", PrimaryID: "vm-web", CreatedAt: at}},
}
if err := store.AddExclusion(ResourceExclusion{ResourceA: "vm-web", ResourceB: "agent-web", CreatedAt: at.Add(time.Second)}); err != nil {
t.Fatal(err)
}
registry := NewRegistry(store)
registry.IngestResources([]Resource{
{ID: "vm-web", Type: ResourceTypeVM, Name: "web", Status: StatusOnline, LastSeen: at, Sources: []DataSource{SourceProxmox}},
{ID: "agent-web", Type: ResourceTypeAgent, Name: "web-agent", Status: StatusOnline, LastSeen: at, Sources: []DataSource{SourceAgent}, Agent: &AgentData{AgentID: "agent-web"}},
})
if got := len(registry.List()); got != 2 {
t.Fatalf("registry folded an unlinked pair: %d resources", got)
}
}
func TestNewSQLiteResourceStore_MigratesLegacyResourceChangesTable(t *testing.T) {
dataDir := t.TempDir()
legacyPath := filepath.Join(dataDir, "resources", resourceDBFileName)