refactor(1828): split handleObserverAnalytics into 5 helpers + byTxID fast-path (#1839)

## 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 <bot@corescope.local>
Co-authored-by: clawbot <bot@corescope>
This commit is contained in:
authored and GitHub committed 2026-07-09 19:14:44 -07:00
1 parent 9d47a4adea
commit d60188e481
3 files changed
+433 -104

No files matched your search

+207
View File
@@ -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
// consumed its slot (via `continue`) and could not be replaced by a later
// good-ts obs. Result can be <limit when bad-ts obs sit in the head of
// `filtered`. We reproduce that exact semantic here to keep output byte-
// identical.
func buildRecentPackets(store *PacketStore, filtered []*StoreObs, limit int) []map[string]interface{} {
if limit <= 0 {
return []map[string]interface{}{}
}
out := make([]map[string]interface{}, 0, limit)
for i, obs := range filtered {
if i >= limit {
break
}
if _, ok := obs.ParsedTime(); !ok {
continue
}
out = append(out, store.enrichObs(obs))
}
return out
}
+218
View File
@@ -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)
}
}
+8 -104
View File
@@ -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),
})
}