mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-15 20:29:56 +00:00
## Problem `spTxIndex` kept every 2–8 hop run of every stored path as a map key, with one transmission pointer per occurrence. On a busy regional network's 7-day window (59k transmissions) that's about 0.9–1.07 M keys and 130–160 MB of live heap, roughly a quarter of the store. It was also rebuilt at the end of every startup, which is when memory peaks. Two smaller issues came with it: - Eviction never decremented `spTotalPaths`, so the percentages were divided by paths that were no longer stored. - Route Patterns took 1.2–3.6 s uncached, because the fast path still split, resolved and re-joined about a million raw keys per request. ## Change - **Count on request.** Subpaths are counted when the analytics are asked for, from the stored paths. Results stay cached for `rfCacheTTL`, as before. - **Pointer-free tallies.** `tallyNamedSubpaths` counts runs in one pointer-free `map[uint64]subpathTally`, keyed by a `maphash` of the resolved name built in a reused buffer. The GC doesn't scan the map, and counting allocates nothing but the map. - **Names only for the top entries.** Each tally records the first transmission and offset where its run appeared, so names and raw hops are built only for each group's top entries. `topSubpathTallies` picks those with a `container/heap` min-heap of `limit` entries instead of sorting every name, and the existing `rankSubpaths` formats the response, so no new response maps are added. - **Hash collisions.** If two names shared a hash their counts would merge; the odds are about 1 in 10⁷ for a few million names. - **Subpath detail.** `GetSubpathDetail` finds the run with `txsWithSubpath`, a scan of the stored paths. - **Counts unchanged:** runs of 2–8 hops, counted once per occurrence, case-folded. The tests pin this. - **Totals.** `spTotalPaths` stays, and eviction now decrements it. - **Memory estimate.** The O(path²) subpath term is dropped from the estimate. ## Perf justification (AGENTS.md rule 0) - **Complexity.** A cold Route Patterns request is O(Σ runs), about 1.5 M for this window: one pass over the stored paths, with one map lookup per run against a reused buffer. Ranking is O(tallies · log limit). Results are cached for `rfCacheTTL`, so this runs at most once per TTL per group set. - **Subpath detail** is O(stored hops), one pass with no allocation per packet. It's an on-click endpoint, not a hot path. - **Ingest and eviction** no longer touch a subpath index at all. Removal used to be a linear search of every key's transmission list. **Measured on a copy of that window**, upstream master vs this branch, same queries: | | master | this PR | |---|---|---| | peak RSS (`GOMEMLIMIT=750MiB`) | 948 MB | 767 MB | | in-use heap after load (750MiB) | 592 MB | 456 MB | | `/api/analytics/subpaths` cold (750MiB) | 3.56 s | 0.64 s | | `/api/analytics/subpaths-bulk` cold (750MiB) | 3.33 s | 0.70 s | | `/api/analytics/subpaths` cold (1GiB) | 1.18 s | 0.61 s | | `/api/analytics/subpath-detail` | 0.6 ms | 8 ms | | cached responses | 0.6 ms | 0.5 ms | At 750 MiB, master's request time is mostly GC pressure from the index. On the production-shaped fixture that `subpath_index_memory_test.go` builds (41k transmissions): | | master | this PR | |---|---|---| | retained by the store for subpaths | 197 MB | 0 MB | | allocated per bulk request | 336 MB | 111 MB | This has also run on a staging instance with live MQTT ingest for about nine hours, with Route Patterns answering in about 1 s. ## Tests - `subpaths_on_demand_test.go`: - per-occurrence, case-folded counts; - the 8-hop cap; - bulk matches single; - detail counts, including repeats, case and the length limits; - eviction drops evicted paths from the totals. - `TestSubpathIndexRetainedHeap`: the bound is now 5 MB (it was 225 MB). - `TestSubpathRequestAllocations`: 150 MB per request. - Each change has its red commit first, failing on assertions. - Tests that reached into `spTxIndex` now assert through `txsWithSubpath` and the public counts. - `cd cmd/server && go test ./...` passes. - SonarQube: quality gate passed, no issues on added lines. **Overlap with #2181:** both edit the same memory-estimate comment block in `store.go`. Whichever lands second needs a trivial rebase there; there's no code dependency. Backend only: no UI, API shape or config change. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01KaR4xu8gVykQBkH8yG8m6H --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
199 lines
6.6 KiB
Go
199 lines
6.6 KiB
Go
package main
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// TestEstimateStoreTxBytes_ReasonableValues verifies the estimate function
|
|
// returns reasonable values for different packet sizes.
|
|
func TestEstimateStoreTxBytes_ReasonableValues(t *testing.T) {
|
|
tx := &StoreTx{
|
|
Hash: "abcdef1234567890",
|
|
RawHex: "deadbeef",
|
|
DecodedJSON: `{"type":"GRP_TXT"}`,
|
|
PathJSON: `["hop1","hop2","hop3"]`,
|
|
parsedPath: []string{"hop1", "hop2", "hop3"},
|
|
pathParsed: true,
|
|
}
|
|
got := estimateStoreTxBytes(tx)
|
|
|
|
// Should be at least base (384) + maps (200) + indexes + path/subpath costs
|
|
if got < 700 {
|
|
t.Errorf("estimate too low for 3-hop tx: %d", got)
|
|
}
|
|
if got > 5000 {
|
|
t.Errorf("estimate unreasonably high for 3-hop tx: %d", got)
|
|
}
|
|
}
|
|
|
|
// TestEstimateStoreTxBytes_HopsAreLinear verifies that a longer path costs
|
|
// one byPathHop entry per hop. Subpaths are counted per request, so the
|
|
// store no longer pays O(path²) for them.
|
|
func TestEstimateStoreTxBytes_HopsAreLinear(t *testing.T) {
|
|
tx2 := &StoreTx{
|
|
Hash: "aabb",
|
|
parsedPath: []string{"a", "b"},
|
|
pathParsed: true,
|
|
}
|
|
tx10 := &StoreTx{
|
|
Hash: "aabb",
|
|
parsedPath: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"},
|
|
pathParsed: true,
|
|
}
|
|
est2 := estimateStoreTxBytes(tx2)
|
|
est10 := estimateStoreTxBytes(tx10)
|
|
if want := est2 + 8*perPathHopBytes; est10 != want {
|
|
t.Errorf("10-hop = %d, want %d (2-hop %d + 8 hops x %d)", est10, want, est2, perPathHopBytes)
|
|
}
|
|
}
|
|
|
|
// TestEstimateStoreObsBytes_AfterRefactor verifies that after #800 refactor,
|
|
// observations no longer have ResolvedPath overhead in their estimate.
|
|
func TestEstimateStoreObsBytes_AfterRefactor(t *testing.T) {
|
|
obs := &StoreObs{
|
|
ObserverID: "obs1",
|
|
PathJSON: `["a","b"]`,
|
|
}
|
|
|
|
est := estimateStoreObsBytes(obs)
|
|
if est <= 0 {
|
|
t.Errorf("estimate should be positive, got %d", est)
|
|
}
|
|
// After #800, all obs estimates should be the same (no RP field variation)
|
|
obs2 := &StoreObs{
|
|
ObserverID: "obs1",
|
|
PathJSON: `["a","b"]`,
|
|
}
|
|
est2 := estimateStoreObsBytes(obs2)
|
|
if est != est2 {
|
|
t.Errorf("estimates should be equal after #800 (no RP field), got %d vs %d", est, est2)
|
|
}
|
|
}
|
|
|
|
// TestEstimateStoreObsBytes_ManyObservations verifies that 15 observations
|
|
// estimate significantly more than 1.
|
|
func TestEstimateStoreObsBytes_ManyObservations(t *testing.T) {
|
|
est1 := estimateStoreObsBytes(&StoreObs{ObserverID: "a", PathJSON: `["x"]`})
|
|
est15 := int64(0)
|
|
for i := 0; i < 15; i++ {
|
|
est15 += estimateStoreObsBytes(&StoreObs{ObserverID: "a", PathJSON: `["x"]`})
|
|
}
|
|
if est15 <= est1*10 {
|
|
t.Errorf("15 obs total (%d) should be >10x single obs (%d)", est15, est1)
|
|
}
|
|
}
|
|
|
|
// TestTrackedBytesMatchesSumAfterInsert verifies that trackedBytes equals the
|
|
// sum of individual estimates after inserting packets via makeTestStore.
|
|
func TestTrackedBytesMatchesSumAfterInsert(t *testing.T) {
|
|
store := makeTestStore(20, time.Now().Add(-2*time.Hour), 5)
|
|
|
|
// Manually compute trackedBytes as sum of estimates
|
|
var expectedSum int64
|
|
for _, tx := range store.packets {
|
|
expectedSum += estimateStoreTxBytes(tx)
|
|
for _, obs := range tx.Observations {
|
|
expectedSum += estimateStoreObsBytes(obs)
|
|
}
|
|
}
|
|
|
|
if store.trackedBytes != expectedSum {
|
|
t.Errorf("trackedBytes=%d, expected sum=%d", store.trackedBytes, expectedSum)
|
|
}
|
|
}
|
|
|
|
// TestEvictionTriggersWithImprovedEstimates verifies that eviction triggers
|
|
// at the right point with the improved (higher) estimates.
|
|
func TestEvictionTriggersWithImprovedEstimates(t *testing.T) {
|
|
store := makeTestStore(100, time.Now().Add(-10*time.Hour), 5)
|
|
|
|
// trackedBytes for 100 packets is small — artificially set maxMemoryMB
|
|
// so highWatermark is just below trackedBytes to trigger eviction.
|
|
highWatermarkBytes := store.trackedBytes - 1000
|
|
if highWatermarkBytes < 1 {
|
|
highWatermarkBytes = 1
|
|
}
|
|
// maxMemoryMB * 1048576 = highWatermark, so maxMemoryMB = ceil(highWatermarkBytes / 1048576)
|
|
// But that'll be 0 for small values. Instead, directly set trackedBytes high.
|
|
store.trackedBytes = 6 * 1048576 // 6MB
|
|
store.maxMemoryMB = 3 // 3MB limit
|
|
|
|
beforeCount := len(store.packets)
|
|
store.RunEviction()
|
|
afterCount := len(store.packets)
|
|
|
|
if afterCount >= beforeCount {
|
|
t.Errorf("expected eviction to remove packets: before=%d, after=%d, trackedBytes=%d, maxMB=%d",
|
|
beforeCount, afterCount, store.trackedBytes, store.maxMemoryMB)
|
|
}
|
|
// trackedBytes should have decreased
|
|
if store.trackedBytes >= 6*1048576 {
|
|
t.Errorf("trackedBytes should have decreased after eviction")
|
|
}
|
|
}
|
|
|
|
// BenchmarkEstimateStoreTxBytes verifies the estimate function is fast.
|
|
func BenchmarkEstimateStoreTxBytes(b *testing.B) {
|
|
tx := &StoreTx{
|
|
Hash: "abcdef1234567890",
|
|
RawHex: "deadbeefdeadbeef",
|
|
DecodedJSON: `{"type":"GRP_TXT","payload":"hello"}`,
|
|
PathJSON: `["hop1","hop2","hop3","hop4","hop5"]`,
|
|
parsedPath: []string{"hop1", "hop2", "hop3", "hop4", "hop5"},
|
|
pathParsed: true,
|
|
}
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
estimateStoreTxBytes(tx)
|
|
}
|
|
}
|
|
|
|
// BenchmarkEstimateStoreObsBytes verifies the obs estimate function is fast.
|
|
func BenchmarkEstimateStoreObsBytes(b *testing.B) {
|
|
obs := &StoreObs{
|
|
ObserverID: "observer1234",
|
|
PathJSON: `["a","b","c"]`,
|
|
}
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
estimateStoreObsBytes(obs)
|
|
}
|
|
}
|
|
|
|
// TestTrackedBytes_PathKnownAfterCreate reproduces the load/ingest order: a tx
|
|
// is charged when it is created, before its observations set the path. Once
|
|
// pickBestObservation has run, recharging must pick up the path costs, and
|
|
// eviction must subtract exactly what was charged.
|
|
func TestTrackedBytes_PathKnownAfterCreate(t *testing.T) {
|
|
store := makeTestStore(0, time.Now().UTC(), 0)
|
|
store.retentionHours = 1
|
|
|
|
tx := &StoreTx{ID: 1, Hash: "aabb", FirstSeen: time.Now().UTC().Add(-2 * time.Hour).Format(time.RFC3339)}
|
|
store.trackedBytes += rechargeTx(tx)
|
|
bare := store.trackedBytes
|
|
|
|
obs := &StoreObs{ID: 1, TransmissionID: 1, ObserverID: "o1", PathJSON: `["a1","b2","c3","d4","e5","f6"]`}
|
|
tx.Observations = append(tx.Observations, obs)
|
|
store.trackedBytes += estimateStoreObsBytes(obs)
|
|
pickBestObservation(tx)
|
|
store.trackedBytes += rechargeTx(tx)
|
|
store.packets = append(store.packets, tx)
|
|
store.byHash[tx.Hash] = tx
|
|
store.byTxID[tx.ID] = tx
|
|
store.byObsID[obs.ID] = obs
|
|
|
|
if want := estimateStoreTxBytes(tx) + estimateStoreObsBytes(obs); store.trackedBytes != want {
|
|
t.Fatalf("trackedBytes = %d, want %d (path costs missing?)", store.trackedBytes, want)
|
|
}
|
|
if tx.accountedBytes <= bare {
|
|
t.Fatalf("recharge did not add the path costs: %d <= %d", tx.accountedBytes, bare)
|
|
}
|
|
if n := store.EvictStale(); n != 1 {
|
|
t.Fatalf("evicted %d, want 1", n)
|
|
}
|
|
if store.trackedBytes != 0 {
|
|
t.Fatalf("trackedBytes after evicting everything = %d, want 0", store.trackedBytes)
|
|
}
|
|
}
|