neighbors: follow-ups from #541 (graph-age clamp, cache/publish tests, atomic.Pointer)

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"`
 }
 

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 {

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

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()
 	}
 }

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)
+			}
+		}
+	}
+}