diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index a595a8cf2..ff1ab4719 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -27,6 +27,14 @@ Monitoring owns source freshness cadence for Proxmox, PBS, and PMG resources: the stale threshold is derived from the configured polling interval with a minimum floor, so API-facing resource status must not degrade merely because a healthy source is between normal poll cycles. +Proxmox guest enumeration is a generation boundary. VM and LXC collection and +enrichment must finish before one `State.UpdateGuestsForInstance` publication, +so readers never observe a VM-only or LXC-only intermediate snapshot. A failed +online cluster member retains only that member's last coherent guests and their +source-native `{instance}:{node}:{vmid}` IDs; a successful empty member +enumeration is authoritative and removes genuinely deleted guests. Collection +failure remains visible through source freshness/error state and must not be +converted into an authoritative empty inventory. Host-agent report liveness is server-observed, not agent-clock-observed: `ApplyHostReport` must stamp `Host.LastSeen`, agent-sourced Ceph cluster freshness, and host-agent cluster sensor freshness from Pulse receipt time, so @@ -213,6 +221,7 @@ node-local Agent evidence. 36. `internal/dockeragent/docker_client.go` 37. `pkg/agents/docker/report.go` 38. `internal/models/models.go` +38a. `internal/models/proxmox_guest_state.go` 39. `internal/models/models_frontend.go` 40. `internal/models/converters.go` 41. `internal/models/deepcopy.go` diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 4b1e2c70f..638d03e3c 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -854,6 +854,19 @@ shell clickable behind another overlay. ## Current State +### Workload refreshes retain one coherent paged generation + +The workload polling cache accepts a REST refresh only when every advertised +page succeeds. A failed later page leaves the prior array and its stable row +identities in place, ends loading, and exposes the refresh error; it must not +publish a partial list, clear rows as a loading transition, or silently hide +the error. Proxmox workload filtering consumes the source-authored +`proxmox.runtimeStatus` before aggregate health so an availability-only +freshness update cannot collapse a Running view. A later complete response may +still remove an authoritatively deleted guest. These rules preserve sort, +selection, drawer, and virtualized viewport state without adding another +resource scan, websocket subscription, or browser-local source of truth. + ### Canonical mutation-plane dependency Router wiring now exposes only typed action planning for model-originated diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index 590c0cd3a..51401af61 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -5186,6 +5186,7 @@ "internal/models/deepcopy.go", "internal/models/models.go", "internal/models/models_frontend.go", + "internal/models/proxmox_guest_state.go", "internal/proxmoxidentity/backup_identity.go", "pkg/agents/docker/report.go", "pkg/agents/host/report.go", @@ -5398,6 +5399,7 @@ "internal/models/deepcopy.go", "internal/models/models.go", "internal/models/models_frontend.go", + "internal/models/proxmox_guest_state.go", "pkg/agents/docker/report.go" ], "allow_same_subsystem_tests": false, diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 125f6c028..2b0c01fed 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -1917,6 +1917,17 @@ that Safe auto-fix or Autopilot remediation is verified. ## Current State +### Proxmox runtime continuity is not protection evidence + +The additive `ProxmoxData.RuntimeStatus` field preserves VM/LXC power-state +presentation while a platform snapshot is stale or refreshing. Storage and +recovery consumers must not interpret `running`, `stopped`, a retained guest +row, or an availability facet as backup freshness, protection coverage, +restore-chain verification, or recoverability. Authoritative guest deletion +may remove the inventory row and its canonical history identity, while +Recovery Assurance remains governed solely by its own backup and verification +evidence. + ### Storage history remount state is bounded Storage summary history no longer treats every node/range combination visited diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 66a5d498e..8146e0315 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -43,6 +43,18 @@ comparable fallback because LXC observes the shared kernel and QEMU accounting is independently scoped. Agent CPU may fill the field only when the platform has no CPU observation. Agent-only fields that the platform does not provide remain eligible per metric. +Proxmox guest power state is likewise source-authored semantic state, separate +from aggregate resource health. `ProxmoxData.RuntimeStatus` retains the last +coherent `running`/`stopped` observation while `Resource.Status` and +`SourceStatus` continue to report stale, warning, offline, availability, and +error evidence honestly. Availability checks may add facets and influence +aggregate health after Proxmox becomes stale, but they must not overwrite or +erase the Proxmox runtime state used by workload power filters. Full monitor +registry generations and incremental agent/availability mutations serialize +at the adapter boundary; a mutation that arrives during a rebuild must apply to +the newly published registry rather than a superseded pointer. Authoritative +snapshot omissions still remove resources and emit the normal canonical +history change. Physical-disk resources own cross-source disk identity. When Proxmox inventory and host-agent SMART telemetry describe the same device, the merged resource must retain Proxmox node/instance source payloads while carrying SMART diff --git a/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts b/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts index a8bdf6d92..83b06950d 100644 --- a/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts +++ b/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts @@ -228,6 +228,126 @@ describe('useWorkloads', () => { dispose(); }); + it('keeps Proxmox power state stable while aggregate freshness changes, then removes an authoritative deletion', async () => { + const guest = (vmid: number, status: string, runtimeStatus: string) => ({ + ...sampleResource, + id: `cluster-a-pve1-${vmid}`, + type: 'system-container', + name: `lxc-${vmid}`, + status, + vmid, + proxmox: { + vmid, + nodeName: 'pve1', + instance: 'cluster-a', + runtimeStatus, + }, + }); + + apiFetchJSONMock.mockResolvedValueOnce({ + data: [ + guest(101, 'online', 'running'), + guest(102, 'online', 'running'), + guest(103, 'online', 'running'), + ], + meta: { totalPages: 1 }, + }); + + let dispose = () => {}; + let result: ReturnType | undefined; + createRoot((d) => { + dispose = d; + const [enabled] = createSignal(true); + result = useWorkloads(enabled); + }); + + await waitForWorkloadCount(() => result!.workloads().length, 3); + const initialIds = result!.workloads().map((workload) => workload.id); + + apiFetchJSONMock.mockResolvedValueOnce({ + data: [ + guest(101, 'warning', 'running'), + { + ...guest(102, 'online', 'running'), + availability: { targetId: 'probe-102', protocol: 'icmp', enabled: true, available: true }, + }, + guest(103, 'warning', 'running'), + ], + meta: { totalPages: 1 }, + }); + await result!.refetch(); + + expect(result!.workloads().map((workload) => workload.id)).toEqual(initialIds); + expect(result!.workloads().map((workload) => workload.status)).toEqual([ + 'running', + 'running', + 'running', + ]); + + apiFetchJSONMock.mockResolvedValueOnce({ + data: [guest(101, 'online', 'running'), guest(102, 'online', 'running')], + meta: { totalPages: 1 }, + }); + await result!.refetch(); + + expect(result!.workloads().map((workload) => workload.id)).toEqual(initialIds.slice(0, 2)); + expect(result!.error()).toBeUndefined(); + + dispose(); + }); + + it('rejects a partial paged refresh and retains the last coherent snapshot', async () => { + const secondResource = { + ...sampleResource, + id: 'cluster-a-pve1-102', + name: 'vm-102', + vmid: 102, + }; + apiFetchJSONMock.mockResolvedValueOnce({ + data: [sampleResource, secondResource], + meta: { totalPages: 1 }, + }); + + let dispose = () => {}; + let result: ReturnType | undefined; + createRoot((d) => { + dispose = d; + const [enabled] = createSignal(true); + result = useWorkloads(enabled); + }); + + await waitForWorkloadCount(() => result!.workloads().length, 2); + const coherentSnapshot = result!.workloads(); + + apiFetchJSONMock + .mockResolvedValueOnce({ + data: [sampleResource], + meta: { totalPages: 2 }, + }) + .mockRejectedValueOnce(new Error('page 2 unavailable')); + + await expect(result!.refetch()).rejects.toThrow('page 2 unavailable'); + expect(result!.workloads()).toBe(coherentSnapshot); + expect(result!.workloads()).toHaveLength(2); + expect(result!.error()).toBeInstanceOf(Error); + + apiFetchJSONMock + .mockResolvedValueOnce({ + data: [sampleResource], + meta: { totalPages: 2 }, + }) + .mockResolvedValueOnce({ + data: [], + meta: { totalPages: 2 }, + }); + await result!.refetch(); + + expect(result!.workloads()).toHaveLength(1); + expect(result!.error()).toBeUndefined(); + + dispose(); + }); + it('does not apply in-flight workload results after the hook is disabled', async () => { const pendingFetch = deferred(); apiFetchJSONMock.mockImplementationOnce(() => pendingFetch.promise as Promise); diff --git a/frontend-modern/src/hooks/useWorkloads.ts b/frontend-modern/src/hooks/useWorkloads.ts index c0e375112..c3d5e7bad 100644 --- a/frontend-modern/src/hooks/useWorkloads.ts +++ b/frontend-modern/src/hooks/useWorkloads.ts @@ -87,6 +87,7 @@ type APIResource = { node?: string; instance?: string; proxmox?: { + runtimeStatus?: string; nodeName?: string; clusterName?: string; instance?: string; @@ -451,7 +452,9 @@ const mapResourceToWorkload = (resource: APIResource): WorkloadGuest | null => { node, instance, status: normalizeWorkloadStatus( - resource.status || (platformType === 'vmware-vsphere' ? resource.vmware?.powerState : null), + resource.proxmox?.runtimeStatus || + resource.status || + (platformType === 'vmware-vsphere' ? resource.vmware?.powerState : null), ), type: workloadType === 'vm' @@ -587,10 +590,9 @@ async function fetchWorkloads(): Promise { for (let page = 2; page <= totalPages; page++) { pageRequests.push(apiFetchJSON(buildWorkloadsUrl(page), { cache: 'no-store' })); } - const settled = await Promise.allSettled(pageRequests); - for (const result of settled) { - if (result.status !== 'fulfilled') continue; - const pageData = resolveWorkloadsPayload(result.value); + const responses = await Promise.all(pageRequests); + for (const response of responses) { + const pageData = resolveWorkloadsPayload(response); allResources.push(...pageData.data); } } @@ -775,8 +777,12 @@ export function useWorkloads(enabled: Accessor = () => true) { } applyWorkloads(data, scope); setError(undefined); - } catch { - // Silently ignore poll errors; keep showing last data + } catch (err) { + // Keep the last coherent snapshot while making the failed refresh + // observable to the surface. + if (scope === resolveActiveOrgScope()) { + setError(err); + } } }, DEFAULT_POLL_INTERVAL_MS); onCleanup(() => clearInterval(id)); diff --git a/frontend-modern/src/types/__tests__/resource.test.ts b/frontend-modern/src/types/__tests__/resource.test.ts index 512a07bf7..5246a23e9 100644 --- a/frontend-modern/src/types/__tests__/resource.test.ts +++ b/frontend-modern/src/types/__tests__/resource.test.ts @@ -240,6 +240,7 @@ describe('Resource Helper Functions', () => { vmid: 101, nodeName: 'pve-a', instance: 'cluster-a', + runtimeStatus: 'running', diskStatusReason: 'agent-not-running', guestAgentStatus: 'expected-unreachable', guestAgentExpected: true, @@ -262,6 +263,7 @@ describe('Resource Helper Functions', () => { expect(resource.proxmox?.guestAgentStatus).toBe('expected-unreachable'); expect(resource.proxmox?.guestAgentExpected).toBe(true); + expect(resource.proxmox?.runtimeStatus).toBe('running'); expect(resource.proxmox?.diskStatusReason).toBe('agent-not-running'); expect(resource.incidents?.[0]?.source).toBe('qemu-guest-agent'); }); diff --git a/frontend-modern/src/types/resource.ts b/frontend-modern/src/types/resource.ts index ff9488be6..a2ec33409 100644 --- a/frontend-modern/src/types/resource.ts +++ b/frontend-modern/src/types/resource.ts @@ -644,6 +644,8 @@ export interface ResourceAgentMeta { } export interface ResourceProxmoxMeta { + /** Proxmox-authored VM/LXC power state, kept separate from collection freshness. */ + runtimeStatus?: string; vmid?: number; node?: string; nodeName?: string; diff --git a/internal/api/resources_test.go b/internal/api/resources_test.go index e142a154b..09d8dae4b 100644 --- a/internal/api/resources_test.go +++ b/internal/api/resources_test.go @@ -779,11 +779,12 @@ func TestResourceListDerivesProxmoxWorkloadParentFromUnifiedSeed(t *testing.T) { Sources: []unified.DataSource{unified.SourceProxmox}, Identity: unified.ResourceIdentity{Hostnames: []string{"cloudflared"}}, Proxmox: &unified.ProxmoxData{ - SourceID: "delly:delly:104", - NodeName: "delly", - ClusterName: "homelab", - Instance: "delly", - VMID: 104, + SourceID: "delly:delly:104", + RuntimeStatus: "running", + NodeName: "delly", + ClusterName: "homelab", + Instance: "delly", + VMID: 104, }, }, }, @@ -815,6 +816,9 @@ func TestResourceListDerivesProxmoxWorkloadParentFromUnifiedSeed(t *testing.T) { if resource.Proxmox == nil || resource.Proxmox.NodeName != "delly" || resource.Proxmox.ClusterName != "homelab" { t.Fatalf("expected cloudflared Proxmox node metadata, got %+v", resource.Proxmox) } + if resource.Proxmox.RuntimeStatus != "running" { + t.Fatalf("expected source-authored runtime status, got %+v", resource.Proxmox) + } } func TestStateEndpointDerivesProxmoxWorkloadParentFromSupplementalRecords(t *testing.T) { diff --git a/internal/models/proxmox_guest_state.go b/internal/models/proxmox_guest_state.go new file mode 100644 index 000000000..467b4d65e --- /dev/null +++ b/internal/models/proxmox_guest_state.go @@ -0,0 +1,34 @@ +package models + +import "time" + +// UpdateGuestsForInstance replaces the VM and container projections for one +// Proxmox instance under a single state lock. Pollers collect and enrich both +// guest kinds before calling this method so readers cannot observe a mixed +// generation while a refresh is in flight. +func (s *State) UpdateGuestsForInstance(instanceName string, vms []VM, containers []Container) { + s.mu.Lock() + defer s.mu.Unlock() + + s.VMs = updateSliceByInstanceWithBackup( + s.VMs, vms, instanceName, + func(vm VM) string { return vm.ID }, + func(vm VM) string { return vm.Instance }, + func(vm VM) int { return vm.VMID }, + func(vm VM) time.Time { return vm.LastBackup }, + func(vm VM, t time.Time) VM { vm.LastBackup = t; return vm }, + cloneVM, + func(items []VM, i, j int) bool { return items[i].VMID < items[j].VMID }, + ) + s.Containers = updateSliceByInstanceWithBackup( + s.Containers, containers, instanceName, + func(ct Container) string { return ct.ID }, + func(ct Container) string { return ct.Instance }, + func(ct Container) int { return ct.VMID }, + func(ct Container) time.Time { return ct.LastBackup }, + func(ct Container, t time.Time) Container { ct.LastBackup = t; return ct }, + cloneContainer, + func(items []Container, i, j int) bool { return items[i].VMID < items[j].VMID }, + ) + s.LastUpdate = time.Now() +} diff --git a/internal/models/state_host_test.go b/internal/models/state_host_test.go index 9d3b249a2..e57f77437 100644 --- a/internal/models/state_host_test.go +++ b/internal/models/state_host_test.go @@ -1327,6 +1327,44 @@ func TestSyncGuestBackupTimesVMContainerCollision(t *testing.T) { } } +func TestUpdateGuestsForInstancePublishesCoherentGeneration(t *testing.T) { + state := NewState() + previousBackup := time.Now().Add(-2 * time.Hour).UTC() + + state.UpdateVMs([]VM{ + {ID: "lab-a:node-a:101", VMID: 101, Instance: "lab-a", Node: "node-a", LastBackup: previousBackup}, + {ID: "lab-b:node-b:201", VMID: 201, Instance: "lab-b", Node: "node-b"}, + }) + state.UpdateContainers([]Container{ + {ID: "lab-a:node-a:102", VMID: 102, Instance: "lab-a", Node: "node-a", LastBackup: previousBackup}, + {ID: "lab-b:node-b:202", VMID: 202, Instance: "lab-b", Node: "node-b"}, + }) + before := state.GetSnapshot().LastUpdate + + state.UpdateGuestsForInstance( + "lab-a", + []VM{{ID: "lab-a:node-a:101", VMID: 101, Instance: "lab-a", Node: "node-a", Status: "running"}}, + []Container{{ID: "lab-a:node-a:103", VMID: 103, Instance: "lab-a", Node: "node-a", Status: "running"}}, + ) + + snapshot := state.GetSnapshot() + if !snapshot.LastUpdate.After(before) { + t.Fatalf("LastUpdate did not advance: before=%v after=%v", before, snapshot.LastUpdate) + } + if len(snapshot.VMs) != 2 || len(snapshot.Containers) != 2 { + t.Fatalf("unexpected coherent generation sizes: vms=%d containers=%d", len(snapshot.VMs), len(snapshot.Containers)) + } + if snapshot.VMs[0].ID != "lab-a:node-a:101" || !snapshot.VMs[0].LastBackup.Equal(previousBackup) { + t.Fatalf("updated VM did not retain backup state: %+v", snapshot.VMs[0]) + } + if snapshot.Containers[0].ID != "lab-a:node-a:103" { + t.Fatalf("authoritatively deleted container remained in state: %+v", snapshot.Containers) + } + if snapshot.VMs[1].Instance != "lab-b" || snapshot.Containers[1].Instance != "lab-b" { + t.Fatalf("other instance was not isolated: vms=%+v containers=%+v", snapshot.VMs, snapshot.Containers) + } +} + func TestUpdateStorageBackupsForInstance(t *testing.T) { state := NewState() diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index f57874ee5..86e0e44d7 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -1110,11 +1110,14 @@ func TestProxmoxGuestDockerInventoryUsesCanonicalReportIngestPath(t *testing.T) }, "monitor_pve_guest_poll.go": { "m.CollectProxmoxGuestDockerInventory(ctx, allContainers)", - "m.state.UpdateContainersForInstance(instanceName, allContainers)", + "m.state.UpdateGuestsForInstance(instanceName, allVMs, allContainers)", }, "monitor_polling_containers.go": { "m.CollectProxmoxGuestDockerInventory(ctx, allContainers)", - "m.state.UpdateContainersForInstance(instanceName, allContainers)", + "return allContainers", + }, + "monitor_pve.go": { + "m.state.UpdateGuestsForInstance(instanceName, vms, containers)", }, } diff --git a/internal/monitoring/memory_trust_characterization_test.go b/internal/monitoring/memory_trust_characterization_test.go index b48ef5d6b..e3947e1c1 100644 --- a/internal/monitoring/memory_trust_characterization_test.go +++ b/internal/monitoring/memory_trust_characterization_test.go @@ -22,6 +22,38 @@ type vmMemoryTrustStubClient struct { vmAgentMemCalls int } +func TestCollectVMsWithNodesRetainsFailedNodeRuntimeState(t *testing.T) { + monitor := newTestPVEMonitor("lab") + defer monitor.alertManager.Stop() + defer monitor.notificationMgr.Stop() + monitor.state.UpdateVMsForInstance("lab", []models.VM{{ + ID: "lab:node-b:201", + VMID: 201, + Name: "database", + Node: "node-b", + Instance: "lab", + Status: "running", + }}) + + client := &partialNodeGuestClient{ + stubPVEClient: &stubPVEClient{}, + failedNodes: map[string]bool{"node-b": true}, + } + vms := monitor.collectVMsWithNodes( + context.Background(), + "lab", + "", + false, + client, + []proxmox.Node{{Node: "node-b", Status: "online"}}, + map[string]string{"node-b": "online"}, + ) + + if len(vms) != 1 || vms[0].ID != "lab:node-b:201" || vms[0].Status != "running" { + t.Fatalf("failed-node VM continuity = %+v, want retained source ID and running state", vms) + } +} + func (s *vmMemoryTrustStubClient) GetVMs(ctx context.Context, node string) ([]proxmox.VM, error) { return s.vms, nil } diff --git a/internal/monitoring/monitor_extra_coverage_test.go b/internal/monitoring/monitor_extra_coverage_test.go index d3da293eb..2f49adcf1 100644 --- a/internal/monitoring/monitor_extra_coverage_test.go +++ b/internal/monitoring/monitor_extra_coverage_test.go @@ -1172,12 +1172,12 @@ func TestMonitor_PreviousGuestContextForInstance_Extra(t *testing.T) { if len(prev.vms) != 1 || prev.vms[0].VMID != 101 || prev.vms[0].Instance != "pve1" || prev.vms[0].Name != "vm1" { t.Fatalf("expected only pve1 VMs, got %#v", prev.vms) } - canonicalID := prev.vms[0].ID - if len(prev.vmsByID) != 2 || prev.vmsByID[canonicalID].VMID != 101 || prev.vmsByID[makeGuestID("pve1", "", 101)].VMID != 101 { - t.Fatalf("expected previous VM lookup to be indexed by canonical and runtime guest IDs, got %#v", prev.vmsByID) + guestID := makeGuestID("pve1", "", 101) + if prev.vms[0].ID != guestID || len(prev.vmsByID) != 1 || prev.vmsByID[guestID].VMID != 101 { + t.Fatalf("expected previous VM lookup to retain the source-authored guest ID, got %#v", prev.vmsByID) } - if prev.vmsByID[canonicalID].Disk.Total != 100 || prev.vmsByID[canonicalID].Disk.Used != 40 { - t.Fatalf("expected previous VM projection to preserve aggregate disk summary, got %#v", prev.vmsByID[canonicalID].Disk) + if prev.vmsByID[guestID].Disk.Total != 100 || prev.vmsByID[guestID].Disk.Used != 40 { + t.Fatalf("expected previous VM projection to preserve aggregate disk summary, got %#v", prev.vmsByID[guestID].Disk) } if len(prev.containers) != 2 { t.Fatalf("expected only pve1 containers, got %#v", prev.containers) diff --git a/internal/monitoring/monitor_polling_containers.go b/internal/monitoring/monitor_polling_containers.go index 9cbad2e16..6df34aaa7 100644 --- a/internal/monitoring/monitor_polling_containers.go +++ b/internal/monitoring/monitor_polling_containers.go @@ -13,7 +13,7 @@ import ( "github.com/rs/zerolog/log" ) -func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) { +func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) []models.Container { startTime := time.Now() // Channel to collect container results from each node @@ -303,10 +303,12 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri lxcTemplateSubjects := make(map[string]struct{}) successfulNodes := 0 failedNodes := 0 + failedNodeNames := make(map[string]struct{}) for result := range resultChan { if result.err != nil { failedNodes++ + failedNodeNames[result.node] = struct{}{} } else { successfulNodes++ allContainers = append(allContainers, result.containers...) @@ -319,28 +321,27 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects) } - // If we got ZERO containers but had containers before (likely cluster health issue), - // preserve previous containers instead of clearing them - if len(allContainers) == 0 && len(nodes) > 0 { - allContainers = append(allContainers, prevGuests.containers...) - prevContainerCount := len(prevGuests.containers) - if prevContainerCount > 0 { - log.Warn(). - Str("instance", instanceName). - Int("prevContainers", prevContainerCount). - Int("successfulNodes", successfulNodes). - Int("totalNodes", len(nodes)). - Msg("Traditional polling returned zero containers but had containers before - preserving previous containers") + preservedContainers := 0 + if len(failedNodeNames) > 0 { + for _, container := range prevGuests.containers { + if _, failed := failedNodeNames[container.Node]; failed { + allContainers = append(allContainers, container) + preservedContainers++ + } } } + if preservedContainers > 0 { + log.Warn(). + Str("instance", instanceName). + Int("preservedContainers", preservedContainers). + Int("failedNodes", failedNodes). + Msg("Preserved prior containers for nodes whose enumeration failed") + } // Check Docker presence for containers that need it (new, restarted, started) allContainers = m.CheckContainersForDocker(ctx, allContainers) m.CollectProxmoxGuestDockerInventory(ctx, allContainers) - // Update state with all containers - m.state.UpdateContainersForInstance(instanceName, allContainers) - // Record guest metrics history for running containers (enables sparkline/trends view) if !shouldSkipNativeMockStateMetricWrites() { now := time.Now() @@ -362,6 +363,16 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri Int("failedNodes", failedNodes). Dur("duration", duration). Msg("Parallel container polling completed") + + return allContainers +} + +// pollContainersWithNodes retains the focused single-kind polling entry point +// used by tests and maintenance callers. The production guest cycle uses +// collectContainersWithNodes and publishes both guest kinds atomically. +func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) { + containers := m.collectContainersWithNodes(ctx, instanceName, clusterName, isCluster, client, nodes, nodeEffectiveStatus) + m.state.UpdateContainersForInstance(instanceName, containers) } // pollStorageWithNodes polls storage from all nodes in parallel using goroutines diff --git a/internal/monitoring/monitor_polling_vm.go b/internal/monitoring/monitor_polling_vm.go index 6fbc8ace3..d4a6c0f87 100644 --- a/internal/monitoring/monitor_polling_vm.go +++ b/internal/monitoring/monitor_polling_vm.go @@ -12,7 +12,7 @@ import ( "github.com/rs/zerolog/log" ) -func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) { +func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) []models.VM { startTime := time.Now() type nodeResult struct { @@ -85,10 +85,12 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu qemuTemplateSubjects := make(map[string]struct{}) successfulNodes := 0 failedNodes := 0 + failedNodeNames := make(map[string]struct{}) for result := range resultChan { if result.err != nil { failedNodes++ + failedNodeNames[result.node] = struct{}{} continue } successfulNodes++ @@ -101,20 +103,22 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects) } - if len(allVMs) == 0 && len(nodes) > 0 { - allVMs = append(allVMs, prevGuests.vms...) - prevVMCount := len(prevGuests.vms) - if prevVMCount > 0 { - log.Warn(). - Str("instance", instanceName). - Int("prevVMs", prevVMCount). - Int("successfulNodes", successfulNodes). - Int("totalNodes", len(nodes)). - Msg("Traditional polling returned zero VMs but had VMs before - preserving previous VMs") + preservedVMs := 0 + if len(failedNodeNames) > 0 { + for _, vm := range prevGuests.vms { + if _, failed := failedNodeNames[vm.Node]; failed { + allVMs = append(allVMs, vm) + preservedVMs++ + } } } - - m.state.UpdateVMsForInstance(instanceName, allVMs) + if preservedVMs > 0 { + log.Warn(). + Str("instance", instanceName). + Int("preservedVMs", preservedVMs). + Int("failedNodes", failedNodes). + Msg("Preserved prior VMs for nodes whose enumeration failed") + } if !shouldSkipNativeMockStateMetricWrites() { now := time.Now() @@ -136,4 +140,14 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu Int("failedNodes", failedNodes). Dur("duration", duration). Msg("Parallel VM polling completed") + + return allVMs +} + +// pollVMsWithNodes retains the focused single-kind polling entry point used by +// tests and maintenance callers. The production guest cycle uses +// collectVMsWithNodes and publishes both guest kinds atomically. +func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) { + vms := m.collectVMsWithNodes(ctx, instanceName, clusterName, isCluster, client, nodes, nodeEffectiveStatus) + m.state.UpdateVMsForInstance(instanceName, vms) } diff --git a/internal/monitoring/monitor_previous_state.go b/internal/monitoring/monitor_previous_state.go index b8ce0bfa9..f7a03d6e4 100644 --- a/internal/monitoring/monitor_previous_state.go +++ b/internal/monitoring/monitor_previous_state.go @@ -93,14 +93,17 @@ func previousVMFromView(vm *unifiedresources.VMView) models.VM { if vm == nil { return models.VM{} } + instance := vm.Instance() + node := vm.Node() + vmid := vm.VMID() return models.VM{ - ID: vm.ID(), - Instance: vm.Instance(), - Node: vm.Node(), - VMID: vm.VMID(), + ID: makeGuestID(instance, node, vmid), + Instance: instance, + Node: node, + VMID: vmid, Name: vm.Name(), Type: "qemu", - Status: string(vm.Status()), + Status: vm.RuntimeStatus(), IPAddresses: vm.IPAddresses(), OSName: vm.OSName(), OSVersion: vm.OSVersion(), @@ -122,13 +125,16 @@ func previousContainerFromView(ct *unifiedresources.ContainerView) models.Contai if ct == nil { return models.Container{} } + instance := ct.Instance() + node := ct.Node() + vmid := ct.VMID() return models.Container{ - ID: ct.ID(), - Instance: ct.Instance(), - Node: ct.Node(), - VMID: ct.VMID(), + ID: makeGuestID(instance, node, vmid), + Instance: instance, + Node: node, + VMID: vmid, Name: ct.Name(), - Status: string(ct.Status()), + Status: ct.RuntimeStatus(), Type: ct.ContainerType(), IsOCI: ct.IsOCI(), LastSeen: ct.LastSeen(), diff --git a/internal/monitoring/monitor_pve.go b/internal/monitoring/monitor_pve.go index 6ad887906..4fa2984c4 100644 --- a/internal/monitoring/monitor_pve.go +++ b/internal/monitoring/monitor_pve.go @@ -803,12 +803,16 @@ func (m *Monitor) pollGuestsWithFallback( } // Use optimized parallel polling for better performance + previous := m.previousGuestContextForInstance(instanceName) + vms := previous.vms + containers := previous.containers if instanceCfg.MonitorVMs { - m.pollVMsWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus) + vms = m.collectVMsWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus) } if instanceCfg.MonitorContainers { - m.pollContainersWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus) + containers = m.collectContainersWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus) } + m.state.UpdateGuestsForInstance(instanceName, vms, containers) } return nil diff --git a/internal/monitoring/monitor_pve_guest_poll.go b/internal/monitoring/monitor_pve_guest_poll.go index c993b19fe..cd9cae89f 100644 --- a/internal/monitoring/monitor_pve_guest_poll.go +++ b/internal/monitoring/monitor_pve_guest_poll.go @@ -44,15 +44,14 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam allVMs, allContainers = m.preserveGuestsForGracePeriod(instanceName, resources, prevGuests.vms, prevGuests.containers, nodeEffectiveStatus, allVMs, allContainers) - // Always update state when using efficient polling path - // Even if arrays are empty, we need to update to clear out VMs from genuinely offline nodes - m.state.UpdateVMsForInstance(instanceName, allVMs) - // Check Docker presence for containers that need it (new, restarted, started) allContainers = m.CheckContainersForDocker(ctx, allContainers) m.CollectProxmoxGuestDockerInventory(ctx, allContainers) - m.state.UpdateContainersForInstance(instanceName, allContainers) + // Publish the complete guest generation only after both VM and container + // collection/enrichment has finished. Empty authoritative results still + // remove genuinely deleted guests. + m.state.UpdateGuestsForInstance(instanceName, allVMs, allContainers) m.recordGuestMetrics(allVMs, allContainers) diff --git a/internal/monitoring/proxmox_guest_snapshot_coherence_test.go b/internal/monitoring/proxmox_guest_snapshot_coherence_test.go new file mode 100644 index 000000000..3734866da --- /dev/null +++ b/internal/monitoring/proxmox_guest_snapshot_coherence_test.go @@ -0,0 +1,116 @@ +package monitoring + +import ( + "context" + "errors" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +type partialNodeGuestClient struct { + *stubPVEClient + failedNodes map[string]bool + vmsByNode map[string][]proxmox.VM + containersByNode map[string][]proxmox.Container +} + +func (c *partialNodeGuestClient) GetClusterResources(context.Context, string) ([]proxmox.ClusterResource, error) { + return nil, errors.New("cluster resources unavailable") +} + +func (c *partialNodeGuestClient) GetVMs(_ context.Context, node string) ([]proxmox.VM, error) { + if c.failedNodes[node] { + return nil, errors.New("VM enumeration unavailable") + } + return c.vmsByNode[node], nil +} + +func (c *partialNodeGuestClient) GetContainers(_ context.Context, node string) ([]proxmox.Container, error) { + if c.failedNodes[node] { + return nil, errors.New("container enumeration unavailable") + } + return c.containersByNode[node], nil +} + +func TestPollGuestsWithFallbackRetainsOnlyFailedNodeGeneration(t *testing.T) { + monitor := newTestPVEMonitor("lab") + defer monitor.alertManager.Stop() + defer monitor.notificationMgr.Stop() + + monitor.state.UpdateGuestsForInstance( + "lab", + []models.VM{ + {ID: "lab:node-a:101", VMID: 101, Name: "deleted-vm", Node: "node-a", Instance: "lab", Status: "running"}, + {ID: "lab:node-b:201", VMID: 201, Name: "retained-vm", Node: "node-b", Instance: "lab", Status: "running"}, + }, + []models.Container{ + {ID: "lab:node-a:102", VMID: 102, Name: "deleted-ct", Node: "node-a", Instance: "lab", Status: "running"}, + {ID: "lab:node-b:202", VMID: 202, Name: "retained-ct", Node: "node-b", Instance: "lab", Status: "running"}, + }, + ) + monitor.state.UpdateGuestsForInstance( + "other", + []models.VM{{ID: "other:node-c:301", VMID: 301, Node: "node-c", Instance: "other"}}, + []models.Container{{ID: "other:node-c:302", VMID: 302, Node: "node-c", Instance: "other"}}, + ) + + client := &partialNodeGuestClient{ + stubPVEClient: &stubPVEClient{}, + failedNodes: map[string]bool{"node-b": true}, + vmsByNode: map[string][]proxmox.VM{"node-a": {}}, + containersByNode: map[string][]proxmox.Container{"node-a": {}}, + } + nodes := []proxmox.Node{ + {Node: "node-a", Status: "online"}, + {Node: "node-b", Status: "online"}, + } + nodeStatus := map[string]string{"node-a": "online", "node-b": "online"} + cfg := &config.PVEInstance{MonitorVMs: true, MonitorContainers: true} + + if err := monitor.pollGuestsWithFallback(context.Background(), "lab", cfg, client, nodes, nodeStatus); err != nil { + t.Fatalf("partial poll failed: %v", err) + } + + snapshot := monitor.GetState() + assertGuestIDs(t, snapshot.VMs, []string{"lab:node-b:201", "other:node-c:301"}) + assertGuestIDs(t, snapshot.Containers, []string{"lab:node-b:202", "other:node-c:302"}) + if snapshot.VMs[0].Status != "running" || snapshot.Containers[0].Status != "running" { + t.Fatalf("failed-node power state was not retained: vm=%q container=%q", snapshot.VMs[0].Status, snapshot.Containers[0].Status) + } + + client.failedNodes["node-b"] = false + client.vmsByNode["node-b"] = []proxmox.VM{} + client.containersByNode["node-b"] = []proxmox.Container{} + + if err := monitor.pollGuestsWithFallback(context.Background(), "lab", cfg, client, nodes, nodeStatus); err != nil { + t.Fatalf("recovery poll failed: %v", err) + } + + snapshot = monitor.GetState() + assertGuestIDs(t, snapshot.VMs, []string{"other:node-c:301"}) + assertGuestIDs(t, snapshot.Containers, []string{"other:node-c:302"}) +} + +func assertGuestIDs[T models.VM | models.Container](t *testing.T, guests []T, want []string) { + t.Helper() + got := make([]string, 0, len(guests)) + for _, guest := range guests { + switch typed := any(guest).(type) { + case models.VM: + got = append(got, typed.ID) + case models.Container: + got = append(got, typed.ID) + } + } + if len(got) != len(want) { + t.Fatalf("guest ids = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("guest ids = %v, want %v", got, want) + } + } +} diff --git a/internal/unifiedresources/adapter_coverage_test.go b/internal/unifiedresources/adapter_coverage_test.go index 4611e9cad..3ca54555d 100644 --- a/internal/unifiedresources/adapter_coverage_test.go +++ b/internal/unifiedresources/adapter_coverage_test.go @@ -200,6 +200,19 @@ func TestMonitorAdapterUsesConfiguredProxmoxStaleThreshold(t *testing.T) { if defaultVMs[0].Status() != StatusWarning { t.Fatalf("default stale threshold status = %q, want warning", defaultVMs[0].Status()) } + if defaultVMs[0].RuntimeStatus() != "running" { + t.Fatalf("default stale runtime status = %q, want running", defaultVMs[0].RuntimeStatus()) + } + defaultResources := defaultAdapter.GetByType(ResourceTypeVM) + if len(defaultResources) != 1 || defaultResources[0].Proxmox == nil { + t.Fatalf("expected one Proxmox VM resource, got %+v", defaultResources) + } + if got := defaultResources[0].Proxmox.RuntimeStatus; got != "running" { + t.Fatalf("Proxmox runtime status = %q, want running while collection is stale", got) + } + if got := defaultResources[0].SourceStatus[SourceProxmox].Status; got != "stale" { + t.Fatalf("Proxmox source status = %q, want stale", got) + } adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), map[DataSource]time.Duration{ SourceProxmox: 10 * time.Minute, diff --git a/internal/unifiedresources/adapters.go b/internal/unifiedresources/adapters.go index 3eb62a824..2986fa30b 100644 --- a/internal/unifiedresources/adapters.go +++ b/internal/unifiedresources/adapters.go @@ -1676,6 +1676,7 @@ func resourceFromVM(vm models.VM) (Resource, ResourceIdentity) { metrics := metricsFromVM(vm) proxmox := &ProxmoxData{ SourceID: sourceID, + RuntimeStatus: vm.Status, NodeName: vm.Node, Pool: vm.Pool, Instance: vm.Instance, @@ -1742,6 +1743,7 @@ func resourceFromContainer(ct models.Container) (Resource, ResourceIdentity) { metrics := metricsFromContainer(ct) proxmox := &ProxmoxData{ SourceID: sourceID, + RuntimeStatus: ct.Status, NodeName: ct.Node, Pool: ct.Pool, Instance: ct.Instance, diff --git a/internal/unifiedresources/canonical_ids_types_test.go b/internal/unifiedresources/canonical_ids_types_test.go index 9843328de..f01e9625a 100644 --- a/internal/unifiedresources/canonical_ids_types_test.go +++ b/internal/unifiedresources/canonical_ids_types_test.go @@ -22,6 +22,23 @@ func TestCanonicalResourceTypeDoesNotAliasHost(t *testing.T) { } } +func TestProxmoxRuntimeStatusJSONContract(t *testing.T) { + payload := ProxmoxData{ + SourceID: "lab:node-a:101", + RuntimeStatus: "running", + NodeName: "node-a", + VMID: 101, + } + + data, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal ProxmoxData: %v", err) + } + if !strings.Contains(string(data), `"runtimeStatus":"running"`) { + t.Fatalf("ProxmoxData JSON did not carry runtimeStatus: %s", data) + } +} + func TestHostSMARTMetaCarriesSizeBytesJSONContract(t *testing.T) { payload := HostSMARTMeta{ Device: "/dev/sda", diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index 33957eb70..4e3e8e2dd 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -14,6 +14,11 @@ import ( type MonitorAdapter struct { registry *ResourceRegistry + // mutationMu serializes complete registry generations with incremental + // supplemental updates. Registry construction intentionally happens while + // readers keep using the prior pointer, but two writers must never publish + // out of order or mutate a registry after it has been replaced. + mutationMu sync.Mutex mu sync.RWMutex activeAlerts []models.Alert lastRebuiltAt time.Time @@ -142,6 +147,12 @@ func (a *MonitorAdapter) ResolveCanonicalResourceID(ref string) (string, bool) { } func (a *MonitorAdapter) replaceRegistry(snapshot models.StateSnapshot, recordsBySource map[DataSource][]IngestRecord) { + if a == nil { + return + } + a.mutationMu.Lock() + defer a.mutationMu.Unlock() + registry := a.currentRegistry() if registry == nil { return @@ -360,6 +371,12 @@ func (a *MonitorAdapter) PopulateSnapshotAndSupplemental(snapshot models.StateSn // PopulateSupplementalRecords ingests source-native records emitted outside the // legacy state snapshot pipeline. func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records []IngestRecord) { + if a == nil { + return + } + a.mutationMu.Lock() + defer a.mutationMu.Unlock() + registry := a.currentRegistry() if registry == nil || len(records) == 0 || strings.TrimSpace(string(source)) == "" { return diff --git a/internal/unifiedresources/monitor_adapter_generation_test.go b/internal/unifiedresources/monitor_adapter_generation_test.go new file mode 100644 index 000000000..8d8f42082 --- /dev/null +++ b/internal/unifiedresources/monitor_adapter_generation_test.go @@ -0,0 +1,101 @@ +package unifiedresources + +import ( + "sync" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +type blockingChangeStore struct { + *MemoryStore + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (s *blockingChangeStore) RecordChange(change ResourceChange) error { + s.once.Do(func() { + close(s.entered) + <-s.release + }) + return s.MemoryStore.RecordChange(change) +} + +func TestMonitorAdapterSerializesSupplementalMutationAfterSnapshotPublication(t *testing.T) { + store := &blockingChangeStore{ + MemoryStore: NewMemoryStore(), + entered: make(chan struct{}), + release: make(chan struct{}), + } + adapter := NewMonitorAdapter(NewRegistry(store)) + + rebuildDone := make(chan struct{}) + go func() { + adapter.PopulateFromSnapshot(models.StateSnapshot{ + LastUpdate: time.Now().UTC(), + VMs: []models.VM{{ + ID: "lab:node-a:101", + VMID: 101, + Name: "database", + Node: "node-a", + Instance: "lab", + Status: "running", + LastSeen: time.Now().UTC(), + }}, + }) + close(rebuildDone) + }() + + select { + case <-store.entered: + case <-time.After(time.Second): + t.Fatal("snapshot rebuild did not reach change publication") + } + + supplementalDone := make(chan struct{}) + go func() { + adapter.PopulateSupplementalRecords(SourceAgent, []IngestRecord{{ + SourceID: "host-supplemental", + Resource: Resource{ + Type: ResourceTypeAgent, + Name: "host-supplemental", + Status: StatusOnline, + LastSeen: time.Now().UTC(), + }, + }}) + close(supplementalDone) + }() + + select { + case <-supplementalDone: + t.Fatal("supplemental mutation bypassed the in-flight snapshot generation") + case <-time.After(20 * time.Millisecond): + } + + close(store.release) + select { + case <-rebuildDone: + case <-time.After(time.Second): + t.Fatal("snapshot rebuild did not complete") + } + select { + case <-supplementalDone: + case <-time.After(time.Second): + t.Fatal("supplemental mutation did not resume") + } + + resources := adapter.GetAll() + if len(resources) != 2 { + t.Fatalf("final generation contains %d resources, want snapshot plus supplemental record: %+v", len(resources), resources) + } + var foundVM, foundSupplemental bool + for _, resource := range resources { + foundVM = foundVM || resource.Name == "database" + foundSupplemental = foundSupplemental || resource.Name == "host-supplemental" + } + if !foundVM || !foundSupplemental { + t.Fatalf("final generation lost a writer: %+v", resources) + } +} diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go index 972f8a0f1..4ae8722cc 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -1,6 +1,7 @@ package unifiedresources import ( + "fmt" "testing" "time" @@ -525,6 +526,89 @@ func TestMonitorAdapterIngestsAvailabilityAfterCorrelatableSupplementalSources(t } } +func TestMonitorAdapterStalenessDoesNotEmitRemovalButAuthoritativeOmissionDoes(t *testing.T) { + store := NewMemoryStore() + adapter := NewMonitorAdapter(NewRegistry(store)) + now := time.Now().UTC() + container := func(vmid int, name string, seen time.Time) models.Container { + return models.Container{ + ID: fmt.Sprintf("lab:node-a:%d", vmid), + VMID: vmid, + Name: name, + Node: "node-a", + Instance: "lab", + Status: "running", + Type: "lxc", + LastSeen: seen, + } + } + + adapter.PopulateFromSnapshot(models.StateSnapshot{ + LastUpdate: now, + Containers: []models.Container{ + container(101, "alpha", now), + container(102, "beta", now), + }, + }) + initial := adapter.GetByType(ResourceTypeSystemContainer) + if len(initial) != 2 { + t.Fatalf("initial container count = %d, want 2", len(initial)) + } + var removedID string + for _, resource := range initial { + if resource.Name == "beta" { + removedID = resource.ID + } + } + if removedID == "" { + t.Fatal("beta canonical ID not found") + } + + staleSeen := now.Add(-5 * time.Minute) + adapter.PopulateFromSnapshot(models.StateSnapshot{ + LastUpdate: now.Add(time.Second), + Containers: []models.Container{ + container(101, "alpha", staleSeen), + container(102, "beta", staleSeen), + }, + }) + if got := len(adapter.GetByType(ResourceTypeSystemContainer)); got != 2 { + t.Fatalf("stale refresh container count = %d, want 2", got) + } + if changes, err := store.GetRecentChanges(removedID, time.Time{}, 20); err != nil { + t.Fatalf("GetRecentChanges before deletion: %v", err) + } else { + for _, change := range changes { + if change.Metadata["changeType"] == "resource_removed" { + t.Fatalf("staleness emitted a removal: %+v", change) + } + } + } + + adapter.PopulateFromSnapshot(models.StateSnapshot{ + LastUpdate: now.Add(2 * time.Second), + Containers: []models.Container{ + container(101, "alpha", now.Add(2*time.Second)), + }, + }) + if got := len(adapter.GetByType(ResourceTypeSystemContainer)); got != 1 { + t.Fatalf("authoritative deletion container count = %d, want 1", got) + } + changes, err := store.GetRecentChanges(removedID, time.Time{}, 20) + if err != nil { + t.Fatalf("GetRecentChanges after deletion: %v", err) + } + removals := 0 + for _, change := range changes { + if change.Metadata["changeType"] == "resource_removed" { + removals++ + } + } + if removals != 1 { + t.Fatalf("resource removal history count = %d, want 1: %+v", removals, changes) + } +} + func TestMonitorAdapterRecordChangeForwardsToStore(t *testing.T) { store := NewMemoryStore() adapter := NewMonitorAdapter(NewRegistry(store)) diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 8369819a0..7ad0bcda5 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -3372,6 +3372,9 @@ func mergeProxmoxData(existing *ProxmoxData, incoming *ProxmoxData) *ProxmoxData if incoming.HostURL != "" { merged.HostURL = incoming.HostURL } + if incoming.RuntimeStatus != "" { + merged.RuntimeStatus = incoming.RuntimeStatus + } if incoming.VMID != 0 { merged.VMID = incoming.VMID } diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index b63cd4eeb..35bca312e 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -5306,6 +5306,71 @@ func TestMarkStaleRecomputesFromRemainingFreshSources(t *testing.T) { } } +func TestMarkStaleKeepsProxmoxRuntimeStateIndependentOfAvailabilityFacet(t *testing.T) { + rr := NewRegistry(nil) + staleSeen := time.Now().UTC().Add(-5 * time.Minute) + freshSeen := time.Now().UTC() + + for _, sourceID := range []string{"lab:node-a:101", "lab:node-a:102"} { + rr.IngestRecords(SourceProxmox, []IngestRecord{{ + SourceID: sourceID, + Resource: Resource{ + Type: ResourceTypeSystemContainer, + Name: sourceID, + Status: StatusOnline, + LastSeen: staleSeen, + Proxmox: &ProxmoxData{RuntimeStatus: "running", NodeName: "node-a"}, + }, + }}) + } + + resources := rr.ListByType(ResourceTypeSystemContainer) + if len(resources) != 2 { + t.Fatalf("expected two Proxmox containers, got %d", len(resources)) + } + var checkedID string + for _, resource := range resources { + if resource.Name == "lab:node-a:102" { + checkedID = resource.ID + break + } + } + if checkedID == "" { + t.Fatal("availability target container not found") + } + + rr.IngestRecords(SourceAvailability, []IngestRecord{{ + SourceID: "probe-102", + Resource: Resource{ + Type: ResourceTypeNetworkEndpoint, + Name: "probe-102", + Status: StatusOnline, + LastSeen: freshSeen, + Availability: &AvailabilityData{ + TargetID: "probe-102", LinkedResourceID: checkedID, + Address: "192.0.2.102", Protocol: "icmp", Enabled: true, Available: true, + }, + }, + }}) + + rr.MarkStale(freshSeen, nil) + + resources = rr.ListByType(ResourceTypeSystemContainer) + statuses := make(map[string]ResourceStatus, len(resources)) + for _, resource := range resources { + statuses[resource.Name] = resource.Status + if resource.Proxmox == nil || resource.Proxmox.RuntimeStatus != "running" { + t.Fatalf("runtime status changed with source freshness for %s: %+v", resource.Name, resource.Proxmox) + } + } + if got := statuses["lab:node-a:101"]; got != StatusWarning { + t.Fatalf("unfaceted stale container status = %q, want warning", got) + } + if got := statuses["lab:node-a:102"]; got != StatusOnline { + t.Fatalf("availability-faceted stale container status = %q, want online", got) + } +} + func TestResourceRegistryUsesConfiguredProxmoxStaleThresholds(t *testing.T) { seen := time.Now().UTC().Add(-90 * time.Second).Truncate(time.Millisecond) snapshot := models.StateSnapshot{ diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index 4c50d3e25..a0cd9fe33 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -331,7 +331,8 @@ type MetricValue struct { // ProxmoxData contains Proxmox-specific data for a resource. type ProxmoxData struct { - SourceID string `json:"sourceId,omitempty"` // raw model ID from source snapshot + SourceID string `json:"sourceId,omitempty"` // raw model ID from source snapshot + RuntimeStatus string `json:"runtimeStatus,omitempty"` // source-authored VM/LXC power state, independent of collection freshness NodeName string `json:"nodeName,omitempty"` Pool string `json:"pool,omitempty"` ClusterName string `json:"clusterName,omitempty"` diff --git a/internal/unifiedresources/views.go b/internal/unifiedresources/views.go index 5b7c91f39..e22d56e7b 100644 --- a/internal/unifiedresources/views.go +++ b/internal/unifiedresources/views.go @@ -97,6 +97,13 @@ func (v VMView) Status() ResourceStatus { return v.r.Status } +func (v VMView) RuntimeStatus() string { + if v.r == nil || v.r.Proxmox == nil { + return "" + } + return strings.TrimSpace(v.r.Proxmox.RuntimeStatus) +} + func (v VMView) VMID() int { if v.r == nil || v.r.Proxmox == nil { return 0 @@ -375,6 +382,13 @@ func (v ContainerView) Status() ResourceStatus { return v.r.Status } +func (v ContainerView) RuntimeStatus() string { + if v.r == nil || v.r.Proxmox == nil { + return "" + } + return strings.TrimSpace(v.r.Proxmox.RuntimeStatus) +} + func (v ContainerView) VMID() int { if v.r == nil || v.r.Proxmox == nil { return 0 diff --git a/tests/integration/tests/64-workloads-proxmox-refresh-stability.spec.ts b/tests/integration/tests/64-workloads-proxmox-refresh-stability.spec.ts index c94265da2..35d4e40af 100644 --- a/tests/integration/tests/64-workloads-proxmox-refresh-stability.spec.ts +++ b/tests/integration/tests/64-workloads-proxmox-refresh-stability.spec.ts @@ -1,7 +1,12 @@ import fs from "node:fs"; import path from "node:path"; import { fileURLToPath } from "node:url"; -import { expect, test as base, type Locator, type Page } from "@playwright/test"; +import { + expect, + test as base, + type Locator, + type Page, +} from "@playwright/test"; import { createAuthenticatedStorageState, @@ -113,6 +118,49 @@ async function readGuestDrawerActiveTab(detailRow: Locator): Promise { return active; } +function issue1611Container( + vmid: number, + name: string, + status: "online" | "warning", + availability = false, +) { + return { + id: `lab-node-a-${vmid}`, + type: "system-container", + name, + status, + lastSeen: "2026-07-24T08:00:00Z", + vmid, + node: "node-a", + instance: "lab", + sources: ["proxmox", ...(availability ? ["availability"] : [])], + platformScopes: ["proxmox-pve"], + metrics: { + cpu: { percent: 0.12 }, + memory: { used: 1024, total: 4096, percent: 25 }, + disk: { used: 2048, total: 8192, percent: 25 }, + }, + proxmox: { + runtimeStatus: "running", + nodeName: "node-a", + instance: "lab", + vmid, + cpus: 2, + uptime: 3600, + }, + ...(availability + ? { + availability: { + targetId: `probe-${vmid}`, + protocol: "icmp", + enabled: true, + available: true, + }, + } + : {}), + }; +} + test.describe.serial("Workloads Proxmox refresh stability", () => { test.setTimeout(180_000); @@ -172,7 +220,9 @@ test.describe.serial("Workloads Proxmox refresh stability", () => { await expect(discoveryButton).toBeVisible(); await discoveryButton.click(); - await expect.poll(() => readGuestDrawerActiveTab(detailRow)).toBe("discovery"); + await expect + .poll(() => readGuestDrawerActiveTab(detailRow)) + .toBe("discovery"); const beforePollScrollTop = await readPrimaryViewportScrollTop(page); @@ -188,4 +238,95 @@ test.describe.serial("Workloads Proxmox refresh stability", () => { Math.max(10, beforePollScrollTop - 80), ); }); + + test("retains running LXC rows through stale availability projections and removes a confirmed deletion", async ({ + page, + }, testInfo) => { + test.skip( + testInfo.project.name.startsWith("mobile-"), + "Desktop-only workload refresh proof", + ); + + await ensureMockModeEnabled(page); + + let workloadRequests = 0; + let staleProjectionResponses = 0; + let deletionResponses = 0; + let publishDeletion = false; + await page.route("**/api/resources?**", async (route) => { + const url = new URL(route.request().url()); + if ( + url.pathname !== "/api/resources" || + url.searchParams.get("type") !== "vm,system-container,app-container,pod" + ) { + await route.continue(); + return; + } + + workloadRequests += 1; + let data; + if (workloadRequests === 1) { + data = [ + issue1611Container(101, "lxc-alpha", "online"), + issue1611Container(102, "lxc-beta", "online", true), + issue1611Container(103, "lxc-gamma", "online"), + ]; + } else if (!publishDeletion) { + staleProjectionResponses += 1; + data = [ + issue1611Container(101, "lxc-alpha", "warning"), + issue1611Container(102, "lxc-beta", "online", true), + issue1611Container(103, "lxc-gamma", "warning"), + ]; + } else { + deletionResponses += 1; + data = [ + issue1611Container(101, "lxc-alpha", "online"), + issue1611Container(102, "lxc-beta", "online", true), + ]; + } + + await route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ + data, + meta: { page: 1, limit: 200, total: data.length, totalPages: 1 }, + }), + }); + }); + + await page.goto( + "/proxmox/workloads?type=system-container&platform=proxmox-pve&status=running", + { waitUntil: "domcontentloaded" }, + ); + + const rows = page.locator("tr[data-guest-id]"); + await expect(rows).toHaveCount(3, { timeout: 60_000 }); + await expect(rows.first()).toBeVisible(); + + await page.locator("th").filter({ hasText: "Name" }).last().click(); + const retainedRow = rows.filter({ hasText: "lxc-beta" }); + await retainedRow.click(); + const detailRow = page.locator( + 'tr[data-inline-detail-for="lab:node-a:102"]', + ); + await expect(detailRow).toBeVisible(); + + await expect + .poll(() => staleProjectionResponses, { timeout: 15_000 }) + .toBeGreaterThan(0); + await expect(rows).toHaveCount(3); + await expect(rows.filter({ hasText: "lxc-alpha" })).toBeVisible(); + await expect(rows.filter({ hasText: "lxc-gamma" })).toBeVisible(); + await expect(detailRow).toBeVisible(); + + publishDeletion = true; + await expect + .poll(() => deletionResponses, { timeout: 15_000 }) + .toBeGreaterThan(0); + await expect(rows).toHaveCount(2); + await expect(rows.filter({ hasText: "lxc-gamma" })).toHaveCount(0); + await expect(detailRow).toBeVisible(); + }); });