From 4cd84452338c1abd4fbd8654e7a72aed50a4164b Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Mon, 18 May 2026 12:27:44 -0700 Subject: [PATCH] perf(#1265): wire /api/observers/clock-skew + /api/nodes/clock-skew into analytics recomputer (#1266) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RED: 97f49a0c · CI: https://github.com/Kpa-clawbot/CoreScope/actions/runs/26046530920 Fixes #1265. ## Problem On staging two clock-skew endpoints serve compute-on-request: - `/api/observers/clock-skew` — 3.3s - `/api/nodes/clock-skew` — 8.9s Both drive a full `clockSkew.Recompute` over 100k+ adverts while holding `s.mu.RLock`, blocking under concurrent reader load. ## Fix Wire both endpoints into the established `analytics_recomputer.go` pattern (PRs #1248 / #1259 / #1263). Two new slots: - `recompObserversClockSkew` — wraps `computeObserverCalibrations()` - `recompNodesClockSkew` — wraps `computeFleetClockSkew()` Accessors `GetObserverCalibrations` / `GetFleetClockSkew` now prefer the atomic-pointer snapshot; on-request compute is fallback-only for the brief window before initial sync compute lands (and for tests that skip the recomputer). Default interval **300s**, overridable via: ```json "analytics": { "recomputeIntervalSeconds": { "observersClockSkew": 300, "nodesClockSkew": 300 } } ``` `config.example.json` + the `_comment_analytics` doc updated. ## TDD - RED `97f49a0c` — `TestClockSkewRecomputersRegistered` + `TestClockSkewHandlersSteadyStateLatency` (8 concurrent readers × 25 reqs per endpoint, p99 < 100ms gate). Fails on master: recomputer slots nil. - GREEN `19599375` — wire + accessor switch. p99 well under 5ms on the test fixture. ## Verification ``` cd cmd/server && go test ./... -count=1 # ok 42s bash ~/.openclaw/skills/pr-preflight/scripts/run-all.sh origin/master # all gates pass ``` --------- Co-authored-by: CoreScope Bot --- cmd/server/analytics_recomputer.go | 25 ++++-- cmd/server/clock_skew.go | 44 ++++++++++- cmd/server/clock_skew_recompute_test.go | 101 ++++++++++++++++++++++++ cmd/server/config.go | 4 +- cmd/server/store.go | 2 + config.example.json | 6 +- 6 files changed, 168 insertions(+), 14 deletions(-) create mode 100644 cmd/server/clock_skew_recompute_test.go diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go index 82a51128..fbc13518 100644 --- a/cmd/server/analytics_recomputer.go +++ b/cmd/server/analytics_recomputer.go @@ -146,13 +146,15 @@ func (r *analyticsRecomputer) ComputeRuns() int64 { // per-endpoint recompute interval from config.json. Zero values fall // back to the defaultInterval passed to StartAnalyticsRecomputers. type AnalyticsRecomputeIntervals struct { - Topology time.Duration - RF time.Duration - Distance time.Duration - Channels time.Duration - HashCollisions time.Duration - HashSizes time.Duration - Roles time.Duration + Topology time.Duration + RF time.Duration + Distance time.Duration + Channels time.Duration + HashCollisions time.Duration + HashSizes time.Duration + Roles time.Duration + ObserversClockSkew time.Duration + NodesClockSkew time.Duration } func pickInterval(override, def time.Duration) time.Duration { @@ -224,10 +226,19 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o "roles", pickInterval(ov.Roles, defaultInterval), func() interface{} { return s.computeAnalyticsRoles() }, ) + s.recompObserversClockSkew = newAnalyticsRecomputer( + "observers-clock-skew", pickInterval(ov.ObserversClockSkew, defaultInterval), + func() interface{} { return s.computeObserverCalibrations() }, + ) + s.recompNodesClockSkew = newAnalyticsRecomputer( + "nodes-clock-skew", pickInterval(ov.NodesClockSkew, defaultInterval), + func() interface{} { return s.computeFleetClockSkew() }, + ) all := []*analyticsRecomputer{ s.recompTopology, s.recompRF, s.recompDistance, s.recompChannels, s.recompHashCollisions, s.recompHashSizes, s.recompRoles, + s.recompObserversClockSkew, s.recompNodesClockSkew, } s.analyticsRecomputerMu.Unlock() diff --git a/cmd/server/clock_skew.go b/cmd/server/clock_skew.go index a6a8c649..4683544e 100644 --- a/cmd/server/clock_skew.go +++ b/cmd/server/clock_skew.go @@ -694,9 +694,27 @@ func (s *PacketStore) getNodeClockSkewLocked(pubkey string) *NodeClockSkew { } } -// GetFleetClockSkew returns clock skew data for all nodes that have skew data. -// Must NOT be called with s.mu held. +// GetFleetClockSkew returns clock skew data for all nodes, preferring +// the steady-state recomputer snapshot (issue #1265). Falls back to an +// on-request compute if the recomputer is not yet running. func (s *PacketStore) GetFleetClockSkew() []*NodeClockSkew { + s.analyticsRecomputerMu.RLock() + rc := s.recompNodesClockSkew + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if r, ok := v.([]*NodeClockSkew); ok { + return r + } + } + } + return s.computeFleetClockSkew() +} + +// computeFleetClockSkew is the underlying compute used by the +// recomputer and the on-request fallback. Must NOT be called with +// s.mu held. +func (s *PacketStore) computeFleetClockSkew() []*NodeClockSkew { s.mu.RLock() defer s.mu.RUnlock() @@ -707,7 +725,7 @@ func (s *PacketStore) GetFleetClockSkew() []*NodeClockSkew { nameMap[ni.PublicKey] = ni } - var results []*NodeClockSkew + var results = []*NodeClockSkew{} for pubkey := range s.byNode { cs := s.getNodeClockSkewLocked(pubkey) if cs == nil { @@ -727,8 +745,26 @@ func (s *PacketStore) GetFleetClockSkew() []*NodeClockSkew { return results } -// GetObserverCalibrations returns the current observer clock offsets. +// GetObserverCalibrations returns the current observer clock offsets, +// preferring the steady-state recomputer snapshot (issue #1265). Falls +// back to an on-request compute when the recomputer is not running. func (s *PacketStore) GetObserverCalibrations() []ObserverCalibration { + s.analyticsRecomputerMu.RLock() + rc := s.recompObserversClockSkew + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + if v := rc.Load(); v != nil { + if r, ok := v.([]ObserverCalibration); ok { + return r + } + } + } + return s.computeObserverCalibrations() +} + +// computeObserverCalibrations is the underlying compute used by the +// recomputer and on-request fallback. Must NOT be called with s.mu held. +func (s *PacketStore) computeObserverCalibrations() []ObserverCalibration { s.mu.RLock() defer s.mu.RUnlock() diff --git a/cmd/server/clock_skew_recompute_test.go b/cmd/server/clock_skew_recompute_test.go new file mode 100644 index 00000000..7cabe9fd --- /dev/null +++ b/cmd/server/clock_skew_recompute_test.go @@ -0,0 +1,101 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "sort" + "sync" + "testing" + "time" +) + +// Issue #1265: /api/observers/clock-skew (3.3s) and /api/nodes/clock-skew (8.9s) +// must be wired into the steady-state analytics recomputer so reads serve +// from an atomic-pointer snapshot in <100ms p99 under concurrent load. + +func TestClockSkewRecomputersRegistered(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) + + store.analyticsRecomputerMu.RLock() + rcObs := store.recompObserversClockSkew + rcNodes := store.recompNodesClockSkew + store.analyticsRecomputerMu.RUnlock() + + if rcObs == nil { + t.Fatalf("recompObserversClockSkew not registered after StartAnalyticsRecomputers (issue #1265 not fixed)") + } + if rcNodes == nil { + t.Fatalf("recompNodesClockSkew not registered after StartAnalyticsRecomputers (issue #1265 not fixed)") + } + if rcObs.Load() == nil { + t.Fatalf("recompObserversClockSkew snapshot is nil after initial compute") + } + if rcNodes.Load() == nil { + t.Fatalf("recompNodesClockSkew snapshot is nil after initial compute") + } +} + +func TestClockSkewHandlersSteadyStateLatency(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} + + endpoints := []struct { + name string + path string + handler http.HandlerFunc + }{ + {"observers", "/api/observers/clock-skew", s.handleObserverClockSkew}, + {"nodes", "/api/nodes/clock-skew", s.handleFleetClockSkew}, + } + + for _, ep := range endpoints { + ep := ep + t.Run(ep.name, func(t *testing.T) { + const readers = 8 + const perReader = 25 + var ( + mu sync.Mutex + samples []time.Duration + wg sync.WaitGroup + ) + wg.Add(readers) + for r := 0; r < readers; r++ { + go func() { + defer wg.Done() + for i := 0; i < perReader; i++ { + rr := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, ep.path, nil) + t0 := time.Now() + ep.handler(rr, req) + dt := time.Since(t0) + if rr.Code != http.StatusOK { + t.Errorf("%s status = %d, want 200", ep.path, rr.Code) + } + mu.Lock() + samples = append(samples, dt) + mu.Unlock() + } + }() + } + wg.Wait() + + sort.Slice(samples, func(i, j int) bool { return samples[i] < samples[j] }) + p99 := samples[int(float64(len(samples))*0.99)] + if p99 > 100*time.Millisecond { + t.Fatalf("%s p99 latency = %v over %d reqs, want <100ms (recomputer snapshot)", ep.path, p99, len(samples)) + } + }) + } +} diff --git a/cmd/server/config.go b/cmd/server/config.go index 87ffeb98..047ac041 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, roles +// topology, rf, distance, channels, hashCollisions, hashSizes, roles, observersClockSkew, nodesClockSkew type AnalyticsConfig struct { DefaultIntervalSeconds int `json:"defaultIntervalSeconds,omitempty"` RecomputeIntervalSeconds map[string]int `json:"recomputeIntervalSeconds,omitempty"` @@ -520,5 +520,7 @@ func (c *Config) AnalyticsRecomputeIntervals() AnalyticsRecomputeIntervals { out.HashCollisions = get("hashCollisions") out.HashSizes = get("hashSizes") out.Roles = get("roles") + out.ObserversClockSkew = get("observersClockSkew") + out.NodesClockSkew = get("nodesClockSkew") return out } diff --git a/cmd/server/store.go b/cmd/server/store.go index cd382844..f19676a0 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -165,6 +165,8 @@ type PacketStore struct { recompHashCollisions *analyticsRecomputer recompHashSizes *analyticsRecomputer recompRoles *analyticsRecomputer + recompObserversClockSkew *analyticsRecomputer + recompNodesClockSkew *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 187d56c2..52aa5d79 100644 --- a/config.example.json +++ b/config.example.json @@ -259,8 +259,10 @@ "channels": 300, "hashCollisions": 300, "hashSizes": 300, - "roles": 300 + "roles": 300, + "observersClockSkew": 300, + "nodesClockSkew": 300 } }, - "_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." + "_comment_analytics": "Issue #1240 + #1256 + #1265. Each analytics endpoint (topology, rf, distance, channels, hashCollisions, hashSizes, roles, observersClockSkew, nodesClockSkew) 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." }