From f81ed5b3cf37cc4fa55bbb55695e36bced5832f7 Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Mon, 18 May 2026 07:36:28 -0700 Subject: [PATCH] perf(#1256): wire /api/analytics/roles into steady-state recomputer (#1259) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- cmd/server/analytics_recomputer.go | 6 ++ cmd/server/config.go | 3 +- cmd/server/role_analytics.go | 41 ++++++++- cmd/server/role_analytics_recompute_test.go | 97 +++++++++++++++++++++ cmd/server/store.go | 1 + config.example.json | 5 +- 6 files changed, 146 insertions(+), 7 deletions(-) create mode 100644 cmd/server/role_analytics_recompute_test.go diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go index 9917930d..82a51128 100644 --- a/cmd/server/analytics_recomputer.go +++ b/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() diff --git a/cmd/server/config.go b/cmd/server/config.go index e1cc8de3..87ffeb98 100644 --- a/cmd/server/config.go +++ b/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 } diff --git a/cmd/server/role_analytics.go b/cmd/server/role_analytics.go index 1519a854..6760b3e3 100644 --- a/cmd/server/role_analytics.go +++ b/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) } diff --git a/cmd/server/role_analytics_recompute_test.go b/cmd/server/role_analytics_recompute_test.go new file mode 100644 index 00000000..8fd1ab7d --- /dev/null +++ b/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) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index dd09823e..50188d24 100644 --- a/cmd/server/store.go +++ b/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) diff --git a/config.example.json b/config.example.json index f3b43c47..187d56c2 100644 --- a/config.example.json +++ b/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." }