From 51965868b2557507316b1b1ac039796f085eae38 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 07:35:41 +0200 Subject: [PATCH] perf(neighbors): rebuild published graphs in the background --- cmd/server/config.go | 7 +- cmd/server/neighbor_api.go | 62 +++++-- cmd/server/neighbor_background_build_test.go | 183 +++++++++++++++++++ cmd/server/neighbor_graph_cache.go | 1 + cmd/server/routes.go | 7 +- config.example.json | 2 +- 6 files changed, 240 insertions(+), 22 deletions(-) create mode 100644 cmd/server/neighbor_background_build_test.go diff --git a/cmd/server/config.go b/cmd/server/config.go index b05417fb2..21dfd41d1 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -493,9 +493,10 @@ 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 the default-shape neighbor-graph - // response (#1481 P0-1). 0/missing = default 300 (5 min). - // Lower = fresher data, more CPU per minute. #1483. + // 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. CacheRecomputeIntervalSeconds int `json:"cacheRecomputeIntervalSeconds,omitempty"` } diff --git a/cmd/server/neighbor_api.go b/cmd/server/neighbor_api.go index b0d0bd815..7e05d8a8f 100644 --- a/cmd/server/neighbor_api.go +++ b/cmd/server/neighbor_api.go @@ -80,24 +80,56 @@ type GraphStats struct { // ─── Graph accessor on Server ────────────────────────────────────────────────── -// getNeighborGraph returns the current neighbor graph, rebuilding if stale. +// 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() - defer s.neighborMu.Unlock() - - if s.neighborGraph == nil || s.neighborGraph.IsStale() { - if s.store != nil { - opts := BuildOptions{MaxEdgeKm: DefaultMaxEdgeKm} - if s.cfg != nil { - opts.EnableLog = s.cfg.DebugAffinity - opts.MaxEdgeKm = s.cfg.NeighborMaxEdgeKm() - } - s.neighborGraph = BuildFromStoreWithOptions(s.store, opts) - } else { - s.neighborGraph = NewNeighborGraph() - } + g := s.neighborGraph + s.neighborMu.Unlock() + if g != nil { + return g + } + + s.neighborBuildMu.Lock() + defer s.neighborBuildMu.Unlock() + s.neighborMu.Lock() + g = s.neighborGraph + s.neighborMu.Unlock() + if g != nil { + return g // a concurrent initial request or recomputer built it + } + g = s.buildNeighborGraph() + s.neighborMu.Lock() + s.neighborGraph = g + s.neighborMu.Unlock() + 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. +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() +} + +func (s *Server) buildNeighborGraph() *NeighborGraph { + if s.store == nil { + return NewNeighborGraph() + } + opts := BuildOptions{MaxEdgeKm: DefaultMaxEdgeKm} + if s.cfg != nil { + opts.EnableLog = s.cfg.DebugAffinity + opts.MaxEdgeKm = s.cfg.NeighborMaxEdgeKm() } - return s.neighborGraph + return BuildFromStoreWithOptions(s.store, opts) } // ─── Handlers ────────────────────────────────────────────────────────────────── diff --git a/cmd/server/neighbor_background_build_test.go b/cmd/server/neighbor_background_build_test.go new file mode 100644 index 000000000..4a26c1762 --- /dev/null +++ b/cmd/server/neighbor_background_build_test.go @@ -0,0 +1,183 @@ +package main + +import ( + "context" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" +) + +func neighborBackgroundTestStore() *PacketStore { + return ngTestStore([]nodeInfo{ + {PublicKey: "aaaa1111", Role: "repeater"}, + {PublicKey: "bbbb2222", Role: "repeater"}, + {PublicKey: "cccc3333", Role: "repeater"}, + }, []*StoreTx{ngMakeTx(1, PayloadADVERT, ngFromNodeJSON("aaaa1111"), []*StoreObs{ + ngMakeObs("cccc3333", `["bbbb"]`, "2026-10-01T12:00:00Z", nil), + })}) +} + +func TestNeighborGraphExistingStaleSnapshotDoesNotRebuild(t *testing.T) { + old := NewNeighborGraph() + old.builtAt = time.Now().Add(-2 * neighborGraphTTL) + srv := &Server{store: neighborBackgroundTestStore(), neighborGraph: old} + if got := srv.getNeighborGraph(); got != old { + t.Fatal("request rebuilt an existing stale graph instead of serving it") + } +} + +func TestNeighborGraphTickRebuildsEvenFreshSnapshot(t *testing.T) { + old := NewNeighborGraph() + old.builtAt = time.Now() + srv := &Server{ + store: neighborBackgroundTestStore(), + neighborGraph: old, + computeNeighborGraphResponseFn: func(int, float64, string, string) NeighborGraphResponse { + return NeighborGraphResponse{} + }, + } + 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") + } +} + +// Hold the real SQL pool so the expensive build remains in flight. Serving +// the old graph must not wait for that build or hold the pointer mutex for it. +func TestNeighborGraphReadersContinueDuringBlockedRebuild(t *testing.T) { + srv, _ := setupTestServer(t) + old := NewNeighborGraph() + old.builtAt = time.Now().Add(-2 * neighborGraphTTL) + srv.neighborGraph = old + srv.store.InvalidateNodeCache() + conn, err := srv.db.conn.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + buildDone := make(chan struct{}) + readDone := make(chan *NeighborGraph, 1) + readerStarted := false + t.Cleanup(func() { + conn.Close() + select { + case <-buildDone: + case <-time.After(5 * time.Second): + t.Error("build did not finish after releasing SQL") + } + if readerStarted { + select { + case <-readDone: + case <-time.After(5 * time.Second): + t.Error("reader did not finish after releasing SQL") + } + } + }) + waits := srv.db.conn.Stats().WaitCount + go func() { + defer close(buildDone) + srv.recomputeNeighborGraphCache() + }() + deadline := time.Now().Add(3 * time.Second) + for srv.db.conn.Stats().WaitCount == waits { + select { + case <-buildDone: + t.Fatal("recompute did not attempt a graph build") + default: + } + if time.Now().After(deadline) { + t.Fatal("build never reached the occupied SQL pool") + } + runtime.Gosched() + } + readerStarted = true + go func() { readDone <- srv.getNeighborGraph() }() + select { + case got := <-readDone: + readerStarted = false + if got != old { + t.Fatal("reader did not receive the old graph during the blocked build") + } + case <-time.After(time.Second): + t.Fatal("reader waited for the background graph build") + } +} + +func TestNeighborGraphConcurrentFirstReadersShareBuild(t *testing.T) { + srv := &Server{store: neighborBackgroundTestStore()} + const readers = 16 + start := make(chan struct{}) + results := make(chan *NeighborGraph, readers) + var wg sync.WaitGroup + for range readers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + results <- srv.getNeighborGraph() + }() + } + close(start) + wg.Wait() + close(results) + var first *NeighborGraph + for got := range results { + if first == nil { + first = got + } + if got == nil || got != first || len(got.AllEdges()) == 0 { + t.Fatal("concurrent initial requests did not share one populated graph") + } + } +} + +func TestNeighborGraphRecomputeWithoutStoreKeepsInjectedGraph(t *testing.T) { + injected := NewNeighborGraph() + srv := &Server{ + neighborGraph: injected, + computeNeighborGraphResponseFn: func(int, float64, string, string) NeighborGraphResponse { + return NeighborGraphResponse{} + }, + } + srv.recomputeNeighborGraphCache() + if srv.getNeighborGraph() != injected { + t.Fatal("recompute discarded an injected graph without a packet store") + } +} + +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, + } + srv.recomputeNeighborGraphCache() + if got := srv.getNeighborGraph(); got != old { + t.Fatal("failed background build replaced the last published graph") + } + if failures := atomic.LoadUint64(&srv.neighborGraphCacheRebuildFailures); failures != 1 { + t.Fatalf("rebuild failures = %d, want 1", failures) + } +} + +func BenchmarkNeighborGraphStaleRead30K(b *testing.B) { + store := neighborBackgroundTestStore() + packet := store.packets[0] + store.packets = make([]*StoreTx, 30000) + for i := range store.packets { + store.packets[i] = ngMakeTx(i+1, PayloadADVERT, packet.DecodedJSON, packet.Observations) + } + old := NewNeighborGraph() + old.builtAt = time.Now().Add(-2 * neighborGraphTTL) + srv := &Server{store: store, neighborGraph: old} + b.ReportAllocs() + b.ResetTimer() + for range b.N { + srv.neighborMu.Lock() + srv.neighborGraph = old + srv.neighborMu.Unlock() + srv.getNeighborGraph() + } +} diff --git a/cmd/server/neighbor_graph_cache.go b/cmd/server/neighbor_graph_cache.go index 319af56b1..ef5313b3b 100644 --- a/cmd/server/neighbor_graph_cache.go +++ b/cmd/server/neighbor_graph_cache.go @@ -73,6 +73,7 @@ func (s *Server) recomputeNeighborGraphCache() { } }() start := time.Now() + s.rebuildNeighborGraph() resp := s.buildDefaultNeighborGraphResponse() var buf bytes.Buffer if err := json.NewEncoder(&buf).Encode(resp); err != nil { diff --git a/cmd/server/routes.go b/cmd/server/routes.go index de6850110..521b69017 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -74,9 +74,10 @@ type Server struct { // can't race on the .tmp file or interleave disk/memory updates. saveMu sync.Mutex - // Neighbor affinity graph (lazy-built, cached with TTL) - neighborMu sync.Mutex - neighborGraph *NeighborGraph + // 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 // #2073 node-detail advert route breakdown, per pubkey // (node_advert_routes_cache.go). diff --git a/config.example.json b/config.example.json index 20350f852..1baa24b4b 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 default-shape /api/analytics/neighbor-graph response (#1481 P0-1). Default 300 (5 min). Lower = fresher data, more CPU per minute. Issue #1483." + "_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." }, "estimatedPositions": { "enabled": true,