spb/internetpressure

1// Package sched schedules (target, check-family) jobs with a single priority timer wheel.2//3// Guarantees (docs/PROBE-PROTOCOL.md "Rules the agent follows" + SPEC ethics):4//   - per-family intervals: http → schedule.tiers[tier], dns → dns_every, ping/tcp → ping_every,5//     traceroute → traceroute_every; boost multiplies the interval for listed targets until `until`;6//   - ±10 % jitter on every interval, initial offsets spread uniformly over the interval (no start-up burst);7//   - never below MinInterval (10 s) even if the server asks for less (clamped, logged);8//   - a global concurrency limit, at most one in-flight check per hostname, at most one traceroute at a time;9//   - config changes (targets added/removed/edited, intervals) are applied live without restarting.10//11// All state is owned by the Run loop goroutine; Apply/SetBoost communicate through channels, so no locks.12package sched1314import (15	"container/heap"16	"context"17	"log/slog"18	"math/rand/v2"19	"runtime/debug"20	"strconv"21	"sync/atomic"22	"time"2324	"internetpressure.io/probe-agent/internal/protocol"25)2627// Family is a check family.28type Family string2930// Families.31const (32	FamilyHTTP       Family = "http"33	FamilyDNS        Family = "dns"34	FamilyPing       Family = "ping"35	FamilyTCP        Family = "tcp"36	FamilyTraceroute Family = "traceroute"37)3839// Defaults when the server omits a schedule value.40const (41	MinInterval          = 10 * time.Second42	DefaultDNSEvery      = 60 * time.Second43	DefaultPingEvery     = 30 * time.Second44	DefaultTraceEvery    = 15 * time.Minute45	DefaultTierInterval  = 180 * time.Second46	JitterFraction       = 0.1047	busyRetryBase        = 500 * time.Millisecond48	DefaultMaxConcurrent = 849)5051// Job is one schedulable unit.52type Job struct {53	Key      string // target_id + "/" + family54	Target   protocol.Target55	Family   Family56	Interval time.Duration // base interval (clamped, before boost/jitter)57}5859// Runner executes a job. It must return when ctx is done.60type Runner func(ctx context.Context, job Job)6162// Options tune the scheduler.63type Options struct {64	MaxConcurrency int65	MinInterval    time.Duration // default MinInterval; tests lower it66	Logger         *slog.Logger67	Seed           uint64 // 0 → random68	// Capabilities: when false the family is not scheduled at all.69	CanPing       bool70	CanTraceroute bool71	// IntervalScale multiplies every configured interval before the clamp (tests only; 0 → 1).72	IntervalScale float6473}7475type entry struct {76	job     Job77	gen     uint6478	running bool79}8081type item struct {82	key string83	gen uint6484	at  time.Time85}8687type itemHeap []item8889func (h itemHeap) Len() int           { return len(h) }90func (h itemHeap) Less(i, j int) bool { return h[i].at.Before(h[j].at) }91func (h itemHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }92func (h *itemHeap) Push(x any)        { *h = append(*h, x.(item)) }93func (h *itemHeap) Pop() any          { old := *h; n := len(old); it := old[n-1]; *h = old[:n-1]; return it }94func (h itemHeap) peek() (item, bool) {95	if len(h) == 0 {96		return item{}, false97	}98	return h[0], true99}100101type doneMsg struct {102	key      string103	gen      uint64104	host     string105	trace    bool106	panicked bool107}108109// Scheduler is the timer wheel.110type Scheduler struct {111	run  Runner112	opts Options113	rng  *rand.Rand114	log  *slog.Logger115116	applyCh chan *protocol.RemoteConfig117	boostCh chan *protocol.Boost118	doneCh  chan doneMsg119120	// loop-owned121	entries  map[string]*entry122	heap     itemHeap123	hostBusy map[string]int124	inflight int125	trBusy   bool126	boost    *protocol.Boost127	gen      uint64128129	targets atomic.Int32130	jobs    atomic.Int32131	panics  atomic.Uint64132	runs    atomic.Uint64133}134135// New creates a scheduler; call Run to start it.136func New(run Runner, opts Options) *Scheduler {137	if opts.MaxConcurrency <= 0 {138		opts.MaxConcurrency = DefaultMaxConcurrent139	}140	if opts.MinInterval <= 0 {141		opts.MinInterval = MinInterval142	}143	if opts.Logger == nil {144		opts.Logger = slog.Default()145	}146	seed := opts.Seed147	if seed == 0 {148		seed = rand.Uint64()149	}150	return &Scheduler{151		run:      run,152		opts:     opts,153		rng:      rand.New(rand.NewPCG(seed, seed^0x9e3779b97f4a7c15)),154		log:      opts.Logger,155		applyCh:  make(chan *protocol.RemoteConfig, 1),156		boostCh:  make(chan *protocol.Boost, 1),157		doneCh:   make(chan doneMsg, 64),158		entries:  map[string]*entry{},159		hostBusy: map[string]int{},160	}161}162163// Apply installs a new remote config (targets + schedule + boost). Non-blocking; the latest wins.164func (s *Scheduler) Apply(cfg *protocol.RemoteConfig) {165	for {166		select {167		case s.applyCh <- cfg:168			return169		default:170			select {171			case <-s.applyCh: // drop the stale pending config172			default:173			}174		}175	}176}177178// SetBoost replaces the active boost (nil clears it).179func (s *Scheduler) SetBoost(b *protocol.Boost) {180	select {181	case s.boostCh <- b:182	default:183		select {184		case <-s.boostCh:185		default:186		}187		s.boostCh <- b188	}189}190191// Targets returns the number of distinct scheduled targets.192func (s *Scheduler) Targets() int { return int(s.targets.Load()) }193194// Jobs returns the number of scheduled (target, family) jobs.195func (s *Scheduler) Jobs() int { return int(s.jobs.Load()) }196197// Panics returns the number of recovered check panics.198func (s *Scheduler) Panics() uint64 { return s.panics.Load() }199200// Runs returns the number of completed job executions.201func (s *Scheduler) Runs() uint64 { return s.runs.Load() }202203// Run drives the wheel until ctx is cancelled. In-flight jobs are cancelled through ctx too.204func (s *Scheduler) Run(ctx context.Context) {205	timer := time.NewTimer(time.Hour)206	timer.Stop()207	for {208		// Dispatch everything that is due, then arm the timer for the next item.209		now := time.Now()210		for {211			it, ok := s.heap.peek()212			if !ok || it.at.After(now) {213				break214			}215			heap.Pop(&s.heap)216			s.dispatch(ctx, it, now)217		}218		if it, ok := s.heap.peek(); ok {219			timer.Reset(time.Until(it.at))220		} else {221			timer.Stop()222		}223		select {224		case <-ctx.Done():225			timer.Stop()226			return227		case cfg := <-s.applyCh:228			s.apply(cfg, time.Now())229		case b := <-s.boostCh:230			s.boost = b231		case d := <-s.doneCh:232			s.complete(d, time.Now())233		case <-timer.C:234		}235	}236}237238func (s *Scheduler) dispatch(ctx context.Context, it item, now time.Time) {239	e, ok := s.entries[it.key]240	if !ok || e.gen != it.gen {241		return // removed or replaced242	}243	host := e.job.Target.Hostname244	trace := e.job.Family == FamilyTraceroute245	if e.running || s.hostBusy[host] > 0 || s.inflight >= s.opts.MaxConcurrency || (trace && s.trBusy) {246		// Busy: try again shortly, spread so that many deferred jobs don't collide. The retry window is247		// 0.5–1.5 s in production (MinInterval 10 s) and scales down with MinInterval in tests.248		base := busyRetryBase249		if b := s.opts.MinInterval / 20; b < base {250			base = b251		}252		s.push(it.key, it.gen, now.Add(base+time.Duration(s.rng.Int64N(int64(2*base)))))253		return254	}255	e.running = true256	s.hostBusy[host]++257	s.inflight++258	if trace {259		s.trBusy = true260	}261	job := e.job262	gen := e.gen263	go func() {264		msg := doneMsg{key: job.Key, gen: gen, host: host, trace: trace}265		defer func() {266			if r := recover(); r != nil {267				msg.panicked = true268				s.panics.Add(1)269				s.log.Error("check panicked", "job", job.Key, "panic", r, "stack", string(debug.Stack()))270			}271			s.runs.Add(1)272			select {273			case s.doneCh <- msg:274			case <-ctx.Done():275				// Loop has exited; nothing to release.276			}277		}()278		s.run(ctx, job)279	}()280}281282func (s *Scheduler) complete(d doneMsg, now time.Time) {283	s.inflight--284	if s.hostBusy[d.host] > 1 {285		s.hostBusy[d.host]--286	} else {287		delete(s.hostBusy, d.host)288	}289	if d.trace {290		s.trBusy = false291	}292	e, ok := s.entries[d.key]293	if !ok {294		return // removed while running295	}296	if e.gen != d.gen {297		return // replaced while running: the new generation already has a heap item298	}299	e.running = false300	s.push(d.key, e.gen, now.Add(s.nextDelay(e.job, now)))301}302303func (s *Scheduler) push(key string, gen uint64, at time.Time) {304	heap.Push(&s.heap, item{key: key, gen: gen, at: at})305}306307// nextDelay = interval × boost (if active) clamped to MinInterval, with ±10 % jitter.308func (s *Scheduler) nextDelay(job Job, now time.Time) time.Duration {309	base := job.Interval310	if s.boost != nil && s.boost.Active(job.Target.TargetID, now) {311		base = time.Duration(float64(base) * s.boost.Factor)312	}313	if base < s.opts.MinInterval {314		base = s.opts.MinInterval315	}316	return Jitter(base, s.rng)317}318319// Jitter returns d × U(0.9, 1.1).320func Jitter(d time.Duration, rng *rand.Rand) time.Duration {321	f := 1 + (rng.Float64()*2-1)*JitterFraction322	return time.Duration(float64(d) * f)323}324325// InitialOffset returns U[0, d): uniform spread of first runs over one interval.326func InitialOffset(d time.Duration, rng *rand.Rand) time.Duration {327	if d <= 0 {328		return 0329	}330	return time.Duration(rng.Int64N(int64(d)))331}332333// apply computes the desired job set from cfg and reconciles it with the current entries.334func (s *Scheduler) apply(cfg *protocol.RemoteConfig, now time.Time) {335	desired := map[string]Job{}336	targets := 0337	if cfg != nil && cfg.Probe.Enabled {338		for _, j := range JobsFor(cfg, s.opts) {339			desired[j.Key] = j340		}341		targets = len(cfg.Targets)342	}343	// Remove vanished jobs (heap items are dropped lazily on pop).344	for key := range s.entries {345		if _, keep := desired[key]; !keep {346			delete(s.entries, key)347		}348	}349	added, changed := 0, 0350	for key, job := range desired {351		e, ok := s.entries[key]352		if !ok {353			s.gen++354			s.entries[key] = &entry{job: job, gen: s.gen}355			s.push(key, s.gen, now.Add(InitialOffset(job.Interval, s.rng)))356			added++357			continue358		}359		if targetChanged(e.job.Target, job.Target) {360			// Re-key the generation: the old heap item is discarded on pop, the job restarts spread out.361			s.gen++362			e.job, e.gen = job, s.gen363			s.push(key, s.gen, now.Add(InitialOffset(job.Interval, s.rng)))364			changed++365			continue366		}367		e.job.Interval = job.Interval // takes effect after the next run368	}369	s.targets.Store(int32(targets))370	s.jobs.Store(int32(len(s.entries)))371	if cfg != nil {372		s.log.Info("schedule applied", "config_version", cfg.ConfigVersion, "enabled", cfg.Probe.Enabled,373			"targets", targets, "jobs", len(s.entries), "added", added, "changed", changed)374	}375}376377func targetChanged(a, b protocol.Target) bool {378	if a.Hostname != b.Hostname || a.URL != b.URL || a.Port != b.Port || a.FixedIP() != b.FixedIP() || a.Tier != b.Tier {379		return true380	}381	return false382}383384// JobsFor expands a remote config into jobs (one per target × family), applying defaults, capability filters385// and the MinInterval clamp.386func JobsFor(cfg *protocol.RemoteConfig, opts Options) []Job {387	minI := opts.MinInterval388	if minI <= 0 {389		minI = MinInterval390	}391	log := opts.Logger392	if log == nil {393		log = slog.Default()394	}395	scale := opts.IntervalScale396	if scale <= 0 {397		scale = 1398	}399	sc := cfg.Schedule400	secs := func(v int, def time.Duration) time.Duration {401		if v <= 0 {402			return time.Duration(float64(def) * scale)403		}404		return time.Duration(float64(v) * scale * float64(time.Second))405	}406	clamp := func(d time.Duration, what string) time.Duration {407		if d < minI {408			log.Warn("interval below minimum, clamped", "what", what, "requested", d, "min", minI)409			return minI410		}411		return d412	}413	dnsEvery := clamp(secs(sc.DNSEvery, DefaultDNSEvery), "dns_every")414	pingEvery := clamp(secs(sc.PingEvery, DefaultPingEvery), "ping_every")415	trEvery := clamp(secs(sc.TracerouteEvery, DefaultTraceEvery), "traceroute_every")416417	var jobs []Job418	for _, t := range cfg.Targets {419		if t.TargetID == "" || t.Hostname == "" {420			continue421		}422		tierI := secs(sc.Tiers[strconv.Itoa(t.Tier)], DefaultTierInterval)423		tierI = clamp(tierI, "tier "+strconv.Itoa(t.Tier))424		add := func(f Family, d time.Duration) {425			jobs = append(jobs, Job{Key: t.TargetID + "/" + string(f), Target: t, Family: f, Interval: d})426		}427		for _, c := range t.Checks {428			switch Family(c) {429			case FamilyHTTP:430				add(FamilyHTTP, tierI)431			case FamilyDNS:432				add(FamilyDNS, dnsEvery)433			case FamilyPing:434				if opts.CanPing {435					add(FamilyPing, pingEvery)436				} else {437					add(FamilyTCP, pingEvery) // icmp unavailable → tcp connects on the same cadence438				}439			case FamilyTCP:440				if !t.HasCheck("ping") || opts.CanPing { // avoid a duplicate tcp job when ping already fell back441					add(FamilyTCP, pingEvery)442				}443			}444		}445		if t.Traceroute && opts.CanTraceroute {446			add(FamilyTraceroute, trEvery)447		}448	}449	return jobs450}451