fix(agent): skip scheduled monitor probes while the hub isn't requesting stats

The hub collects monitor results only with stats requests, which stop
while
a system is paused or the agent is disconnected. Paused WebSocket agents
stay
connected, so they kept probing. Skip scheduled probes after three
default
update intervals without a stats request and resume on the next one.

Co-authored-by: Sven van Ginkel <svenvanginkel@icloud.com>

agent/agent.go

@@ -60,6 +60,7 @@ type Agent struct {
 	monitorManager            *MonitorManager                                       // Manages network monitors
 	storagePoolManager        *StoragePoolManager                                   // Manages storage pool and dataset data
 	packageUpdates            *packageUpdatesManager                                // Checks for pending package updates
+	lastHubRequest            atomic.Int64                                          // Wall clock Unix nanoseconds of the last hub stats request
 }
 
 // NewAgent creates a new agent with the given data directory for persisting data.
@@ -141,7 +142,7 @@ func NewAgent(dataDir ...string) (agent *Agent, err error) {
 	agent.handlerRegistry = NewHandlerRegistry()
 
 	// initialize monitor manager
-	agent.monitorManager = newMonitorManager()
+	agent.monitorManager = newMonitorManager(agent)
 
 	agent.storagePoolManager = newStoragePoolManager()
 
@@ -191,6 +192,18 @@ func NewAgent(dataDir ...string) (agent *Agent, err error) {
 	return agent, nil
 }
 
+// markHubRequest records a hub stats request. It doesn't use the agent lock,
+// which gatherStats holds through collection.
+func (a *Agent) markHubRequest() {
+	a.lastHubRequest.Store(time.Now().UnixNano())
+}
+
+// hubIdleFor reports whether the hub hasn't requested stats for at least d.
+// Wall time counts host suspend toward idle.
+func (a *Agent) hubIdleFor(d time.Duration) bool {
+	return time.Now().UnixNano()-a.lastHubRequest.Load() >= int64(d)
+}
+
 func (a *Agent) gatherStats(options common.DataRequestOptions) *system.CombinedData {
 	a.Lock()
 	defer a.Unlock()

agent/handlers.go

@@ -103,6 +103,7 @@ func (h *GetDataHandler) Handle(hctx *HandlerContext) error {
 	var options common.DataRequestOptions
 	_ = cbor.Unmarshal(hctx.Request.Data, &options)
 
+	hctx.Agent.markHubRequest()
 	sysStats := hctx.Agent.gatherStats(options)
 	// Cached stats may be shared by concurrent SSH and WebSocket requests.
 	// Set the transport on the response copy, not on the cached data.

agent/network_monitor.go

@@ -4,13 +4,20 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"log/slog"
 	"net/http"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/henrygd/beszel/internal/entities/monitor"
 )
 
+// monitorIdleTimeout is three default hub update intervals. The hub collects
+// monitor results only with stats, which it stops requesting while the system is
+// paused or the agent is disconnected, so scheduled probes stop after this long.
+const monitorIdleTimeout = 3 * time.Minute
+
 // MonitorManager manages network monitor configurations and task lifetimes.
 type MonitorManager struct {
 	mu          sync.RWMutex
@@ -18,14 +25,18 @@ type MonitorManager struct {
 	probe       monitorProbe
 	certCheck   certChecker
 	resumeGuard monitorResumeGuard
+	agent       *Agent // Reference to the parent agent; nil means the hub is never idle
+	idle        atomic.Bool
 }
 
-func newMonitorManager() *MonitorManager {
+func newMonitorManager(agent *Agent) *MonitorManager {
 	client := &http.Client{Timeout: monitor.MaxProbeTimeout}
 	insecureTransport := http.DefaultTransport.(*http.Transport).Clone()
 	insecureTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
 	insecureClient := &http.Client{Timeout: monitor.MaxProbeTimeout, Transport: insecureTransport}
-	return newMonitorManagerWithProbe(networkMonitorProbe(client, insecureClient))
+	pm := newMonitorManagerWithProbe(networkMonitorProbe(client, insecureClient))
+	pm.agent = agent
+	return pm
 }
 
 func newMonitorManagerWithProbe(probe monitorProbe) *MonitorManager {
@@ -188,6 +199,20 @@ func (pm *MonitorManager) GetResults(durationMs uint16) map[string]monitor.Resul
 	return results
 }
 
+// hubIdle reports whether scheduled probes should be skipped because the hub
+// hasn't requested stats recently. Immediate probes requested by the hub still run.
+func (pm *MonitorManager) hubIdle() bool {
+	idle := pm.agent != nil && pm.agent.hubIdleFor(monitorIdleTimeout)
+	if pm.idle.Swap(idle) != idle {
+		if idle {
+			slog.Debug("pausing network monitors until the hub requests stats")
+		} else {
+			slog.Debug("resuming network monitors")
+		}
+	}
+	return idle
+}
+
 // Stop stops all monitor tasks.
 func (pm *MonitorManager) Stop() {
 	pm.mu.Lock()

agent/network_monitor_schedule.go

@@ -17,6 +17,9 @@ func (pm *MonitorManager) startMonitor(task *monitorTask) {
 	// Certificate checks piggyback on probe ticks, so they run at most once per
 	// probe interval after they become due.
 	go runMonitorSchedule(task.ctx, interval, delay, func() {
+		if pm.hubIdle() {
+			return
+		}
 		if _, allowed := task.resumeGuard.snapshot(); allowed {
 			task.runProbe(pm.probe)
 			task.refreshCert(pm.certCheck)

agent/network_monitor_schedule_test.go

@@ -165,3 +165,50 @@ func TestMonitorInjectedProbeTimeoutRecordsLoss(t *testing.T) {
 		assert.NoError(t, pm.monitors["test"].ctx.Err())
 	})
 }
+
+func TestMonitorScheduleSkipsProbesWhileHubIdle(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		var calls atomic.Int32
+		pm := newMonitorManagerWithProbe(func(context.Context, monitor.Config) (int64, error) {
+			calls.Add(1)
+			return 42, nil
+		})
+		defer pm.Stop()
+		a := &Agent{}
+		pm.agent = a
+		cfg := monitor.Config{ID: "test", Interval: 10}
+
+		// Configs synced before the first stats request don't probe on schedule.
+		pm.SyncMonitors([]monitor.Config{cfg})
+		time.Sleep(time.Minute)
+		synctest.Wait()
+		assert.Zero(t, calls.Load(), "monitors must not probe before the hub requests stats")
+
+		// Immediate probes requested by the hub still run.
+		result, err := pm.UpsertMonitor(cfg, true)
+		require.NoError(t, err)
+		require.NotNil(t, result)
+		assert.Equal(t, 1, int(calls.Load()))
+
+		a.markHubRequest()
+		time.Sleep(time.Minute)
+		synctest.Wait()
+		active := calls.Load()
+		assert.GreaterOrEqual(t, int(active), 6, "monitors must probe while the hub requests stats")
+
+		// Without further requests, probing stops once the idle timeout passes.
+		time.Sleep(monitorIdleTimeout)
+		synctest.Wait()
+		idle := calls.Load()
+		time.Sleep(10 * time.Minute)
+		synctest.Wait()
+		assert.Equal(t, idle, calls.Load(), "monitors must not probe while the hub is idle")
+		assert.NotNil(t, pm.monitors[cfg.ID], "idle monitors keep their config")
+
+		// The next stats request resumes the schedule without a resync.
+		a.markHubRequest()
+		time.Sleep(time.Minute)
+		synctest.Wait()
+		assert.Greater(t, calls.Load(), idle, "monitors must resume when the hub requests stats")
+	})
+}

agent/network_monitor_test.go

@@ -26,7 +26,7 @@ func TestMonitorManagerGetResultsIncludesHourResponseRange(t *testing.T) {
 	task.history.addSampleLocked(monitorSample{responseUs: 30, timestamp: now.Add(-50 * time.Second)})
 	task.history.addSampleLocked(monitorSample{responseUs: -1, timestamp: now.Add(-30 * time.Second)})
 
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.monitors = map[string]*monitorTask{"icmp:example.com": task}
 
 	results := pm.GetResults(uint16(time.Minute / time.Millisecond))
@@ -48,7 +48,7 @@ func TestMonitorManagerGetResultsIncludesLossOnlyHourData(t *testing.T) {
 	task.history.addSampleLocked(monitorSample{responseUs: -1, timestamp: now.Add(-30 * time.Second)})
 	task.history.addSampleLocked(monitorSample{responseUs: -1, timestamp: now.Add(-10 * time.Second)})
 
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.monitors = map[string]*monitorTask{"icmp:example.com": task}
 
 	results := pm.GetResults(uint16(time.Minute / time.Millisecond))
@@ -73,7 +73,7 @@ func TestMonitorManagerSyncMonitorsSkipsConfigsWithoutStableID(t *testing.T) {
 	validCfg := monitor.Config{ID: "monitor-1", Target: "ignored", Protocol: "noop", Interval: 10}
 	invalidCfg := monitor.Config{Target: "ignored", Protocol: "noop", Interval: 10}
 
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.SyncMonitors([]monitor.Config{validCfg, invalidCfg})
 	defer pm.Stop()
 
@@ -89,7 +89,7 @@ func TestMonitorManagerSyncMonitorsStopsRemovedTasksButKeepsExisting(t *testing.
 
 	keptTask := newMonitorTask(keepCfg)
 	removedTask := newMonitorTask(removeCfg)
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.monitors = map[string]*monitorTask{
 		keepCfg.ID:   keptTask,
 		removeCfg.ID: removedTask,
@@ -118,7 +118,7 @@ func TestMonitorManagerSyncMonitorsRestartsChangedConfig(t *testing.T) {
 	originalCfg := monitor.Config{ID: "monitor-1", Target: "ignored-a", Protocol: "noop", Interval: 10}
 	updatedCfg := monitor.Config{ID: "monitor-1", Target: "ignored-b", Protocol: "noop", Interval: 10}
 	originalTask := newMonitorTask(originalCfg)
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.monitors = map[string]*monitorTask{
 		originalCfg.ID: originalTask,
 	}
@@ -176,7 +176,7 @@ func TestMonitorManagerUpsertMonitorKeepsHistoryWhenOnlyIntervalChanges(t *testi
 	existingTask.history.addSampleLocked(monitorSample{responseUs: 12, timestamp: now.Add(-50 * time.Minute)})
 	existingTask.history.addSampleLocked(monitorSample{responseUs: 24, timestamp: now.Add(-30 * time.Second)})
 
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.monitors = map[string]*monitorTask{originalCfg.ID: existingTask}
 
 	result, err := pm.UpsertMonitor(updatedCfg, false)
@@ -211,7 +211,7 @@ func TestMonitorManagerUpsertMonitorKeepsHistoryWhenOnlyIntervalChanges(t *testi
 func TestMonitorManagerApplySyncDeleteRemovesTask(t *testing.T) {
 	config := monitor.Config{ID: "monitor-1", Target: "1.1.1.1", Protocol: "icmp", Interval: 10}
 	task := newMonitorTask(config)
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.monitors = map[string]*monitorTask{config.ID: task}
 
 	_, err := pm.HandleSyncRequest(monitor.SyncRequest{
@@ -267,7 +267,7 @@ func TestMonitorHTTP(t *testing.T) {
 			w.WriteHeader(http.StatusNoContent)
 		}))
 		defer server.Close()
-		probe := newMonitorManager().probe
+		probe := newMonitorManager(nil).probe
 		config := monitor.Config{Protocol: "http", Target: server.URL}
 
 		responseUs, err := probe(context.Background(), config)
@@ -488,7 +488,7 @@ func TestMonitorManagerCancelsActiveProbe(t *testing.T) {
 			}))
 			defer server.Close()
 			defer close(release)
-			pm := newMonitorManager()
+			pm := newMonitorManager(nil)
 			defer pm.Stop()
 			cfg := monitor.Config{ID: "test", Protocol: "http", Target: server.URL, Interval: 3600}
 			task := newMonitorTask(cfg)
@@ -593,7 +593,7 @@ func TestMonitorProbeTimeoutRecordsLoss(t *testing.T) {
 	}))
 	defer server.Close()
 	defer close(release)
-	pm := newMonitorManager()
+	pm := newMonitorManager(nil)
 	pm.probe = networkMonitorProbe(&http.Client{Timeout: 20 * time.Millisecond}, nil)
 	task := newMonitorTask(monitor.Config{ID: "timeout", Protocol: "http", Target: server.URL})
 	defer task.cancel()