diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index d3a52b602..1f47be0f2 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -7861,29 +7861,38 @@ symmetric bridges would miss removal of filtering from one inventory. This proof does not establish safety for arbitrarily renamed Docker bridges or repair persisted links, and is not reporter or installed-release validation. -### Manual host/node association safety boundary (6 September 2026) +### Durable host/node association intent (6 September 2026) -`TestManualHostLinkSurvivesUnmatchedReportsAndProviderRefresh` protects an -operator-created association between `nas.example` and an otherwise unmatched -`pve` node through repeated reports, `UpdateNodesForInstance`, and explicit -unlink. Reapplying the rejected stale-link cleanup plus provider-name matcher -change makes the first report fail this regression. Neither change is included -with this test. +The earlier restart reproduction showed that the link API only changed memory, +and that provider-name cleanup erased unmatched manual links. The replacement +uses internal `NodeLinkSource` provenance in the host continuity journal: +`manual`, `unlinked`, `automatic`, or empty (legacy/unknown). This is not a new +API or UI surface. -Restart remains an independently reproduced gap, not a supported invariant: -using `newHostRemovalLifecycleMonitor` with a temporary data directory, submit -`hostRemovalLifecycleReport("machine", "machine", "agent", "nas.example", -"linux", now)`, populate the unmatched `pve-node`, and call `LinkHostAgent`. -Reconstructing the monitor from the same directory immediately yields a host -without the link. Sending another report before reconstruction persists the -link, but refreshing the provider on the reconstructed monitor and submitting -the next report still returns an empty `LinkedNodeID`. +Link and unlink take the host lifecycle write lock and commit all affected +journal entries while holding the state lock, before publishing visible state. +A failed write (or unavailable store) returns an error without changing either +direction of the link. Reassigning an occupied node writes the displaced host's +explicit unlink in the same transaction. Identity, report ordering and removal +metadata are preserved. -`LinkHostAgent` currently changes state only. `HostContinuityEntry` stores the -target ID but no manual/automatic provenance. Legacy manual and automatic links -can therefore have identical persisted representations. A repair must not -classify every unmarked legacy link as automatic. Durable operator intent, -failed-write handling, provider identity changes, and explicit unlink across -restart need coverage before enabling destructive stale-link cleanup. This is -maintenance evidence, not demand for a new user-visible linking surface or a -claim of resolution of issue #1930. +Manual links are pinned to the operator-selected provider ID, including through +unmatched reports and restart before another report. Provider identity replacement +does not authorise redirection: intent stays dormant if that ID disappears and +reattaches if it returns. A persisted reservation also prevents another host's +automatic match from taking the node before its owner reconnects. Replacing a +dormant owner explicitly persists that owner's unlink too. Explicit unlink persists and suppresses automatic node +reassociation even when a subsequent report matches. An explicit link can select +a replacement ID. + +Only known automatic associations are re-evaluated destructively against provider +names and network evidence. Obsolete reverse links are removed with host updates. +An unmarked persisted link is not assumed automatic: report ingestion retains it, +so this is deliberately **not** a blanket repair of existing v6.4.1 associations. +Legacy provider reconciliation otherwise retains its previous behaviour. + +`host_manual_link_regression_test.go` covers immediate restart, unlink with +positive matching evidence, provider replacement/return, write failure and +automatic versus unknown-provenance cleanup. State and config tests cover atomic +replacement and preservation of lifecycle evidence. These are synthetic local +proofs, not reporter confirmation or installed-release resolution of #1930. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 6bab17f70..ad279d435 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -39,6 +39,54 @@ ambiguous-name tests remain required. These checks prevent a reproduced synthetic misassociation, not all private-address collisions or unknown custom bridge names, and do not prove the cause of #1930. +### Host/node associations retain durable operator intent + +Monitoring owns the provenance used to distinguish an operator selection from +an automatic association. The host continuity journal records `manual`, +`unlinked` or `automatic`; an absent value is legacy/unknown, not evidence that +the link was automatic. This internal provenance does not add a public API or +frontend field. + +The existing link/unlink APIs hold the host lifecycle write lock and persist +the complete affected intent transaction under the state lock before publishing +either direction of the association. Missing storage or a failed journal write +must return an error without changing visible links or the journal's in-memory +entries. Replacing a node's owner persists the displaced owner's explicit unlink +too, including an owner which has not yet reconnected after restart. Link writes +preserve identity, report watermarks, credential and removal evidence. + +A manual selection remains pinned to the selected provider ID through unmatched +reports, provider refresh and restart before another report. A replacement +provider ID must not inherit that selection by name: intent remains dormant +until the original ID returns or the operator explicitly selects another ID. +The persisted reservation prevents another host's automatic match taking that +node before its owner reconnects. Explicit unlink survives restart and suppresses +automatic node reassociation even when the next report supplies matching evidence. + +Automatic name matching uses the provider's node name, never the display name +merged from an already-linked agent. Known automatic associations may be +re-evaluated and cleared; provenance-bearing host updates remove obsolete +reverse links without clearing another agent's link. Unmarked legacy forward +links are retained during report ingestion, and unmarked one-way reverse links +must survive host updates because SMART fallback also consumes them. Legacy +provider reconciliation otherwise keeps its existing behaviour. This deliberately +does not claim to repair every persisted v6.4.1 association or resolve #1930. + +Verification: +- `internal/monitoring/host_manual_link_regression_test.go` covers immediate + restart, explicit unlink with positive matching evidence, provider-ID + replacement/return, failed writes, automatic versus unknown provenance, + dormant reservations and concurrent reports/provider refresh. +- `internal/config/host_continuity_test.go` pins journal preservation and + replacement of a dormant owner. +- `internal/models/state_additional_test.go` pins transaction rollback, + obsolete reverse-link cleanup and retention of unknown legacy reverse links. +- `TestMaybePollPhysicalDisksAsync_AgentFallbackWhenDiskQueryFails` in + `internal/monitoring/monitor_pve_disk_fallback_test.go` remains required: + association cleanup must not remove the one-way link used by SMART fallback. + +### Physical-disk observation cadence + Physical disk inventory has an independent collector schedule. The PVE poller carries its default five-minute or configured interval with each disk record, while keeping the last successful observation timestamp on retained records. diff --git a/internal/config/host_continuity.go b/internal/config/host_continuity.go index 2f9e0b344..c05444855 100644 --- a/internal/config/host_continuity.go +++ b/internal/config/host_continuity.go @@ -28,6 +28,7 @@ type HostContinuityEntry struct { DeniedTokenIDs []string `json:"deniedTokenIds,omitempty"` AgentVersion string `json:"agentVersion,omitempty"` Platform string `json:"platform,omitempty"` + NodeLinkSource string `json:"nodeLinkSource,omitempty"` LinkedNodeID string `json:"linkedNodeId,omitempty"` LinkedVMID string `json:"linkedVmId,omitempty"` LinkedContainerID string `json:"linkedContainerId,omitempty"` @@ -401,3 +402,56 @@ func uniqueTrimmedStrings(values ...string) []string { } return out } + +// SetNodeLinkIntents writes all affected links as one journal transaction. +// It preserves report watermarks, credentials and removal tombstones. +func (s *HostContinuityStore) SetNodeLinkIntents(links []HostContinuityEntry) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.loadErr != nil { + return s.loadErr + } + previous := make(map[string]HostContinuityEntry, len(s.entries)) + for id, entry := range s.entries { + previous[id] = entry + } + for _, link := range links { + // An operator can replace an owner which has not reported since restart. + // Persist that owner's unlink too, so it cannot reclaim the selection. + if link.NodeLinkSource == "manual" && link.LinkedNodeID != "" { + for id, prior := range s.entries { + if id != link.HostID && prior.LinkedNodeID == link.LinkedNodeID && prior.RemovedAt.IsZero() { + prior.LinkedNodeID, prior.NodeLinkSource = "", "unlinked" + s.entries[id] = prior + } + } + } + entry, ok := s.entries[link.HostID] + if !ok { + entry = link + } + entry.LinkedNodeID = link.LinkedNodeID + entry.LinkedVMID, entry.LinkedContainerID = "", "" + entry.NodeLinkSource = link.NodeLinkSource + s.entries[link.HostID] = entry + } + if err := s.save(); err != nil { + s.entries = previous + return err + } + return nil +} + +// NodeLinkReservedByOther protects an operator-selected node even before its +// owner reconnects after restart. Removal releases the reservation. +func (s *HostContinuityStore) NodeLinkReservedByOther(hostID, nodeID string) bool { + s.mu.RLock() + defer s.mu.RUnlock() + for id, entry := range s.entries { + if id != hostID && entry.NodeLinkSource == "manual" && + entry.LinkedNodeID == nodeID && entry.RemovedAt.IsZero() { + return true + } + } + return false +} diff --git a/internal/config/host_continuity_test.go b/internal/config/host_continuity_test.go index 348a07f39..9de597675 100644 --- a/internal/config/host_continuity_test.go +++ b/internal/config/host_continuity_test.go @@ -260,3 +260,51 @@ func TestHostContinuityStoreReportsLoadFailure(t *testing.T) { t.Fatal("LoadError = nil, want continuity read failure") } } + +func TestHostContinuityNodeLinkIntentsPreserveJournal(t *testing.T) { + dir := t.TempDir() + s := NewHostContinuityStore(dir, nil) + entry := HostContinuityEntry{HostID: "a", LinkedNodeID: "n", NodeLinkSource: "manual", + TokenID: "synthetic-token", DeniedTokenIDs: []string{"synthetic-denied"}, + ReportStreamID: "stream", ReportSequence: 42, RemovedAt: time.Now().UTC()} + if err := s.Upsert(entry); err != nil { + t.Fatal(err) + } + if err := s.SetNodeLinkIntents([]HostContinuityEntry{ + {HostID: "a", NodeLinkSource: "unlinked"}, + {HostID: "b", LinkedNodeID: "n", NodeLinkSource: "manual"}, + }); err != nil { + t.Fatal(err) + } + reloaded := NewHostContinuityStore(dir, nil) + a, _ := reloaded.Get("a") + b, _ := reloaded.Get("b") + if a.NodeLinkSource != "unlinked" || a.LinkedNodeID != "" || b.NodeLinkSource != "manual" || b.LinkedNodeID != "n" { + t.Fatal("replacement intent lost across reload") + } + if a.TokenID != entry.TokenID || len(a.DeniedTokenIDs) != 1 || a.ReportSequence != 42 || !a.RemovedAt.Equal(entry.RemovedAt) { + t.Fatal("link update changed lifecycle evidence") + } +} + +func TestHostContinuityNodeLinkReplacesDormantOwner(t *testing.T) { + dir := t.TempDir() + s := NewHostContinuityStore(dir, nil) + if err := s.SetNodeLinkIntents([]HostContinuityEntry{{HostID: "a", LinkedNodeID: "n", NodeLinkSource: "manual"}}); err != nil { + t.Fatal(err) + } + if !s.NodeLinkReservedByOther("b", "n") || s.NodeLinkReservedByOther("a", "n") { + t.Fatal("reservation ownership incorrect") + } + if err := s.SetNodeLinkIntents([]HostContinuityEntry{{HostID: "b", LinkedNodeID: "n", NodeLinkSource: "manual"}}); err != nil { + t.Fatal(err) + } + s = NewHostContinuityStore(dir, nil) + a, _ := s.Get("a") + if a.LinkedNodeID != "" || a.NodeLinkSource != "unlinked" { + t.Fatal("dormant owner can reclaim replacement link") + } + if !s.NodeLinkReservedByOther("a", "n") || s.NodeLinkReservedByOther("b", "n") { + t.Fatal("replacement reservation incorrect") + } +} diff --git a/internal/models/models.go b/internal/models/models.go index ee61bc8b9..082f8ca36 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -363,6 +363,7 @@ type Host struct { DiskWriteRate float64 `json:"diskWriteRate,omitempty"` // Linking: When this host agent is running on a known PVE node/VM/container + NodeLinkSource string `json:"-"` // manual, unlinked, automatic; empty means legacy/unknown LinkedNodeID string `json:"linkedNodeId,omitempty"` // ID of the PVE node this agent is running on LinkedVMID string `json:"linkedVmId,omitempty"` // ID of the VM this agent is running inside LinkedContainerID string `json:"linkedContainerId,omitempty"` // ID of the container this agent is running inside @@ -3859,6 +3860,21 @@ func preferNodeForMerge(existing Node, candidate Node) Node { } func reconcileHostNodeLinksLocked(hosts []Host, nodes []Node) { + // Explicit operator intent is pinned to the selected provider ID, never + // transferred by hostname to a replacement provider identity. + for _, host := range hosts { + if host.NodeLinkSource != "manual" && host.NodeLinkSource != "unlinked" { + continue + } + for i := range nodes { + if nodes[i].LinkedAgentID == host.ID { + nodes[i].LinkedAgentID = "" + } + if host.NodeLinkSource == "manual" && nodes[i].ID == host.LinkedNodeID { + nodes[i].LinkedAgentID = host.ID + } + } + } linkedNodeByHostID := make(map[string]string) multipleNodeLinksByHostID := make(map[string]struct{}) for _, node := range nodes { @@ -3880,6 +3896,9 @@ func reconcileHostNodeLinksLocked(hosts []Host, nodes []Node) { continue } + if hosts[i].NodeLinkSource == "manual" || hosts[i].NodeLinkSource == "unlinked" { + continue + } nodeID, hasLinkedNode := linkedNodeByHostID[hostID] _, ambiguous := multipleNodeLinksByHostID[hostID] switch { @@ -5315,6 +5334,13 @@ func (s *State) UpsertHost(host Host) { defer s.mu.Unlock() host = cloneHost(host) + for i, node := range s.Nodes { + // Legacy one-way links have unknown intent; an unmarked host update + // cannot authorise their removal (SMART fallback also uses these links). + if host.NodeLinkSource != "" && host.ID != "" && node.LinkedAgentID == host.ID && node.ID != host.LinkedNodeID { + s.Nodes[i].LinkedAgentID = "" + } + } updated := false for i, existing := range s.Hosts { @@ -6077,3 +6103,39 @@ func (s *State) UpdatePollStats(pollDuration float64, uptime int64, wsClients in s.Stats.Uptime = uptime s.Stats.WebSocketClients = wsClients } + +// SetHostNodeLinkIntent commits operator intent only after durable storage +// accepts the complete set of changed hosts. persist must not call State. +func (s *State) SetHostNodeLinkIntent(hostID, nodeID string, persist func([]Host) error) error { + s.mu.Lock() + defer s.mu.Unlock() + next := &State{Hosts: cloneHosts(s.Hosts), Nodes: append([]Node(nil), s.Nodes...)} + if nodeID != "" { + if err := next.LinkHostAgentToNode(hostID, nodeID); err != nil { + return err + } + } else if !next.UnlinkHostAgent(hostID) { + return fmt.Errorf("host not found or not linked to a node") + } + var changed []Host + for i := range next.Hosts { + h := &next.Hosts[i] + if h.ID == hostID { + h.NodeLinkSource = "manual" + if nodeID == "" { + h.NodeLinkSource = "unlinked" + } + changed = append(changed, *h) + } else if h.LinkedNodeID != s.Hosts[i].LinkedNodeID { + h.NodeLinkSource = "unlinked" + changed = append(changed, *h) + } + } + if persist != nil { + if err := persist(changed); err != nil { + return err + } + } + s.Hosts, s.Nodes, s.LastUpdate = next.Hosts, next.Nodes, next.LastUpdate + return nil +} diff --git a/internal/models/state_additional_test.go b/internal/models/state_additional_test.go index 399d2eb99..a2edd35f2 100644 --- a/internal/models/state_additional_test.go +++ b/internal/models/state_additional_test.go @@ -685,3 +685,56 @@ func TestStateContainers(t *testing.T) { t.Fatalf("expected false for missing container") } } + +func TestSetHostNodeLinkIntentTransaction(t *testing.T) { + s := &State{ + Hosts: []Host{{ID: "a", LinkedNodeID: "n", NodeLinkSource: "manual"}, {ID: "b"}}, + Nodes: []Node{{ID: "n", LinkedAgentID: "a"}}, + } + failure := errors.New("disk unavailable") + err := s.SetHostNodeLinkIntent("b", "n", func(changed []Host) error { + if len(changed) != 2 { + t.Fatalf("changed hosts = %d", len(changed)) + } + if changed[0].NodeLinkSource != "unlinked" || changed[1].NodeLinkSource != "manual" { + t.Fatalf("replacement intent not explicit: %+v", changed) + } + return failure + }) + if !errors.Is(err, failure) { + t.Fatalf("error = %v", err) + } + if s.Hosts[0].LinkedNodeID != "n" || s.Hosts[1].LinkedNodeID != "" || s.Nodes[0].LinkedAgentID != "a" { + t.Fatal("failed persistence changed visible state") + } + if err := s.SetHostNodeLinkIntent("b", "n", func([]Host) error { return nil }); err != nil { + t.Fatal(err) + } + if s.Hosts[0].NodeLinkSource != "unlinked" || s.Hosts[1].NodeLinkSource != "manual" || s.Nodes[0].LinkedAgentID != "b" { + t.Fatal("replacement did not commit both intents") + } + called := false + if err := s.SetHostNodeLinkIntent("b", "missing", func([]Host) error { called = true; return nil }); err == nil || called { + t.Fatal("invalid target reached persistence") + } +} + +func TestUpsertHostClearsOnlyObsoleteReverseLinks(t *testing.T) { + s := &State{Nodes: []Node{ + {ID: "old", LinkedAgentID: "agent"}, + {ID: "current", LinkedAgentID: "agent"}, + {ID: "other", LinkedAgentID: "other-agent"}, + }} + s.UpsertHost(Host{ID: "agent", LinkedNodeID: "current", NodeLinkSource: "automatic"}) + if s.Nodes[0].LinkedAgentID != "" || s.Nodes[1].LinkedAgentID != "agent" || s.Nodes[2].LinkedAgentID != "other-agent" { + t.Fatalf("incorrect reverse-link cleanup: %+v", s.Nodes) + } +} + +func TestUpsertHostPreservesUnknownLegacyReverseLink(t *testing.T) { + s := &State{Nodes: []Node{{ID: "legacy-node", LinkedAgentID: "legacy-agent"}}} + s.UpsertHost(Host{ID: "legacy-agent", Hostname: "legacy"}) + if s.Nodes[0].LinkedAgentID != "legacy-agent" { + t.Fatal("unmarked host update erased unknown legacy reverse link") + } +} diff --git a/internal/monitoring/host_manual_link_regression_test.go b/internal/monitoring/host_manual_link_regression_test.go index 0daf6a0b3..63818e2d6 100644 --- a/internal/monitoring/host_manual_link_regression_test.go +++ b/internal/monitoring/host_manual_link_regression_test.go @@ -1,6 +1,8 @@ package monitoring import ( + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "os" "testing" "time" @@ -11,6 +13,7 @@ import ( // repairing old automatic associations must not erase a manual association. func TestManualHostLinkSurvivesUnmatchedReportsAndProviderRefresh(t *testing.T) { monitor := issue1654Monitor() + monitor.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil) report := issue1654Report(time.Now().UTC()) report.Host.Hostname = "nas.example" report.Host.ReportIP = "" @@ -52,3 +55,250 @@ func TestManualHostLinkSurvivesUnmatchedReportsAndProviderRefresh(t *testing.T) t.Fatalf("explicit unlink retained node link %q", got) } } + +// A different provider ID is not permission to redirect an operator's intent. +// Keep the original selection dormant until that ID returns or explicit unlink. +func TestManualHostLinkDurableIntent(t *testing.T) { + dir := t.TempDir() + newMonitor := func() *Monitor { + m := issue1654Monitor() + m.hostContinuityStore = config.NewHostContinuityStore(dir, nil) + return m + } + m := newMonitor() + report := issue1654Report(time.Now().UTC()) + report.Host.Hostname, report.Host.ReportIP, report.Network = "nas.example", "", nil + host, err := m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + node := models.Node{ID: "selected", Name: "pve", Instance: "cluster"} + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + if err := m.LinkHostAgent(host.ID, node.ID); err != nil { + t.Fatal(err) + } + // Restart immediately, with no intervening report to accidentally save intent. + m = newMonitor() + for i := 0; i < 3; i++ { + provider := node + if i == 1 { + provider.ID = "replacement" + } + m.state.UpdateNodesForInstance("cluster", []models.Node{provider}) + report.Timestamp = report.Timestamp.Add(time.Second) + host, err = m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + m.state.UpdateNodesForInstance("cluster", []models.Node{provider}) + if got := m.state.GetHosts()[0]; got.LinkedNodeID != node.ID || got.NodeLinkSource != "manual" { + t.Fatalf("step %d lost intent: %+v", i, got) + } + want := host.ID + if i == 1 { + want = "" + } + if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != want { + t.Fatalf("step %d reverse = %q, want %q", i, got, want) + } + } + if err := m.UnlinkHostAgent(host.ID); err != nil { + t.Fatal(err) + } + m = newMonitor() + // Positive matching evidence must not undo an explicit unlink. + report.Host.Hostname = node.Name + for i := 0; i < 2; i++ { + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + report.Timestamp = report.Timestamp.Add(time.Second) + host, err = m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + if host.LinkedNodeID != "" || host.NodeLinkSource != "unlinked" { + t.Fatalf("explicit unlink undone: %+v", host) + } + if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != "" { + t.Fatalf("reverse = %q", got) + } + } +} + +func TestHostLinkProvenanceCleanupAcrossRestart(t *testing.T) { + for _, source := range []string{"automatic", ""} { + t.Run("source="+source, func(t *testing.T) { + dir := t.TempDir() + m := issue1654Monitor() + m.hostContinuityStore = config.NewHostContinuityStore(dir, nil) + report := issue1654Report(time.Now().UTC()) + report.Host.Hostname, report.Host.ReportIP, report.Network = "nas.example", "", nil + host, err := m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + node := models.Node{ID: "pve", Name: "pve", Instance: "cluster", LinkedAgentID: host.ID} + host.LinkedNodeID, host.NodeLinkSource = node.ID, source + m.state.UpsertHost(host) + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + entry, _ := m.hostContinuityStore.Get(host.ID) + entry.LinkedNodeID, entry.NodeLinkSource = node.ID, source + if err := m.hostContinuityStore.Upsert(entry); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if i == 1 { + m = issue1654Monitor() + m.hostContinuityStore = config.NewHostContinuityStore(dir, nil) + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + } + report.Timestamp = report.Timestamp.Add(time.Second) + host, err = m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + want := "" + if source == "" { + want = node.ID + } + if host.LinkedNodeID != want { + t.Fatalf("step %d link = %q, want %q", i, host.LinkedNodeID, want) + } + reverse := "" + if want != "" { + reverse = host.ID + } + if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != reverse { + t.Fatalf("reverse = %q, want %q", got, reverse) + } + } + }) + } +} + +func TestHostLinkIntentPersistenceFailure(t *testing.T) { + dir := t.TempDir() + m := issue1654Monitor() + m.hostContinuityStore = config.NewHostContinuityStore(dir, nil) + report := issue1654Report(time.Now().UTC()) + host, err := m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + node := models.Node{ID: "selected", Name: "pve", Instance: "cluster"} + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + // Replace the directory with a file after the store has loaded successfully. + if err := os.RemoveAll(dir); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(dir, []byte("not a directory"), 0600); err != nil { + t.Fatal(err) + } + if err := m.LinkHostAgent(host.ID, node.ID); err == nil { + t.Fatal("link acknowledged without persistence") + } + if got := m.state.GetHosts()[0].LinkedNodeID; got != "" { + t.Fatalf("failed link changed state: %q", got) + } + entry, _ := m.hostContinuityStore.Get(host.ID) + if entry.LinkedNodeID != "" { + t.Fatal("failed link changed journal memory") + } + if err := os.Remove(dir); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(dir, 0700); err != nil { + t.Fatal(err) + } + if err := m.LinkHostAgent(host.ID, node.ID); err != nil { + t.Fatal(err) + } + if err := os.RemoveAll(dir); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(dir, []byte("not a directory"), 0600); err != nil { + t.Fatal(err) + } + if err := m.UnlinkHostAgent(host.ID); err == nil { + t.Fatal("unlink acknowledged without persistence") + } + if got := m.state.GetHosts()[0].LinkedNodeID; got != node.ID { + t.Fatalf("failed unlink changed state: %q", got) + } + if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != host.ID { + t.Fatal("failed unlink changed reverse link") + } + entry, _ = m.hostContinuityStore.Get(host.ID) + if entry.LinkedNodeID != node.ID || entry.NodeLinkSource != "manual" { + t.Fatal("failed unlink changed journal memory") + } +} + +func TestManualHostLinkConcurrentReportsAndRefresh(t *testing.T) { + m := issue1654Monitor() + m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil) + report := issue1654Report(time.Now().UTC()) + report.Host.Hostname, report.Host.ReportIP, report.Network = "nas.example", "", nil + host, err := m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + node := models.Node{ID: "selected", Name: "pve", Instance: "cluster"} + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + if err := m.LinkHostAgent(host.ID, node.ID); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + for i := 0; i < 20; i++ { + report.Timestamp = report.Timestamp.Add(time.Second) + if _, err := m.ApplyHostReport(report, nil); err != nil { + done <- err + return + } + m.state.UpdateNodesForInstance("cluster", []models.Node{node}) + } + done <- nil + }() + for i := 0; i < 20; i++ { + if err := m.UnlinkHostAgent(host.ID); err != nil { + t.Error(err) + break + } + if err := m.LinkHostAgent(host.ID, node.ID); err != nil { + t.Error(err) + break + } + } + if err := <-done; err != nil { + t.Fatal(err) + } + entry, _ := m.hostContinuityStore.Get(host.ID) + live := m.state.GetHosts()[0] + if live.NodeLinkSource != "manual" || live.LinkedNodeID != node.ID || + entry.NodeLinkSource != "manual" || entry.LinkedNodeID != node.ID || + m.state.GetSnapshot().Nodes[0].LinkedAgentID != host.ID { + t.Fatal("concurrent reports or provider refresh overwrote committed intent") + } +} + +func TestManualHostLinkReservesNodeBeforeOwnerReconnects(t *testing.T) { + dir := t.TempDir() + m := issue1654Monitor() + m.hostContinuityStore = config.NewHostContinuityStore(dir, nil) + if err := m.hostContinuityStore.SetNodeLinkIntents([]config.HostContinuityEntry{{ + HostID: "dormant-owner", LinkedNodeID: "selected", NodeLinkSource: "manual", + }}); err != nil { + t.Fatal(err) + } + m.hostContinuityStore = config.NewHostContinuityStore(dir, nil) + m.state.UpdateNodesForInstance("cluster", []models.Node{{ID: "selected", Name: "pve", Instance: "cluster"}}) + report := issue1654Report(time.Now().UTC()) + report.Host.Hostname = "pve" + host, err := m.ApplyHostReport(report, nil) + if err != nil { + t.Fatal(err) + } + if host.LinkedNodeID != "" || m.state.GetSnapshot().Nodes[0].LinkedAgentID != "" { + t.Fatal("automatic match stole operator reservation") + } +} diff --git a/internal/monitoring/monitor_additional_test.go b/internal/monitoring/monitor_additional_test.go index 86778197c..a98494ff1 100644 --- a/internal/monitoring/monitor_additional_test.go +++ b/internal/monitoring/monitor_additional_test.go @@ -64,7 +64,7 @@ func TestMonitorGetDockerHostsNilReceiver(t *testing.T) { } func TestMonitorLinkHostAgent(t *testing.T) { - monitor := &Monitor{state: models.NewState()} + monitor := &Monitor{state: models.NewState(), hostContinuityStore: config.NewHostContinuityStore(t.TempDir(), nil)} if err := monitor.LinkHostAgent("", "node-1"); err == nil { t.Fatalf("expected error on empty host ID") diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 4465f8ce7..0d5b8809b 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -974,6 +974,8 @@ func hostAgentIdentifiersMatch(left, right string) bool { // This is used when auto-linking can't disambiguate (e.g., multiple nodes with hostname "pve"). // After linking, the host agent's temperature/sensor data will appear on the correct node. func (m *Monitor) LinkHostAgent(hostID, nodeID string) error { + m.hostAgentLifecycleMu.Lock() + defer m.hostAgentLifecycleMu.Unlock() hostID = strings.TrimSpace(hostID) nodeID = strings.TrimSpace(nodeID) if hostID == "" { @@ -983,7 +985,7 @@ func (m *Monitor) LinkHostAgent(hostID, nodeID string) error { return fmt.Errorf("node id is required") } - if err := m.state.LinkHostAgentToNode(hostID, nodeID); err != nil { + if err := m.state.SetHostNodeLinkIntent(hostID, nodeID, m.persistNodeLinkIntents); err != nil { return fmt.Errorf("link host agent %q to node %q: %w", hostID, nodeID, err) } @@ -999,13 +1001,15 @@ func (m *Monitor) LinkHostAgent(hostID, nodeID string) error { // The agent will continue to report but will appear in the Managed Agents table // instead of being merged with the PVE node in the Dashboard. func (m *Monitor) UnlinkHostAgent(hostID string) error { + m.hostAgentLifecycleMu.Lock() + defer m.hostAgentLifecycleMu.Unlock() hostID = strings.TrimSpace(hostID) if hostID == "" { return fmt.Errorf("host id is required") } - if !m.state.UnlinkHostAgent(hostID) { - return fmt.Errorf("host not found or not linked to a node") + if err := m.state.SetHostNodeLinkIntent(hostID, "", m.persistNodeLinkIntents); err != nil { + return err } log.Info(). @@ -1664,6 +1668,7 @@ func (m *Monitor) persistHostContinuity(host models.Host, report agentshost.Repo TokenID: strings.TrimSpace(host.TokenID), AgentVersion: strings.TrimSpace(host.AgentVersion), Platform: strings.TrimSpace(host.Platform), + NodeLinkSource: host.NodeLinkSource, LinkedNodeID: strings.TrimSpace(host.LinkedNodeID), LinkedVMID: strings.TrimSpace(host.LinkedVMID), LinkedContainerID: strings.TrimSpace(host.LinkedContainerID), @@ -1782,6 +1787,7 @@ func hostFromContinuityEntry(entry config.HostContinuityEntry) models.Host { Platform: platformsupport.NormalizeAgentReportedPlatform(entry.Platform), IsLegacy: entry.IsLegacy, IntervalSeconds: entry.IntervalSeconds, + NodeLinkSource: entry.NodeLinkSource, LinkedNodeID: strings.TrimSpace(entry.LinkedNodeID), LinkedVMID: strings.TrimSpace(entry.LinkedVMID), LinkedContainerID: strings.TrimSpace(entry.LinkedContainerID), @@ -3561,6 +3567,25 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. report.Host.ReportIP, report.Network, ) + // Unknown legacy provenance is not permission to erase an operator link. + // Only associations created with explicit automatic provenance are cleaned. + prior, exists := m.hostByID(host.ID) + if !exists && m.hostContinuityStore != nil { + if entry, ok := m.hostContinuityStore.Get(host.ID); ok { + prior = hostFromContinuityEntry(entry) + } + } + host.NodeLinkSource = "automatic" + if prior.NodeLinkSource == "manual" || prior.NodeLinkSource == "unlinked" || + (prior.NodeLinkSource == "" && prior.LinkedNodeID != "") { + host.NodeLinkSource = prior.NodeLinkSource + linkedNodeID = prior.LinkedNodeID + linkedVMID, linkedContainerID = "", "" + } + if linkedNodeID != "" && host.NodeLinkSource != "manual" && m.hostContinuityStore != nil && + m.hostContinuityStore.NodeLinkReservedByOther(host.ID, linkedNodeID) { + linkedNodeID = "" + } if linkedNodeID != "" { host.LinkedNodeID = linkedNodeID log.Debug(). @@ -4578,7 +4603,8 @@ func (m *Monitor) findLinkedProxmoxEntityWithHints( // Check PVE nodes first - but detect ambiguity when multiple nodes match var matchingNodes []linkedEntityMatch for _, node := range nodes { - if matchHostname(node.Name()) { + // Merged display names can come from the linked agent, not the provider. + if matchHostname(node.NodeName()) { matchingNodes = append(matchingNodes, linkedEntityMatch{ id: node.SourceID(), instance: node.Instance(), @@ -5144,4 +5170,19 @@ func sharedSystemAlertCorrelationForHost(host models.Host, nodes []models.Node) ) } -// sortContent sorts comma-separated content values for consistent display +// Called with the lifecycle write lock and state lock held. No report or +// provider refresh can interleave the journal commit and visible link change. +func (m *Monitor) persistNodeLinkIntents(hosts []models.Host) error { + if m.hostContinuityStore == nil { + return fmt.Errorf("host continuity storage unavailable") + } + entries := make([]config.HostContinuityEntry, 0, len(hosts)) + for _, h := range hosts { + entries = append(entries, config.HostContinuityEntry{ + HostID: h.ID, Hostname: h.Hostname, MachineID: h.MachineID, + TokenID: h.TokenID, LastSeen: h.LastSeen, LinkedNodeID: h.LinkedNodeID, + NodeLinkSource: h.NodeLinkSource, + }) + } + return m.hostContinuityStore.SetNodeLinkIntents(entries) +} diff --git a/internal/monitoring/monitor_full_coverage_test.go b/internal/monitoring/monitor_full_coverage_test.go index d78afc044..bce1b5d0b 100644 --- a/internal/monitoring/monitor_full_coverage_test.go +++ b/internal/monitoring/monitor_full_coverage_test.go @@ -536,6 +536,7 @@ func TestMonitor_HostAgentManagement(t *testing.T) { m.nodePendingUpdatesCache = make(map[string]pendingUpdatesCache) // Test UnlinkHostAgent + m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil) err := m.UnlinkHostAgent("host1") if err != nil { t.Errorf("UnlinkHostAgent failed: %v", err)