spb/internetpressure

1package sched23import (4	"context"5	"log/slog"6	"math/rand/v2"7	"strconv"8	"sync"9	"testing"10	"time"1112	"internetpressure.io/probe-agent/internal/protocol"13)1415func TestJitterBounds(t *testing.T) {16	rng := rand.New(rand.NewPCG(1, 2))17	d := 20 * time.Second18	var sum time.Duration19	for i := 0; i < 20000; i++ {20		j := Jitter(d, rng)21		if j < 18*time.Second || j > 22*time.Second {22			t.Fatalf("jitter out of ±10 %%: %v", j)23		}24		sum += j25	}26	mean := sum / 2000027	if mean < 19900*time.Millisecond || mean > 20100*time.Millisecond {28		t.Fatalf("jitter mean drifted: %v", mean)29	}30}3132func TestInitialOffsetSpread(t *testing.T) {33	rng := rand.New(rand.NewPCG(3, 4))34	d := 60 * time.Second35	buckets := make([]int, 6)36	n := 6000037	for i := 0; i < n; i++ {38		o := InitialOffset(d, rng)39		if o < 0 || o >= d {40			t.Fatalf("offset out of [0, d): %v", o)41		}42		buckets[int(o/(10*time.Second))]++43	}44	for i, b := range buckets {45		if b < n/6*9/10 || b > n/6*11/10 {46			t.Fatalf("bucket %d not uniform: %d of %d", i, b, n)47		}48	}49	if InitialOffset(0, rng) != 0 {50		t.Fatal("zero interval must give zero offset")51	}52}5354func cfgWith(n int, checks []string, tier int) *protocol.RemoteConfig {55	cfg := &protocol.RemoteConfig{ConfigVersion: "v1"}56	cfg.Probe.Enabled = true57	cfg.Schedule = protocol.Schedule{Tiers: map[string]int{"1": 20, "2": 45, "3": 180}, DNSEvery: 60, PingEvery: 30, TracerouteEvery: 900}58	for i := 0; i < n; i++ {59		id := "t" + strconv.Itoa(i)60		cfg.Targets = append(cfg.Targets, protocol.Target{TargetID: id, Hostname: id + ".example", URL: "https://" + id + ".example/",61			Port: 443, Tier: tier, Checks: checks, Traceroute: true})62	}63	return cfg64}6566func TestJobsForIntervalsClampAndCapabilities(t *testing.T) {67	cfg := cfgWith(1, []string{"http", "dns", "ping"}, 1)68	cfg.Schedule.Tiers["1"] = 3 // below the 10 s floor69	cfg.Schedule.PingEvery = 0  // default 30 s70	jobs := JobsFor(cfg, Options{CanPing: true, CanTraceroute: true, Logger: slog.Default()})71	got := map[Family]time.Duration{}72	for _, j := range jobs {73		got[j.Family] = j.Interval74	}75	if len(jobs) != 4 {76		t.Fatalf("expected 4 jobs, got %d: %+v", len(jobs), jobs)77	}78	if got[FamilyHTTP] != MinInterval {79		t.Errorf("tier 3 s should be clamped to 10 s, got %v", got[FamilyHTTP])80	}81	if got[FamilyDNS] != 60*time.Second || got[FamilyPing] != 30*time.Second || got[FamilyTraceroute] != 900*time.Second {82		t.Errorf("intervals: %v", got)83	}84	// No ICMP → ping becomes tcp; no traceroute binary → no traceroute job.85	jobs = JobsFor(cfg, Options{CanPing: false, CanTraceroute: false})86	fams := map[Family]int{}87	for _, j := range jobs {88		fams[j.Family]++89	}90	if fams[FamilyPing] != 0 || fams[FamilyTCP] != 1 || fams[FamilyTraceroute] != 0 || len(jobs) != 3 {91		t.Errorf("capability filter: %v", fams)92	}93	// Explicit "tcp" next to "ping" without ICMP must not duplicate.94	cfg = cfgWith(1, []string{"ping", "tcp"}, 2)95	jobs = JobsFor(cfg, Options{CanPing: false})96	if len(jobs) != 1 || jobs[0].Family != FamilyTCP {97		t.Errorf("ping+tcp without icmp: %+v", jobs)98	}99	jobs = JobsFor(cfg, Options{CanPing: true})100	if len(jobs) != 2 {101		t.Errorf("ping+tcp with icmp: %+v", jobs)102	}103}104105func TestBoost(t *testing.T) {106	s := New(func(context.Context, Job) {}, Options{MinInterval: time.Second, Seed: 7})107	job := Job{Target: protocol.Target{TargetID: "a"}, Interval: 100 * time.Second}108	now := time.Now()109	s.boost = &protocol.Boost{Targets: []string{"a"}, Factor: 0.5, Until: protocol.FormatTime(now.Add(time.Hour))}110	if d := s.nextDelay(job, now); d < 45*time.Second || d > 55*time.Second {111		t.Fatalf("boosted delay = %v", d)112	}113	s.boost.Until = protocol.FormatTime(now.Add(-time.Minute)) // expired114	if d := s.nextDelay(job, now); d < 90*time.Second || d > 110*time.Second {115		t.Fatalf("expired boost delay = %v", d)116	}117	s.boost = &protocol.Boost{Targets: []string{"b"}, Factor: 0.1, Until: protocol.FormatTime(now.Add(time.Hour))}118	if d := s.nextDelay(job, now); d < 90*time.Second {119		t.Fatalf("boost for another target applied: %v", d)120	}121	// Boost can never push below the floor.122	s.boost = &protocol.Boost{Targets: []string{"a"}, Factor: 0.001, Until: protocol.FormatTime(now.Add(time.Hour))}123	if d := s.nextDelay(job, now); d < 900*time.Millisecond {124		t.Fatalf("boost broke the floor: %v", d)125	}126}127128// Live run with tiny intervals: every job runs repeatedly, never two checks on the same host at once, never129// more than MaxConcurrency in flight, and a config change adds/removes jobs without restart.130func TestRunConcurrencyAndLiveApply(t *testing.T) {131	var (132		mu       sync.Mutex133		perHost  = map[string]int{}134		inflight int135		maxIn    int136		runs     = map[string]int{}137	)138	runner := func(ctx context.Context, j Job) {139		mu.Lock()140		perHost[j.Target.Hostname]++141		if perHost[j.Target.Hostname] > 1 {142			t.Errorf("two checks in flight for %s", j.Target.Hostname)143		}144		inflight++145		if inflight > maxIn {146			maxIn = inflight147		}148		runs[j.Key]++149		mu.Unlock()150		time.Sleep(15 * time.Millisecond)151		mu.Lock()152		perHost[j.Target.Hostname]--153		inflight--154		mu.Unlock()155	}156	// IntervalScale 0.001 → configured seconds become milliseconds: 40 ms floor, 60 ms intervals.157	s := New(runner, Options{MaxConcurrency: 3, MinInterval: 40 * time.Millisecond, CanPing: true, CanTraceroute: true,158		Seed: 11, IntervalScale: 0.001})159	ctx, cancel := context.WithCancel(context.Background())160	defer cancel()161	go s.Run(ctx)162163	cfg := cfgWith(5, []string{"http", "dns", "ping"}, 1)164	cfg.Schedule.Tiers["1"], cfg.Schedule.DNSEvery, cfg.Schedule.PingEvery, cfg.Schedule.TracerouteEvery = 60, 60, 60, 60165	s.Apply(cfg)166	time.Sleep(600 * time.Millisecond)167	if s.Targets() != 5 || s.Jobs() != 20 {168		t.Fatalf("targets=%d jobs=%d", s.Targets(), s.Jobs())169	}170	mu.Lock()171	for _, tg := range cfg.Targets {172		for _, f := range []Family{FamilyHTTP, FamilyDNS, FamilyPing, FamilyTraceroute} {173			if runs[tg.TargetID+"/"+string(f)] < 2 {174				t.Errorf("job %s/%s ran %d times", tg.TargetID, f, runs[tg.TargetID+"/"+string(f)])175			}176		}177	}178	if maxIn > 3 {179		t.Errorf("max in flight %d > 3", maxIn)180	}181	mu.Unlock()182183	// Shrink to 2 targets with http only; the rest must stop running.184	cfg2 := cfgWith(2, []string{"http"}, 1)185	cfg2.Schedule.Tiers["1"] = 60186	for i := range cfg2.Targets {187		cfg2.Targets[i].Traceroute = false188	}189	s.Apply(cfg2)190	time.Sleep(150 * time.Millisecond)191	mu.Lock()192	before := runs["t4/dns"]193	mu.Unlock()194	time.Sleep(200 * time.Millisecond)195	mu.Lock()196	after := runs["t4/dns"]197	mu.Unlock()198	if after != before {199		t.Errorf("removed job kept running: %d → %d", before, after)200	}201	if s.Targets() != 2 || s.Jobs() != 2 {202		t.Errorf("after shrink: targets=%d jobs=%d", s.Targets(), s.Jobs())203	}204205	// Disabled probe → idle.206	cfg2.Probe.Enabled = false207	s.Apply(cfg2)208	time.Sleep(100 * time.Millisecond)209	if s.Jobs() != 0 {210		t.Errorf("disabled probe should have no jobs, has %d", s.Jobs())211	}212	if s.Panics() != 0 {213		t.Errorf("unexpected panics: %d", s.Panics())214	}215}216217func TestPanicRecovered(t *testing.T) {218	s := New(func(context.Context, Job) { panic("boom") }, Options{MinInterval: 30 * time.Millisecond, Seed: 5,219		Logger: slog.New(slog.DiscardHandler), IntervalScale: 0.001})220	ctx, cancel := context.WithCancel(context.Background())221	defer cancel()222	go s.Run(ctx)223	cfg := cfgWith(1, []string{"http"}, 1)224	cfg.Schedule.Tiers["1"] = 30225	s.Apply(cfg)226	time.Sleep(200 * time.Millisecond)227	if s.Panics() < 2 {228		t.Fatalf("panics should be recovered and the job rescheduled, got %d", s.Panics())229	}230}231