diff --git a/cmd/server/analytics_pause_when_idle.go b/cmd/server/analytics_pause_when_idle.go new file mode 100644 index 000000000..0734064ba --- /dev/null +++ b/cmd/server/analytics_pause_when_idle.go @@ -0,0 +1,57 @@ +// Package main: snapshot freshness headers for analytics.pauseWhenIdle +// (issue #492). +// +// With pauseWhenIdle on, a recomputer nobody reads stops refreshing, and +// the first read after that is served the snapshot it already holds while +// one refresh runs. That snapshot can be older than the recompute +// interval, so every default-shape response served from a recomputer says +// how old its snapshot is and whether it missed a scheduled refresh. With +// pauseWhenIdle off no header is set: responses are unchanged. +package main + +import ( + "net/http" + "strconv" + "time" +) + +const ( + // analyticsSnapshotHeader is "current" or "stale" ("stale": the + // snapshot outlived a refresh skipped while idle; a refresh is on + // its way). + analyticsSnapshotHeader = "X-Analytics-Snapshot" + // analyticsSnapshotAgeHeader is the snapshot's age in whole seconds. + analyticsSnapshotAgeHeader = "X-Analytics-Snapshot-Age-Seconds" +) + +// recomputer returns *field under analyticsRecomputerMu (nil before +// StartAnalyticsRecomputers). +func (s *PacketStore) recomputer(field **analyticsRecomputer) *analyticsRecomputer { + s.analyticsRecomputerMu.RLock() + defer s.analyticsRecomputerMu.RUnlock() + return *field +} + +// setAnalyticsSnapshotHeaders labels a response about to be served from +// rc's snapshot. Call it BEFORE reading the snapshot: a refresh that lands +// in between only makes the served snapshot newer than labelled, never +// older. No-op when rc is nil, has no snapshot or pauseWhenIdle is off. +func setAnalyticsSnapshotHeaders(w http.ResponseWriter, rc *analyticsRecomputer) { + if rc == nil || !rc.pauseWhenIdle { + return + } + at, stale := rc.SnapshotInfo() + if at.IsZero() { + return + } + state := "current" + if stale { + state = "stale" + } + age := rc.now().Sub(at) + if age < 0 { + age = 0 + } + w.Header().Set(analyticsSnapshotHeader, state) + w.Header().Set(analyticsSnapshotAgeHeader, strconv.FormatInt(int64(age/time.Second), 10)) +} diff --git a/cmd/server/analytics_pause_when_idle_measure_test.go b/cmd/server/analytics_pause_when_idle_measure_test.go new file mode 100644 index 000000000..1b5c93a24 --- /dev/null +++ b/cmd/server/analytics_pause_when_idle_measure_test.go @@ -0,0 +1,144 @@ +package main + +import ( + "fmt" + "io" + "os" + "path/filepath" + "runtime" + "runtime/metrics" + "sort" + "testing" + "time" +) + +// measurePauseWhenIdle backs TestPauseWhenIdle_Measure (#492): on a copy of +// test-fixtures/e2e-fixture.db it runs the nine analytics recomputers at a +// short interval for a fixed window, once per (pauseWhenIdle, traffic) +// pair, and prints compute passes, allocations, process CPU and read +// latency. "visible" reads every default-shape snapshot once per interval +// (an open analytics page polling); "sparse" reads them once every ten +// intervals (each read is a first read after idle); "idle" reads nothing. +func measurePauseWhenIdle(t *testing.T) { + interval := 100 * time.Millisecond + window := 5 * time.Second + if v := os.Getenv("PAUSE_IDLE_WINDOW"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + window = d + } + } + src := filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db") + fmt.Printf("| pauseWhenIdle | traffic | compute passes | alloc MB | mallocs (k) | CPU s | read p50 | read p99 | read max |\n") + fmt.Printf("|---|---|---|---|---|---|---|---|---|\n") + for _, pause := range []bool{false, true} { + for _, mode := range []struct { + name string + readEvery int + }{{"idle", 0}, {"sparse", 10}, {"visible", 1}} { + dbPath := filepath.Join(t.TempDir(), "fixture.db") + copyFileForMeasure(t, src, dbPath) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + s := NewPacketStore(db, nil) + s.graph.Store(loadNeighborEdgesFromDB(db.conn)) // main.go order: graph before packets + if err := s.Load(); err != nil { + t.Fatal(err) + } + s.WaitIndexesReady(30 * time.Second) + s.signalStartupLoadDone() + stop := s.StartAnalyticsRecomputers(interval, AnalyticsRecomputeIntervals{PauseWhenIdle: pause}) + rcs := storeRecomputerFields(t, s) + for _, rc := range rcs { + rc.RecomputeNow() // settle past the post-load pass before measuring + } + runtime.GC() + runs0 := totalRuns(rcs) + var m0, m1 runtime.MemStats + runtime.ReadMemStats(&m0) + cpu0 := processCPUSeconds() + + var lat []time.Duration + deadline := time.Now().Add(window) + tk := time.NewTicker(interval) + for n := 1; time.Now().Before(deadline); n++ { + <-tk.C + if mode.readEvery > 0 && n%mode.readEvery == 0 { + t0 := time.Now() + readAllDefaultSnapshots(s) + lat = append(lat, time.Since(t0)) + } + } + tk.Stop() + + cpu1 := processCPUSeconds() + runtime.ReadMemStats(&m1) + runs := totalRuns(rcs) - runs0 + stop() + db.Close() + + p50, p99, pmax := "-", "-", "-" + if len(lat) > 0 { + sort.Slice(lat, func(i, j int) bool { return lat[i] < lat[j] }) + p50 = lat[len(lat)/2].String() + p99 = lat[len(lat)*99/100].String() + pmax = lat[len(lat)-1].String() + } + fmt.Printf("| %v | %s | %d | %.1f | %.1f | %.2f | %s | %s | %s |\n", + pause, mode.name, runs, + float64(m1.TotalAlloc-m0.TotalAlloc)/1e6, + float64(m1.Mallocs-m0.Mallocs)/1e3, + cpu1-cpu0, p50, p99, pmax) + } + } + fmt.Printf("\ninterval=%s window=%s recomputers=9 GOMAXPROCS=%d\n", interval, window, runtime.GOMAXPROCS(0)) +} + +func totalRuns(rcs map[string]*analyticsRecomputer) int64 { + var n int64 + for _, rc := range rcs { + n += rc.ComputeRuns() + } + return n +} + +// readAllDefaultSnapshots is what an open analytics page reads per refresh. +func readAllDefaultSnapshots(s *PacketStore) { + s.GetAnalyticsRF("", "") + s.GetAnalyticsTopology("", "") + s.GetAnalyticsChannels("", "") + s.GetAnalyticsDistance("", "") + s.GetAnalyticsHashSizes("", "") + s.GetAnalyticsHashCollisions("", "") + s.GetAnalyticsRoles() + s.GetObserverCalibrations() + s.GetFleetClockSkew("") +} + +// processCPUSeconds is the runtime's estimate of total CPU time used by the +// process (user + GC + idle-excluded), portable across CI platforms. +func processCPUSeconds() float64 { + samples := []metrics.Sample{{Name: "/cpu/classes/total:cpu-seconds"}, {Name: "/cpu/classes/idle:cpu-seconds"}} + metrics.Read(samples) + return samples[0].Value.Float64() - samples[1].Value.Float64() +} + +func copyFileForMeasure(t *testing.T, src, dst string) { + t.Helper() + in, err := os.Open(src) + if err != nil { + t.Fatal(err) + } + defer in.Close() + out, err := os.Create(dst) + if err != nil { + t.Fatal(err) + } + if _, err := io.Copy(out, in); err != nil { + t.Fatal(err) + } + if err := out.Close(); err != nil { + t.Fatal(err) + } +} diff --git a/cmd/server/analytics_pause_when_idle_test.go b/cmd/server/analytics_pause_when_idle_test.go new file mode 100644 index 000000000..be9863bac --- /dev/null +++ b/cmd/server/analytics_pause_when_idle_test.go @@ -0,0 +1,777 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Issue #492: opt-in analytics.pauseWhenIdle. A scheduled recompute is +// skipped when nothing read the snapshot since the previous pass; the +// next read is served the existing snapshot at once and wakes ONE +// coalesced refresh. +// +// Every test here is driven by a manual ticker and the loop's event hook, +// never by sleeping: a tick is delivered by a channel send, and the test +// waits for the loop to report what it did with it. + +type manualRecomputeTicker struct { + c chan time.Time + resets atomic.Int64 + // resetC gets one (non-blocking) send per Reset. The loop calls Reset + // after a kick or RecomputeNow pass, so a receive means that pass is + // done. + resetC chan struct{} +} + +func newManualRecomputeTicker() *manualRecomputeTicker { + return &manualRecomputeTicker{c: make(chan time.Time), resetC: make(chan struct{}, 16)} +} + +func (m *manualRecomputeTicker) C() <-chan time.Time { return m.c } +func (m *manualRecomputeTicker) Stop() {} +func (m *manualRecomputeTicker) Reset(time.Duration) { + m.resets.Add(1) + select { + case m.resetC <- struct{}{}: + default: + } +} + +// snapshotClock is an injectable analyticsRecomputer.now. +type snapshotClock struct{ ns atomic.Int64 } + +func newSnapshotClock() *snapshotClock { + c := &snapshotClock{} + c.ns.Store(time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC).UnixNano()) + return c +} + +func (c *snapshotClock) now() time.Time { return time.Unix(0, c.ns.Load()) } +func (c *snapshotClock) advance(d time.Duration) { c.ns.Add(int64(d)) } + +const pauseTestDeadline = 5 * time.Second + +type pauseHarness struct { + t *testing.T + r *analyticsRecomputer + tk *manualRecomputeTicker + events chan string +} + +// newPauseHarness builds an unstarted recomputer whose compute is fn +// (called with the 1-based pass number) and whose loop reports every +// event on h.events. +func newPauseHarness(t *testing.T, name string, pause bool, fn func(pass int64) interface{}) *pauseHarness { + t.Helper() + h := &pauseHarness{t: t, tk: newManualRecomputeTicker(), events: make(chan string, 256)} + var passes atomic.Int64 + h.r = newAnalyticsRecomputer(name, time.Hour, func() interface{} { + return fn(passes.Add(1)) + }) + h.r.pauseWhenIdle = pause + h.r.newTicker = func(string, time.Duration) recomputeTicker { return h.tk } + h.r.onLoopEvent = func(ev string) { h.events <- ev } + return h +} + +func (h *pauseHarness) start() *pauseHarness { + h.r.Start() + h.t.Cleanup(h.r.Stop) + return h +} + +func countingPass(pass int64) interface{} { return pass } + +func (h *pauseHarness) next() string { + h.t.Helper() + select { + case ev := <-h.events: + return ev + case <-time.After(pauseTestDeadline): + h.t.Fatalf("%s: no loop event within %s", h.r.name, pauseTestDeadline) + return "" + } +} + +func (h *pauseHarness) tick() string { + h.t.Helper() + select { + case h.tk.c <- time.Now(): + case <-time.After(pauseTestDeadline): + h.t.Fatalf("%s: loop did not accept a tick", h.r.name) + } + return h.next() +} + +func (h *pauseHarness) expect(want string) { + h.t.Helper() + if got := h.next(); got != want { + h.t.Fatalf("%s: loop event = %q, want %q", h.r.name, got, want) + } +} + +func (h *pauseHarness) expectTick(want string) { + h.t.Helper() + if got := h.tick(); got != want { + h.t.Fatalf("%s: tick event = %q, want %q", h.r.name, got, want) + } +} + +func (h *pauseHarness) noPendingEvent() { + h.t.Helper() + select { + case ev := <-h.events: + h.t.Fatalf("%s: unexpected loop event %q", h.r.name, ev) + default: + } +} + +// --- config ------------------------------------------------------------- + +func TestPauseWhenIdle_ConfigDefaultsOff(t *testing.T) { + if (*Config)(nil).AnalyticsPauseWhenIdle() || (*Config)(nil).AnalyticsRecomputeIntervals().PauseWhenIdle { + t.Fatal("nil config must leave pauseWhenIdle off") + } + for _, raw := range []string{ + `{}`, + `{"analytics":{}}`, + `{"analytics":{"defaultIntervalSeconds":300}}`, + `{"analytics":{"pauseWhenIdle":false,"recomputeIntervalSeconds":{"rf":60}}}`, + } { + var c Config + if err := json.Unmarshal([]byte(raw), &c); err != nil { + t.Fatal(err) + } + if c.AnalyticsPauseWhenIdle() || c.AnalyticsRecomputeIntervals().PauseWhenIdle { + t.Errorf("%s: pauseWhenIdle must default off", raw) + } + } +} + +// main.go passes cfg.AnalyticsRecomputeIntervals() straight to +// StartAnalyticsRecomputers, so the flag must survive the early return +// for a config without per-endpoint intervals. +func TestPauseWhenIdle_ConfigTrueReachesIntervals(t *testing.T) { + for _, raw := range []string{ + `{"analytics":{"pauseWhenIdle":true}}`, + `{"analytics":{"pauseWhenIdle":true,"recomputeIntervalSeconds":{"rf":60}}}`, + } { + var c Config + if err := json.Unmarshal([]byte(raw), &c); err != nil { + t.Fatal(err) + } + if !c.AnalyticsPauseWhenIdle() { + t.Errorf("%s: AnalyticsPauseWhenIdle = false", raw) + } + if !c.AnalyticsRecomputeIntervals().PauseWhenIdle { + t.Errorf("%s: AnalyticsRecomputeIntervals().PauseWhenIdle = false", raw) + } + } +} + +// storeRecomputerFields returns every *analyticsRecomputer field on +// PacketStore by reflection, so a recomputer added later without the +// knob fails here. +func storeRecomputerFields(t *testing.T, s *PacketStore) map[string]*analyticsRecomputer { + t.Helper() + s.analyticsRecomputerMu.RLock() + defer s.analyticsRecomputerMu.RUnlock() + out := map[string]*analyticsRecomputer{} + v := reflect.ValueOf(s).Elem() + rt := reflect.TypeOf((*analyticsRecomputer)(nil)) + for i := 0; i < v.NumField(); i++ { + f := v.Field(i) + if f.Type() != rt { + continue + } + rc := (*analyticsRecomputer)(f.UnsafePointer()) + out[v.Type().Field(i).Name] = rc + } + return out +} + +func TestPauseWhenIdle_ReachesEveryRecomputer(t *testing.T) { + for _, on := range []bool{false, true} { + db := setupTestDB(t) + s := NewPacketStore(db, nil) + stop := s.StartAnalyticsRecomputers(time.Hour, AnalyticsRecomputeIntervals{PauseWhenIdle: on}) + fields := storeRecomputerFields(t, s) + if len(fields) != 9 { + t.Fatalf("expected 9 analytics recomputer fields, found %d: %v", len(fields), fields) + } + s.analyticsRecomputerMu.RLock() + listed := s.analyticsRecomputersLocked() + s.analyticsRecomputerMu.RUnlock() + if len(listed) != len(fields) { + t.Fatalf("analyticsRecomputersLocked lists %d, store has %d", len(listed), len(fields)) + } + for name, rc := range fields { + if rc == nil { + t.Fatalf("%s not started", name) + } + if rc.pauseWhenIdle != on { + t.Errorf("pause=%v: %s.pauseWhenIdle = %v", on, name, rc.pauseWhenIdle) + } + } + // roles reads the nodes-clock-skew snapshot (GetFleetClockSkew): + // that dependency must be declared so a paused clock-skew is + // refreshed before roles computes over it. + if deps := fields["recompRoles"].deps; len(deps) != 1 || deps[0] != fields["recompNodesClockSkew"] { + t.Errorf("roles deps = %v, want [nodes-clock-skew]", deps) + } + stop() + db.Close() + } +} + +// --- loop behaviour ----------------------------------------------------- + +func TestPauseWhenIdle_DefaultOffComputesEveryUnreadTick(t *testing.T) { + h := newPauseHarness(t, "off", false, countingPass).start() + for i := 0; i < 5; i++ { + h.expectTick("tick-run") + } + if got := h.r.ComputeRuns(); got != 6 { + t.Fatalf("ComputeRuns = %d, want 6 (start + 5 ticks)", got) + } + if h.r.paused.Load() { + t.Fatal("pause flag set with pauseWhenIdle off") + } +} + +func TestPauseWhenIdle_UnreadTicksSkip(t *testing.T) { + h := newPauseHarness(t, "idle", true, countingPass).start() + for i := 0; i < 5; i++ { + h.expectTick("tick-skip") + } + if got := h.r.ComputeRuns(); got != 1 { + t.Fatalf("ComputeRuns = %d, want 1 (only the start pass)", got) + } + if !h.r.paused.Load() { + t.Fatal("an unread recomputer is not marked paused") + } +} + +func TestPauseWhenIdle_ContinuousReadsKeepComputing(t *testing.T) { + h := newPauseHarness(t, "read", true, countingPass).start() + for i := 0; i < 5; i++ { + if v := h.r.Load(); v != int64(i+1) { + t.Fatalf("read %d got %v, want the latest pass %d", i, v, i+1) + } + h.expectTick("tick-run") + } + if got := h.r.ComputeRuns(); got != 6 { + t.Fatalf("ComputeRuns = %d, want 6", got) + } + if len(h.r.kick) != 0 { + t.Fatal("a recomputer that was never paused queued a wake-up") + } + h.noPendingEvent() +} + +// A read while a pass is computing may have got the previous snapshot, so +// it counts towards the next tick: the pass clears the read flag before +// computing, not after. +func TestPauseWhenIdle_ReadDuringPassCounts(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + h := newPauseHarness(t, "read-during-pass", true, func(pass int64) interface{} { + if pass == 2 { + close(started) + <-release + } + return pass + }).start() + h.r.Load() // so the next tick computes + select { + case h.tk.c <- time.Now(): + case <-time.After(pauseTestDeadline): + t.Fatal("loop did not accept a tick") + } + <-started + if v := h.r.Load(); v != int64(1) { // read while pass 2 runs + t.Fatalf("read during the pass got %v, want the previous snapshot", v) + } + close(release) + h.expect("tick-run") + h.expectTick("tick-run") // the read during pass 2 is not forgotten + h.expectTick("tick-skip") +} + +// The first read after idle returns the existing snapshot while the wake +// pass is still blocked, a burst of concurrent readers coalesces into ONE +// pass, and later reads see the refreshed snapshot. +func TestPauseWhenIdle_WakeServesSnapshotAndCoalesces(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}, 4) + h := newPauseHarness(t, "wake", true, func(pass int64) interface{} { + if pass > 1 { + started <- struct{}{} + <-release + } + return pass + }).start() + h.expectTick("tick-skip") + + // 64 concurrent readers, each must return the old snapshot without + // waiting for the blocked pass. + const readers = 64 + var wg sync.WaitGroup + var wrong atomic.Int64 + for i := 0; i < readers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if h.r.Load() != int64(1) { + wrong.Add(1) + } + }() + } + readersDone := make(chan struct{}) + go func() { wg.Wait(); close(readersDone) }() + select { + case <-readersDone: + case <-time.After(pauseTestDeadline): + t.Fatal("readers waited for the heavy computation") + } + if wrong.Load() != 0 { + t.Fatalf("%d readers did not get the existing snapshot", wrong.Load()) + } + select { + case <-started: + case <-time.After(pauseTestDeadline): + t.Fatal("a read on a paused recomputer did not wake it") + } + if c := cap(h.r.kick); c != 1 { + t.Fatalf("wake-up channel capacity = %d, want 1", c) + } + close(release) + h.expect("kick-run") + // The wake pass restarts the schedule, so the next tick comes a full + // interval after it instead of marking the new snapshot stale at once. + if got := h.tk.resets.Load(); got != 1 { + t.Fatalf("ticker resets after the wake pass = %d, want 1", got) + } + + // Barrier: RecomputeNow is served after any wake-up still queued, so + // every event before it belongs to the burst. + h.r.RecomputeNow() + runs, ignored := 1, 0 + for ev := h.next(); ev != "recompute-now"; ev = h.next() { + switch ev { + case "kick-run": + runs++ + case "kick-ignored": + ignored++ + default: + t.Fatalf("unexpected event %q during the burst", ev) + } + } + if runs != 1 || ignored > 1 { + t.Fatalf("burst of %d reads gave %d wake passes and %d ignored wake-ups, want 1 and <=1", readers, runs, ignored) + } + if got := h.r.ComputeRuns(); got != 3 { + t.Fatalf("ComputeRuns = %d, want 3 (start + 1 wake + barrier)", got) + } + if v := h.r.Load(); v != int64(3) { + t.Fatalf("read after the refresh got %v, want the refreshed snapshot", v) + } +} + +// A read landing between "pause armed" and "read checked" on the loop +// goroutine must still be served: either the tick computes, or the read +// queues a wake-up. +func TestPauseWhenIdle_NoLostWakeupInPauseWindow(t *testing.T) { + h := newPauseHarness(t, "window", true, countingPass) + var once sync.Once + h.r.testHookPauseWindow = func() { once.Do(func() { h.r.Load() }) } + h.start() + switch ev := h.tick(); ev { + case "tick-run": + case "tick-skip": + if next := h.next(); next != "kick-run" { + t.Fatalf("read in the pause window was lost: skip followed by %q", next) + } + default: + t.Fatalf("unexpected tick event %q", ev) + } + if got := h.r.ComputeRuns(); got != 2 { + t.Fatalf("ComputeRuns = %d, want 2", got) + } +} + +func TestPauseWhenIdle_FirstComputationNeverPauses(t *testing.T) { + // Start computes synchronously even with the opt-in on. + h := newPauseHarness(t, "first", true, func(pass int64) interface{} { + if pass <= 3 { + return nil // no snapshot yet + } + return pass + }).start() + if h.r.ComputeRuns() != 1 { + t.Fatalf("Start did not run the initial compute") + } + // Without a successful first snapshot unread ticks keep computing. + h.expectTick("tick-run") + h.expectTick("tick-run") + if h.r.Load() != nil { + t.Fatal("snapshot present before the first successful pass") + } + h.expectTick("tick-run") // pass 4 succeeds + h.r.cache.Load() // not a read: only Load counts + h.expectTick("tick-skip") + if v := h.r.Load(); v != int64(4) { + t.Fatalf("snapshot = %v, want pass 4", v) + } +} + +// Until a successful pass starts after the startup load, unread ticks keep +// computing; the post-load RecomputeNow runs even on a paused recomputer. +func TestPauseWhenIdle_StartupLoad(t *testing.T) { + loaded := make(chan struct{}) + h := newPauseHarness(t, "startup", true, countingPass) + h.r.pauseReadyGate = func() bool { return isClosed(loaded) } + h.start() + h.expectTick("tick-run") // load not done: no pause + h.expectTick("tick-run") + close(loaded) + h.expectTick("tick-run") // first pass started after the load: settles + h.expectTick("tick-skip") + + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { defer close(done); recomputeWhenLoaded(loaded, stop, []*analyticsRecomputer{h.r}) }() + h.expect("recompute-now") + <-done + if got := h.r.ComputeRuns(); got != 5 { + t.Fatalf("ComputeRuns = %d, want 5: the post-load pass was skipped", got) + } + h.expectTick("tick-skip") +} + +// A failed pass is a panic or a nil result; both keep the recomputer from +// pausing. +func TestPauseWhenIdle_FailedComputationRecovers(t *testing.T) { + for _, mode := range []string{"panic", "nil"} { + t.Run(mode, func(t *testing.T) { + fail := atomic.Bool{} + h := newPauseHarness(t, "fail-"+mode, true, func(pass int64) interface{} { + if !fail.Load() { + return pass + } + if mode == "panic" { + panic("compute failed") + } + return nil + }).start() + h.expectTick("tick-skip") + + // A failed wake pass keeps the old snapshot, still marked + // stale, and the next tick retries even though nothing read it. + fail.Store(true) + if v := h.r.Load(); v != int64(1) { + t.Fatalf("read got %v, want the existing snapshot", v) + } + h.expect("kick-run") + if _, stale := h.r.SnapshotInfo(); !stale { + t.Fatal("a failed refresh made the old snapshot look fresh") + } + h.expectTick("tick-run") + h.expectTick("tick-run") + if v := h.r.Load(); v != int64(1) { + t.Fatalf("read after failed passes got %v, want the existing snapshot", v) + } + fail.Store(false) + h.expectTick("tick-run") // recovers + h.expectTick("tick-skip") + if v := h.r.Load(); v != int64(5) { + t.Fatalf("snapshot = %v, want pass 5", v) + } + }) + } +} + +func TestPauseWhenIdle_Shutdown(t *testing.T) { + t.Run("while paused", func(t *testing.T) { + h := newPauseHarness(t, "stop-paused", true, countingPass).start() + h.expectTick("tick-skip") + h.r.Stop() + if !isClosed(h.r.done) { + t.Fatal("Stop returned before the loop exited") + } + for i := 0; i < 10; i++ { + if h.r.Load() != int64(1) { + t.Fatal("Load after Stop lost the snapshot") + } + } + if len(h.r.kick) > 1 { + t.Fatal("wake-up queue grew past its bound") + } + h.r.RecomputeNow() // returns, does not hang + }) + t.Run("during a wake pass", func(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + h := newPauseHarness(t, "stop-wake", true, func(pass int64) interface{} { + if pass == 2 { + close(started) + <-release + } + return pass + }).start() + h.expectTick("tick-skip") + h.r.Load() + <-started + stopped := make(chan struct{}) + go func() { h.r.Stop(); close(stopped) }() + close(release) + <-stopped + if !isClosed(h.r.done) { + t.Fatal("Stop did not join the loop") + } + if h.r.ComputeRuns() != 2 { + t.Fatalf("ComputeRuns = %d, want 2", h.r.ComputeRuns()) + } + }) + t.Run("dependency already stopped", func(t *testing.T) { + dep := newPauseHarness(t, "dep", true, countingPass).start() + r := newPauseHarness(t, "dependent", true, countingPass) + r.r.deps = []*analyticsRecomputer{dep.r} + r.start() + dep.expectTick("tick-skip") + r.expectTick("tick-skip") + dep.r.Stop() + r.r.Load() + r.expect("kick-run") + }) +} + +// A dependent that wakes refreshes its paused dependency FIRST, and the +// dependent reading the dependency's snapshot keeps the dependency awake. +func TestPauseWhenIdle_DependencyNotStarved(t *testing.T) { + var mu sync.Mutex + var order []string + note := func(n string) { mu.Lock(); order = append(order, n); mu.Unlock() } + + dep := newPauseHarness(t, "nodes-clock-skew", true, func(pass int64) interface{} { + note("dep") + return pass + }).start() + var depSeen atomic.Int64 + rh := newPauseHarness(t, "roles", true, func(pass int64) interface{} { + note("roles") + depSeen.Store(dep.r.Load().(int64)) + return pass + }) + rh.r.deps = []*analyticsRecomputer{dep.r} + rh.start() + + dep.expectTick("tick-run") // roles' start pass read it + dep.expectTick("tick-skip") + rh.expectTick("tick-skip") + mu.Lock() + order = nil + mu.Unlock() + + rh.r.Load() // a reader of roles only + dep.expect("recompute-now") + rh.expect("kick-run") + mu.Lock() + got := append([]string(nil), order...) + mu.Unlock() + if !reflect.DeepEqual(got, []string{"dep", "roles"}) { + t.Fatalf("compute order = %v, want [dep roles]", got) + } + if want := dep.r.ComputeRuns(); depSeen.Load() != want { + t.Fatalf("roles computed over dependency pass %d, want the refreshed pass %d", depSeen.Load(), want) + } + + // roles keeps being read; its computes count as reads of the + // dependency, so the dependency keeps ticking too. + for i := 0; i < 3; i++ { + rh.r.Load() + rh.expectTick("tick-run") + dep.expectTick("tick-run") + } +} + +// --- freshness ---------------------------------------------------------- + +func TestPauseWhenIdle_SnapshotInfoNeverFalselyFresh(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + h := newPauseHarness(t, "fresh", true, func(pass int64) interface{} { + if pass == 2 { + close(started) + <-release + } + return pass + }).start() + at1, stale := h.r.SnapshotInfo() + if stale || at1.IsZero() { + t.Fatalf("start snapshot: at=%v stale=%v, want a fresh dated snapshot", at1, stale) + } + h.expectTick("tick-skip") + if _, stale := h.r.SnapshotInfo(); !stale { + t.Fatal("snapshot that missed a scheduled refresh is reported fresh") + } + h.r.Load() + <-started // the wake pass is running, the snapshot is still the old one + if at, stale := h.r.SnapshotInfo(); !stale || !at.Equal(at1) { + t.Fatalf("during the wake pass: at=%v stale=%v, want the old snapshot reported stale", at, stale) + } + close(release) + h.expect("kick-run") + if at, stale := h.r.SnapshotInfo(); stale || !at.After(at1) && !at.Equal(at1) { + t.Fatalf("after the refresh: at=%v stale=%v, want a fresh snapshot", at, stale) + } +} + +func TestPauseWhenIdle_DefaultOffNeverStale(t *testing.T) { + h := newPauseHarness(t, "off-fresh", false, countingPass).start() + h.expectTick("tick-run") + if _, stale := h.r.SnapshotInfo(); stale { + t.Fatal("stale with pauseWhenIdle off") + } +} + +// startRecomputersWithManualTickers starts s's analytics recomputers with +// one manual ticker per recomputer name (names must list all of them), so +// no scheduled tick fires on its own. +func startRecomputersWithManualTickers(t *testing.T, s *PacketStore, names []string, ov AnalyticsRecomputeIntervals) (func(), map[string]*manualRecomputeTicker) { + t.Helper() + tickers := make(map[string]*manualRecomputeTicker, len(names)) + for _, n := range names { + tickers[n] = newManualRecomputeTicker() + } + prev := newRecomputeTicker + // Called later on each loop goroutine; tickers is read-only by then. + newRecomputeTicker = func(name string, _ time.Duration) recomputeTicker { + tk, ok := tickers[name] + if !ok { + panic("no manual ticker for recomputer " + name) + } + return tk + } + // Each recomputer copies the factory when it is constructed. + stop := s.StartAnalyticsRecomputers(time.Hour, ov) + newRecomputeTicker = prev + return stop, tickers +} + +// Every default-shape endpoint served from a recomputer carries the +// snapshot headers when the opt-in is on, reports a skipped refresh as +// stale and gives the snapshot's exact age; non-default shapes (region, +// area, window) carry none. With the opt-in off no response has them. +func TestPauseWhenIdle_SnapshotHeaders(t *testing.T) { + endpoints := []struct { + path, field, name string + other []string // non-default shapes, not recomputer-served + }{ + {"/api/analytics/rf", "recompRF", "rf", []string{"region=SJC", "area=fyn", "window=1h"}}, + {"/api/analytics/topology", "recompTopology", "topology", []string{"region=SJC", "area=fyn", "window=1h"}}, + {"/api/analytics/channels", "recompChannels", "channels", []string{"region=SJC", "area=fyn", "window=1h"}}, + {"/api/analytics/distance", "recompDistance", "distance", []string{"region=SJC", "area=fyn"}}, + {"/api/analytics/hash-sizes", "recompHashSizes", "hash-sizes", []string{"region=SJC", "area=fyn"}}, + {"/api/analytics/hash-collisions", "recompHashCollisions", "hash-collisions", []string{"region=SJC", "area=fyn"}}, + {"/api/analytics/roles", "recompRoles", "roles", nil}, + {"/api/observers/clock-skew", "recompObserversClockSkew", "observers-clock-skew", nil}, + {"/api/nodes/clock-skew", "recompNodesClockSkew", "nodes-clock-skew", []string{"area=fyn"}}, + } + names := make([]string, 0, len(endpoints)) + for _, e := range endpoints { + names = append(names, e.name) + } + for _, on := range []bool{false, true} { + srv, router := setupTestServer(t) + srv.cfg.Areas = map[string]AreaEntry{"fyn": {}} + stop, tickers := startRecomputersWithManualTickers(t, srv.store, names, AnalyticsRecomputeIntervals{PauseWhenIdle: on}) + srv.store.signalStartupLoadDone() + // Wait for the post-load pass of every recomputer (#116): it + // stores a new snapshot, which would clear a stale mark set below. + // It is the first Reset on each ticker, and with manual tickers + // nothing else runs a pass from here on except this test. + for _, n := range names { + select { + case <-tickers[n].resetC: + case <-time.After(pauseTestDeadline): + t.Fatalf("pause=%v %s: no post-load pass", on, n) + } + } + srv.store.distLazyMu.Lock() + srv.store.distLazyBuilding = true + srv.store.distLazyMu.Unlock() + srv.store.runDistanceIndexBuild() // synchronous lazy build (#1011) + if !srv.store.DistanceIndexBuilt() { + t.Fatal("distance index not built") + } + fields := storeRecomputerFields(t, srv.store) + if len(fields) != len(endpoints) { + t.Fatalf("%d recomputers, %d endpoints under test", len(fields), len(endpoints)) + } + // Every loop is idle (its last pass is done), so swapping the clock + // does not race; the passes below store snapshots dated clock.now(). + clock := newSnapshotClock() + for _, rc := range fields { + rc.now = clock.now + rc.RecomputeNow() + } + clock.advance(125*time.Second + 900*time.Millisecond) + get := func(path string) *httptest.ResponseRecorder { + t.Helper() + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) + if w.Code != http.StatusOK { + t.Fatalf("pause=%v %s: status %d", on, path, w.Code) + } + return w + } + for _, e := range endpoints { + for _, q := range e.other { + w := get(e.path + "?" + q) + if h := w.Header(); h.Get(analyticsSnapshotHeader) != "" || h.Get(analyticsSnapshotAgeHeader) != "" { + t.Errorf("pause=%v %s?%s: snapshot headers on a shape not served from the recomputer", on, e.path, q) + } + } + w := get(e.path) + state, age := w.Header().Get(analyticsSnapshotHeader), w.Header().Get(analyticsSnapshotAgeHeader) + if !on { + if state != "" || age != "" { + t.Errorf("%s: snapshot headers set with pauseWhenIdle off", e.path) + } + continue + } + if state != "current" || age != "125" { + t.Errorf("%s: %s = %q, %s = %q, want current, 125", e.path, analyticsSnapshotHeader, state, analyticsSnapshotAgeHeader, age) + } + fields[e.field].missedRefresh.Store(true) // as after a skipped tick + w = get(e.path) + state, age = w.Header().Get(analyticsSnapshotHeader), w.Header().Get(analyticsSnapshotAgeHeader) + if state != "stale" || age != "125" { + t.Errorf("%s after a skipped refresh: %s = %q, %s = %q, want stale, 125", e.path, analyticsSnapshotHeader, state, analyticsSnapshotAgeHeader, age) + } + } + stop() + } +} + +// --- measurement (manual) ----------------------------------------------- + +// TestPauseWhenIdle_Measure prints computation counts, allocations, CPU and +// read latency for idle vs visible traffic with the opt-in off and on. It +// sleeps by design (real tickers), so it only runs when asked: +// +// PAUSE_IDLE_MEASURE=1 go test -run TestPauseWhenIdle_Measure -v ./cmd/server +func TestPauseWhenIdle_Measure(t *testing.T) { + if testing.Short() || os.Getenv("PAUSE_IDLE_MEASURE") != "1" { + t.Skip("set PAUSE_IDLE_MEASURE=1 to measure") + } + measurePauseWhenIdle(t) +} diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go index d4acc3a04..9e8fd7e84 100644 --- a/cmd/server/analytics_recomputer.go +++ b/cmd/server/analytics_recomputer.go @@ -37,7 +37,7 @@ type analyticsRecomputer struct { interval time.Duration compute func() interface{} - cache atomic.Value // holds interface{} — the latest snapshot + cache atomic.Value // holds *recomputerSnapshot — the latest snapshot stop chan struct{} done chan struct{} // recomputeReq carries RecomputeNow requests to the loop goroutine, @@ -47,6 +47,37 @@ type analyticsRecomputer struct { startOnce sync.Once stopOnce sync.Once + // Issue #492 — opt-in pauseWhenIdle (analytics.pauseWhenIdle). Set + // before Start. A tick with no Load since the previous pass is skipped + // once the recomputer has settled; the next Load serves the existing + // snapshot and queues one wake-up on kick. See pauseTick / markRead. + pauseWhenIdle bool + // deps are recomputers whose snapshot this one reads while computing + // (roles reads nodes-clock-skew). A paused dependency is refreshed + // before this one computes, so it never computes over a snapshot + // that stopped refreshing. + deps []*analyticsRecomputer + // pauseReadyGate: pausing is allowed only after a successful pass that + // STARTED with this gate open (the startup load has terminated). nil + // means no gate. + pauseReadyGate func() bool + settled atomic.Bool // a successful pass started with pauseReadyGate open + lastPassOK atomic.Bool // the last pass returned a snapshot without panicking + readSinceCompute atomic.Bool + paused atomic.Bool + // missedRefresh: the current snapshot outlived a scheduled refresh + // that was skipped. Cleared only when a new snapshot is stored, so a + // stale snapshot is never reported fresh while a wake pass runs or + // after one fails. + missedRefresh atomic.Bool + kick chan struct{} // buffered(1): coalesced wake-ups + + newTicker func(name string, d time.Duration) recomputeTicker + now func() time.Time + // Test hooks; nil in production. + onLoopEvent func(event string) + testHookPauseWindow func() + // Stats (atomic). computeRuns atomic.Int64 lastComputeNs atomic.Int64 // duration of last compute in nanoseconds @@ -75,6 +106,36 @@ func newAnalyticsRecomputer(name string, interval time.Duration, compute func() done: make(chan struct{}), recomputeReq: make(chan chan struct{}), + kick: make(chan struct{}, 1), + newTicker: newRecomputeTicker, + now: time.Now, + } +} + +// recomputerSnapshot is one stored compute result and when it was stored. +type recomputerSnapshot struct { + v interface{} + at time.Time +} + +// recomputeTicker is the loop's schedule; tests inject a manual one. +type recomputeTicker interface { + C() <-chan time.Time + Reset(d time.Duration) + Stop() +} + +type timeRecomputeTicker struct{ *time.Ticker } + +func (t timeRecomputeTicker) C() <-chan time.Time { return t.Ticker.C } + +var newRecomputeTicker = func(_ string, d time.Duration) recomputeTicker { + return timeRecomputeTicker{time.NewTicker(d)} +} + +func (r *analyticsRecomputer) event(ev string) { + if r.onLoopEvent != nil { + r.onLoopEvent(ev) } } @@ -99,19 +160,87 @@ func (r *analyticsRecomputer) Start() { func (r *analyticsRecomputer) loop() { defer close(r.done) - t := time.NewTicker(r.interval) + t := r.newTicker(r.name, r.interval) defer t.Stop() for { select { - case <-t.C: + case <-t.C(): + if r.pauseTick() { + r.event("tick-skip") + continue + } + r.refreshPausedDeps() + r.runOnce() + r.event("tick-run") + case <-r.kick: + // #492: a paused recomputer was read. A wake-up that arrives + // after a pass already ran (the burst raced it) is dropped, + // so a burst costs at most one pass. + if !r.paused.Load() { + r.event("kick-ignored") + continue + } + r.refreshPausedDeps() r.runOnce() + t.Reset(r.interval) + r.event("kick-run") case ack := <-r.recomputeReq: // #116: on-demand pass, on this goroutine so it never runs // concurrently with a periodic one; restart the ticker so no - // periodic pass follows right behind it. + // periodic pass follows right behind it. Runs whether or not + // the recomputer is paused (startup load, distance index). r.runOnce() t.Reset(r.interval) close(ack) + r.event("recompute-now") + case <-r.stop: + return + } + } +} + +// pauseTick decides whether a scheduled pass is skipped (#492): only with +// pauseWhenIdle on, only once a successful pass started after the startup +// load, never while the last pass failed, and only when nothing read the +// snapshot since the previous pass. +// +// paused is published BEFORE readSinceCompute is checked, and markRead +// stores readSinceCompute BEFORE checking paused, so a read racing this +// tick is either seen here (the tick computes) or sees paused (it queues a +// wake-up): it is never lost. +func (r *analyticsRecomputer) pauseTick() bool { + if !r.pauseWhenIdle || !r.settled.Load() || !r.lastPassOK.Load() { + return false + } + r.paused.Store(true) + if r.testHookPauseWindow != nil { + r.testHookPauseWindow() + } + if r.readSinceCompute.Load() { + return false + } + r.missedRefresh.Store(true) + return true +} + +// refreshPausedDeps runs one pass of every paused dependency before this +// recomputer computes over its snapshot. Gives up on either side stopping. +func (r *analyticsRecomputer) refreshPausedDeps() { + for _, d := range r.deps { + if d == nil || !d.paused.Load() { + continue + } + ack := make(chan struct{}) + select { + case d.recomputeReq <- ack: + case <-d.stop: + continue + case <-r.stop: + return + } + select { + case <-ack: + case <-d.stop: case <-r.stop: return } @@ -122,6 +251,7 @@ func (r *analyticsRecomputer) runOnce() { if r.compute == nil { return } + ok := false defer func() { // Don't let a compute panic kill the background goroutine. // The previous snapshot remains valid. Even on panic, we @@ -129,18 +259,32 @@ func (r *analyticsRecomputer) runOnce() { // safety net (a perpetually panicking compute would never // reach markFirstPassDone otherwise). _ = recover() + // #492: a failed pass (panic or nil) keeps the recomputer from + // pausing, so the next tick retries. + r.lastPassOK.Store(ok) }() // #116: sample the warm-up readiness gate BEFORE computing, so a // pass that started on a partially loaded store never ends the // warm-up, even if the load finishes while it runs. ready := r.warmupReadyGateOpen_1659() + // #492: same rule for pausing: only a pass that started after the + // startup load may settle the recomputer. + pauseReady := r.pauseReadyGate == nil || r.pauseReadyGate() + // #492: reads from here on count towards the next tick. + r.readSinceCompute.Store(false) + r.paused.Store(false) t0 := time.Now() r.lastStartNs.Store(t0.UnixNano()) result := r.compute() r.lastComputeNs.Store(int64(time.Since(t0))) r.computeRuns.Add(1) if result != nil { - r.cache.Store(result) + r.cache.Store(&recomputerSnapshot{v: result, at: r.now()}) + r.missedRefresh.Store(false) + ok = true + if pauseReady { + r.settled.Store(true) + } } // Issue #1659: mark the first-pass clock so the warmup gate // in GetAnalyticsRFWithWindow / Topology / Channels handlers @@ -209,15 +353,46 @@ func recomputeWhenLoaded(loaded, stop <-chan struct{}, rcs []*analyticsRecompute len(rcs), time.Since(t0).Round(time.Millisecond), strings.Join(parts, " ")) } +// markRead counts a read for pauseWhenIdle (#492): the next tick +// computes, and a paused recomputer gets one non-blocking wake-up (a +// wake-up already queued absorbs the rest of a burst). A no-op when +// pauseWhenIdle is off. See pauseTick for the ordering. +func (r *analyticsRecomputer) markRead() { + if !r.pauseWhenIdle { + return + } + r.readSinceCompute.Store(true) + if r.paused.Load() { + select { + case r.kick <- struct{}{}: + default: + } + } +} + // Load returns the most recently computed snapshot, or nil if Start // has not been called (or the very first compute returned nil). -// Never blocks beyond a single atomic load. +// Never blocks: beyond the atomic load, a paused recomputer +// (pauseWhenIdle) gets a non-blocking wake-up and the snapshot it already +// holds is returned while the refresh runs. func (r *analyticsRecomputer) Load() interface{} { - v := r.cache.Load() - if v == nil { + r.markRead() + snap, _ := r.cache.Load().(*recomputerSnapshot) + if snap == nil { return nil } - return v + return snap.v +} + +// SnapshotInfo reports when the current snapshot was stored and whether it +// is stale: it outlived a scheduled refresh that pauseWhenIdle skipped and +// no newer snapshot has been stored yet (#492). It is not a read. Always +// fresh with pauseWhenIdle off. +func (r *analyticsRecomputer) SnapshotInfo() (at time.Time, stale bool) { + if snap, _ := r.cache.Load().(*recomputerSnapshot); snap != nil { + at = snap.at + } + return at, r.missedRefresh.Load() } // Stop signals the background goroutine to exit and waits for it. @@ -257,6 +432,10 @@ type AnalyticsRecomputeIntervals struct { Roles time.Duration ObserversClockSkew time.Duration NodesClockSkew time.Duration + + // PauseWhenIdle is applied to every recomputer (#492, + // AnalyticsConfig.PauseWhenIdle). + PauseWhenIdle bool } func pickInterval(override, def time.Duration) time.Duration { @@ -336,15 +515,11 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o "nodes-clock-skew", pickInterval(ov.NodesClockSkew, defaultInterval), func() interface{} { return s.computeFleetClockSkew() }, ) - // Start and post-load recompute order (#116): the three warm-up-gated - // recomputers first, and roles after nodes-clock-skew because - // computeAnalyticsRoles reads that snapshot (GetFleetClockSkew). - all := []*analyticsRecomputer{ - s.recompRF, s.recompTopology, s.recompChannels, - s.recompDistance, s.recompHashCollisions, s.recompHashSizes, - s.recompObserversClockSkew, s.recompNodesClockSkew, - s.recompRoles, - } + // computeAnalyticsRoles reads the nodes-clock-skew snapshot + // (GetFleetClockSkew): with pauseWhenIdle a paused clock-skew is + // refreshed before roles computes (#492). + s.recompRoles.deps = []*analyticsRecomputer{s.recompNodesClockSkew} + all := s.analyticsRecomputersLocked() s.analyticsRecomputerMu.Unlock() // Issue #1659 (PR #1688 r1, munger #5): wire the loader readiness @@ -366,6 +541,10 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o s.recompRF.setWarmupReadyGate_1659(loadedGate) s.recompTopology.setWarmupReadyGate_1659(loadedGate) s.recompChannels.setWarmupReadyGate_1659(loadedGate) + for _, rc := range all { + rc.pauseWhenIdle = ov.PauseWhenIdle + rc.pauseReadyGate = loadedGate + } for _, rc := range all { rc.Start() @@ -393,3 +572,17 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o }) } } + +// analyticsRecomputersLocked lists every analytics recomputer in start and +// post-load recompute order (#116): the three warm-up-gated recomputers +// first, and roles after nodes-clock-skew because computeAnalyticsRoles +// reads that snapshot (GetFleetClockSkew). Caller holds +// analyticsRecomputerMu. +func (s *PacketStore) analyticsRecomputersLocked() []*analyticsRecomputer { + return []*analyticsRecomputer{ + s.recompRF, s.recompTopology, s.recompChannels, + s.recompDistance, s.recompHashCollisions, s.recompHashSizes, + s.recompObserversClockSkew, s.recompNodesClockSkew, + s.recompRoles, + } +} diff --git a/cmd/server/config.go b/cmd/server/config.go index d6badca9b..a39a58cae 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -1501,6 +1501,13 @@ func (c *Config) IsObserverBlacklisted(id string) bool { type AnalyticsConfig struct { DefaultIntervalSeconds int `json:"defaultIntervalSeconds,omitempty"` RecomputeIntervalSeconds map[string]int `json:"recomputeIntervalSeconds,omitempty"` + // PauseWhenIdle (#492) skips a scheduled recompute when nothing read + // that endpoint's default snapshot since the previous recompute. The + // first read after a pause is served the existing snapshot at once + // (it can be older than the interval and is marked stale in the + // X-Analytics-Snapshot header) and wakes one refresh. Off by default: + // every endpoint recomputes on its interval whether read or not. + PauseWhenIdle bool `json:"pauseWhenIdle,omitempty"` // LoRaPreset is the assumed PHY preset used by the relay-airtime-share // metric to compute true Time-on-Air (issue #1768). Defaults to the // EU MeshCore deployment: 869.6 MHz / BW 62.5 kHz / SF 8 / CR 4/5. @@ -1527,11 +1534,17 @@ func (c *Config) AnalyticsDefaultRecomputeInterval() time.Duration { return 5 * time.Minute } +// AnalyticsPauseWhenIdle reports whether idle analytics recomputes are +// skipped (#492). False for a nil config or an absent analytics block. +func (c *Config) AnalyticsPauseWhenIdle() bool { + return c != nil && c.Analytics != nil && c.Analytics.PauseWhenIdle +} + // AnalyticsRecomputeIntervals returns the per-endpoint override map. // Returns the zero value (all defaults) if the analytics block is // absent or empty. func (c *Config) AnalyticsRecomputeIntervals() AnalyticsRecomputeIntervals { - out := AnalyticsRecomputeIntervals{} + out := AnalyticsRecomputeIntervals{PauseWhenIdle: c.AnalyticsPauseWhenIdle()} if c == nil || c.Analytics == nil || c.Analytics.RecomputeIntervalSeconds == nil { return out } diff --git a/cmd/server/main.go b/cmd/server/main.go index 6f0b69e9c..8839a17d6 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -398,7 +398,7 @@ func main() { cfg.AnalyticsRecomputeIntervals(), ) defer stopAnalyticsRecomp() - log.Printf("[analytics-recompute] background recompute enabled (default=%s)", cfg.AnalyticsDefaultRecomputeInterval()) + log.Printf("[analytics-recompute] background recompute enabled (default=%s, pauseWhenIdle=%v)", cfg.AnalyticsDefaultRecomputeInterval(), cfg.AnalyticsPauseWhenIdle()) // #1481 P0-1: background recomputer for the default-shape // /api/analytics/neighbor-graph response (5 min cadence). Reads diff --git a/cmd/server/role_analytics.go b/cmd/server/role_analytics.go index 56a07e0eb..46a4fdc4c 100644 --- a/cmd/server/role_analytics.go +++ b/cmd/server/role_analytics.go @@ -120,6 +120,7 @@ func (s *Server) handleAnalyticsRoles(w http.ResponseWriter, r *http.Request) { writeJSON(w, RoleAnalyticsResponse{Roles: []RoleStats{}}) return } + setAnalyticsSnapshotHeaders(w, s.store.recomputer(&s.store.recompRoles)) // #492 writeJSON(w, s.store.GetAnalyticsRoles()) } diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 9d349571b..9aec81470 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -2882,6 +2882,7 @@ func (s *Server) handleObserverClockSkew(w http.ResponseWriter, r *http.Request) writeJSON(w, []ObserverCalibration{}) return } + setAnalyticsSnapshotHeaders(w, s.store.recomputer(&s.store.recompObserversClockSkew)) // #492 writeJSON(w, s.store.GetObserverCalibrations()) } @@ -2891,6 +2892,9 @@ func (s *Server) handleFleetClockSkew(w http.ResponseWriter, r *http.Request) { return } area := r.URL.Query().Get("area") + if area == "" { + setAnalyticsSnapshotHeaders(w, s.store.recomputer(&s.store.recompNodesClockSkew)) // #492 + } writeJSON(w, s.store.GetFleetClockSkew(area)) } @@ -2916,6 +2920,7 @@ func (s *Server) handleAnalyticsRF(w http.ResponseWriter, r *http.Request) { writeAnalyticsWarmup503(w) return } + setAnalyticsSnapshotHeaders(w, rc) // #492 } writeJSON(w, s.store.GetAnalyticsRFWithWindow(region, area, window)) return @@ -2966,6 +2971,7 @@ func (s *Server) handleAnalyticsTopology(w http.ResponseWriter, r *http.Request) writeAnalyticsWarmup503(w) return } + setAnalyticsSnapshotHeaders(w, rc) // #492 } // The store hands out its shared cached result; the filter never // writes to it and returns a filtered copy when anything is hidden. @@ -3005,6 +3011,7 @@ func (s *Server) handleAnalyticsChannels(w http.ResponseWriter, r *http.Request) writeAnalyticsWarmup503(w) return } + setAnalyticsSnapshotHeaders(w, rc) // #492 } writeJSON(w, s.store.GetAnalyticsChannelsWithWindow(region, area, window)) return @@ -3043,6 +3050,9 @@ func (s *Server) handleAnalyticsDistance(w http.ResponseWriter, r *http.Request) _, _ = w.Write([]byte(`{"status":"building","retry_after_seconds":5,"detail":"distance index is being computed (lazy build, #1011). Retry after Retry-After seconds."}`)) return } + if region == "" && area == "" { + setAnalyticsSnapshotHeaders(w, s.store.recomputer(&s.store.recompDistance)) // #492 + } writeJSON(w, s.store.GetAnalyticsDistance(region, area)) return } @@ -3063,6 +3073,9 @@ func (s *Server) handleAnalyticsHashSizes(w http.ResponseWriter, r *http.Request if s.rejectAnalyticsFilters(w, region, area) { return } + if region == "" && area == "" { + setAnalyticsSnapshotHeaders(w, s.store.recomputer(&s.store.recompHashSizes)) // #492 + } writeJSON(w, s.store.GetAnalyticsHashSizes(region, area)) return } @@ -3083,6 +3096,9 @@ func (s *Server) handleAnalyticsHashCollisions(w http.ResponseWriter, r *http.Re if s.rejectAnalyticsFilters(w, region, area) { return } + if region == "" && area == "" { + setAnalyticsSnapshotHeaders(w, s.store.recomputer(&s.store.recompHashCollisions)) // #492 + } writeJSON(w, s.store.GetAnalyticsHashCollisions(region, area)) return } diff --git a/config.example.json b/config.example.json index 1c0b8ec5b..e90f1878c 100644 --- a/config.example.json +++ b/config.example.json @@ -439,6 +439,8 @@ "analytics": { "defaultIntervalSeconds": 300, + "pauseWhenIdle": false, + "_comment_pauseWhenIdle": "Issue #492. Opt-in, default false (every analytics snapshot is recomputed on its interval whether read or not). When true, a scheduled recompute is skipped if nothing read that endpoint's default snapshot since the previous recompute, so an instance whose analytics pages are rarely open stops recomputing and allocating for them. Freshness trade-off: the first read after a pause is served the existing snapshot at once — it can be much older than the interval — and wakes one refresh; reads after that see fresh data. While pauseWhenIdle is on, responses served from a snapshot carry X-Analytics-Snapshot (current|stale) and X-Analytics-Snapshot-Age-Seconds, so a stale snapshot is never presented as freshly computed. Startup load, warm-up gates and failed recomputes are unaffected: a recomputer only pauses after a successful pass that started once the startup load finished.", "recomputeIntervalSeconds": { "topology": 300, "rf": 300,