diff --git a/cmd/server/config.go b/cmd/server/config.go index 21dfd41d1..d6badca9b 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -493,10 +493,13 @@ type NeighborGraphConfig struct { MaxEdgeKm float64 `json:"maxEdgeKm"` // geo-implausibility threshold (km); 0 = default 500; negative disables (#1228) // CacheRecomputeIntervalSeconds: cadence for the background - // recomputer that rebuilds both the neighbor graph and its cached - // responses. Requests keep the graph until the next build finishes. - // 0/missing = default 300 (5 min). Lower = fresher data, with a full - // graph build on every tick; higher = an older graph between ticks. + // recomputer that refreshes the cached default-shape neighbor-graph + // responses (#1481 P0-1). The graph itself is rebuilt at most once per + // neighborGraphTTL (5 min), so a shorter interval refreshes the cached + // responses more often without adding graph-build CPU, while a longer one + // makes the graph as old as the interval. Requests keep the published + // graph until the next build finishes. 0/missing = default 300 (5 min). + // #1483, #545. CacheRecomputeIntervalSeconds int `json:"cacheRecomputeIntervalSeconds,omitempty"` } diff --git a/cmd/server/neighbor_api.go b/cmd/server/neighbor_api.go index 7e05d8a8f..8a39d403f 100644 --- a/cmd/server/neighbor_api.go +++ b/cmd/server/neighbor_api.go @@ -83,41 +83,30 @@ type GraphStats struct { // getNeighborGraph serves the published graph until the background recomputer // replaces it. Only an initial request with no graph waits for a build. func (s *Server) getNeighborGraph() *NeighborGraph { - s.neighborMu.Lock() - g := s.neighborGraph - s.neighborMu.Unlock() - if g != nil { + if g := s.neighborGraph.Load(); g != nil { return g } s.neighborBuildMu.Lock() defer s.neighborBuildMu.Unlock() - s.neighborMu.Lock() - g = s.neighborGraph - s.neighborMu.Unlock() - if g != nil { + if g := s.neighborGraph.Load(); g != nil { return g // a concurrent initial request or recomputer built it } - g = s.buildNeighborGraph() - s.neighborMu.Lock() - s.neighborGraph = g - s.neighborMu.Unlock() + g := s.buildNeighborGraph() + s.neighborGraph.Store(g) return g } -// rebuildNeighborGraph runs on every recompute tick, not just after the old -// TTL. Builds are serialized, but readers retain the previous snapshot while -// the potentially expensive store scan runs outside the pointer mutex. +// rebuildNeighborGraph builds a replacement graph and atomically publishes it. +// Builds are serialized, but readers retain the previous snapshot while the +// potentially expensive store scan runs. func (s *Server) rebuildNeighborGraph() { if s.store == nil { return // preserve injected graphs when there is no source to rebuild } s.neighborBuildMu.Lock() defer s.neighborBuildMu.Unlock() - g := s.buildNeighborGraph() - s.neighborMu.Lock() - s.neighborGraph = g - s.neighborMu.Unlock() + s.neighborGraph.Store(s.buildNeighborGraph()) } func (s *Server) buildNeighborGraph() *NeighborGraph { diff --git a/cmd/server/neighbor_api_test.go b/cmd/server/neighbor_api_test.go index a34d1220e..d79349dfe 100644 --- a/cmd/server/neighbor_api_test.go +++ b/cmd/server/neighbor_api_test.go @@ -19,7 +19,7 @@ func makeTestServer(graph *NeighborGraph) *Server { srv := &Server{ perfStats: NewPerfStats(), } - srv.neighborGraph = graph + srv.neighborGraph.Store(graph) return srv } @@ -364,7 +364,7 @@ func TestNeighborAPI_DistanceKm_WithGPS(t *testing.T) { srv.store = NewPacketStore(db, nil) now := time.Now() - srv.neighborGraph = makeTestGraph(newEdge("aaaa", "bbbb", "bb", 50, now)) + srv.neighborGraph.Store(makeTestGraph(newEdge("aaaa", "bbbb", "bb", 50, now))) rr := serveRequest(srv, "GET", "/api/nodes/aaaa/neighbors") var resp NeighborResponse @@ -398,7 +398,7 @@ func TestNeighborAPI_DistanceKm_NoGPS(t *testing.T) { srv.store = NewPacketStore(db, nil) now := time.Now() - srv.neighborGraph = makeTestGraph(newEdge("aaaa", "bbbb", "bb", 50, now)) + srv.neighborGraph.Store(makeTestGraph(newEdge("aaaa", "bbbb", "bb", 50, now))) rr := serveRequest(srv, "GET", "/api/nodes/aaaa/neighbors") var resp NeighborResponse diff --git a/cmd/server/neighbor_background_build_test.go b/cmd/server/neighbor_background_build_test.go index 4a26c1762..e1e11337a 100644 --- a/cmd/server/neighbor_background_build_test.go +++ b/cmd/server/neighbor_background_build_test.go @@ -22,7 +22,8 @@ func neighborBackgroundTestStore() *PacketStore { func TestNeighborGraphExistingStaleSnapshotDoesNotRebuild(t *testing.T) { old := NewNeighborGraph() old.builtAt = time.Now().Add(-2 * neighborGraphTTL) - srv := &Server{store: neighborBackgroundTestStore(), neighborGraph: old} + srv := &Server{store: neighborBackgroundTestStore()} + srv.neighborGraph.Store(old) if got := srv.getNeighborGraph(); got != old { t.Fatal("request rebuilt an existing stale graph instead of serving it") } @@ -32,12 +33,12 @@ func TestNeighborGraphTickRebuildsEvenFreshSnapshot(t *testing.T) { old := NewNeighborGraph() old.builtAt = time.Now() srv := &Server{ - store: neighborBackgroundTestStore(), - neighborGraph: old, + store: neighborBackgroundTestStore(), computeNeighborGraphResponseFn: func(int, float64, string, string) NeighborGraphResponse { return NeighborGraphResponse{} }, } + srv.neighborGraph.Store(old) srv.recomputeNeighborGraphCache() if got := srv.getNeighborGraph(); got == old || len(got.AllEdges()) == 0 { t.Fatal("tick did not replace a fresh graph with a newly built graph") @@ -50,7 +51,7 @@ func TestNeighborGraphReadersContinueDuringBlockedRebuild(t *testing.T) { srv, _ := setupTestServer(t) old := NewNeighborGraph() old.builtAt = time.Now().Add(-2 * neighborGraphTTL) - srv.neighborGraph = old + srv.neighborGraph.Store(old) srv.store.InvalidateNodeCache() conn, err := srv.db.conn.Conn(context.Background()) if err != nil { @@ -135,11 +136,11 @@ func TestNeighborGraphConcurrentFirstReadersShareBuild(t *testing.T) { func TestNeighborGraphRecomputeWithoutStoreKeepsInjectedGraph(t *testing.T) { injected := NewNeighborGraph() srv := &Server{ - neighborGraph: injected, computeNeighborGraphResponseFn: func(int, float64, string, string) NeighborGraphResponse { return NeighborGraphResponse{} }, } + srv.neighborGraph.Store(injected) srv.recomputeNeighborGraphCache() if srv.getNeighborGraph() != injected { t.Fatal("recompute discarded an injected graph without a packet store") @@ -150,9 +151,9 @@ func TestNeighborGraphFailedRebuildKeepsPublishedSnapshot(t *testing.T) { old := NewNeighborGraph() old.builtAt = time.Now() srv := &Server{ - store: ngTestStore(nil, []*StoreTx{nil}), // fault injection: graph build panics - neighborGraph: old, + store: ngTestStore(nil, []*StoreTx{nil}), // fault injection: graph build panics } + srv.neighborGraph.Store(old) srv.recomputeNeighborGraphCache() if got := srv.getNeighborGraph(); got != old { t.Fatal("failed background build replaced the last published graph") @@ -171,13 +172,12 @@ func BenchmarkNeighborGraphStaleRead30K(b *testing.B) { } old := NewNeighborGraph() old.builtAt = time.Now().Add(-2 * neighborGraphTTL) - srv := &Server{store: store, neighborGraph: old} + srv := &Server{store: store} + srv.neighborGraph.Store(old) b.ReportAllocs() b.ResetTimer() for range b.N { - srv.neighborMu.Lock() - srv.neighborGraph = old - srv.neighborMu.Unlock() + srv.neighborGraph.Store(old) srv.getNeighborGraph() } } diff --git a/cmd/server/neighbor_followups_test.go b/cmd/server/neighbor_followups_test.go new file mode 100644 index 000000000..dedebafcb --- /dev/null +++ b/cmd/server/neighbor_followups_test.go @@ -0,0 +1,227 @@ +package main + +import ( + "os" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Follow-ups from #541 (issue #545). + +func followupNodes() []nodeInfo { + return []nodeInfo{ + {Role: "repeater", PublicKey: "aaaa1111", Name: "A_SF", Lat: 37.77, Lon: -122.41, HasGPS: true}, + {Role: "repeater", PublicKey: "cccc3333", Name: "C_SJ", Lat: 37.34, Lon: -121.89, HasGPS: true}, // ~68 km from A + {Role: "repeater", PublicKey: "obs00001", Name: "Observer", Lat: 37.77, Lon: -122.41, HasGPS: true}, + } +} + +func followupStore() *PacketStore { + // Enough adverts to pass the default-shape filter (minCount=5, minScore=0.1). + var txs []*StoreTx + for i := 1; i <= 20; i++ { + txs = append(txs, ngMakeTx(i, PayloadADVERT, ngFromNodeJSON("aaaa1111"), []*StoreObs{ + ngMakeObs("obs00001", `["cccc"]`, nowStr, ngFloatPtr(-10)), + })) + } + return ngTestStore(followupNodes(), txs) +} + +// F1: the rebuild cadence is derived from the TTL, not the response-cache interval. +func TestNeighborGraphRebuildEvery(t *testing.T) { + cases := []struct { + interval time.Duration + want int + }{ + {0, 1}, + {-time.Second, 1}, + {neighborGraphCacheInterval, 1}, // default behaves as before + {neighborGraphTTL, 1}, + {2 * neighborGraphTTL, 1}, // slower than the TTL: every tick + {time.Minute, 5}, + {90 * time.Second, 4}, // ceil(300/90): never rebuilds more often than the TTL + {10 * time.Second, 30}, + } + for _, c := range cases { + if got := neighborGraphRebuildEvery(c.interval); got != c.want { + t.Errorf("neighborGraphRebuildEvery(%v) = %d, want %d", c.interval, got, c.want) + } + } +} + +// F1: a fast response-cache cadence must not rebuild the graph on every tick. +func TestNeighborGraphFastCacheCadenceDoesNotRebuildGraphEveryTick(t *testing.T) { + srv := &Server{store: followupStore()} + stop := make(chan struct{}) + defer close(stop) + srv.startNeighborGraphRecomputer(5*time.Millisecond, stop) + + deadline := time.Now().Add(5 * time.Second) + for srv.neighborGraphCache.ptr.Load() == nil { + if time.Now().After(deadline) { + t.Fatal("first recompute never published") + } + time.Sleep(time.Millisecond) + } + first := srv.neighborGraph.Load() + if first == nil { + t.Fatal("first tick did not publish a graph") + } + seen := map[*neighborGraphCacheEntry]bool{} + for len(seen) < 4 { + if time.Now().After(deadline) { + t.Fatalf("only saw %d cache ticks", len(seen)) + } + seen[srv.neighborGraphCache.ptr.Load()] = true + time.Sleep(time.Millisecond) + } + if srv.neighborGraph.Load() != first { + t.Fatal("graph was rebuilt on a cache tick well inside neighborGraphTTL") + } +} + +// F2: the cached response is built from the graph the same tick just produced. +func TestNeighborGraphCacheBuiltFromSameTickGraph(t *testing.T) { + stale := NewNeighborGraph() // empty, as a previous tick's graph would be before new traffic + stale.builtAt = time.Now() + srv := &Server{store: followupStore()} + srv.neighborGraph.Store(stale) + srv.recomputeNeighborGraphCache() + + entry := srv.neighborGraphCache.unfilteredPtr.Load() + if entry == nil { + t.Fatal("no unfiltered cache entry published") + } + if len(entry.resp.Edges) == 0 { + t.Fatal("cached response was built from the previous graph, not the one this tick produced") + } + def := srv.neighborGraphCache.ptr.Load() + if def == nil || len(def.resp.Edges) == 0 { + t.Fatal("default-shape cached response was built from the previous graph") + } +} + +// F2 (skipped rebuild): a tick that skips the graph rebuild still builds from the published graph. +func TestNeighborGraphCacheTickWithoutRebuildKeepsGraph(t *testing.T) { + g := BuildFromStore(followupStore()) + srv := &Server{store: followupStore()} + srv.neighborGraph.Store(g) + srv.recomputeNeighborGraphCacheTick(false) + if srv.neighborGraph.Load() != g { + t.Fatal("tick with rebuild=false replaced the graph") + } + if e := srv.neighborGraphCache.unfilteredPtr.Load(); e == nil || len(e.resp.Edges) == 0 { + t.Fatal("response not built from the published graph") + } +} + +// F3/F7: concurrent publishers and readers are race-free. The previous plain +// pointer field raced here under -race when the mutex was dropped; the +// atomic.Pointer field makes that mutant unrepresentable. +func TestNeighborGraphPublishIsRaceFree(t *testing.T) { + srv := &Server{store: followupStore()} + srv.neighborGraph.Store(NewNeighborGraph()) + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(2) + go func() { + defer wg.Done() + for j := 0; j < 20; j++ { + srv.rebuildNeighborGraph() + } + }() + go func() { + defer wg.Done() + for j := 0; j < 200; j++ { + if srv.getNeighborGraph() == nil { + t.Error("nil graph") + return + } + } + }() + } + wg.Wait() +} + +// F6: Config.NeighborMaxEdgeKm semantics. +func TestConfigNeighborMaxEdgeKm(t *testing.T) { + var nilCfg *Config + cases := []struct { + name string + cfg *Config + want float64 + }{ + {"nil config", nilCfg, DefaultMaxEdgeKm}, + {"no neighborGraph block", &Config{}, DefaultMaxEdgeKm}, + {"zero = default", &Config{NeighborGraph: &NeighborGraphConfig{}}, DefaultMaxEdgeKm}, + {"explicit", &Config{NeighborGraph: &NeighborGraphConfig{MaxEdgeKm: 42}}, 42}, + {"negative disables", &Config{NeighborGraph: &NeighborGraphConfig{MaxEdgeKm: -1}}, 0}, + } + for _, c := range cases { + if got := c.cfg.NeighborMaxEdgeKm(); got != c.want { + t.Errorf("%s: got %v, want %v", c.name, got, c.want) + } + } +} + +// F6: Server.buildNeighborGraph passes the configured maxEdgeKm into the build. +func TestBuildNeighborGraphWiresMaxEdgeKm(t *testing.T) { + edges := func(cfg *Config) (int, uint64) { + srv := &Server{store: followupStore(), cfg: cfg} + g := srv.buildNeighborGraph() + return len(g.AllEdges()), g.RejectedEdgesGeoFar + } + if n, rej := edges(&Config{NeighborGraph: &NeighborGraphConfig{MaxEdgeKm: 10}}); n != 0 || rej == 0 { + t.Errorf("maxEdgeKm=10: edges=%d rejected=%d, want 0 edges and >0 rejected", n, rej) + } + if n, rej := edges(&Config{}); n == 0 || rej != 0 { + t.Errorf("default: edges=%d rejected=%d, want edges kept", n, rej) + } + if n, rej := edges(&Config{NeighborGraph: &NeighborGraphConfig{MaxEdgeKm: -1}}); n == 0 || rej != 0 { + t.Errorf("filter disabled: edges=%d rejected=%d, want edges kept", n, rej) + } + if n, _ := edges(nil); n == 0 { + t.Error("nil cfg: edges dropped") + } +} + +// F6: Config.DebugAffinity enables the build's disambiguation log. +func TestBuildNeighborGraphWiresDebugAffinity(t *testing.T) { + on := &Server{store: followupStore(), cfg: &Config{DebugAffinity: true}} + if on.buildNeighborGraph().logFn == nil { + t.Error("DebugAffinity=true did not enable build logging") + } + off := &Server{store: followupStore(), cfg: &Config{}} + if off.buildNeighborGraph().logFn != nil { + t.Error("DebugAffinity=false enabled build logging") + } +} + +// F3/F7: the publish pointer is an atomic.Pointer (house pattern, like +// PacketStore.graph and neighborGraphCache.ptr), so a mutex-less plain-pointer +// publish cannot be written. +func TestNeighborGraphFieldIsAtomicPointer(t *testing.T) { + f, ok := reflect.TypeOf(Server{}).FieldByName("neighborGraph") + if !ok || f.Type != reflect.TypeOf(atomic.Pointer[NeighborGraph]{}) { + t.Fatalf("Server.neighborGraph must be atomic.Pointer[NeighborGraph], got %v", f.Type) + } +} + +// F4: the cacheRecomputeIntervalSeconds docs keep their #1481 P0-1 / #1483 provenance. +func TestNeighborCacheIntervalDocsKeepIssueRefs(t *testing.T) { + for _, path := range []string{"config.go", "../../config.example.json"} { + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + for _, ref := range []string{"#1481 P0-1", "#1483"} { + if !strings.Contains(string(b), "cacheRecomputeIntervalSeconds") || !strings.Contains(string(b), ref) { + t.Errorf("%s lost reference %q", path, ref) + } + } + } +} diff --git a/cmd/server/neighbor_graph_cache.go b/cmd/server/neighbor_graph_cache.go index ef5313b3b..16f843ab8 100644 --- a/cmd/server/neighbor_graph_cache.go +++ b/cmd/server/neighbor_graph_cache.go @@ -21,6 +21,19 @@ import ( const neighborGraphCacheInterval = 5 * time.Minute +// neighborGraphRebuildEvery returns how many recompute ticks make up one graph +// rebuild: ceil(neighborGraphTTL / interval), at least 1. The response cache is +// refreshed on every tick, but the ~seconds-long graph build runs at most once +// per neighborGraphTTL, so cacheRecomputeIntervalSeconds neither multiplies the +// rebuild CPU nor changes the graph's age below the TTL. Intervals at or above +// the TTL rebuild on every tick (graph age == interval). +func neighborGraphRebuildEvery(interval time.Duration) int { + if interval <= 0 || interval >= neighborGraphTTL { + return 1 + } + return int((neighborGraphTTL + interval - 1) / interval) +} + // neighborGraphCacheEntry holds both the response struct (kept for // tests / structured access) and the pre-marshaled bytes that the // handler writes verbatim. @@ -46,14 +59,15 @@ func (s *Server) startNeighborGraphRecomputer(interval time.Duration, stop <-cha if interval <= 0 { interval = neighborGraphCacheInterval } + every := neighborGraphRebuildEvery(interval) go func() { s.recomputeNeighborGraphCache() t := time.NewTicker(interval) defer t.Stop() - for { + for tick := 1; ; tick++ { select { case <-t.C: - s.recomputeNeighborGraphCache() + s.recomputeNeighborGraphCacheTick(tick%every == 0) case <-stop: return } @@ -66,6 +80,13 @@ func (s *Server) startNeighborGraphRecomputer(interval time.Duration, stop <-cha // rebuild doesn't kill the background goroutine — but logs the panic // and increments a counter so operators see the failure (#1483 follow-up). func (s *Server) recomputeNeighborGraphCache() { + s.recomputeNeighborGraphCacheTick(true) +} + +// recomputeNeighborGraphCacheTick refreshes the cached responses, first +// rebuilding the graph when rebuildGraph is set. The responses are always built +// from the graph published by this call (or the existing one when skipped). +func (s *Server) recomputeNeighborGraphCacheTick(rebuildGraph bool) { defer func() { if r := recover(); r != nil { log.Printf("[neighbor-graph-cache] rebuild panic: %v\n%s", r, debug.Stack()) @@ -73,7 +94,9 @@ func (s *Server) recomputeNeighborGraphCache() { } }() start := time.Now() - s.rebuildNeighborGraph() + if rebuildGraph { + s.rebuildNeighborGraph() + } resp := s.buildDefaultNeighborGraphResponse() var buf bytes.Buffer if err := json.NewEncoder(&buf).Encode(resp); err != nil { diff --git a/cmd/server/resolve_context_test.go b/cmd/server/resolve_context_test.go index c05018503..a79b1176f 100644 --- a/cmd/server/resolve_context_test.go +++ b/cmd/server/resolve_context_test.go @@ -242,9 +242,7 @@ func TestResolveHopsAPI_WithAffinityContext(t *testing.T) { graph.upsertEdge("c0c0c0c0c0", "dd1aaaaaaa", "dd1", "obs1", nil, time.Now()) } graph.builtAt = time.Now() - srv.neighborMu.Lock() - srv.neighborGraph = graph - srv.neighborMu.Unlock() + srv.neighborGraph.Store(graph) req := httptest.NewRequest("GET", "/api/resolve-hops?hops=dd1&from_node=c0c0c0c0c0", nil) rr := httptest.NewRecorder() diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 9f401a8eb..9d349571b 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -75,9 +75,8 @@ type Server struct { saveMu sync.Mutex // Neighbor affinity graph: background-built, with one synchronous first build. - neighborMu sync.Mutex // protects the published pointer, never a graph build - neighborBuildMu sync.Mutex // serializes initial and background builds - neighborGraph *NeighborGraph + neighborBuildMu sync.Mutex // serializes initial and background builds + neighborGraph atomic.Pointer[NeighborGraph] // published snapshot; readers never lock // #2073 node-detail advert route breakdown, per pubkey // (node_advert_routes_cache.go). diff --git a/config.example.json b/config.example.json index 1baa24b4b..1c0b8ec5b 100644 --- a/config.example.json +++ b/config.example.json @@ -383,7 +383,7 @@ "maxAgeDays": 5, "maxEdgeKm": 500, "cacheRecomputeIntervalSeconds": 300, - "_comment": "maxAgeDays: neighbor edges older than this many days are pruned on startup and daily. Default 5. maxEdgeKm: geo-implausibility filter — when both endpoints have GPS, edges with haversine distance > maxEdgeKm are rejected at build time to prevent disambiguator self-reinforcement on wide-geo MQTT deployments. Default 500 km (well above any plausible terrestrial LoRa hop). 0 ⇒ use default; set negative to disable the filter. Rejected count surfaces in /api/analytics/neighbor-graph stats. Issue #1228. cacheRecomputeIntervalSeconds: how often the background recomputer rebuilds the neighbor graph and its cached responses. Default 300 (5 min). Requests keep the old graph until the next build finishes. Lower means fresher data and a full build on every tick; higher means an older graph between ticks." + "_comment": "maxAgeDays: neighbor edges older than this many days are pruned on startup and daily. Default 5. maxEdgeKm: geo-implausibility filter — when both endpoints have GPS, edges with haversine distance > maxEdgeKm are rejected at build time to prevent disambiguator self-reinforcement on wide-geo MQTT deployments. Default 500 km (well above any plausible terrestrial LoRa hop). 0 ⇒ use default; set negative to disable the filter. Rejected count surfaces in /api/analytics/neighbor-graph stats. Issue #1228. cacheRecomputeIntervalSeconds: how often the background recomputer refreshes the cached default-shape /api/analytics/neighbor-graph responses (#1481 P0-1). Default 300 (5 min). The graph itself is rebuilt at most once per 5 minutes, so a lower value refreshes the cached responses more often without extra graph-build CPU; a higher value makes the graph as old as the interval. Requests keep the published graph until the next build finishes. Issue #1483, #545." }, "estimatedPositions": { "enabled": true,