Fix intermittent VM disappearance from dashboard (#555)

Two root causes: (1) When Proxmox cluster/resources returns a partial
response (e.g. during migration or transient API issue), VMs missing
from a responsive node were silently dropped because the node appeared
in nodesWithResources, bypassing grace-period preservation. Now
preserves recently-seen guests from online nodes for up to the grace
window. (2) The task queue allowed overlapping polls for the same PVE
instance — a slower stale poll could overwrite a newer complete VM list.
Added per-instance execution lock to skip duplicate scheduled tasks.

internal/monitoring/monitor.go

@@ -879,6 +879,7 @@ type Monitor struct {
 	kubernetesTokenBindings    map[string]string             // Track token ID -> agent ID bindings to enforce uniqueness
 	hostTokenBindings          map[string]string             // Track tokenID:hostname -> host identity bindings
 	lastHostRuntimePersist     map[string]time.Time          // Track last persisted host runtime write per host ID
+	runningTasks               map[string]struct{}           // Prevent overlapping polls for the same instance
 	dockerCommands             map[string]*dockerHostCommand
 	dockerCommandIndex         map[string]string
 	guestMetadataMu            sync.RWMutex
@@ -984,6 +985,103 @@ func makeGuestID(instanceName string, node string, vmid int) string {
 	return fmt.Sprintf("%s:%s:%d", instanceName, node, vmid)
 }
 
+func (m *Monitor) beginTaskExecution(task ScheduledTask) bool {
+	if m == nil {
+		return true
+	}
+
+	key := schedulerKey(task.InstanceType, task.InstanceName)
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.runningTasks == nil {
+		m.runningTasks = make(map[string]struct{})
+	}
+
+	if _, exists := m.runningTasks[key]; exists {
+		return false
+	}
+
+	m.runningTasks[key] = struct{}{}
+	return true
+}
+
+func (m *Monitor) finishTaskExecution(task ScheduledTask) {
+	if m == nil {
+		return
+	}
+
+	key := schedulerKey(task.InstanceType, task.InstanceName)
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.runningTasks != nil {
+		delete(m.runningTasks, key)
+	}
+}
+
+func preserveRecentMissingGuestsFromResponsiveNodes(
+	instanceName string,
+	prevState models.StateSnapshot,
+	currentVMs []models.VM,
+	currentContainers []models.Container,
+	nodeEffectiveStatus map[string]string,
+	nodesWithResources map[string]bool,
+	now time.Time,
+) ([]models.VM, []models.Container, int, int) {
+	currentGuestVMIDs := make(map[int]struct{}, len(currentVMs)+len(currentContainers))
+	for _, vm := range currentVMs {
+		currentGuestVMIDs[vm.VMID] = struct{}{}
+	}
+	for _, ct := range currentContainers {
+		currentGuestVMIDs[ct.VMID] = struct{}{}
+	}
+
+	preservedVMs := 0
+	for _, vm := range prevState.VMs {
+		if vm.Instance != instanceName {
+			continue
+		}
+		if nodeEffectiveStatus[vm.Node] != "online" || !nodesWithResources[vm.Node] {
+			continue
+		}
+		if !vm.LastSeen.IsZero() && now.Sub(vm.LastSeen) >= nodeOfflineGracePeriod {
+			continue
+		}
+		if _, exists := currentGuestVMIDs[vm.VMID]; exists {
+			continue
+		}
+
+		currentVMs = append(currentVMs, vm)
+		currentGuestVMIDs[vm.VMID] = struct{}{}
+		preservedVMs++
+	}
+
+	preservedContainers := 0
+	for _, ct := range prevState.Containers {
+		if ct.Instance != instanceName {
+			continue
+		}
+		if nodeEffectiveStatus[ct.Node] != "online" || !nodesWithResources[ct.Node] {
+			continue
+		}
+		if !ct.LastSeen.IsZero() && now.Sub(ct.LastSeen) >= nodeOfflineGracePeriod {
+			continue
+		}
+		if _, exists := currentGuestVMIDs[ct.VMID]; exists {
+			continue
+		}
+
+		currentContainers = append(currentContainers, ct)
+		currentGuestVMIDs[ct.VMID] = struct{}{}
+		preservedContainers++
+	}
+
+	return currentVMs, currentContainers, preservedVMs, preservedContainers
+}
+
 // parseDurationEnv parses a duration from an environment variable, returning defaultVal if not set or invalid
 func parseDurationEnv(key string, defaultVal time.Duration) time.Duration {
 	val := os.Getenv(key)
@@ -5315,6 +5413,17 @@ func (m *Monitor) executeScheduledTask(ctx context.Context, task ScheduledTask)
 		return
 	}
 
+	if !m.beginTaskExecution(task) {
+		if logging.IsLevelEnabled(zerolog.DebugLevel) {
+			log.Debug().
+				Str("instance", task.InstanceName).
+				Str("type", string(task.InstanceType)).
+				Msg("Skipping overlapping scheduled task for instance already being polled")
+		}
+		return
+	}
+	defer m.finishTaskExecution(task)
+
 	if m.pollMetrics != nil {
 		wait := time.Duration(0)
 		if !task.NextRun.IsZero() {
@@ -7983,6 +8092,25 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam
 			Msg("Grace period preservation complete")
 	}
 
+	partialPreservedVMs := 0
+	partialPreservedContainers := 0
+	allVMs, allContainers, partialPreservedVMs, partialPreservedContainers = preserveRecentMissingGuestsFromResponsiveNodes(
+		instanceName,
+		prevState,
+		allVMs,
+		allContainers,
+		nodeEffectiveStatus,
+		nodesWithResources,
+		time.Now(),
+	)
+	if partialPreservedVMs > 0 || partialPreservedContainers > 0 {
+		log.Warn().
+			Str("instance", instanceName).
+			Int("preservedMissingVMs", partialPreservedVMs).
+			Int("preservedMissingContainers", partialPreservedContainers).
+			Msg("Preserved recently seen guests missing from partial cluster/resources response")
+	}
+
 	m.logSuspiciousRepeatedVMMemoryUsage(instanceName, allVMs, prevInstanceVMs)
 
 	// Always update state when using efficient polling path

internal/monitoring/monitor_extra_coverage_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -451,6 +452,70 @@ func TestMonitor_PollVMsAndContainersEfficient_UsesVMRRDMemUsedWhenStatusUnavail
 	}
 }
 
+func TestMonitor_PollVMsAndContainersEfficient_PreservesMissingGuestsFromPartialResources(t *testing.T) {
+	now := time.Now()
+
+	m := &Monitor{
+		state:                    models.NewState(),
+		guestAgentFSInfoTimeout:  time.Second,
+		guestAgentRetries:        1,
+		guestAgentNetworkTimeout: time.Second,
+		guestAgentOSInfoTimeout:  time.Second,
+		guestAgentVersionTimeout: time.Second,
+		guestMetadataCache:       make(map[string]guestMetadataCacheEntry),
+		guestMetadataLimiter:     make(map[string]time.Time),
+		rateTracker:              NewRateTracker(),
+		metricsHistory:           NewMetricsHistory(100, time.Hour),
+		alertManager:             alerts.NewManager(),
+		stalenessTracker:         NewStalenessTracker(nil),
+		nodeRRDMemCache:          make(map[string]rrdMemCacheEntry),
+		vmRRDMemCache:            make(map[string]rrdMemCacheEntry),
+		vmAgentMemCache:          make(map[string]agentMemCacheEntry),
+	}
+	defer m.alertManager.Stop()
+
+	m.state.UpdateVMsForInstance("pve1", []models.VM{
+		{ID: "pve1:node1:100", Instance: "pve1", Node: "node1", VMID: 100, Name: "vm100", Type: "qemu", Status: "stopped", LastSeen: now},
+		{ID: "pve1:node1:101", Instance: "pve1", Node: "node1", VMID: 101, Name: "vm101", Type: "qemu", Status: "stopped", LastSeen: now},
+	})
+
+	client := &mockPVEClientExtra{
+		resources: []proxmox.ClusterResource{
+			{Type: "qemu", VMID: 100, Name: "vm100", Node: "node1", Status: "stopped", MaxMem: 2048, Mem: 1024},
+		},
+	}
+
+	success := m.pollVMsAndContainersEfficient(
+		context.Background(),
+		"pve1",
+		"",
+		false,
+		client,
+		map[string]string{"node1": "online"},
+	)
+	if !success {
+		t.Fatal("pollVMsAndContainersEfficient failed")
+	}
+
+	state := m.GetState()
+	if len(state.VMs) != 2 {
+		t.Fatalf("expected 2 VMs after partial preservation, got %d", len(state.VMs))
+	}
+
+	found := false
+	for _, vm := range state.VMs {
+		if vm.VMID == 101 {
+			found = true
+			if vm.LastSeen.IsZero() {
+				t.Fatal("preserved VM should keep last seen timestamp")
+			}
+		}
+	}
+	if !found {
+		t.Fatal("missing VM from partial cluster/resources response was not preserved")
+	}
+}
+
 func TestMonitor_PollVMsWithNodes_UsesVMRRDMemUsedWhenStatusUnavailable(t *testing.T) {
 	const total = uint64(8 << 30)
 	const inflatedUsed = uint64(6 << 30)
@@ -990,11 +1055,29 @@ func (m *mockPVEClientFailNodes) GetVMMemAvailableFromAgent(ctx context.Context,
 }
 
 type mockExecutor struct {
+	mu       sync.Mutex
 	executed []PollTask
+	started  chan struct{}
+	release  chan struct{}
 }
 
 func (m *mockExecutor) Execute(ctx context.Context, task PollTask) {
+	m.mu.Lock()
 	m.executed = append(m.executed, task)
+	started := m.started
+	release := m.release
+	m.mu.Unlock()
+
+	if started != nil {
+		select {
+		case <-started:
+		default:
+			close(started)
+		}
+	}
+	if release != nil {
+		<-release
+	}
 }
 
 func TestMonitor_ExecuteScheduledTask_Extra(t *testing.T) {
@@ -1022,6 +1105,45 @@ func TestMonitor_ExecuteScheduledTask_Extra(t *testing.T) {
 	}
 }
 
+func TestMonitor_ExecuteScheduledTask_SkipsOverlappingInstanceRuns(t *testing.T) {
+	m := &Monitor{
+		pveClients: map[string]PVEClientInterface{"pve1": &mockPVEClientExtra{}},
+	}
+
+	exec := &mockExecutor{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+	m.SetExecutor(exec)
+
+	task := ScheduledTask{InstanceName: "pve1", InstanceType: InstanceTypePVE}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		m.executeScheduledTask(context.Background(), task)
+	}()
+
+	select {
+	case <-exec.started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("first task did not start execution")
+	}
+
+	m.executeScheduledTask(context.Background(), task)
+
+	exec.mu.Lock()
+	executed := len(exec.executed)
+	exec.mu.Unlock()
+	if executed != 1 {
+		t.Fatalf("expected overlapping execution to be skipped, got %d executions", executed)
+	}
+
+	close(exec.release)
+	wg.Wait()
+}
+
 func TestMonitor_Start_Extra(t *testing.T) {
 	t.Setenv("PULSE_MOCK_TRENDS_SEED_DURATION", "5m")
 	t.Setenv("PULSE_MOCK_TRENDS_SAMPLE_INTERVAL", "5m")