diff --git a/internal/api/connections_aggregator.go b/internal/api/connections_aggregator.go index f119b5b95..8bc7423c2 100644 --- a/internal/api/connections_aggregator.go +++ b/internal/api/connections_aggregator.go @@ -139,6 +139,27 @@ type aggregatorInputs struct { pvePollingInterval time.Duration pbsPollingInterval time.Duration pmgPollingInterval time.Duration + + // plannedPollIntervals maps instanceHealth keys ("pve::") to the + // adaptive scheduler's currently planned interval for that instance. + // Adaptive polling deliberately stretches cadence past the configured + // interval while data is fresh, so the active→stale cutoff must follow + // the schedule actually in force or a healthy connection reads as stale + // for the back half of every stretched cycle (#1437). Absent or zero + // entries fall back to the configured cadence. + plannedPollIntervals map[string]time.Duration +} + +// effectivePollInterval picks the cadence the stale cutoff should scale by: +// the adaptive scheduler's planned interval when it stretched beyond the +// configured one, otherwise the configured interval. A planned interval that +// shrank below the configured cadence never tightens the cutoff, so a genuine +// poll outage still trips the connectionStaleThreshold floor on time. +func effectivePollInterval(configured, planned time.Duration) time.Duration { + if planned > configured { + return planned + } + return configured } type connectionAgentDesiredConfig struct { @@ -160,13 +181,16 @@ func buildConnections(in aggregatorInputs) []Connection { len(in.vmwareInstances)+len(in.truenasInstances)+len(in.availabilityTargets)+len(in.hosts)) for _, pve := range in.pveInstances { - out = append(out, buildPVEConnection(pve, in.instanceHealth, now, in.pvePollingInterval)) + interval := effectivePollInterval(in.pvePollingInterval, in.plannedPollIntervals["pve::"+pve.Name]) + out = append(out, buildPVEConnection(pve, in.instanceHealth, now, interval)) } for _, pbs := range in.pbsInstances { - out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, in.pbsPollingInterval, in.pbsReportedNodeNames[pbs.Name])) + interval := effectivePollInterval(in.pbsPollingInterval, in.plannedPollIntervals["pbs::"+pbs.Name]) + out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, interval, in.pbsReportedNodeNames[pbs.Name])) } for _, pmg := range in.pmgInstances { - out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, in.pmgPollingInterval)) + interval := effectivePollInterval(in.pmgPollingInterval, in.plannedPollIntervals["pmg::"+pmg.Name]) + out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, interval)) } for _, vmw := range in.vmwareInstances { out = append(out, buildVMwareConnection(vmw, in.instanceHealth, in.vmwareSummaries, now)) diff --git a/internal/api/connections_aggregator_adaptive_stale_test.go b/internal/api/connections_aggregator_adaptive_stale_test.go new file mode 100644 index 000000000..8a89ec5a7 --- /dev/null +++ b/internal/api/connections_aggregator_adaptive_stale_test.go @@ -0,0 +1,58 @@ +package api + +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +// Regression coverage for #1437: adaptive polling stretches an instance's +// cadence past the configured interval while data is fresh, so the +// active→stale cutoff has to follow the planned schedule. Before the fix a +// healthy connection on a stretched (e.g. 5-minute) cadence read as stale for +// the back half of every cycle and the platform page dropped to Agent-only. +func TestBuildConnectionsAdaptiveIntervalScalesStaleCutoff(t *testing.T) { + now := time.Date(2026, 8, 11, 12, 0, 0, 0, time.UTC) + lastSuccess := now.Add(-4 * time.Minute) + base := aggregatorInputs{ + pveInstances: []config.PVEInstance{{Name: "lab", Host: "https://lab:8006"}}, + instanceHealth: map[string]monitoring.InstanceHealth{ + "pve::lab": {PollStatus: monitoring.InstancePollStatus{LastSuccess: &lastSuccess}}, + }, + pvePollingInterval: 30 * time.Second, + now: now, + } + + pveState := func(in aggregatorInputs) ConnectionState { + t.Helper() + conns := buildConnections(in) + if len(conns) != 1 { + t.Fatalf("connections = %d, want 1", len(conns)) + } + return conns[0].State + } + + // No planned interval known: a 4-minute-old poll on a 30s configured + // cadence is past the floor and correctly reads stale. + if state := pveState(base); state != ConnectionStateStale { + t.Fatalf("state without plan = %q, want %q", state, ConnectionStateStale) + } + + // The scheduler planned a 5-minute cadence, so 4 minutes old is on + // schedule and must stay active. + onSchedule := base + onSchedule.plannedPollIntervals = map[string]time.Duration{"pve::lab": 5 * time.Minute} + if state := pveState(onSchedule); state != ConnectionStateActive { + t.Fatalf("state with 5m plan = %q, want %q", state, ConnectionStateActive) + } + + // A plan tighter than the configured cadence never tightens the cutoff: + // a genuine outage still trips against the configured interval and floor. + tightPlan := base + tightPlan.plannedPollIntervals = map[string]time.Duration{"pve::lab": 5 * time.Second} + if state := pveState(tightPlan); state != ConnectionStateStale { + t.Fatalf("state with 5s plan = %q, want %q", state, ConnectionStateStale) + } +} diff --git a/internal/api/connections_alerts.go b/internal/api/connections_alerts.go index 36089aa5d..dc8e519a5 100644 --- a/internal/api/connections_alerts.go +++ b/internal/api/connections_alerts.go @@ -66,6 +66,7 @@ func buildAggregatorInputsWithRuntimeSources( inputs.instanceHealth = instanceHealthByKey(monitor.SchedulerHealth()) inputs.availabilityStatuses = monitor.AvailabilityStatusSnapshot() inputs.pbsReportedNodeNames = pbsReportedNodeNamesByInstance(monitor.PBSInstancesSnapshot()) + inputs.plannedPollIntervals = plannedPollIntervalsForConfig(monitor, cfg) } else { inputs.hosts = []models.Host{} inputs.instanceHealth = map[string]monitoring.InstanceHealth{} @@ -143,3 +144,29 @@ func buildAlertConnectionSnapshotsWithRuntimeSources( inputs := buildAggregatorInputsWithRuntimeSources(ctx, cfg, persistence, monitor, runtime) return snapshotConnectionsForAlerts(buildConnections(inputs)) } + +// plannedPollIntervalsForConfig collects the adaptive scheduler's currently +// planned interval for every configured PVE/PBS/PMG instance, keyed the same +// way as instanceHealth ("pve::"). Instances the scheduler has no plan +// for are omitted so the aggregator falls back to the configured cadence. +func plannedPollIntervalsForConfig(monitor *monitoring.Monitor, cfg *config.Config) map[string]time.Duration { + if monitor == nil || cfg == nil { + return nil + } + out := make(map[string]time.Duration) + record := func(instanceType monitoring.InstanceType, keyPrefix, name string) { + if interval := monitor.PlannedPollInterval(instanceType, name); interval > 0 { + out[keyPrefix+name] = interval + } + } + for _, inst := range cfg.PVEInstances { + record(monitoring.InstanceTypePVE, "pve::", inst.Name) + } + for _, inst := range cfg.PBSInstances { + record(monitoring.InstanceTypePBS, "pbs::", inst.Name) + } + for _, inst := range cfg.PMGInstances { + record(monitoring.InstanceTypePMG, "pmg::", inst.Name) + } + return out +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 33c82f30d..e50f2f007 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -2636,6 +2636,22 @@ func (m *Monitor) DeadLetterCount() int { return m.deadLetterQueue.Size() } +// PlannedPollInterval reports the adaptive scheduler's currently planned +// interval for one instance, or zero when no plan exists (adaptive polling +// disabled, unknown instance, or startup before the first plan). Consumers +// use it to judge poll freshness against the schedule actually in force +// rather than the configured cadence, which adaptive polling deliberately +// stretches while data is fresh. +func (m *Monitor) PlannedPollInterval(instanceType InstanceType, instanceName string) time.Duration { + if m == nil || m.scheduler == nil { + return 0 + } + if task, ok := m.scheduler.LastScheduled(instanceType, instanceName); ok { + return task.Interval + } + return 0 +} + func (m *Monitor) SchedulerHealth() SchedulerHealthResponse { response := emptySchedulerHealthResponse(m.config != nil && m.config.AdaptivePollingEnabled) diff --git a/internal/monitoring/monitor_planned_interval_test.go b/internal/monitoring/monitor_planned_interval_test.go new file mode 100644 index 000000000..89fb7a6b3 --- /dev/null +++ b/internal/monitoring/monitor_planned_interval_test.go @@ -0,0 +1,40 @@ +package monitoring + +import ( + "testing" + "time" +) + +// PlannedPollInterval feeds the connections aggregator the cadence the +// adaptive scheduler actually promised, so the active→stale cutoff can track +// stretched schedules (#1437). +func TestPlannedPollIntervalReportsLastScheduled(t *testing.T) { + sched := NewAdaptiveScheduler(DefaultSchedulerConfig(), nil, nil, nil) + now := time.Now() + tasks := sched.BuildPlan(now, []InstanceDescriptor{{ + Name: "pve-a", + Type: InstanceTypePVE, + LastSuccess: now, + }}, 0) + if len(tasks) != 1 { + t.Fatalf("planned tasks = %d, want 1", len(tasks)) + } + if tasks[0].Interval <= 0 { + t.Fatalf("planned interval = %v, want > 0", tasks[0].Interval) + } + + m := &Monitor{scheduler: sched} + if got := m.PlannedPollInterval(InstanceTypePVE, "pve-a"); got != tasks[0].Interval { + t.Fatalf("PlannedPollInterval = %v, want %v", got, tasks[0].Interval) + } + if got := m.PlannedPollInterval(InstanceTypePVE, "unknown"); got != 0 { + t.Fatalf("unknown instance interval = %v, want 0", got) + } + if got := (&Monitor{}).PlannedPollInterval(InstanceTypePVE, "pve-a"); got != 0 { + t.Fatalf("schedulerless monitor interval = %v, want 0", got) + } + var nilMonitor *Monitor + if got := nilMonitor.PlannedPollInterval(InstanceTypePVE, "pve-a"); got != 0 { + t.Fatalf("nil monitor interval = %v, want 0", got) + } +}