diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go new file mode 100644 index 00000000..9917930d --- /dev/null +++ b/cmd/server/analytics_recomputer.go @@ -0,0 +1,237 @@ +// Package main: analytics recomputer (issue #1240). +// +// Steady-state background recompute loop for expensive analytics +// endpoints. Reads always hit an atomic-pointer cache; compute runs +// on a fixed ticker in a goroutine. This eliminates the on-request +// compute-then-cache pattern where the first reader after expiry pays +// the full compute cost and blocks under writer contention. +// +// See issue #1240 and AGENTS.md "Performance is a feature". +package main + +import ( + "sync" + "sync/atomic" + "time" +) + +// analyticsRecomputer holds the latest snapshot of an analytics result +// in an atomic.Value, refreshed periodically by a background goroutine. +// +// Lifecycle: +// 1. Construct via newAnalyticsRecomputer(...) +// 2. Call Start() — runs initial compute synchronously, then launches +// the recompute goroutine. Initial compute is synchronous so the +// first Load() after Start returns never sees a nil cache. +// 3. Call Load() any number of times concurrently — never blocks +// beyond an atomic-pointer load. +// 4. Call Stop() to terminate the background goroutine cleanly. +// +// Compute func is called WITHOUT any lock held by this struct, so it +// may freely take any application-level locks it needs. +type analyticsRecomputer struct { + name string + interval time.Duration + compute func() interface{} + + cache atomic.Value // holds interface{} — the latest snapshot + stop chan struct{} + done chan struct{} + + startOnce sync.Once + stopOnce sync.Once + + // Stats (atomic). + computeRuns atomic.Int64 + lastComputeNs atomic.Int64 // duration of last compute in nanoseconds +} + +// newAnalyticsRecomputer constructs an unstarted recomputer. +// interval must be > 0; compute must be non-nil. +func newAnalyticsRecomputer(name string, interval time.Duration, compute func() interface{}) *analyticsRecomputer { + if interval <= 0 { + interval = 5 * time.Minute + } + return &analyticsRecomputer{ + name: name, + interval: interval, + compute: compute, + stop: make(chan struct{}), + done: make(chan struct{}), + } +} + +// Start runs the initial compute synchronously (so the first Load +// after Start returns a populated snapshot, never nil), then launches +// a background goroutine to periodically recompute. +// +// Calling Start multiple times is a no-op after the first call. +func (r *analyticsRecomputer) Start() { + r.startOnce.Do(func() { + // Initial synchronous compute — first read must NOT see empty + // or uninitialized data (acceptance criterion #1240). + r.runOnce() + go r.loop() + }) +} + +func (r *analyticsRecomputer) loop() { + defer close(r.done) + t := time.NewTicker(r.interval) + defer t.Stop() + for { + select { + case <-t.C: + r.runOnce() + case <-r.stop: + return + } + } +} + +func (r *analyticsRecomputer) runOnce() { + if r.compute == nil { + return + } + defer func() { + // Don't let a compute panic kill the background goroutine. + // The previous snapshot remains valid. + _ = recover() + }() + t0 := time.Now() + result := r.compute() + r.lastComputeNs.Store(int64(time.Since(t0))) + r.computeRuns.Add(1) + if result != nil { + r.cache.Store(result) + } +} + +// 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. +func (r *analyticsRecomputer) Load() interface{} { + v := r.cache.Load() + if v == nil { + return nil + } + return v +} + +// Stop signals the background goroutine to exit and waits for it. +// Safe to call multiple times. Safe to call before Start (no-op). +func (r *analyticsRecomputer) Stop() { + r.stopOnce.Do(func() { + close(r.stop) + }) + // Only wait if the goroutine was actually started. + select { + case <-r.done: + case <-time.After(5 * time.Second): + // Defensive timeout: shouldn't happen in practice. + } +} + +// LastComputeDuration returns the duration of the most recent compute. +func (r *analyticsRecomputer) LastComputeDuration() time.Duration { + return time.Duration(r.lastComputeNs.Load()) +} + +// ComputeRuns returns the total number of compute invocations. +func (r *analyticsRecomputer) ComputeRuns() int64 { + return r.computeRuns.Load() +} + +// AnalyticsRecomputeIntervals lets callers (main.go) override the +// per-endpoint recompute interval from config.json. Zero values fall +// back to the defaultInterval passed to StartAnalyticsRecomputers. +type AnalyticsRecomputeIntervals struct { + Topology time.Duration + RF time.Duration + Distance time.Duration + Channels time.Duration + HashCollisions time.Duration + HashSizes time.Duration +} + +func pickInterval(override, def time.Duration) time.Duration { + if override > 0 { + return override + } + return def +} + +// StartAnalyticsRecomputers wires each analytics endpoint to a +// background recompute goroutine. Each runs an initial compute +// synchronously (so the first read after startup is a cache hit, never +// cold) and then refreshes on a ticker. +// +// All recomputers serve the DEFAULT query shape only: region="" and +// zero-window (no ?since= / ?until= params). Region-keyed or windowed +// queries continue to use the legacy on-request compute + TTL cache — +// the recomputer count would explode if we maintained one per +// (endpoint × region × window) combination, and region filtering is +// fast read-time work anyway. +// +// Returns a stop closure that signals all goroutines and blocks until +// they exit. Safe to call once per PacketStore. Idempotent if called +// multiple times (subsequent calls return the first stop closure). +func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, overrides ...AnalyticsRecomputeIntervals) func() { + if defaultInterval <= 0 { + defaultInterval = 5 * time.Minute + } + var ov AnalyticsRecomputeIntervals + if len(overrides) > 0 { + ov = overrides[0] + } + + s.analyticsRecomputerMu.Lock() + if s.recompTopology != nil { + // Already started; return a no-op so the caller's defer is harmless. + s.analyticsRecomputerMu.Unlock() + return func() {} + } + + // Each recomputer wraps the underlying compute* function with the + // default arguments. We use computeAnalytics* (not GetAnalytics*) to + // bypass the legacy TTL cache layer — the recomputer IS the cache. + s.recompTopology = newAnalyticsRecomputer( + "topology", pickInterval(ov.Topology, defaultInterval), + func() interface{} { return s.computeAnalyticsTopology("", TimeWindow{}) }, + ) + s.recompRF = newAnalyticsRecomputer( + "rf", pickInterval(ov.RF, defaultInterval), + func() interface{} { return s.computeAnalyticsRF("", TimeWindow{}) }, + ) + s.recompDistance = newAnalyticsRecomputer( + "distance", pickInterval(ov.Distance, defaultInterval), + func() interface{} { return s.computeAnalyticsDistance("") }, + ) + s.recompChannels = newAnalyticsRecomputer( + "channels", pickInterval(ov.Channels, defaultInterval), + func() interface{} { return s.computeAnalyticsChannels("", TimeWindow{}) }, + ) + s.recompHashCollisions = newAnalyticsRecomputer( + "hash-collisions", pickInterval(ov.HashCollisions, defaultInterval), + func() interface{} { return s.computeHashCollisions("") }, + ) + s.recompHashSizes = newAnalyticsRecomputer( + "hash-sizes", pickInterval(ov.HashSizes, defaultInterval), + func() interface{} { return s.computeAnalyticsHashSizesWithCapability("") }, + ) + all := []*analyticsRecomputer{ + s.recompTopology, s.recompRF, s.recompDistance, + s.recompChannels, s.recompHashCollisions, s.recompHashSizes, + } + s.analyticsRecomputerMu.Unlock() + + for _, rc := range all { + rc.Start() + } + + return func() { + for _, rc := range all { + rc.Stop() + } + } +} diff --git a/cmd/server/analytics_recomputer_test.go b/cmd/server/analytics_recomputer_test.go new file mode 100644 index 00000000..0b2c1f91 --- /dev/null +++ b/cmd/server/analytics_recomputer_test.go @@ -0,0 +1,174 @@ +package main + +import ( + "runtime" + "sort" + "sync" + "sync/atomic" + "testing" + "time" +) + +func numGoroutinesForTest() int { return runtime.NumGoroutine() } + +// TestAnalyticsRecomputerSteadyStateLatency asserts that issue #1240's +// steady-state background recompute is in place: reads of the common +// analytics endpoints (region="") return from cache in <50ms p99 even +// under simulated ingest load. +// +// On master (pre-fix), GetAnalyticsTopology holds s.mu.RLock for the +// entire compute. Concurrent ingest writers (s.mu.Lock) starve readers +// or vice versa, producing per-read latencies in the hundreds of +// milliseconds. The cache TTL doesn't help: after every expiry one +// reader still pays the full compute cost. +// +// Post-fix, GetAnalyticsTopology with region="" and zero window must +// Load() from the background-refreshed atomic snapshot — never blocking +// under writer contention. +func TestAnalyticsRecomputerSteadyStateLatency(t *testing.T) { + if testing.Short() { + t.Skip("skipping latency timing test in -short mode") + } + + db := setupTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + + // Populate with enough records to make on-request compute non-trivial. + const N = 20000 + hops := make([]distHopRecord, N) + for i := 0; i < N; i++ { + hops[i] = distHopRecord{ + FromName: "A", FromPk: "aa", + ToName: "B", ToPk: "bb", + Dist: float64(i%500) + 0.5, + Type: []string{"R↔R", "C↔R", "C↔C"}[i%3], + Hash: "h", + Timestamp: "2024-01-01T00:00:00Z", + HourBucket: "2024-01-01-00", + } + } + store.mu.Lock() + store.distHops = hops + store.mu.Unlock() + + // Start the recomputer infrastructure. On master this method + // doesn't exist, so this test won't compile until the GREEN commit + // lands; the RED commit lands the test + a stub. Stub returns + // without wiring background recompute, so the test still fails on + // the latency assertion below. + stop := store.StartAnalyticsRecomputers(10 * time.Millisecond) + defer stop() + + // Give the initial compute a moment to populate. + time.Sleep(50 * time.Millisecond) + + // Simulated writer: contend for s.mu.Lock. This is what makes the + // non-recomputer path miss the latency target — the old + // GetAnalyticsTopology grabs s.mu.RLock for the entire compute and + // blocks behind every writer cycle. + var stopWriters atomic.Bool + var writerWg sync.WaitGroup + const Writers = 4 + writerWg.Add(Writers) + for w := 0; w < Writers; w++ { + go func() { + defer writerWg.Done() + for !stopWriters.Load() { + store.mu.Lock() + // Trivial mutation: extend distHops by one and shrink back. + store.distHops = append(store.distHops, distHopRecord{ + Dist: 1, Hash: "x", Timestamp: "2024-01-01T00:00:00Z", + }) + store.distHops = store.distHops[:len(store.distHops)-1] + store.mu.Unlock() + // Brief pause to keep the lock-cycle rate realistic. + time.Sleep(100 * time.Microsecond) + } + }() + } + + // 100 concurrent reads. + const Readers = 100 + latencies := make([]time.Duration, Readers) + var rwg sync.WaitGroup + rwg.Add(Readers) + for i := 0; i < Readers; i++ { + i := i + go func() { + defer rwg.Done() + t0 := time.Now() + r := store.GetAnalyticsDistance("") + latencies[i] = time.Since(t0) + if r == nil { + t.Errorf("reader %d got nil result", i) + } + }() + } + rwg.Wait() + stopWriters.Store(true) + writerWg.Wait() + + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + p50 := latencies[Readers/2] + p99 := latencies[(Readers*99)/100] + + t.Logf("analytics distance read latency: p50=%v p99=%v max=%v", + p50, p99, latencies[Readers-1]) + + // p99 budget: 50ms. Atomic-pointer load + JSON-shape map return + // should be sub-millisecond; 50ms leaves margin for goroutine + // scheduling jitter under concurrent test runs. + const budget = 50 * time.Millisecond + if p99 > budget { + t.Fatalf("p99 read latency %v exceeds %v budget (issue #1240 not in effect)", p99, budget) + } +} + +// TestAnalyticsRecomputerShutdownNoLeak asserts the background +// goroutines started by StartAnalyticsRecomputers exit cleanly when +// the returned stop function is called — no leak across server +// shutdown (issue #1240 acceptance criterion). +func TestAnalyticsRecomputerShutdownNoLeak(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + + // Use a tight tick so we know recompute is actually running (not + // just blocked on the ticker). + stop := store.StartAnalyticsRecomputers(20 * time.Millisecond) + + // Snapshot active goroutines a beat after start. + time.Sleep(80 * time.Millisecond) + startGoroutines := runtimeNumGoroutine() + + stop() + + // After stop returns, give the scheduler a beat to reap exits. + deadline := time.Now().Add(2 * time.Second) + var endGoroutines int + for time.Now().Before(deadline) { + endGoroutines = runtimeNumGoroutine() + if endGoroutines <= startGoroutines-5 { // we started 6 recomputers + break + } + time.Sleep(20 * time.Millisecond) + } + + // We expect ~6 fewer goroutines than the snapshot taken DURING + // recompute (one per registered recomputer). Allow some slack + // since test runners can have flaky goroutine counts. + if endGoroutines >= startGoroutines { + t.Fatalf("goroutine leak after stop: %d → %d (expected fewer)", + startGoroutines, endGoroutines) + } + t.Logf("goroutines: during=%d after=%d (Δ=%d)", + startGoroutines, endGoroutines, startGoroutines-endGoroutines) +} + +// runtimeNumGoroutine is wrapped to keep the imports section of the +// production file minimal. +func runtimeNumGoroutine() int { + // imported below + return numGoroutinesForTest() +} diff --git a/cmd/server/config.go b/cmd/server/config.go index 10a78007..e1cc8de3 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -7,6 +7,7 @@ import ( "path/filepath" "strings" "sync" + "time" "github.com/meshcore-analyzer/dbconfig" "github.com/meshcore-analyzer/geofilter" @@ -90,6 +91,9 @@ type Config struct { ResolvedPath *ResolvedPathConfig `json:"resolvedPath,omitempty"` NeighborGraph *NeighborGraphConfig `json:"neighborGraph,omitempty"` + // Analytics steady-state background recompute (issue #1240). + Analytics *AnalyticsConfig `json:"analytics,omitempty"` + // BatteryThresholds: voltage cutoffs for low/critical alerts (#663). BatteryThresholds *BatteryThresholdsConfig `json:"batteryThresholds,omitempty"` } @@ -468,3 +472,52 @@ func (c *Config) IsObserverBlacklisted(id string) bool { } return c.obsBlacklistSet()[strings.ToLower(strings.TrimSpace(id))] } + +// AnalyticsConfig controls steady-state background recompute of +// analytics endpoints (issue #1240). +// +// DefaultIntervalSeconds applies to every endpoint that does not have +// an explicit per-endpoint override in RecomputeIntervalSeconds. The +// project default is 300 (5 minutes): the operator's guiding principle +// is "serving slightly stale data quickly is better than real-time +// data slowly." Lower values give fresher data at higher CPU cost. +// +// RecomputeIntervalSeconds keys (all optional): +// topology, rf, distance, channels, hashCollisions, hashSizes +type AnalyticsConfig struct { + DefaultIntervalSeconds int `json:"defaultIntervalSeconds,omitempty"` + RecomputeIntervalSeconds map[string]int `json:"recomputeIntervalSeconds,omitempty"` +} + +// AnalyticsDefaultRecomputeInterval returns the configured default +// recompute interval, or 5 minutes if unset/invalid. +func (c *Config) AnalyticsDefaultRecomputeInterval() time.Duration { + if c != nil && c.Analytics != nil && c.Analytics.DefaultIntervalSeconds > 0 { + return time.Duration(c.Analytics.DefaultIntervalSeconds) * time.Second + } + return 5 * time.Minute +} + +// 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{} + if c == nil || c.Analytics == nil || c.Analytics.RecomputeIntervalSeconds == nil { + return out + } + get := func(key string) time.Duration { + v, ok := c.Analytics.RecomputeIntervalSeconds[key] + if !ok || v <= 0 { + return 0 + } + return time.Duration(v) * time.Second + } + out.Topology = get("topology") + out.RF = get("rf") + out.Distance = get("distance") + out.Channels = get("channels") + out.HashCollisions = get("hashCollisions") + out.HashSizes = get("hashSizes") + return out +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 82dd2e99..cd65a6eb 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -350,6 +350,17 @@ func main() { stopEviction := store.StartEvictionTicker() defer stopEviction() + // Steady-state analytics recomputers (issue #1240). Replaces the + // on-request compute-then-cache pattern for the default (region="", + // zero-window) analytics queries with a background refresh loop so + // reads always hit cache in <1ms. + stopAnalyticsRecomp := store.StartAnalyticsRecomputers( + cfg.AnalyticsDefaultRecomputeInterval(), + cfg.AnalyticsRecomputeIntervals(), + ) + defer stopAnalyticsRecomp() + log.Printf("[analytics-recompute] background recompute enabled (default=%s)", cfg.AnalyticsDefaultRecomputeInterval()) + // Auto-prune old packets if retention.packetDays is configured vacuumPages := cfg.IncrementalVacuumPages() var stopPrune func() @@ -529,6 +540,13 @@ func main() { stopEdgePrune() } + // 1c. Stop steady-state analytics recomputers (issue #1240). + // Must happen before dbClose so any in-flight compute that + // reaches into SQLite has finished. + if stopAnalyticsRecomp != nil { + stopAnalyticsRecomp() + } + // 2. Gracefully drain HTTP connections (up to 15s) ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() diff --git a/cmd/server/store.go b/cmd/server/store.go index 773aaebc..6f41315c 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -150,6 +150,20 @@ type PacketStore struct { subpathCache map[string]*cachedResult // params → cached subpaths result rfCacheTTL time.Duration collisionCacheTTL time.Duration + // Steady-state analytics recomputers (issue #1240). Each holds the + // latest snapshot for the default region="" / zero-window query of + // an analytics endpoint in an atomic.Value, refreshed by a + // background goroutine on a fixed interval. When set, the matching + // GetAnalytics* function serves from Load() instead of running the + // on-request compute path. Region/window variants still go through + // the legacy TTL cache (compute-on-miss). + analyticsRecomputerMu sync.RWMutex + recompTopology *analyticsRecomputer + recompRF *analyticsRecomputer + recompDistance *analyticsRecomputer + recompChannels *analyticsRecomputer + recompHashCollisions *analyticsRecomputer + recompHashSizes *analyticsRecomputer cacheHits int64 cacheMisses int64 // Rate-limited invalidation (fixes #533: caches cleared faster than hit) @@ -4649,6 +4663,21 @@ func (s *PacketStore) GetAnalyticsChannels(region string) map[string]interface{} // GetAnalyticsChannelsWithWindow returns channel analytics for the given region, // optionally bounded to a time window (issue #842). Zero TimeWindow = all data. func (s *PacketStore) GetAnalyticsChannelsWithWindow(region string, window TimeWindow) map[string]interface{} { + if region == "" && window.IsZero() { + s.analyticsRecomputerMu.RLock() + rc := s.recompChannels + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if m, ok := v.(map[string]interface{}); ok { + s.cacheMu.Lock() + s.cacheHits++ + s.cacheMu.Unlock() + return m + } + } + } + } cacheKey := region if !window.IsZero() { cacheKey = region + "|" + window.CacheKey() @@ -4896,6 +4925,21 @@ func (s *PacketStore) GetAnalyticsRF(region string) map[string]interface{} { // GetAnalyticsRFWithWindow returns RF analytics bounded by an optional // time window (issue #842). Zero TimeWindow = all data (backwards compatible). func (s *PacketStore) GetAnalyticsRFWithWindow(region string, window TimeWindow) map[string]interface{} { + if region == "" && window.IsZero() { + s.analyticsRecomputerMu.RLock() + rc := s.recompRF + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if m, ok := v.(map[string]interface{}); ok { + s.cacheMu.Lock() + s.cacheHits++ + s.cacheMu.Unlock() + return m + } + } + } + } cacheKey := region if !window.IsZero() { cacheKey = region + "|" + window.CacheKey() @@ -5834,7 +5878,24 @@ func (s *PacketStore) GetAnalyticsTopology(region string) map[string]interface{} } // GetAnalyticsTopologyWithWindow — see issue #842. +// For default (region="", zero window), prefer the steady-state +// recomputer snapshot if registered (issue #1240). func (s *PacketStore) GetAnalyticsTopologyWithWindow(region string, window TimeWindow) map[string]interface{} { + if region == "" && window.IsZero() { + s.analyticsRecomputerMu.RLock() + rc := s.recompTopology + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if m, ok := v.(map[string]interface{}); ok { + s.cacheMu.Lock() + s.cacheHits++ + s.cacheMu.Unlock() + return m + } + } + } + } cacheKey := region if !window.IsZero() { cacheKey = region + "|" + window.CacheKey() @@ -6342,7 +6403,26 @@ func haversineKm(lat1, lon1, lat2, lon2 float64) float64 { return R * 2 * math.Atan2(math.Sqrt(a), math.Sqrt(1-a)) } +// GetAnalyticsDistance returns the distance analytics map. For the +// default (region="") query, prefer the steady-state recomputer +// snapshot if one is registered (issue #1240). Region-keyed variants +// continue to use the legacy TTL cache + on-request compute. func (s *PacketStore) GetAnalyticsDistance(region string) map[string]interface{} { + if region == "" { + s.analyticsRecomputerMu.RLock() + rc := s.recompDistance + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if m, ok := v.(map[string]interface{}); ok { + s.cacheMu.Lock() + s.cacheHits++ + s.cacheMu.Unlock() + return m + } + } + } + } s.cacheMu.Lock() if cached, ok := s.distCache[region]; ok && time.Now().Before(cached.expiresAt) { s.cacheHits++ @@ -6618,6 +6698,21 @@ func (s *PacketStore) computeAnalyticsDistance(region string) map[string]interfa // --- Hash Sizes Analytics --- func (s *PacketStore) GetAnalyticsHashSizes(region string) map[string]interface{} { + if region == "" { + s.analyticsRecomputerMu.RLock() + rc := s.recompHashSizes + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if m, ok := v.(map[string]interface{}); ok { + s.cacheMu.Lock() + s.cacheHits++ + s.cacheMu.Unlock() + return m + } + } + } + } s.cacheMu.Lock() if cached, ok := s.hashCache[region]; ok && time.Now().Before(cached.expiresAt) { s.cacheHits++ @@ -6627,16 +6722,21 @@ func (s *PacketStore) GetAnalyticsHashSizes(region string) map[string]interface{ s.cacheMisses++ s.cacheMu.Unlock() - result := s.computeAnalyticsHashSizes(region) + result := s.computeAnalyticsHashSizesWithCapability(region) - // Multi-byte capability is a NODE property (derived from each node's own - // adverts), not a function of the observing region. The region filter - // should only control which nodes appear in the analytics list, not the - // evidence used to classify their capability. Always compute capability - // against the GLOBAL advert dataset so a region-filtered view doesn't - // downgrade every adopter to "unknown" just because the confirming - // advert was heard by an out-of-region observer (#bug: meshat.se/JKG - // showed 14 unknown vs 0 unknown unfiltered). + s.cacheMu.Lock() + s.hashCache[region] = &cachedResult{data: result, expiresAt: time.Now().Add(s.rfCacheTTL)} + s.cacheMu.Unlock() + + return result +} + +// computeAnalyticsHashSizesWithCapability runs computeAnalyticsHashSizes +// then layers in the multiByteCapability augmentation. Extracted so the +// steady-state recomputer (issue #1240) produces the same shape as the +// cached GetAnalyticsHashSizes call. +func (s *PacketStore) computeAnalyticsHashSizesWithCapability(region string) map[string]interface{} { + result := s.computeAnalyticsHashSizes(region) globalAdopterHS := make(map[string]int) if region == "" { if mbNodes, ok := result["multiByteNodes"].([]map[string]interface{}); ok { @@ -6649,9 +6749,6 @@ func (s *PacketStore) GetAnalyticsHashSizes(region string) map[string]interface{ } } } else { - // Pull the global multiByteNodes set without the region filter. - // Use a separate compute call (not the cached path) to avoid - // recursive locking on hashCache and to keep this side-effect free. globalRes := s.computeAnalyticsHashSizes("") if mbNodes, ok := globalRes["multiByteNodes"].([]map[string]interface{}); ok { for _, n := range mbNodes { @@ -6664,11 +6761,6 @@ func (s *PacketStore) GetAnalyticsHashSizes(region string) map[string]interface{ } } result["multiByteCapability"] = s.computeMultiByteCapability(globalAdopterHS) - - s.cacheMu.Lock() - s.hashCache[region] = &cachedResult{data: result, expiresAt: time.Now().Add(s.rfCacheTTL)} - s.cacheMu.Unlock() - return result } @@ -6959,6 +7051,21 @@ type hashSizeNodeInfo struct { // GetAnalyticsHashCollisions returns pre-computed hash collision analysis. // This moves the O(n²) distance computation from the frontend to the server. func (s *PacketStore) GetAnalyticsHashCollisions(region string) map[string]interface{} { + if region == "" { + s.analyticsRecomputerMu.RLock() + rc := s.recompHashCollisions + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if m, ok := v.(map[string]interface{}); ok { + s.cacheMu.Lock() + s.cacheHits++ + s.cacheMu.Unlock() + return m + } + } + } + } s.cacheMu.Lock() if cached, ok := s.collisionCache[region]; ok && time.Now().Before(cached.expiresAt) { s.cacheHits++ diff --git a/config.example.json b/config.example.json index 64fac6a8..f3b43c47 100644 --- a/config.example.json +++ b/config.example.json @@ -248,5 +248,18 @@ "_comment_hashChannels": "Channel names whose keys are derived via SHA256. Key = SHA256(name)[:16]. Listed here so the ingestor can auto-derive keys.", "_comment_defaultRegion": "IATA code shown by default in region filters.", "_comment_mapDefaults": "Initial map center [lat, lon] and zoom level.", - "_comment_regions": "IATA code → display name mapping for the region filter UI. Each key is a 3-letter IATA code that an observer is tagged with (resolved priority: MQTT payload `region` field > topic-derived region > mqttSources.region). Observers without an IATA tag will not appear under any region filter — only under 'All Regions'. The region filter dropdown shows one entry per code listed here PLUS any extra IATA codes the server discovers from observers at runtime (so you can omit codes here and they will still be selectable, just labelled with the bare IATA code instead of a friendly name). Selecting 'All Regions' (or no region) returns results from every observer including those with no IATA tag; selecting one or more codes restricts results to packets observed by observers tagged with those codes. The reserved value 'All' (case-insensitive) is treated as 'no filter' on the server, so the URL ?region=All behaves identically to omitting the param. Issue #770." + "_comment_regions": "IATA code → display name mapping for the region filter UI. Each key is a 3-letter IATA code that an observer is tagged with (resolved priority: MQTT payload `region` field > topic-derived region > mqttSources.region). Observers without an IATA tag will not appear under any region filter — only under 'All Regions'. The region filter dropdown shows one entry per code listed here PLUS any extra IATA codes the server discovers from observers at runtime (so you can omit codes here and they will still be selectable, just labelled with the bare IATA code instead of a friendly name). Selecting 'All Regions' (or no region) returns results from every observer including those with no IATA tag; selecting one or more codes restricts results to packets observed by observers tagged with those codes. The reserved value 'All' (case-insensitive) is treated as 'no filter' on the server, so the URL ?region=All behaves identically to omitting the param. Issue #770.", + + "analytics": { + "defaultIntervalSeconds": 300, + "recomputeIntervalSeconds": { + "topology": 300, + "rf": 300, + "distance": 300, + "channels": 300, + "hashCollisions": 300, + "hashSizes": 300 + } + }, + "_comment_analytics": "Issue #1240. Each analytics endpoint (topology, rf, distance, channels, hashCollisions, hashSizes) is recomputed in the background on the configured interval and served from an atomic-pointer cache. Reads never block on compute. Default 300s (5 min) per endpoint reflects the operator principle: serving slightly stale data quickly beats real-time data slowly. Lower values = fresher data at higher CPU cost. Only the default query (no region/window) is precomputed; region- and window-filtered requests fall back to the legacy on-request compute + 60s TTL cache." }