mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-09-26 17:47:53 +00:00
## Summary Mirrors the distance-index lazy pattern (#1011): the subpath and path-hop index builds are no longer part of `Load()`'s synchronous critical section. They now run in **two parallel background goroutines** kicked off after `s.loaded = true`, so HTTP comes up immediately even at Cascadia scale (5M observations, previously ~60s blocked on these two builds inside `Load()` under `s.mu`). Fixes #1008. ## Approach Two new `atomic.Bool` fields on `PacketStore` (`subpathReady`, `pathHopReady`) plus a one-shot broadcast channel (`indexReadyChan`) for waiters. `Load()` removes the synchronous `s.buildSubpathIndex()` / `s.buildPathHopIndex()` calls and instead kicks `s.startBackgroundIndexBuilds()` right before returning. That function spawns **two independent goroutines** (review m7), one per index. Each goroutine: 1. acquires `s.mu.Lock()` (blocks until `Load()`'s deferred Unlock fires), 2. runs its builder, releases the lock, stores its `ready = true`, 3. closes the broadcast channel if both flags are now true, 4. logs `[startup] index build complete: subpath (Xs)` (or pathHop). Analytics handlers whose entire response IS the index aggregate — `/api/analytics/subpaths`, `/api/analytics/subpaths-bulk`, `/api/analytics/subpath-detail`, `/api/nodes/{pubkey}/paths` — gate reads behind the corresponding atomic and respond with `503 Service Unavailable`, `Retry-After: 5`, body `{"error":"index loading","retryAfter":5}` until the build completes — matching the triage spec. ### Handler scope (review M2) A second class of handlers also touches these indexes — `/api/nodes`, `/api/nodes/{pubkey}`, the `GetRepeaterRelayInfoMap` / `GetRepeaterUsefulnessScoreMap` / `GetBridgeScore` enrichment helpers, and `repeater_liveness` / `repeater_usefulness`. These are **intentionally NOT 503-gated**: they expose the index via optional enrichment fields that callers already treat as "may be empty", and 503-ing the SPA bootstrap to wait for an index that only affects relay-activity badges would be a worse UX than a 30–60s window of "—" values. The rationale is documented in the package doc-comment at the top of `index_ready_1008.go`. The recomputer's synchronous prewarm path (`StartRepeaterEnrichmentRecomputer`) gates on `WaitIndexesReady(60s)` (review M1) so it never snapshots an empty `byPathHop` into `s.repeaterRelayCache`; on timeout it skips the prewarm and lets the 5-minute ticker pick up the populated index. ## Concurrency safety Each build goroutine acquires `s.mu.Lock()` before calling the existing `buildSubpathIndex()` / `buildPathHopIndex()` helpers, which replace `s.spIndex` / `s.spTxIndex` / `s.byPathHop` with freshly-allocated maps. Visibility of the populated maps to handlers that observe `Ready()==true` is established by Go 1.19+ sync/atomic acquire-release semantics: the atomic store of `true` happens-after `s.mu.Unlock()`, and the handler's atomic load synchronizes-with that store. The handler's subsequent `s.mu.RLock` serializes against concurrent ingest writers, not against the builder. The existing `main.go` boot sequence does not start ingest goroutines until after `store.Load()` returns and graph init completes, so the brief window between `Load()` returning and the two goroutines acquiring `s.mu` does not race with concurrent ingest writes. ## TDD: red → green - **Red** commit `63e79e11`: `cmd/server/index_ready_1008_test.go` adds four assertions; `cmd/server/index_ready_1008.go` adds compile-only stubs returning `true` so the tests fail on assertions, not build errors. - **Green** commit `fb1d22b0`: implements the real atomic gates, the background goroutine, and the four handler 503 branches; also updates four existing tests that read indexes directly post-`Load()` to call `store.WaitIndexesReady(5s)` first. - **Race-fix commit `b77d56eb`** (review m8 — test-infra exemption): adds `WaitIndexesReady` calls in test helpers/setup paths so the race detector no longer flags the read-after-Load() pattern in existing tests. Per AGENTS.md, race-detector flakes are observable evidence (test crashes under `-race`) and qualify for the test-infra exemption from the TDD red-commit requirement; no behavior change in production code. - **Polish round 2 — M1 red `408c7462` / green `85e82c8a`**: `TestIssue1008_M1_PrewarmWaitsForIndexes` asserts the recomputer prewarm SKIPs when indexes are not ready. Red commit adds the assertion + a stub `repeaterEnrichmentPrewarmWait` var; green commit wires `WaitIndexesReady` into the prewarm path and adds the handler-scope docs for M2. - **Polish round 2 — minor cleanups `fd089bd0`** (m3..m7): chunk-loader wires `markIndexesReadySync`, memory-model comment rewritten to cite acquire-release, sentinel deleted, polling replaced with a broadcast channel, two parallel goroutines for the builds. `TestIssue1008_m7_BothFlagsSetAfterParallelStart` covers the parallel path. ## Reproduction ``` git fetch origin fix/issue-1008 git checkout 63e79e11 # red commit cd cmd/server && go test -run TestIssue1008_ -count=1 . # FAILs git checkout fix/issue-1008 # latest green cd cmd/server && go test -run TestIssue1008 -count=1 -race . # all pass cd cmd/server && go test -count=1 -race -short ./... # full suite ok ``` ## Files changed | file | role | |---|---| | `cmd/server/store.go` | atomic.Bool fields + indexReadyChan broadcast field; remove sync build calls in Load(); kick goroutines; wire markIndexesReadySync from chunk loader | | `cmd/server/index_ready_1008.go` | ready flags, two-goroutine background builds, 503 helper, channel-based WaitIndexesReady, handler-scope docs | | `cmd/server/index_ready_1008_test.go` | red-commit contract tests + parallel-start assertion | | `cmd/server/repeater_enrich_recomputer.go` | gate prewarm on WaitIndexesReady (M1) | | `cmd/server/repeater_enrich_recomputer_1008_test.go` | M1 red+green assertions | | `cmd/server/routes.go` | 503 gate on 4 analytics handlers | | `cmd/server/routes_test.go` | setup helpers wait for ready; collision test waits | | `cmd/server/coverage_test.go` | three tests wait for ready before reading indexes | ## Out of scope - Distance index (already deferred in #1011) — untouched. - The `pickBestObservation` + `indexByNode` per-tx loop in `Load()` — kept synchronous per triage Findings (ordering-sensitive, contiguous-memory, fast). --------- Co-authored-by: bot <bot@noreply.local> Co-authored-by: openclaw-bot <bot@openclaw.local> Co-authored-by: mc-bot <mc-bot@users.noreply.github.com>
219 lines
9.0 KiB
Go
219 lines
9.0 KiB
Go
// Issue #1008: background-deferred subpath + pathHop index builds.
|
||
//
|
||
// Pattern mirrors the distance index (#1011) — but where distance is
|
||
// fully lazy (built on first request), these two indexes are kicked off
|
||
// eagerly by Load() in a background goroutine so HTTP becomes ready
|
||
// immediately while the indexes finish populating.
|
||
//
|
||
// Concurrency model:
|
||
//
|
||
// - subpathReady / pathHopReady are atomic.Bool flags written exactly
|
||
// once by the background builder (false → true) and never reset
|
||
// thereafter. Handlers read them via SubpathIndexReady() /
|
||
// PathHopIndexReady() before touching s.spIndex / s.spTxIndex /
|
||
// s.byPathHop. While a flag is false, the handler responds 503 +
|
||
// Retry-After: 5.
|
||
//
|
||
// - The builder itself acquires s.mu.Lock() and calls the existing
|
||
// buildSubpathIndex() / buildPathHopIndex() methods. Those methods
|
||
// replace s.spIndex / s.spTxIndex / s.byPathHop with freshly-
|
||
// allocated maps under the write lock. Visibility of the populated
|
||
// maps to handlers that see Ready()==true is guaranteed by Go's
|
||
// sync/atomic acquire-release semantics (formalized in Go 1.19):
|
||
// the atomic.Store(true) happens-after the s.mu.Unlock() that
|
||
// completes the build, and the handler's atomic.Load()==true
|
||
// synchronizes-with that store. The handler's subsequent s.mu.RLock
|
||
// is not what establishes visibility — it only serializes against
|
||
// concurrent ingest writers — so dropping the RLock would still be
|
||
// safe for the build's "populated map" snapshot (we keep it for
|
||
// ingest serialization).
|
||
//
|
||
// - Ingest-side incremental updates in StoreNewTransmissions /
|
||
// pruning / hash-collision paths continue to write s.spIndex /
|
||
// s.spTxIndex / s.byPathHop directly under s.mu.Lock(). Because
|
||
// the builder also runs under s.mu.Lock() and the builder
|
||
// overwrites whatever is there, the brief window between Load()
|
||
// returning and the goroutine acquiring s.mu means any
|
||
// concurrent ingest writes will be overwritten by the build —
|
||
// this matches the prior behavior where ingest could not start
|
||
// until Load() released s.mu, so in practice ingest does not
|
||
// run during the build window. Documenting this rather than
|
||
// adding a separate gate: the existing main.go boot sequence
|
||
// does not start ingest goroutines until after store.Load()
|
||
// and graph init complete.
|
||
//
|
||
// Handler scope of the ready gate (issue #1008 review M2):
|
||
//
|
||
// - HARD-GATED with 503 + Retry-After: 5 — analytics endpoints whose
|
||
// entire response is the index aggregate. Empty data would be
|
||
// visibly broken (charts, top-N tables). See routes.go:
|
||
// /api/analytics/subpaths, /api/analytics/subpaths-bulk,
|
||
// /api/analytics/subpath-detail, /api/nodes/{pubkey}/paths.
|
||
//
|
||
// - BEST-EFFORT (not gated) — endpoints where the index drives
|
||
// enrichment fields that callers already treat as optional. During
|
||
// the not-ready window these report zero counts / nil scores
|
||
// rather than 503-ing the whole list. Acceptable because:
|
||
//
|
||
// * /api/nodes and /api/nodes/{pubkey} have many other fields
|
||
// (last-seen, position, advert metadata) that callers depend
|
||
// on at startup. 503-ing the SPA bootstrap to wait for an
|
||
// index that exclusively affects "relay activity" badges
|
||
// would be a worse UX than a 30–60s window of "—" badges.
|
||
//
|
||
// * GetRepeaterRelayInfoMap / GetRepeaterUsefulnessScoreMap /
|
||
// GetBridgeScore / repeater_liveness / repeater_usefulness
|
||
// all walk s.byPathHop. During the build window they return
|
||
// empty maps or zero scores; the steady-state recomputer
|
||
// (#1262) refreshes them every 5min once indexes flip ready
|
||
// (prewarm guarded by WaitIndexesReady — see review M1).
|
||
//
|
||
// This is documented rather than gated so operators do not see
|
||
// /api/nodes 503 during routine restarts on Cascadia-scale data.
|
||
package main
|
||
|
||
import (
|
||
"log"
|
||
"net/http"
|
||
"time"
|
||
)
|
||
|
||
// writeIndexLoading503 emits the standard 503 response used by handlers
|
||
// that depend on a not-yet-built index (#1008). Body shape matches the
|
||
// triage spec: {"error":"index loading","retryAfter":5}. The Retry-After
|
||
// header is also set so well-behaved clients back off automatically.
|
||
func writeIndexLoading503(w http.ResponseWriter) {
|
||
w.Header().Set("Retry-After", "5")
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusServiceUnavailable)
|
||
_, _ = w.Write([]byte(`{"error":"index loading","retryAfter":5}`))
|
||
}
|
||
|
||
// SubpathIndexReady reports whether the subpath index build kicked off
|
||
// by Load() has completed (#1008). Until this returns true, callers
|
||
// must NOT read s.spIndex / s.spTxIndex.
|
||
func (s *PacketStore) SubpathIndexReady() bool {
|
||
return s.subpathReady.Load()
|
||
}
|
||
|
||
// PathHopIndexReady reports whether the path-hop index build kicked
|
||
// off by Load() has completed (#1008). Until this returns true,
|
||
// callers must NOT read s.byPathHop.
|
||
func (s *PacketStore) PathHopIndexReady() bool {
|
||
return s.pathHopReady.Load()
|
||
}
|
||
|
||
// indexReadyCh returns the channel that is closed when BOTH indexes
|
||
// have flipped ready. Lazily created on first access. Safe to call
|
||
// concurrently. Used by WaitIndexesReady and any future waiters that
|
||
// want event-driven semantics instead of polling.
|
||
func (s *PacketStore) indexReadyCh() <-chan struct{} {
|
||
s.indexReadyChMu.Lock()
|
||
defer s.indexReadyChMu.Unlock()
|
||
if s.indexReadyChan == nil {
|
||
s.indexReadyChan = make(chan struct{})
|
||
// If both are already ready (e.g. background chunk loader
|
||
// flipped them synchronously before any waiter showed up),
|
||
// close immediately so the channel is usable as a one-shot.
|
||
if s.subpathReady.Load() && s.pathHopReady.Load() {
|
||
close(s.indexReadyChan)
|
||
}
|
||
}
|
||
return s.indexReadyChan
|
||
}
|
||
|
||
// maybeCloseIndexReadyCh closes the ready channel iff both flags are
|
||
// set. Idempotent (a sync.Once on the channel) and safe to call from
|
||
// either builder goroutine on the green-path transitions, as well as
|
||
// from markIndexesReadySync.
|
||
func (s *PacketStore) maybeCloseIndexReadyCh() {
|
||
if !(s.subpathReady.Load() && s.pathHopReady.Load()) {
|
||
return
|
||
}
|
||
s.indexReadyChMu.Lock()
|
||
defer s.indexReadyChMu.Unlock()
|
||
if s.indexReadyChan == nil {
|
||
// Lazily allocate AND close it in one step so any future
|
||
// indexReadyCh() caller gets a pre-closed channel.
|
||
s.indexReadyChan = make(chan struct{})
|
||
close(s.indexReadyChan)
|
||
return
|
||
}
|
||
select {
|
||
case <-s.indexReadyChan:
|
||
// Already closed.
|
||
default:
|
||
close(s.indexReadyChan)
|
||
}
|
||
}
|
||
|
||
// startBackgroundIndexBuilds is called from Load() after s.loaded=true
|
||
// to populate the subpath + path-hop indexes off the critical path
|
||
// (#1008). It returns immediately; the work runs in two background
|
||
// goroutines (one per index — see review m7) that each acquire
|
||
// s.mu.Lock() independently, install their map, then set the
|
||
// corresponding atomic ready flag.
|
||
//
|
||
// At Cascadia scale (~5M observations) this previously blocked HTTP
|
||
// readiness ~60s inside Load() under s.mu. Running the two builds in
|
||
// parallel halves the pathHop-not-ready window since the two builders
|
||
// are independent of each other.
|
||
func (s *PacketStore) startBackgroundIndexBuilds() {
|
||
go func() {
|
||
t0 := time.Now()
|
||
s.mu.Lock()
|
||
s.buildSubpathIndex()
|
||
s.mu.Unlock()
|
||
// Atomic.Store happens-after s.mu.Unlock; handlers that
|
||
// observe Ready()==true synchronize-with this store.
|
||
s.subpathReady.Store(true)
|
||
s.maybeCloseIndexReadyCh()
|
||
log.Printf("[startup] index build complete: subpath (%s)",
|
||
time.Since(t0).Round(time.Millisecond))
|
||
}()
|
||
go func() {
|
||
t1 := time.Now()
|
||
s.mu.Lock()
|
||
s.buildPathHopIndex()
|
||
s.mu.Unlock()
|
||
s.pathHopReady.Store(true)
|
||
s.maybeCloseIndexReadyCh()
|
||
log.Printf("[startup] index build complete: pathHop (%s)",
|
||
time.Since(t1).Round(time.Millisecond))
|
||
}()
|
||
}
|
||
|
||
// markIndexesReadySync is the synchronous-build entry point used by
|
||
// the background chunk loader in store.go (and by tests). The chunk
|
||
// loader rebuilds both indexes under s.mu.Lock(); after the Unlock it
|
||
// calls this to flip the ready flags and close the broadcast channel
|
||
// in one shot, preserving symmetry with the goroutine path above.
|
||
func (s *PacketStore) markIndexesReadySync() {
|
||
s.subpathReady.Store(true)
|
||
s.pathHopReady.Store(true)
|
||
s.maybeCloseIndexReadyCh()
|
||
}
|
||
|
||
// WaitIndexesReady blocks until both background indexes built by
|
||
// startBackgroundIndexBuilds() report ready, or the deadline expires.
|
||
// Returns true if both flipped in time. Intended for tests that read
|
||
// s.spIndex / s.spTxIndex / s.byPathHop directly after Load(); production
|
||
// code paths gate via SubpathIndexReady() / PathHopIndexReady() and
|
||
// respond 503 + Retry-After to clients instead of blocking.
|
||
//
|
||
// Uses the indexReadyCh broadcast channel rather than polling
|
||
// (see review m6) so wake-up is immediate with no poll-interval jitter.
|
||
func (s *PacketStore) WaitIndexesReady(timeout time.Duration) bool {
|
||
if s.SubpathIndexReady() && s.PathHopIndexReady() {
|
||
return true
|
||
}
|
||
ch := s.indexReadyCh()
|
||
select {
|
||
case <-ch:
|
||
return true
|
||
case <-time.After(timeout):
|
||
return s.SubpathIndexReady() && s.PathHopIndexReady()
|
||
}
|
||
}
|
||
|