From 1720060284df7e581ad42bf4c6903b7fbba5b07e Mon Sep 17 00:00:00 2001 From: SaarMesh-Bot Date: Wed, 2 Sep 2026 11:07:08 +0200 Subject: [PATCH] fix(#1827): avoid per-observation SQL fetch in handleObserverAnalytics hot loop (#1829) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Fixes the CPU/DoS issue in #1827: observer detail pages were saturating CPU on busy observers — 6-7 concurrently loaded tabs pegged 12 cores for seconds, and auto-refresh made it self-sustaining. ## Root cause `handleObserverAnalytics` iterated every observation in the requested window and called `enrichObs()` per observation just to read `payload_type` and `decoded_json` for the `packetTypes`/`nodesTimeline` aggregates. `enrichObs()` also runs an on-demand SQL `SELECT resolved_path FROM observations WHERE id=?` (`fetchResolvedPathForObs`) and builds a full response map — both of which are unused by this aggregation loop. `resolved_path` is only actually consumed by the `<=20` kept `recentPackets` entries. Per the triage in #1827 (@carmack): *"Replacing `enrichObs(obs)` with a direct `s.store.byTxID[obs.TransmissionID].PayloadType` read (as sketched in the body) drops a map alloc + interface boxes per obs on the loop that saturated the operator's 12 cores. Byte-identical output. That's ~90% of the value."* This PR implements exactly that fast-path. ## Change - Aggregate loop (`packetTypes`, `nodesTimeline`): read `payload_type`/`decoded_json` directly off the transmission via `s.store.byTxID[obs.TransmissionID]` — no SQL, no per-obs map allocation. - `recentPackets` (`<=20` entries): unchanged, still calls `enrichObs()` since it needs `resolved_path`/`raw_hex`/etc. for display. - Output is unchanged: `packetTypes`/`nodesTimeline` are computed from the exact same underlying fields (`tx.PayloadType`, `tx.DecodedJSON`), just without the O(N) SQL round-trips. ## Scope This is the concrete hot-path fix from #1827's triage — not the broader `/api/observers/{id}/analytics` endpoint-split proposal in #1828, which (per that issue's discussion) is a separate P3 follow-up. #1828's own triage converged on this same `byTxID` fast-path as "the ground-work minimum" before any endpoint splitting. ## Testing - Existing `TestObserverAnalytics` passes unchanged. - Extended `TestObserverAnalytics/default` to assert `packetTypes` counts come out correct (`{"4":2,"5":1}` for the seeded fixture) via the new `byTxID` path, and that `recentPackets` still carries `resolved_path` where present (confirming the `enrichObs()` path for those 20 entries is untouched). - `go build ./...` and `go vet ./...` clean in `cmd/server`. - Full `go test ./...` in `cmd/server`: passes except 4 pre-existing test-order-dependent failures in `TestHandleNodePaths_*` (unrelated to this change — reproduced identically on a fresh, unpatched clone of `upstream/master`). --------- Co-authored-by: SaarMesh-Bot <300107934+SaarMesh-Bot@users.noreply.github.com> Co-authored-by: Claude --- cmd/server/observer_analytics.go | 25 ++++--- cmd/server/observer_analytics_test.go | 8 +-- cmd/server/routes.go | 18 ++++- cmd/server/routes_test.go | 99 ++++++++++++++++++++++++++- cmd/server/store.go | 18 ++++- 5 files changed, 149 insertions(+), 19 deletions(-) diff --git a/cmd/server/observer_analytics.go b/cmd/server/observer_analytics.go index 6d5a7d52..005cf70d 100644 --- a/cmd/server/observer_analytics.go +++ b/cmd/server/observer_analytics.go @@ -10,10 +10,15 @@ // (see #1481 P0-2). Helpers do NOT touch store.mu — the handler owns lock // scoping. // -// Concurrency note (#1839 MINOR): the RLock snapshot only covers the -// *StoreObs pointer slice; reads of store.byTxID inside buildPacketTypes / -// buildNodesTimeline are unsynchronized concurrent-map reads (pre-existing -// behavior — enrichObs did the same). Not introduced by this refactor. +// Concurrency note (#1830): buildPacketTypes / buildNodesTimeline / +// buildRecentPackets take a txByID map instead of *PacketStore for +// transmission lookups. store.byTxID is guarded by store.mu (writes from +// ingest/eviction); reading it here — after the handler's RLock snapshot has +// already been released, to keep JSON decode/enrichment off the hot lock per +// #1481 P0-2 — would race with those writers (Go maps can panic with +// "concurrent map read and map write" during a rehash, not just fail under +// -race). The handler resolves txByID once, under the same RLock that +// snapshots the observation slice, and passes it down instead. // // Perf note (#1839 MINOR): dropping enrichObs on the histogram / nodes- // timeline paths also eliminates N fetchResolvedPathForObs SQL calls per @@ -93,10 +98,10 @@ func buildTimeline(filtered []*StoreObs, days int) []TimeBucket { // needs payload_type, which is a single indirection off the transmission. // This avoids one map allocation + several interface-boxing conversions per // observation (routes.go:2886 hot path pre-#1828). -func buildPacketTypes(store *PacketStore, filtered []*StoreObs) map[string]int { +func buildPacketTypes(filtered []*StoreObs, txByID map[int]*StoreTx) map[string]int { out := map[string]int{} for _, obs := range filtered { - tx := store.byTxID[obs.TransmissionID] + tx := txByID[obs.TransmissionID] if tx == nil || tx.PayloadType == nil { continue } @@ -107,7 +112,7 @@ func buildPacketTypes(store *PacketStore, filtered []*StoreObs) map[string]int { // buildNodesTimeline builds the distinct-node-per-bucket timeline aggregate. // Nodes = path-json hops ∪ decoded_json pubKey/srcHash/destHash. -func buildNodesTimeline(store *PacketStore, filtered []*StoreObs, days int) []TimeBucket { +func buildNodesTimeline(filtered []*StoreObs, days int, txByID map[int]*StoreTx) []TimeBucket { bucketDur := observerAnalyticsBucketDur(days) nodeBucketSets := map[int64]map[string]struct{}{} for _, obs := range filtered { @@ -121,7 +126,7 @@ func buildNodesTimeline(store *PacketStore, filtered []*StoreObs, days int) []Ti } // Legacy handler read decoded_json via enrichObs (which pulls it off // tx.DecodedJSON). Read tx directly for parity + savings. - if tx := store.byTxID[obs.TransmissionID]; tx != nil && tx.DecodedJSON != "" { + if tx := txByID[obs.TransmissionID]; tx != nil && tx.DecodedJSON != "" { var decoded map[string]interface{} if json.Unmarshal([]byte(tx.DecodedJSON), &decoded) == nil { for _, k := range []string{"pubKey", "srcHash", "destHash"} { @@ -189,7 +194,7 @@ func buildSnrDistribution(filtered []*StoreObs) []SnrDistributionEntry { // good-ts obs. Result can be =20). Result len = 17. @@ -208,7 +208,7 @@ func TestBuildNodesTimelineDistinct(t *testing.T) { buildObsForTest(1, base, nil, `["bb","cc"]`), // bucket A: nodes {aaaa, bb, cc} buildObsForTest(1, base.Add(10*time.Minute), nil, `["bb"]`), // same bucket: dedup } - got := buildNodesTimeline(store, filtered, 1) + got := buildNodesTimeline(filtered, 1, store.byTxID) if len(got) != 1 { t.Fatalf("nodes timeline entries = %d, want 1 (got %+v)", len(got), got) } diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 651785b6..0cc7952b 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -2850,6 +2850,18 @@ func (s *Server) handleObserverAnalytics(w http.ResponseWriter, r *http.Request) obsList := s.store.byObserver[id] obsSnapshot := make([]*StoreObs, len(obsList)) copy(obsSnapshot, obsList) + // #1830: also resolve each referenced transmission's *StoreTx under + // this same RLock. s.store.byTxID is guarded by s.store.mu (writes + // from ingest/eviction); reading it after RUnlock — as the loop below + // used to via s.store.byTxID[...] and enrichObs() — races with those + // writers. Keyed by TransmissionID (not by observation index) since + // multiple observations can share one transmission. + txByID := make(map[int]*StoreTx, len(obsSnapshot)) + for _, obs := range obsSnapshot { + if _, ok := txByID[obs.TransmissionID]; !ok { + txByID[obs.TransmissionID] = s.store.byTxID[obs.TransmissionID] + } + } s.store.mu.RUnlock() filtered := make([]*StoreObs, 0, len(obsSnapshot)) for _, obs := range obsSnapshot { @@ -2865,10 +2877,10 @@ func (s *Server) handleObserverAnalytics(w http.ResponseWriter, r *http.Request) writeJSON(w, ObserverAnalyticsResponse{ Timeline: buildTimeline(filtered, days), - PacketTypes: buildPacketTypes(s.store, filtered), - NodesTimeline: buildNodesTimeline(s.store, filtered, days), + PacketTypes: buildPacketTypes(filtered, txByID), + NodesTimeline: buildNodesTimeline(filtered, days, txByID), SnrDistribution: buildSnrDistribution(filtered), - RecentPackets: buildRecentPackets(s.store, filtered, 20), + RecentPackets: buildRecentPackets(s.store, filtered, 20, txByID), }) } diff --git a/cmd/server/routes_test.go b/cmd/server/routes_test.go index 8050ce0f..b4b8ee0c 100644 --- a/cmd/server/routes_test.go +++ b/cmd/server/routes_test.go @@ -9,6 +9,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "testing" "time" @@ -1031,9 +1032,44 @@ func TestObserverAnalytics(t *testing.T) { if body["recentPackets"] == nil { t.Error("expected recentPackets") } - if recent, ok := body["recentPackets"].([]interface{}); !ok || len(recent) == 0 { + recent, ok := body["recentPackets"].([]interface{}) + if !ok || len(recent) == 0 { t.Errorf("expected non-empty recentPackets, got %v", body["recentPackets"]) } + + // #1827: packetTypes must still reflect the real per-transmission + // payload_type after switching the hot loop from enrichObs() to a + // direct s.store.byTxID read. seedTestData gives obs1 three + // observations: tx1 (payload_type=4), tx2 (payload_type=5), tx3 + // (payload_type=4) — i.e. counts {"4":2, "5":1}. + pt, ok := body["packetTypes"].(map[string]interface{}) + if !ok { + t.Fatalf("expected packetTypes to be an object, got %T", body["packetTypes"]) + } + if pt["4"] != float64(2) { + t.Errorf("expected packetTypes[4]=2, got %v", pt["4"]) + } + if pt["5"] != float64(1) { + t.Errorf("expected packetTypes[5]=1, got %v", pt["5"]) + } + + // recentPackets is still built via the unchanged enrichObs() path + // and should retain resolved_path for observations that have one + // (tx1's first observation, resolved via obs1). + foundResolved := false + for _, rp := range recent { + m, ok := rp.(map[string]interface{}) + if !ok { + continue + } + if m["resolved_path"] != nil { + foundResolved = true + break + } + } + if !foundResolved { + t.Error("expected at least one recentPackets entry to carry resolved_path") + } }) t.Run("custom days", func(t *testing.T) { @@ -1057,6 +1093,67 @@ func TestObserverAnalytics(t *testing.T) { }) } +// TestObserverAnalytics_ConcurrentByTxIDMutation_1830 is the -race +// regression requested in #1830: handleObserverAnalytics reads +// s.store.byTxID (directly, and via enrichObsWithTx) for every observation +// in the window. Before the #1830 fix those reads happened after the +// snapshot RLock was released, racing with concurrent ingest/eviction +// writers to that same map. This test hammers the endpoint concurrently +// with goroutines that mutate byTxID exactly like ingest/eviction would, +// and must pass under `go test -race`. +func TestObserverAnalytics_ConcurrentByTxIDMutation_1830(t *testing.T) { + srv, router := setupTestServer(t) + + stop := make(chan struct{}) + var writers sync.WaitGroup + var readers sync.WaitGroup + + // Writer goroutines: mutate byTxID under s.store.mu, like ingest + // (adds) and eviction (deletes) do in production. They run until + // stop is closed, which happens once all readers are done below. + for w := 0; w < 4; w++ { + writers.Add(1) + go func(base int) { + defer writers.Done() + n := 0 + for { + select { + case <-stop: + return + default: + } + srv.store.mu.Lock() + id := base*100000 + n + srv.store.byTxID[id] = &StoreTx{ID: id, PayloadType: intPtr(4)} + delete(srv.store.byTxID, id-1) + srv.store.mu.Unlock() + n++ + } + }(w) + } + + // Reader goroutines: repeatedly hit the endpoint under test. + for r := 0; r < 8; r++ { + readers.Add(1) + go func() { + defer readers.Done() + for i := 0; i < 50; i++ { + req := httptest.NewRequest("GET", "/api/observers/obs1/analytics", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + if w.Code != 200 { + t.Errorf("expected 200, got %d", w.Code) + return + } + } + }() + } + + readers.Wait() + close(stop) + writers.Wait() +} + func TestChannelMessages(t *testing.T) { _, router := setupTestServer(t) req := httptest.NewRequest("GET", "/api/channels/%23test/messages", nil) diff --git a/cmd/server/store.go b/cmd/server/store.go index fd33b7d1..245fadbc 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -3747,9 +3747,25 @@ func (s *PacketStore) computeNodeHomeRegions() map[string]string { } // enrichObs returns a map with observation fields + transmission fields. +// Looks up the transmission in s.byTxID itself — safe only when the caller +// already holds s.mu (directly, or via a defer'd RLock spanning the call). +// Callers that snapshot observations under RLock and then release it before +// iterating (e.g. handleObserverAnalytics, #1830) must use enrichObsWithTx +// with a tx pointer resolved during that same snapshot instead. func (s *PacketStore) enrichObs(obs *StoreObs) map[string]interface{} { - tx := s.byTxID[obs.TransmissionID] + return s.enrichObsWithTx(obs, s.byTxID[obs.TransmissionID]) +} +// enrichObsWithTx is enrichObs with the transmission pointer already +// resolved by the caller, instead of looking it up in s.byTxID here. #1830: +// s.byTxID is guarded by s.mu (writes from ingest/eviction); reading it +// without holding at least RLock races with those writers — Go maps can +// panic with "concurrent map read and map write" during a rehash, not just +// fail under -race. Callers that need to read byTxID after releasing their +// RLock (to keep JSON decode / enrichment off the hot lock, per #1481) +// should resolve the *StoreTx for each observation during their RLock-held +// snapshot and pass it in here. +func (s *PacketStore) enrichObsWithTx(obs *StoreObs, tx *StoreTx) map[string]interface{} { m := map[string]interface{}{ "id": obs.ID, "timestamp": strOrNil(obs.Timestamp),