Avoid estate-wide copies in live node and backup evidence lookups

Read only a uniquely matching node for agent filesystem admission and primary PVE task records for backup intent. Preserve exact identity, ambiguity refusal, detached ownership, task order and freshness rules; capture each replaceable state once. Add parity, concurrency, policy-boundary and allocation-scaling coverage, plus fixed-size benchmarks of both live entry points.

Contract-Neutral: Internal read projections only; collection, alert policy, emitted fields, persistence and frontend behavior are unchanged.
Change-source: pulse-maintainer

internal/models/metadata_snapshot.go

@@ -20,6 +20,15 @@ func (s *State) GetPVEBackups() PVEBackups {
 	return clonePVEBackups(s.PVEBackups)
 }
 
+// GetBackupTasks returns the primary PVE tasks in their existing order, without
+// copying backup artifacts or the resource inventory. BackupTask has only value
+// fields; the returned slice is detached, including when it is empty.
+func (s *State) GetBackupTasks() []BackupTask {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return append([]BackupTask{}, s.PVEBackups.BackupTasks...)
+}
+
 func (s *State) GetPBSBackups() []PBSBackup {
 	s.mu.RLock()
 	defer s.mu.RUnlock()

internal/models/models.go

@@ -5676,6 +5676,28 @@ func (s *State) GetNode(nodeID string) (Node, bool) {
 	return Node{}, false
 }
 
+// GetUniqueNode returns a detached node only when exactly one source record has
+// this ID. Callers that admit agent data must not let a duplicate ID choose its
+// destination, and need not copy unrelated resources to establish uniqueness.
+func (s *State) GetUniqueNode(nodeID string) (Node, bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	match := -1
+	for i := range s.Nodes {
+		if s.Nodes[i].ID == nodeID {
+			if match >= 0 {
+				return Node{}, false
+			}
+			match = i
+		}
+	}
+	if match < 0 {
+		return Node{}, false
+	}
+	return cloneNode(s.Nodes[match]), true
+}
+
 // GetHosts returns a copy of all generic hosts.
 func (s *State) GetHosts() []Host {
 	s.mu.RLock()

internal/models/point_reads_test.go

@@ -0,0 +1,113 @@
+package models
+
+import (
+	"encoding/json"
+	"reflect"
+	"sync"
+	"testing"
+)
+
+func TestPointReadsMatchSnapshotAndDetach(t *testing.T) {
+	enabled := true
+	speed := int64(1000)
+	s := metadataStateFixture()
+	s.Nodes = []Node{{
+		ID: "node-id", Instance: "pve", Name: "node-a",
+		LoadAverage: []float64{1}, NativeNameAliases: []string{"node-a.example"},
+		NetworkInterfaces:            []HostNetworkInterface{{Name: "eth0", Addresses: []string{"192.0.2.1"}, SpeedMbps: &speed}},
+		TemperatureMonitoringEnabled: &enabled,
+	}}
+	before, err := json.Marshal(s.GetSnapshot())
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, ok := s.GetUniqueNode("node-id")
+	if !ok || !reflect.DeepEqual(got, s.GetSnapshot().Nodes[0]) {
+		t.Fatal("unique node differs from its full-snapshot record")
+	}
+	tasks := s.GetBackupTasks()
+	if !reflect.DeepEqual(tasks, s.GetSnapshot().PVEBackups.BackupTasks) {
+		t.Fatal("task order or primary source changed")
+	}
+	got.LoadAverage[0] = 99
+	got.NativeNameAliases[0] = "reader"
+	got.NetworkInterfaces[0].Addresses[0] = "reader"
+	*got.NetworkInterfaces[0].SpeedMbps = 0
+	*got.TemperatureMonitoringEnabled = false
+	tasks[0].ID = "reader"
+	after, err := json.Marshal(s.GetSnapshot())
+	if err != nil || string(before) != string(after) {
+		t.Fatal("point reads share mutable source data")
+	}
+	for _, empty := range []*State{{}, NewState()} {
+		if got := empty.GetBackupTasks(); !reflect.DeepEqual(got, empty.GetSnapshot().PVEBackups.BackupTasks) {
+			t.Fatal("empty task collection changed")
+		}
+	}
+}
+
+func TestPointReadsUniqueNodeKeepsExactIdentityAndAmbiguity(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		nodes []Node
+		id    string
+		ok    bool
+	}{
+		{"missing", []Node{{ID: "other"}}, "node", false},
+		{"unique", []Node{{ID: "other"}, {ID: "node", Name: "chosen"}}, "node", true},
+		{"case-sensitive", []Node{{ID: "NODE"}}, "node", false},
+		{"whitespace-sensitive", []Node{{ID: " node "}}, "node", false},
+		{"different-duplicates", []Node{{ID: "node", Instance: "a"}, {ID: "node", Instance: "b"}}, "node", false},
+		{"identical-duplicates", []Node{{ID: "node", Instance: "a"}, {ID: "node", Instance: "a"}}, "node", false},
+		{"duplicates-after-other", []Node{{ID: "node"}, {ID: "other"}, {ID: "node"}}, "node", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s := NewState()
+			s.UpdateNodes(tc.nodes)
+			got, ok := s.GetUniqueNode(tc.id)
+			if ok != tc.ok || (!ok && !reflect.DeepEqual(got, Node{})) {
+				t.Fatalf("lookup = %+v, %t; want success %t", got, ok, tc.ok)
+			}
+			var matches []Node
+			for _, n := range s.GetSnapshot().Nodes {
+				if n.ID == tc.id {
+					matches = append(matches, n)
+				}
+			}
+			if (len(matches) == 1) != ok || (ok && !reflect.DeepEqual(got, matches[0])) {
+				t.Fatal("point read differs from previous exact-one snapshot rule")
+			}
+		})
+	}
+}
+
+func TestPointReadsConcurrentSourceReplacement(t *testing.T) {
+	s := metadataStateFixture()
+	s.Nodes = []Node{{ID: "node", Instance: "pve", Name: "node-a"}}
+	var workers sync.WaitGroup
+	workers.Add(2)
+	go func() {
+		defer workers.Done()
+		for i := 0; i < 100; i++ {
+			s.UpdateNodes([]Node{{ID: "node", Instance: "pve", Name: "node-a"}})
+			s.UpdateBackupTasksForInstance("pve", []BackupTask{{ID: "task", Instance: "pve", VMID: 100}})
+		}
+	}()
+	go func() {
+		defer workers.Done()
+		for i := 0; i < 100; i++ {
+			node, ok := s.GetUniqueNode("node")
+			if !ok || node.Instance != "pve" {
+				t.Error("unique node lost during locked replacement")
+				return
+			}
+			tasks := s.GetBackupTasks()
+			if len(tasks) != 1 || tasks[0].ID != "task" {
+				t.Error("task read lost during locked replacement")
+				return
+			}
+			tasks[0].ID = "reader"
+		}
+	}()
+	workers.Wait()
+}

internal/monitoring/monitor_agent_lxc_filesystems.go

@@ -59,7 +59,7 @@ func (m *Monitor) applyAgentLXCFilesystems(
 	intervalSeconds int,
 ) {
 	agentID = strings.TrimSpace(agentID)
-	if m == nil || m.state == nil || inventory == nil ||
+	if m == nil || inventory == nil ||
 		strings.TrimSpace(linkedNodeID) == "" || agentID == "" {
 		return
 	}
@@ -67,16 +67,18 @@ func (m *Monitor) applyAgentLXCFilesystems(
 		return
 	}
 
-	var linkedNodes []models.Node
-	for _, node := range m.state.GetSnapshot().Nodes {
-		if node.ID == linkedNodeID {
-			linkedNodes = append(linkedNodes, node)
-		}
-	}
-	if len(linkedNodes) != 1 {
+	// This is live agent admission, not a full inventory or mock-mode read.
+	// Capture the replaceable state once and copy only the unique linked node.
+	m.mu.RLock()
+	state := m.state
+	m.mu.RUnlock()
+	if state == nil {
+		return
+	}
+	node, unique := state.GetUniqueNode(linkedNodeID)
+	if !unique {
 		return
 	}
-	node := linkedNodes[0]
 	if strings.TrimSpace(node.Instance) == "" || strings.TrimSpace(node.Name) == "" {
 		return
 	}

internal/monitoring/monitor_alert_intent.go

@@ -216,7 +216,13 @@ func (i *operatorIntentIdentity) build(generation unifiedresources.RegistryGener
 }
 
 func (m *Monitor) resolveBackupIntentContext(_ string, instance, node string, vmid int, now time.Time) (alerts.BackupIntentContext, bool) {
-	if m == nil || m.state == nil || vmid <= 0 {
+	if m == nil || vmid <= 0 {
+		return alerts.BackupIntentContext{}, false
+	}
+	m.mu.RLock()
+	state := m.state
+	m.mu.RUnlock()
+	if state == nil {
 		return alerts.BackupIntentContext{}, false
 	}
 	instance = strings.TrimSpace(instance)
@@ -225,7 +231,9 @@ func (m *Monitor) resolveBackupIntentContext(_ string, instance, node string, vm
 		now = time.Now().UTC()
 	}
 
-	for _, task := range m.state.GetSnapshot().PVEBackups.BackupTasks {
+	// Policy needs only these primary task records, not every guest, sensor,
+	// backup artifact and Kubernetes object in the monitored estate.
+	for _, task := range state.GetBackupTasks() {
 		if task.VMID != vmid || (instance != "" && task.Instance != instance) || (node != "" && task.Node != "" && task.Node != node) {
 			continue
 		}

internal/monitoring/monitor_live_point_reads_test.go

@@ -0,0 +1,235 @@
+package monitoring
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+	"github.com/rcourtman/pulse-go-rewrite/internal/mock"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
+)
+
+var livePointReadTime = time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+
+func livePointReadFixture(size int) (*Monitor, *agentshost.ProxmoxLXCInventory) {
+	m := metadataMonitorFixture(size)
+	m.state.Nodes = []models.Node{{ID: "node-id", Instance: "pve-a", Name: "node-a", LinkedAgentID: "agent-a"}}
+	m.state.PVEBackups.BackupTasks = []models.BackupTask{{
+		ID: "active", Instance: "pve-a", Node: "node-a", VMID: 100, Status: "running", ObservedAt: livePointReadTime,
+	}}
+	for i := 0; i < size; i++ {
+		m.state.PVEBackups.StorageBackups = append(m.state.PVEBackups.StorageBackups,
+			models.StorageBackup{ID: fmt.Sprint(i), Notes: "unrelated restore point"})
+	}
+	return m, &agentshost.ProxmoxLXCInventory{Status: agentshost.ProxmoxLXCCollectionComplete,
+		Containers: []agentshost.ProxmoxLXCContainer{{VMID: 100, Name: "web", Disks: []agentshost.Disk{
+			{Type: "rootfs", Mountpoint: "/", Device: "local:100/root", TotalBytes: 100, UsedBytes: 25, FreeBytes: 75},
+			{Type: "mp0", Mountpoint: "/srv/data", Device: "tank:100/data", TotalBytes: 1000, UsedBytes: 500, FreeBytes: 500},
+		}}}}
+}
+
+func TestLivePointReadsBackupPolicyBoundaries(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		change   func(*models.BackupTask)
+		instance string
+		node     string
+		vmid     int
+		found    bool
+	}{
+		{"fresh", nil, "pve-a", "node-a", 100, true},
+		{"trimmed-query", nil, " pve-a ", " node-a ", 100, true},
+		{"wrong-instance", nil, "pve-b", "node-a", 100, false},
+		{"wrong-node", nil, "pve-a", "node-b", 100, false},
+		{"wrong-vmid", nil, "pve-a", "node-a", 101, false},
+		{"invalid-vmid", nil, "pve-a", "node-a", 0, false},
+		{"empty-query-hints", nil, "", "", 100, true},
+		{"missing-task-node", func(task *models.BackupTask) { task.Node = "" }, "pve-a", "node-a", 100, true},
+		{"exact-age-boundary", func(task *models.BackupTask) { task.ObservedAt = livePointReadTime.Add(-backupIntentEvidenceMaxAge) }, "pve-a", "node-a", 100, true},
+		{"too-old", func(task *models.BackupTask) {
+			task.ObservedAt = livePointReadTime.Add(-backupIntentEvidenceMaxAge - time.Nanosecond)
+		}, "pve-a", "node-a", 100, false},
+		{"unobserved", func(task *models.BackupTask) { task.ObservedAt = time.Time{} }, "pve-a", "node-a", 100, false},
+		{"future-boundary", func(task *models.BackupTask) { task.ObservedAt = livePointReadTime.Add(time.Minute) }, "pve-a", "node-a", 100, true},
+		{"too-far-future", func(task *models.BackupTask) { task.ObservedAt = livePointReadTime.Add(time.Minute + time.Nanosecond) }, "pve-a", "node-a", 100, false},
+		{"end-time", func(task *models.BackupTask) { task.EndTime = livePointReadTime }, "pve-a", "node-a", 100, false},
+		{"ok", func(task *models.BackupTask) { task.Status = " OK " }, "pve-a", "node-a", 100, false},
+		{"stopped", func(task *models.BackupTask) { task.Status = "Stopped" }, "pve-a", "node-a", 100, false},
+		{"error", func(task *models.BackupTask) { task.Status = "error" }, "pve-a", "node-a", 100, false},
+		{"warning", func(task *models.BackupTask) { task.Status = "WARNING" }, "pve-a", "node-a", 100, false},
+		// Preserve the existing resolver's non-terminal rule, not a new status policy.
+		{"empty-status", func(task *models.BackupTask) { task.Status = "" }, "pve-a", "node-a", 100, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m, _ := livePointReadFixture(0)
+			if tc.change != nil {
+				tc.change(&m.state.PVEBackups.BackupTasks[0])
+			}
+			got, found := m.resolveBackupIntentContext("ignored", tc.instance, tc.node, tc.vmid, livePointReadTime)
+			want := alerts.BackupIntentContext{}
+			if tc.found {
+				want = alerts.BackupIntentContext{Active: true, ObservedAt: m.state.PVEBackups.BackupTasks[0].ObservedAt, Evidence: "pve_vzdump_task:active"}
+			}
+			if found != tc.found || !reflect.DeepEqual(got, want) {
+				t.Fatalf("lookup = %+v, %t; want %+v, %t", got, found, want, tc.found)
+			}
+		})
+	}
+	m, _ := livePointReadFixture(0)
+	m.state.PVEBackups.BackupTasks = append(m.state.PVEBackups.BackupTasks,
+		models.BackupTask{ID: "later", VMID: 100, Instance: "pve-a", Status: "running", ObservedAt: livePointReadTime})
+	got, found := m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime)
+	if !found || got.Evidence != "pve_vzdump_task:active" {
+		t.Fatal("first matching task no longer wins in source order")
+	}
+	m.state.PVEBackups.BackupTasks[0].ObservedAt = time.Now()
+	if _, found := m.resolveBackupIntentContext("", "pve-a", "node-a", 100, time.Time{}); !found {
+		t.Fatal("zero caller clock no longer uses current time")
+	}
+}
+
+func TestLivePointReadsNilAndMockSemantics(t *testing.T) {
+	previous := mock.IsMockEnabled()
+	defer mustSetMockEnabled(t, previous)
+	for _, enabled := range []bool{false, true} {
+		mustSetMockEnabled(t, enabled)
+		for _, m := range []*Monitor{nil, {}, {state: models.NewState()}} {
+			if got, found := m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime); found || got.Active {
+				t.Fatal("absent live state invented backup evidence")
+			}
+			m.applyAgentLXCFilesystems("node-id", "agent-a", &agentshost.ProxmoxLXCInventory{}, livePointReadTime, 30)
+		}
+		// These internal live-state queries have never substituted mock fixtures.
+		m, inventory := livePointReadFixture(0)
+		m.applyAgentLXCFilesystems("node-id", "agent-a", inventory, livePointReadTime, 30)
+		if len(m.proxmoxLXCFilesystemsCache) != 1 {
+			t.Fatal("live admission unexpectedly read a mock inventory")
+		}
+		if _, found := m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime); !found {
+			t.Fatal("live task lookup unexpectedly read a mock inventory")
+		}
+	}
+}
+
+func TestLivePointReadsConcurrentMonitorStateReplacement(t *testing.T) {
+	m, inventory := livePointReadFixture(0)
+	other, _ := livePointReadFixture(0)
+	states := []*models.State{m.state, other.state, nil}
+	var workers sync.WaitGroup
+	workers.Add(2)
+	go func() {
+		defer workers.Done()
+		for i := 0; i < 100; i++ {
+			m.mu.Lock()
+			m.state = states[i%len(states)]
+			m.mu.Unlock()
+		}
+	}()
+	go func() {
+		defer workers.Done()
+		for i := 0; i < 100; i++ {
+			m.applyAgentLXCFilesystems("node-id", "agent-a", inventory, livePointReadTime, 30)
+			m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime)
+		}
+	}()
+	workers.Wait()
+}
+
+// The same benchmark compiles against the candidate and exact former entry
+// point sources. Fixed requested data; only unrelated estate/artifacts grow.
+func BenchmarkLivePointReads(b *testing.B) {
+	for _, size := range []int{0, 100, 1000} {
+		m, inventory := livePointReadFixture(size)
+		b.Run(fmt.Sprintf("agent-filesystems/%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				m.applyAgentLXCFilesystems("node-id", "agent-a", inventory, livePointReadTime, 30)
+			}
+		})
+		b.Run(fmt.Sprintf("backup-intent/%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				livePointReadContext, livePointReadFound = m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime)
+			}
+		})
+	}
+}
+
+var livePointReadContext alerts.BackupIntentContext
+var livePointReadFound bool
+
+func TestLivePointReadsAllocationScaling(t *testing.T) {
+	small, inventory := livePointReadFixture(0)
+	large, _ := livePointReadFixture(1000)
+	for _, read := range []struct {
+		name string
+		call func(*Monitor)
+	}{
+		{"agent-filesystems", func(m *Monitor) { m.applyAgentLXCFilesystems("node-id", "agent-a", inventory, livePointReadTime, 30) }},
+		{"backup-intent", func(m *Monitor) {
+			livePointReadContext, livePointReadFound = m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime)
+		}},
+	} {
+		t.Run(read.name, func(t *testing.T) {
+			baseline := testing.AllocsPerRun(10, func() { read.call(small) })
+			grown := testing.AllocsPerRun(10, func() { read.call(large) })
+			if grown != baseline {
+				t.Fatalf("unrelated estate increased allocations: %.0f -> %.0f", baseline, grown)
+			}
+		})
+	}
+}
+
+// Export actual admission -> enrichment -> frontend-projection bytes, plus the
+// alert resolver result, for an exact baseline/candidate comparison. No browser,
+// installed estate, native helper or reporter incident is represented by this.
+func TestLivePointReadsProducerParityExport(t *testing.T) {
+	for _, scenario := range []string{"complete", "partial", "expired", "ambiguous", "missing", "renamed"} {
+		m, inventory := livePointReadFixture(0)
+		if scenario == "ambiguous" {
+			m.state.Nodes = append(m.state.Nodes, models.Node{ID: "node-id", Instance: "other", Name: "other"})
+		} else if scenario == "missing" {
+			m.state.Nodes = nil
+		} else if scenario == "partial" {
+			inventory.Status = agentshost.ProxmoxLXCCollectionPartial
+			inventory.OmittedVMIDs = []int{101}
+		}
+		m.applyAgentLXCFilesystems("node-id", "agent-a", inventory, livePointReadTime, 30)
+		guest := models.Container{ID: "pve-a:node-a:100", VMID: 100, Instance: "pve-a", Node: "node-a", Name: "web", Status: "running", Disk: models.Disk{Total: 100, Used: 10, Free: 90, Mountpoint: "/"}}
+		readAt := livePointReadTime.Add(time.Minute)
+		if scenario == "expired" {
+			readAt = livePointReadTime.Add(agentLXCFilesystemMinTTL)
+		} else if scenario == "renamed" {
+			guest.Name = "renamed"
+		}
+		m.enrichContainerWithAgentLXCFilesystems("pve-a", "node-a", &guest, readAt)
+		wantDisks := 0
+		if scenario == "complete" || scenario == "partial" {
+			wantDisks = 2
+		}
+		if len(guest.Disks) != wantDisks || (wantDisks == 2 && guest.Disk.Used != 25) || (wantDisks == 0 && guest.Disk.Used != 10) {
+			t.Fatalf("%s changed admission/enrichment: %+v", scenario, guest)
+		}
+		m.state.UpdateContainers([]models.Container{guest})
+		m.state.LastUpdate = livePointReadTime
+		backup, found := m.resolveBackupIntentContext("", "pve-a", "node-a", 100, livePointReadTime)
+		frame := struct {
+			Scenario string                     `json:"scenario"`
+			State    models.StateFrontend       `json:"state"`
+			Backup   alerts.BackupIntentContext `json:"backup"`
+			Found    bool                       `json:"found"`
+		}{scenario, buildFrontendStateFromSnapshot(m.state.GetSnapshot()), backup, found}
+		data, err := json.Marshal(frame)
+		if err != nil {
+			t.Fatal(err)
+		}
+		fmt.Printf("POINT_READ_PRODUCER=%s\n", data)
+	}
+}