perf(#1256): wire /api/analytics/roles into steady-state recomputer (#1259)

RED commit: `0190466d` — failing CI:
https://github.com/Kpa-clawbot/CoreScope/actions (will populate after PR
creation)

## Problem
On staging (commit `d69d9fb`, 78k tx, 2.3M obs), `curl
http://localhost/api/analytics/roles` times out at 60s with 0 bytes —
the Roles tab is unusable. Issue #1256.

PR #1248's steady-state recomputer fan-out (topology / rf / distance /
channels / hash-collisions / hash-sizes) **didn't include roles**. The
legacy handler:

1. Holds `s.mu.RLock` for the entire compute.
2. Calls `GetFleetClockSkew()`, which drives `clockSkew.Recompute(s)`
over all ADVERT transmissions — O(78k) per request.
3. Concurrent ingest writers compound the latency through
writer-starvation.

Result: every request hits the cold path; the response never comes back
inside the 60 s HTTP budget.

## Fix
Add `roles` as the 7th endpoint in the recomputer fan-out — same pattern
as #1248:

- `PacketStore.recompRoles` slot, registered in
`StartAnalyticsRecomputers` with default 5-min interval.
- `PacketStore.GetAnalyticsRoles()` → atomic-pointer load from the
snapshot (sub-ms), with a `computeAnalyticsRoles()` fallback only for
the brief startup window before the initial sync compute completes.
- Handler is now a thin wrapper — no lock-held work on the request path.
- New optional `roles` key under `analytics.recomputeIntervalSeconds` in
config; `config.example.json` and `_comment_analytics` updated.

## Latency (unit-scope benchmark)
- Worst-of-50 handler latency: **<100 ms** (test budget; well under the
2 s p99 acceptance).
- Compute itself is bounded by the existing 5-min recompute window — it
runs once in the background, never on the request path.

## Tests
- RED `0190466d`: asserts `recompRoles` is registered and the handler
returns under the latency budget. Fails on master with `recompRoles not
registered`.
- GREEN `d7784f76`: registers the recomputer + snapshot accessor — both
tests pass.

Fixes #1256

---------

Co-authored-by: openclaw-bot <bot@openclaw.local>

cmd/server/analytics_recomputer.go

@@ -152,6 +152,7 @@ type AnalyticsRecomputeIntervals struct {
 	Channels       time.Duration
 	HashCollisions time.Duration
 	HashSizes      time.Duration
+	Roles          time.Duration
 }
 
 func pickInterval(override, def time.Duration) time.Duration {
@@ -219,9 +220,14 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o
 		"hash-sizes", pickInterval(ov.HashSizes, defaultInterval),
 		func() interface{} { return s.computeAnalyticsHashSizesWithCapability("") },
 	)
+	s.recompRoles = newAnalyticsRecomputer(
+		"roles", pickInterval(ov.Roles, defaultInterval),
+		func() interface{} { return s.computeAnalyticsRoles() },
+	)
 	all := []*analyticsRecomputer{
 		s.recompTopology, s.recompRF, s.recompDistance,
 		s.recompChannels, s.recompHashCollisions, s.recompHashSizes,
+		s.recompRoles,
 	}
 	s.analyticsRecomputerMu.Unlock()
 

cmd/server/config.go

@@ -483,7 +483,7 @@ func (c *Config) IsObserverBlacklisted(id string) bool {
 // data slowly." Lower values give fresher data at higher CPU cost.
 //
 // RecomputeIntervalSeconds keys (all optional):
-//   topology, rf, distance, channels, hashCollisions, hashSizes
+//   topology, rf, distance, channels, hashCollisions, hashSizes, roles
 type AnalyticsConfig struct {
 	DefaultIntervalSeconds    int            `json:"defaultIntervalSeconds,omitempty"`
 	RecomputeIntervalSeconds  map[string]int `json:"recomputeIntervalSeconds,omitempty"`
@@ -519,5 +519,6 @@ func (c *Config) AnalyticsRecomputeIntervals() AnalyticsRecomputeIntervals {
 	out.Channels = get("channels")
 	out.HashCollisions = get("hashCollisions")
 	out.HashSizes = get("hashSizes")
+	out.Roles = get("roles")
 	return out
 }

cmd/server/role_analytics.go

@@ -111,23 +111,56 @@ func computeRoleAnalytics(nodesByPubkey map[string]string, skewByPubkey map[stri
 	return resp
 }
 
-// handleAnalyticsRoles serves /api/analytics/roles.
+// handleAnalyticsRoles serves /api/analytics/roles. Reads from the
+// steady-state recomputer snapshot (issue #1256) so the request never
+// holds s.mu.RLock for a full clock-skew recompute over the advert
+// transmissions — that path hung >60s on staging with 78k tx.
 func (s *Server) handleAnalyticsRoles(w http.ResponseWriter, r *http.Request) {
 	if s.store == nil {
 		writeJSON(w, RoleAnalyticsResponse{Roles: []RoleStats{}})
 		return
 	}
-	nodes, _ := s.store.getCachedNodesAndPM()
+	writeJSON(w, s.store.GetAnalyticsRoles())
+}
+
+// GetAnalyticsRoles returns the role-distribution analytics, preferring
+// the steady-state recomputer snapshot (issue #1256). Falls back to an
+// on-request compute path if the recomputer is not yet running (e.g.
+// during the brief startup window before the initial compute completes
+// — Start runs it synchronously, so this fallback is effectively only
+// hit in tests that skip the recomputer entirely).
+func (s *PacketStore) GetAnalyticsRoles() RoleAnalyticsResponse {
+	s.analyticsRecomputerMu.RLock()
+	rc := s.recompRoles
+	s.analyticsRecomputerMu.RUnlock()
+	if rc != nil {
+		if v := rc.Load(); v != nil {
+			if r, ok := v.(RoleAnalyticsResponse); ok {
+				s.cacheMu.Lock()
+				s.cacheHits++
+				s.cacheMu.Unlock()
+				return r
+			}
+		}
+	}
+	return s.computeAnalyticsRoles()
+}
+
+// computeAnalyticsRoles runs the actual role aggregation. Used by the
+// background recomputer (issue #1256) and as a fallback for callers
+// arriving before the snapshot is populated.
+func (s *PacketStore) computeAnalyticsRoles() RoleAnalyticsResponse {
+	nodes, _ := s.getCachedNodesAndPM()
 	roles := make(map[string]string, len(nodes))
 	for _, n := range nodes {
 		roles[n.PublicKey] = n.Role
 	}
 	skewMap := make(map[string]*NodeClockSkew)
-	for _, cs := range s.store.GetFleetClockSkew() {
+	for _, cs := range s.GetFleetClockSkew() {
 		if cs == nil {
 			continue
 		}
 		skewMap[cs.Pubkey] = cs
 	}
-	writeJSON(w, computeRoleAnalytics(roles, skewMap))
+	return computeRoleAnalytics(roles, skewMap)
 }

cmd/server/role_analytics_recompute_test.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+// TestRolesAnalyticsRecomputerRegistered asserts that the
+// /api/analytics/roles endpoint is backed by the steady-state
+// analytics recomputer (issue #1256). On master, roles was
+// NOT wired into StartAnalyticsRecomputers — every request
+// holds s.mu.RLock for the whole compute and triggers a fleet
+// clock-skew recompute over 78k transmissions, hanging >60s.
+//
+// Post-fix: after StartAnalyticsRecomputers, the store exposes
+// a recomputer for roles whose Load() returns a populated
+// RoleAnalyticsResponse (initial sync compute), and the
+// PacketStore.GetAnalyticsRoles() accessor returns from the
+// snapshot in sub-millisecond time.
+func TestRolesAnalyticsRecomputerRegistered(t *testing.T) {
+	db := setupTestDB(t)
+	defer db.Close()
+	store := NewPacketStore(db, nil)
+
+	stop := store.StartAnalyticsRecomputers(50 * time.Millisecond)
+	defer stop()
+
+	// Give the initial synchronous compute a beat to populate.
+	time.Sleep(100 * time.Millisecond)
+
+	store.analyticsRecomputerMu.RLock()
+	rc := store.recompRoles
+	store.analyticsRecomputerMu.RUnlock()
+	if rc == nil {
+		t.Fatalf("recompRoles not registered after StartAnalyticsRecomputers (issue #1256 not fixed)")
+	}
+	v := rc.Load()
+	if v == nil {
+		t.Fatalf("recompRoles snapshot is nil after initial compute")
+	}
+	if _, ok := v.(RoleAnalyticsResponse); !ok {
+		t.Fatalf("recompRoles snapshot type = %T, want RoleAnalyticsResponse", v)
+	}
+
+	// Accessor must hit the snapshot path.
+	t0 := time.Now()
+	resp := store.GetAnalyticsRoles()
+	dt := time.Since(t0)
+	if dt > 5*time.Millisecond {
+		t.Errorf("GetAnalyticsRoles latency = %v, want <5ms (snapshot path)", dt)
+	}
+	// Just confirm we got the response shape (empty store → empty roles).
+	_ = resp
+}
+
+// TestRolesHandlerUsesRecomputer is a HTTP-level guard that the
+// /api/analytics/roles handler returns from the recomputer snapshot
+// quickly even when no clock skew engine state has been primed (the
+// hang on staging was: every call drove a full clockSkew.Recompute
+// on 78k adverts). With recomputer wired, the handler is an atomic
+// pointer load + JSON encode.
+func TestRolesHandlerSnapshotLatency(t *testing.T) {
+	db := setupTestDB(t)
+	defer db.Close()
+	store := NewPacketStore(db, nil)
+	stop := store.StartAnalyticsRecomputers(50 * time.Millisecond)
+	defer stop()
+	time.Sleep(100 * time.Millisecond)
+
+	s := &Server{store: store}
+
+	// p99 over 50 reads must be well under 2 s (issue acceptance).
+	worst := time.Duration(0)
+	for i := 0; i < 50; i++ {
+		rr := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodGet, "/api/analytics/roles", nil)
+		t0 := time.Now()
+		s.handleAnalyticsRoles(rr, req)
+		dt := time.Since(t0)
+		if dt > worst {
+			worst = dt
+		}
+		if rr.Code != http.StatusOK {
+			t.Fatalf("status = %d, want 200", rr.Code)
+		}
+		var out RoleAnalyticsResponse
+		if err := json.Unmarshal(rr.Body.Bytes(), &out); err != nil {
+			t.Fatalf("invalid json: %v", err)
+		}
+	}
+	if worst > 100*time.Millisecond {
+		t.Fatalf("worst-of-50 handler latency = %v, want <100ms (recomputer snapshot)", worst)
+	}
+}

cmd/server/store.go

@@ -164,6 +164,7 @@ type PacketStore struct {
 	recompChannels        *analyticsRecomputer
 	recompHashCollisions  *analyticsRecomputer
 	recompHashSizes       *analyticsRecomputer
+	recompRoles           *analyticsRecomputer
 	cacheHits    int64
 	cacheMisses  int64
 	// Rate-limited invalidation (fixes #533: caches cleared faster than hit)

config.example.json

@@ -258,8 +258,9 @@
       "distance": 300,
       "channels": 300,
       "hashCollisions": 300,
-      "hashSizes": 300
+      "hashSizes": 300,
+      "roles": 300
     }
   },
-  "_comment_analytics": "Issue #1240. Each analytics endpoint (topology, rf, distance, channels, hashCollisions, hashSizes) is recomputed in the background on the configured interval and served from an atomic-pointer cache. Reads never block on compute. Default 300s (5 min) per endpoint reflects the operator principle: serving slightly stale data quickly beats real-time data slowly. Lower values = fresher data at higher CPU cost. Only the default query (no region/window) is precomputed; region- and window-filtered requests fall back to the legacy on-request compute + 60s TTL cache."
+  "_comment_analytics": "Issue #1240 + #1256. Each analytics endpoint (topology, rf, distance, channels, hashCollisions, hashSizes, roles) is recomputed in the background on the configured interval and served from an atomic-pointer cache. Reads never block on compute. Default 300s (5 min) per endpoint reflects the operator principle: serving slightly stale data quickly beats real-time data slowly. Lower values = fresher data at higher CPU cost. Only the default query (no region/window) is precomputed; region- and window-filtered requests fall back to the legacy on-request compute + 60s TTL cache."
 }