From d60188e4812ef89833bf6091d2bb5062f84193ad Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Thu, 9 Jul 2026 19:14:44 -0700 Subject: [PATCH] refactor(1828): split handleObserverAnalytics into 5 helpers + byTxID fast-path (#1839) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Phase A of #1828: extract the 5 aggregate builders in `handleObserverAnalytics` into pure helpers in a new `cmd/server/observer_analytics.go`. Handler becomes a snapshot + filter + 5 composed calls. Also adopts the `byTxID` direct-read in `buildPacketTypes` (issue body's core observation): the payload-type histogram no longer allocates a full `enrichObs` map + interface-boxed fields just to read `tx.PayloadType`. That's the ~90% perf win the triage called out. Scope is exactly Phase A per the second triage comment. Phase B (sub-endpoints, caching, SQL migration) is deferred to a follow-up. ## Byte-identical output - Timeline / NodesTimeline: same key set, same sort, same labels. - PacketTypes: same keys/counts. Both legacy (`enriched["payload_type"].(int)`) and new (`tx.PayloadType == nil` guard) skip obs whose tx is missing or `PayloadType` is `nil`. - SnrDistribution: same 2-unit floor bucketing (negative-side rounding preserved), same ascending sort. - RecentPackets: still the first 20 enriched observations (`enrichObs` kept only here, where the extra fields are actually needed). ## TDD - Red commit: `9dc62f43` — 7 unit tests fail on assertions (not build errors) against stubs. - Green commit: `8d41011d` — implementations + handler rewire. All new tests + existing `TestObserverAnalytics*` handler tests pass. ## Preflight `bash ~/.openclaw/skills/pr-preflight/scripts/run-all.sh origin/master` → clean (all 8 hard gates + 3 warnings pass). ## Non-goals - No new endpoints. - No SQL migration. - No public API signature change. - Snapshot count unchanged (still one under RLock, per #1481 P0-2). Fixes #1828. --------- Co-authored-by: fix-1828-bot Co-authored-by: clawbot --- cmd/server/observer_analytics.go | 207 ++++++++++++++++++++++++ cmd/server/observer_analytics_test.go | 218 ++++++++++++++++++++++++++ cmd/server/routes.go | 112 +------------ 3 files changed, 433 insertions(+), 104 deletions(-) create mode 100644 cmd/server/observer_analytics.go create mode 100644 cmd/server/observer_analytics_test.go diff --git a/cmd/server/observer_analytics.go b/cmd/server/observer_analytics.go new file mode 100644 index 00000000..6d5a7d52 --- /dev/null +++ b/cmd/server/observer_analytics.go @@ -0,0 +1,207 @@ +// Package main — observer analytics helpers. +// +// #1828 Phase A: extracted from cmd/server/routes.go handleObserverAnalytics +// (routes.go:2819-2953). Split the 5 aggregate builders into standalone +// functions for readability + isolated tests. No behavior change relative to +// the pre-#1828 handler; JSON output is byte-identical. +// +// The `filtered` argument is the time-window-filtered, timestamp-desc sorted +// slice of *StoreObs produced by the handler after snapshotting under RLock +// (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. +// +// Perf note (#1839 MINOR): dropping enrichObs on the histogram / nodes- +// timeline paths also eliminates N fetchResolvedPathForObs SQL calls per +// /analytics request (store.go:fetchResolvedPathForObs), not just the alloc/ +// boxing win — /analytics SQL-load drops materially. +package main + +import ( + "encoding/json" + "fmt" + "sort" + "strconv" + "time" +) + +// observerAnalyticsBucketDur returns the timeline bucket size for a given day +// range. Mirrors the constant table in the legacy handler. +func observerAnalyticsBucketDur(days int) time.Duration { + bucketDur := 24 * time.Hour + if days <= 1 { + bucketDur = time.Hour + } else if days <= 7 { + bucketDur = 4 * time.Hour + } + return bucketDur +} + +// observerAnalyticsFormatLabel returns a closure that formats a bucket time +// as a human-readable label appropriate for the day range. +func observerAnalyticsFormatLabel(days int) func(time.Time) string { + return func(t time.Time) string { + if days <= 1 { + return t.UTC().Format("15:04") + } + if days <= 7 { + return t.UTC().Format("Mon 15:04") + } + return t.UTC().Format("Jan 02") + } +} + +// buildTimelineBuckets is the shared kernel used by both buildTimeline and +// buildNodesTimeline. Sorts the count map by bucket-key and returns +// TimeBucket entries with labels formatted per days. +func buildTimelineBuckets(counts map[int64]int, days int) []TimeBucket { + formatLabel := observerAnalyticsFormatLabel(days) + keys := make([]int64, 0, len(counts)) + for k := range counts { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] }) + out := make([]TimeBucket, 0, len(keys)) + for _, k := range keys { + lbl := formatLabel(time.Unix(k, 0)) + out = append(out, TimeBucket{Label: &lbl, Count: counts[k]}) + } + return out +} + +// buildTimeline builds the packet-timeline aggregate for /analytics. +func buildTimeline(filtered []*StoreObs, days int) []TimeBucket { + bucketDur := observerAnalyticsBucketDur(days) + counts := map[int64]int{} + for _, obs := range filtered { + ts, ok := obs.ParsedTime() + if !ok { + continue + } + bucketStart := ts.UTC().Truncate(bucketDur).Unix() + counts[bucketStart]++ + } + return buildTimelineBuckets(counts, days) +} + +// buildPacketTypes builds the payload-type histogram. Uses a direct +// store.byTxID read (see #1828 triage) rather than enrichObs — the loop only +// 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 { + out := map[string]int{} + for _, obs := range filtered { + tx := store.byTxID[obs.TransmissionID] + if tx == nil || tx.PayloadType == nil { + continue + } + out[strconv.Itoa(*tx.PayloadType)]++ + } + return out +} + +// 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 { + bucketDur := observerAnalyticsBucketDur(days) + nodeBucketSets := map[int64]map[string]struct{}{} + for _, obs := range filtered { + ts, ok := obs.ParsedTime() + if !ok { + continue + } + bucketStart := ts.UTC().Truncate(bucketDur).Unix() + if nodeBucketSets[bucketStart] == nil { + nodeBucketSets[bucketStart] = map[string]struct{}{} + } + // 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 != "" { + var decoded map[string]interface{} + if json.Unmarshal([]byte(tx.DecodedJSON), &decoded) == nil { + for _, k := range []string{"pubKey", "srcHash", "destHash"} { + if v, ok := decoded[k].(string); ok && v != "" { + nodeBucketSets[bucketStart][v] = struct{}{} + } + } + } + } + for _, hop := range parsePathJSON(obs.PathJSON) { + if hop != "" { + nodeBucketSets[bucketStart][hop] = struct{}{} + } + } + } + nodeCounts := make(map[int64]int, len(nodeBucketSets)) + for k, nodes := range nodeBucketSets { + nodeCounts[k] = len(nodes) + } + return buildTimelineBuckets(nodeCounts, days) +} + +// buildSnrDistribution builds the SNR histogram (2-unit buckets, floor). +func buildSnrDistribution(filtered []*StoreObs) []SnrDistributionEntry { + snrBuckets := map[int]*SnrDistributionEntry{} + for _, obs := range filtered { + if obs.SNR == nil { + continue + } + bucket := int(*obs.SNR) / 2 * 2 + if *obs.SNR < 0 && int(*obs.SNR) != bucket { + bucket -= 2 + } + if snrBuckets[bucket] == nil { + snrBuckets[bucket] = &SnrDistributionEntry{Range: fmt.Sprintf("%d to %d", bucket, bucket+2)} + } + snrBuckets[bucket].Count++ + } + keys := make([]int, 0, len(snrBuckets)) + for k := range snrBuckets { + keys = append(keys, k) + } + sort.Ints(keys) + out := make([]SnrDistributionEntry, 0, len(keys)) + for _, k := range keys { + out = append(out, *snrBuckets[k]) + } + return out +} + +// buildRecentPackets builds the "first N enriched observations" list. This is +// the only aggregate that needs the full enrichObs map — recentPackets is a +// UI-facing payload and the extra fields matter here. +// +// Legacy parity (#1839): the pre-#1828 routes.go loop was +// +// for i, obs := range filtered { +// if _, ok := obs.ParsedTime(); !ok { continue } +// ... +// if i < limit { recentPackets = append(..., enriched) } +// } +// +// The `i` in the gate is the RAW slice index — a bad-ts obs at position k= limit { + break + } + if _, ok := obs.ParsedTime(); !ok { + continue + } + out = append(out, store.enrichObs(obs)) + } + return out +} diff --git a/cmd/server/observer_analytics_test.go b/cmd/server/observer_analytics_test.go new file mode 100644 index 00000000..ca60c5a6 --- /dev/null +++ b/cmd/server/observer_analytics_test.go @@ -0,0 +1,218 @@ +package main + +import ( + "sync" + "testing" + "time" +) + +// TestObserverAnalyticsBucketDur pins the bucket-size table copied from the +// legacy handler (#1828 Phase A). +func TestObserverAnalyticsBucketDur(t *testing.T) { + cases := []struct { + days int + want time.Duration + }{ + {1, time.Hour}, + {7, 4 * time.Hour}, + {8, 24 * time.Hour}, + {30, 24 * time.Hour}, + } + for _, c := range cases { + if got := observerAnalyticsBucketDur(c.days); got != c.want { + t.Errorf("bucketDur(%d) = %v, want %v", c.days, got, c.want) + } + } +} + +// TestObserverAnalyticsFormatLabel pins the label formatting per day range. +func TestObserverAnalyticsFormatLabel(t *testing.T) { + ts := time.Date(2026, 3, 4, 15, 30, 0, 0, time.UTC) + if got := observerAnalyticsFormatLabel(1)(ts); got != "15:30" { + t.Errorf("days=1 label = %q, want %q", got, "15:30") + } + if got := observerAnalyticsFormatLabel(7)(ts); got != "Wed 15:30" { + t.Errorf("days=7 label = %q, want %q", got, "Wed 15:30") + } + if got := observerAnalyticsFormatLabel(30)(ts); got != "Mar 04" { + t.Errorf("days=30 label = %q, want %q", got, "Mar 04") + } +} + +// buildObsForTest constructs a StoreObs with the parsed-time cache primed so +// tests don't depend on time.Parse in the helpers. +func buildObsForTest(txID int, ts time.Time, snr *float64, pathJSON string) *StoreObs { + o := &StoreObs{ + TransmissionID: txID, + Timestamp: ts.UTC().Format(time.RFC3339Nano), + SNR: snr, + PathJSON: pathJSON, + } + // Prime the cache by calling ParsedTime once. + o.ParsedTime() + return o +} + +func newStoreForAnalyticsTest() *PacketStore { + return &PacketStore{ + mu: sync.RWMutex{}, + byTxID: map[int]*StoreTx{}, + } +} + +// TestBuildPacketTypesDirectRead verifies the packet-type histogram builds +// from store.byTxID, and skips obs whose tx or tx.PayloadType is missing. +func TestBuildPacketTypesDirectRead(t *testing.T) { + store := newStoreForAnalyticsTest() + pt3, pt5 := 3, 5 + store.byTxID[1] = &StoreTx{ID: 1, PayloadType: &pt3} + store.byTxID[2] = &StoreTx{ID: 2, PayloadType: &pt5} + store.byTxID[3] = &StoreTx{ID: 3, PayloadType: nil} // no type → skip + // tx=4 not in map → skip + + now := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + filtered := []*StoreObs{ + buildObsForTest(1, now, nil, "[]"), + buildObsForTest(1, now, nil, "[]"), + buildObsForTest(2, now, nil, "[]"), + buildObsForTest(3, now, nil, "[]"), + buildObsForTest(4, now, nil, "[]"), + } + got := buildPacketTypes(store, filtered) + if got["3"] != 2 { + t.Errorf("packetTypes[3] = %d, want 2", got["3"]) + } + if got["5"] != 1 { + t.Errorf("packetTypes[5] = %d, want 1", got["5"]) + } + if _, ok := got["0"]; ok { + t.Errorf("packetTypes should not contain missing-type entries: %v", got) + } + if len(got) != 2 { + t.Errorf("packetTypes has %d keys, want 2 (got %v)", len(got), got) + } +} + +// TestBuildTimelineBuckets asserts that the timeline aggregates by bucketDur +// and returns entries sorted by bucket time. +func TestBuildTimelineBuckets(t *testing.T) { + base := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + filtered := []*StoreObs{ + buildObsForTest(1, base, nil, "[]"), + buildObsForTest(1, base.Add(30*time.Minute), nil, "[]"), + buildObsForTest(1, base.Add(2*time.Hour), nil, "[]"), + } + // days=1 → 1-hour buckets → two buckets, counts 2 and 1 + got := buildTimeline(filtered, 1) + if len(got) != 2 { + t.Fatalf("timeline entries = %d, want 2 (got %+v)", len(got), got) + } + if got[0].Count != 2 || got[1].Count != 1 { + t.Errorf("timeline counts = [%d, %d], want [2, 1]", got[0].Count, got[1].Count) + } +} + +// TestBuildSnrDistribution asserts 2-unit floor bucketing over SNR values. +func TestBuildSnrDistribution(t *testing.T) { + f := func(v float64) *float64 { return &v } + now := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + filtered := []*StoreObs{ + buildObsForTest(1, now, f(12.5), "[]"), // bucket 12 + buildObsForTest(1, now, f(13.9), "[]"), // bucket 12 + buildObsForTest(1, now, f(-1.0), "[]"), // bucket -2 + buildObsForTest(1, now, nil, "[]"), // no SNR → skip + } + got := buildSnrDistribution(filtered) + if len(got) != 2 { + t.Fatalf("snr entries = %d, want 2 (got %+v)", len(got), got) + } + // sorted ascending by bucket + if got[0].Range != "-2 to 0" || got[0].Count != 1 { + t.Errorf("snr[0] = %+v, want {Range:'-2 to 0', Count:1}", got[0]) + } + if got[1].Range != "12 to 14" || got[1].Count != 2 { + t.Errorf("snr[1] = %+v, want {Range:'12 to 14', Count:2}", got[1]) + } +} + +// TestBuildRecentPacketsLimit asserts that recentPackets returns at most +// `limit` entries taken from the head of `filtered`. +func TestBuildRecentPacketsLimit(t *testing.T) { + store := newStoreForAnalyticsTest() + pt := 3 + store.byTxID[1] = &StoreTx{ID: 1, PayloadType: &pt} + base := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + filtered := make([]*StoreObs, 0, 25) + for i := 0; i < 25; i++ { + filtered = append(filtered, buildObsForTest(1, base.Add(-time.Duration(i)*time.Minute), nil, "[]")) + } + got := buildRecentPackets(store, filtered, 20) + if len(got) != 20 { + t.Errorf("recentPackets len = %d, want 20", len(got)) + } +} + +// TestBuildRecentPacketsSkipsUnparseableTimestamp asserts obs with an +// unparseable Timestamp are dropped BEFORE the top-N slice — matching the +// legacy pre-refactor loop (routes.go pre-#1828: `if !ok { continue }` sits +// above the `i < 20` gate). Regression guard for #1839 MAJOR. +func TestBuildRecentPacketsSkipsUnparseableTimestamp(t *testing.T) { + store := newStoreForAnalyticsTest() + pt := 3 + store.byTxID[1] = &StoreTx{ID: 1, PayloadType: &pt} + base := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + + filtered := make([]*StoreObs, 0, 23) + // 3 head observations with an unparseable timestamp — legacy skipped them + // before the index-gate, so they must NOT consume slots in the top-20. + for i := 0; i < 3; i++ { + filtered = append(filtered, &StoreObs{ + TransmissionID: 1, + Timestamp: "not-a-timestamp", + PathJSON: "[]", + }) + } + // 20 good observations after them. + for i := 0; i < 20; i++ { + filtered = append(filtered, buildObsForTest(1, base.Add(-time.Duration(i)*time.Minute), nil, "[]")) + } + + got := buildRecentPackets(store, filtered, 20) + + // Legacy loop: index 0-2 skipped (bad ts), index 3-19 appended under the + // i<20 gate (17 entries), index 20-22 dropped (i>=20). Result len = 17. + if len(got) != 17 { + t.Errorf("recentPackets len = %d, want 17 (unparseable-ts obs at head must be skipped before top-N gate)", len(got)) + } + // Sanity: no entry should carry the bad Timestamp string. + for i, e := range got { + if ts, _ := e["timestamp"].(string); ts == "not-a-timestamp" { + t.Errorf("recentPackets[%d] contains unparseable-ts obs (timestamp=%q)", i, ts) + } + } +} + +// TestBuildNodesTimelineDistinct asserts that nodes-timeline counts distinct +// nodes per bucket (path hops + decoded pubKey/srcHash/destHash). +func TestBuildNodesTimelineDistinct(t *testing.T) { + store := newStoreForAnalyticsTest() + pt := 4 + // tx=1 has decoded_json with a pubKey + store.byTxID[1] = &StoreTx{ + ID: 1, + PayloadType: &pt, + DecodedJSON: `{"pubKey":"aaaa"}`, + } + base := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + filtered := []*StoreObs{ + 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) + if len(got) != 1 { + t.Fatalf("nodes timeline entries = %d, want 1 (got %+v)", len(got), got) + } + if got[0].Count != 3 { + t.Errorf("nodes timeline count = %d, want 3 (distinct: aaaa, bb, cc)", got[0].Count) + } +} diff --git a/cmd/server/routes.go b/cmd/server/routes.go index f857d654..aaa23ed3 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -2840,11 +2840,12 @@ func (s *Server) handleObserverAnalytics(w http.ResponseWriter, r *http.Request) return } + // #1828 Phase A: aggregate builders extracted into observer_analytics.go. + // Single snapshot under RLock (#1481 P0-2), then five composable helpers + // off the snapshot. No behavior change; JSON output is byte-identical. since := time.Now().Add(-time.Duration(days) * 24 * time.Hour) s.store.mu.RLock() obsList := s.store.byObserver[id] - // #1481 P0-2: snapshot pointer slice and release RLock immediately — - // don't iterate + json-decode + time-parse under the lock. obsSnapshot := make([]*StoreObs, len(obsList)) copy(obsSnapshot, obsList) s.store.mu.RUnlock() @@ -2860,109 +2861,12 @@ func (s *Server) handleObserverAnalytics(w http.ResponseWriter, r *http.Request) } sort.Slice(filtered, func(i, j int) bool { return filtered[i].Timestamp > filtered[j].Timestamp }) - bucketDur := 24 * time.Hour - if days <= 1 { - bucketDur = time.Hour - } else if days <= 7 { - bucketDur = 4 * time.Hour - } - formatLabel := func(t time.Time) string { - if days <= 1 { - return t.UTC().Format("15:04") - } - if days <= 7 { - return t.UTC().Format("Mon 15:04") - } - return t.UTC().Format("Jan 02") - } - - packetTypes := map[string]int{} - timelineCounts := map[int64]int{} - nodeBucketSets := map[int64]map[string]struct{}{} - snrBuckets := map[int]*SnrDistributionEntry{} - recentPackets := make([]map[string]interface{}, 0, 20) - - for i, obs := range filtered { - ts, ok := obs.ParsedTime() - if !ok { - continue - } - bucketStart := ts.UTC().Truncate(bucketDur).Unix() - timelineCounts[bucketStart]++ - if nodeBucketSets[bucketStart] == nil { - nodeBucketSets[bucketStart] = map[string]struct{}{} - } - - enriched := s.store.enrichObs(obs) - if pt, ok := enriched["payload_type"].(int); ok { - packetTypes[strconv.Itoa(pt)]++ - } - if decodedRaw, ok := enriched["decoded_json"].(string); ok && decodedRaw != "" { - var decoded map[string]interface{} - if json.Unmarshal([]byte(decodedRaw), &decoded) == nil { - for _, k := range []string{"pubKey", "srcHash", "destHash"} { - if v, ok := decoded[k].(string); ok && v != "" { - nodeBucketSets[bucketStart][v] = struct{}{} - } - } - } - } - for _, hop := range parsePathJSON(obs.PathJSON) { - if hop != "" { - nodeBucketSets[bucketStart][hop] = struct{}{} - } - } - if obs.SNR != nil { - bucket := int(*obs.SNR) / 2 * 2 - if *obs.SNR < 0 && int(*obs.SNR) != bucket { - bucket -= 2 - } - if snrBuckets[bucket] == nil { - snrBuckets[bucket] = &SnrDistributionEntry{Range: fmt.Sprintf("%d to %d", bucket, bucket+2)} - } - snrBuckets[bucket].Count++ - } - if i < 20 { - recentPackets = append(recentPackets, enriched) - } - } - // #1481 P0-2: RLock was released earlier after snapshotting the - // observation pointer slice; no Unlock needed here. - - buildTimeline := func(counts map[int64]int) []TimeBucket { - keys := make([]int64, 0, len(counts)) - for k := range counts { - keys = append(keys, k) - } - sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] }) - out := make([]TimeBucket, 0, len(keys)) - for _, k := range keys { - lbl := formatLabel(time.Unix(k, 0)) - out = append(out, TimeBucket{Label: &lbl, Count: counts[k]}) - } - return out - } - - nodeCounts := make(map[int64]int, len(nodeBucketSets)) - for k, nodes := range nodeBucketSets { - nodeCounts[k] = len(nodes) - } - snrKeys := make([]int, 0, len(snrBuckets)) - for k := range snrBuckets { - snrKeys = append(snrKeys, k) - } - sort.Ints(snrKeys) - snrDistribution := make([]SnrDistributionEntry, 0, len(snrKeys)) - for _, k := range snrKeys { - snrDistribution = append(snrDistribution, *snrBuckets[k]) - } - writeJSON(w, ObserverAnalyticsResponse{ - Timeline: buildTimeline(timelineCounts), - PacketTypes: packetTypes, - NodesTimeline: buildTimeline(nodeCounts), - SnrDistribution: snrDistribution, - RecentPackets: recentPackets, + Timeline: buildTimeline(filtered, days), + PacketTypes: buildPacketTypes(s.store, filtered), + NodesTimeline: buildNodesTimeline(s.store, filtered, days), + SnrDistribution: buildSnrDistribution(filtered), + RecentPackets: buildRecentPackets(s.store, filtered, 20), }) }