beszel/agent/network_monitor_schedule_test.go

//go:build testing

package agent

import (
	"context"
	"sync/atomic"
	"testing"
	"testing/synctest"
	"time"

	"github.com/henrygd/beszel/internal/entities/monitor"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestMonitorScheduleTiming(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()
		var calls atomic.Int32
		go runMonitorSchedule(ctx, 10*time.Second, 5*time.Second, func() { calls.Add(1) })
		synctest.Wait()
		time.Sleep(4 * time.Second)
		synctest.Wait()
		assert.Equal(t, 0, int(calls.Load()))
		time.Sleep(time.Second)
		synctest.Wait()
		assert.Equal(t, 1, int(calls.Load()))
		time.Sleep(10 * time.Second)
		synctest.Wait()
		assert.Equal(t, 2, int(calls.Load()))
		cancel()
		synctest.Wait()
		time.Sleep(time.Minute)
		synctest.Wait()
		assert.Equal(t, 2, int(calls.Load()))
	})
}

func TestMonitorScheduleSlowProbe(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()
		var calls atomic.Int32
		release := make(chan struct{})
		go runMonitorSchedule(ctx, time.Second, 0, func() {
			calls.Add(1)
			select {
			case <-release:
			case <-ctx.Done():
			}
		})
		synctest.Wait()
		assert.Equal(t, 1, int(calls.Load()))
		time.Sleep(time.Minute)
		synctest.Wait()
		assert.Equal(t, 1, int(calls.Load()), "a slow probe must not spawn overlapping checks")
		close(release)
		synctest.Wait()
		assert.Equal(t, 1, int(calls.Load()), "missed intervals must not accumulate a backlog")
		time.Sleep(time.Second)
		synctest.Wait()
		assert.Equal(t, 2, int(calls.Load()))
		cancel()
		synctest.Wait()
	})
}

func TestMonitorScheduledAndImmediateRequestsShareProbe(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		var calls atomic.Int32
		release := make(chan struct{})
		cfg := monitor.Config{ID: "test", Interval: 10}
		pm := newMonitorManagerWithProbe(func(ctx context.Context, config monitor.Config) (int64, error) {
			assert.Equal(t, cfg, config)
			calls.Add(1)
			<-release
			return 42, nil
		})
		defer pm.Stop()
		task := newMonitorTask(cfg)
		pm.monitors[cfg.ID] = task
		go runMonitorSchedule(task.ctx, 10*time.Second, 0, func() { task.runProbe(pm.probe) })
		synctest.Wait()
		results := make(chan *monitor.Result, 2)
		for range 2 {
			go func() {
				result, _ := pm.UpsertMonitor(cfg, true)
				results <- result
			}()
		}
		synctest.Wait()
		assert.Equal(t, 1, int(calls.Load()))
		assert.Empty(t, pm.GetResults(1000), "reading history must not wait for network I/O")
		close(release)
		synctest.Wait()
		first, second := <-results, <-results
		require.NotNil(t, first)
		require.NotNil(t, second)
		assert.Equal(t, int64(42), first.AvgResponse)
		assert.Equal(t, first, second)
		assert.NotSame(t, first, second, "callers must not share mutable result pointers")
		assert.Len(t, task.history.samples, 1)
		// A later explicit request must still perform a fresh probe.
		_, err := pm.UpsertMonitor(cfg, true)
		require.NoError(t, err)
		assert.Equal(t, 2, int(calls.Load()))
		assert.Len(t, task.history.samples, 2)
	})
}

func TestMonitorReplacementCancelsSharedProbe(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		cfg := monitor.Config{ID: "test", Interval: 10}
		pm := newMonitorManagerWithProbe(func(ctx context.Context, config monitor.Config) (int64, error) {
			if config.Interval == 10 {
				<-ctx.Done()
				return 0, ctx.Err()
			}
			return 30, nil
		})
		defer pm.Stop()
		task := newMonitorTask(cfg)
		task.history.record(monitorSample{responseUs: 10, timestamp: time.Now()})
		pm.monitors[cfg.ID] = task
		results := make(chan *monitor.Result, 2)
		for range 2 {
			go func() {
				result, _ := pm.UpsertMonitor(cfg, true)
				results <- result
			}()
		}
		synctest.Wait()
		updated := cfg
		updated.Interval = 20
		result, err := pm.UpsertMonitor(updated, true)
		require.NoError(t, err)
		require.NotNil(t, result)
		assert.Equal(t, int64(20), result.AvgResponse)
		assert.Zero(t, result.PacketLoss)
		synctest.Wait()
		assert.Nil(t, <-results)
		assert.Nil(t, <-results)
		assert.Len(t, task.history.samples, 1)
		assert.Len(t, pm.monitors[cfg.ID].history.samples, 2)
	})
}

func TestMonitorInjectedProbeTimeoutRecordsLoss(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		pm := newMonitorManagerWithProbe(func(ctx context.Context, _ monitor.Config) (int64, error) {
			ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
			defer cancel()
			<-ctx.Done()
			return 0, ctx.Err()
		})
		defer pm.Stop()
		start := time.Now()
		result, err := pm.UpsertMonitor(monitor.Config{ID: "test", Interval: 3600}, true)
		require.NoError(t, err)
		require.NotNil(t, result)
		assert.Equal(t, 3*time.Second, time.Since(start))
		assert.Equal(t, 100.0, result.PacketLoss)
		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")
	})
}