mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-15 19:10:26 +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>
237 lines
8.4 KiB
Go
237 lines
8.4 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
var _ = time.Second // suppress unused import
|
|
|
|
// Helper to create a minimal PacketStore with GRP_TXT packets for channel analytics testing.
|
|
func newChannelTestStore(packets []*StoreTx) *PacketStore {
|
|
ps := &PacketStore{
|
|
packets: packets,
|
|
byHash: make(map[string]*StoreTx),
|
|
byTxID: make(map[int]*StoreTx),
|
|
byObsID: make(map[int]*StoreObs),
|
|
byObserver: make(map[string][]*StoreObs),
|
|
byNode: make(map[string][]*StoreTx),
|
|
byPathHop: make(map[string][]*StoreTx),
|
|
nodeHashes: make(map[string]map[string]bool),
|
|
byPayloadType: make(map[int][]*StoreTx),
|
|
rfCache: make(map[string]*cachedResult),
|
|
topoCache: make(map[string]*cachedResult),
|
|
hashCache: make(map[string]*cachedResult),
|
|
collisionCache: make(map[string]*cachedResult),
|
|
chanCache: make(map[string]*cachedResult),
|
|
distCache: make(map[string]*cachedResult),
|
|
subpathCache: make(map[string]*cachedResult),
|
|
advertPubkeys: make(map[string]int),
|
|
clockSkew: NewClockSkewEngine(),
|
|
}
|
|
ps.byPayloadType[5] = packets
|
|
return ps
|
|
}
|
|
|
|
func makeGrpTx(channelHash int, channel, text, sender string) *StoreTx {
|
|
return makeGrpTxWithStatus(channelHash, channel, text, sender, "")
|
|
}
|
|
|
|
// makeGrpTxWithStatus is like makeGrpTx but lets the caller set the ingestor's
|
|
// decryptionStatus ("decrypted" / "no_key" / "decryption_failed" / ""), which
|
|
// computeAnalyticsChannels consults to decide whether a channel name is
|
|
// trustworthy (see #1729).
|
|
func makeGrpTxWithStatus(channelHash int, channel, text, sender, decryptionStatus string) *StoreTx {
|
|
decoded := map[string]interface{}{
|
|
"type": "CHAN",
|
|
"channelHash": float64(channelHash),
|
|
"channel": channel,
|
|
"text": text,
|
|
"sender": sender,
|
|
}
|
|
if decryptionStatus != "" {
|
|
decoded["decryptionStatus"] = decryptionStatus
|
|
}
|
|
b, _ := json.Marshal(decoded)
|
|
pt := 5
|
|
return &StoreTx{
|
|
ID: 1,
|
|
DecodedJSON: string(b),
|
|
FirstSeen: "2026-05-01T12:00:00Z",
|
|
PayloadType: &pt,
|
|
}
|
|
}
|
|
|
|
// TestComputeAnalyticsChannels_MergesEncryptedAndDecrypted verifies that packets
|
|
// with the same hash byte but different decryption status merge into ONE bucket.
|
|
func TestComputeAnalyticsChannels_MergesEncryptedAndDecrypted(t *testing.T) {
|
|
// Hash 129 is the real hash for #wardriving: SHA256(SHA256("#wardriving")[:16])[0] = 129
|
|
// Some packets are decrypted (have channel name), some are not (encrypted)
|
|
packets := []*StoreTx{
|
|
makeGrpTx(129, "#wardriving", "hello", "alice"),
|
|
makeGrpTx(129, "#wardriving", "world", "bob"),
|
|
makeGrpTx(129, "", "", ""), // encrypted — no channel name
|
|
makeGrpTx(129, "", "", ""), // encrypted
|
|
}
|
|
|
|
store := newChannelTestStore(packets)
|
|
result := store.computeAnalyticsChannels("", "", TimeWindow{})
|
|
|
|
channels := result["channels"].([]map[string]interface{})
|
|
if len(channels) != 1 {
|
|
t.Fatalf("expected 1 channel bucket, got %d: %+v", len(channels), channels)
|
|
}
|
|
ch := channels[0]
|
|
if ch["name"] != "#wardriving" {
|
|
t.Errorf("expected name '#wardriving', got %q", ch["name"])
|
|
}
|
|
if ch["messages"] != 4 {
|
|
t.Errorf("expected 4 messages, got %v", ch["messages"])
|
|
}
|
|
if ch["encrypted"] != false {
|
|
t.Errorf("expected encrypted=false (some packets decrypted), got %v", ch["encrypted"])
|
|
}
|
|
}
|
|
|
|
// TestComputeAnalyticsChannels_RejectsRainbowTableMismatch verifies that a packet
|
|
// with channelHash=72 but channel="#wardriving" (mismatch) does NOT create a
|
|
// "#wardriving" bucket — it falls into "ch72" instead.
|
|
func TestComputeAnalyticsChannels_RejectsRainbowTableMismatch(t *testing.T) {
|
|
// Hash 72 is NOT the correct hash for #wardriving (which is 129).
|
|
// This simulates a rainbow-table collision/mismatch.
|
|
packets := []*StoreTx{
|
|
makeGrpTx(72, "#wardriving", "ghost", "eve"), // mismatch: hash 72 != wardriving's real hash
|
|
makeGrpTx(129, "#wardriving", "real", "alice"), // correct match
|
|
}
|
|
|
|
store := newChannelTestStore(packets)
|
|
result := store.computeAnalyticsChannels("", "", TimeWindow{})
|
|
|
|
channels := result["channels"].([]map[string]interface{})
|
|
if len(channels) != 2 {
|
|
t.Fatalf("expected 2 channel buckets, got %d: %+v", len(channels), channels)
|
|
}
|
|
|
|
// Find the buckets
|
|
var ch72, ch129 map[string]interface{}
|
|
for _, ch := range channels {
|
|
if ch["hash"] == "72" {
|
|
ch72 = ch
|
|
} else if ch["hash"] == "129" {
|
|
ch129 = ch
|
|
}
|
|
}
|
|
|
|
if ch72 == nil {
|
|
t.Fatal("expected a bucket for hash 72")
|
|
}
|
|
if ch129 == nil {
|
|
t.Fatal("expected a bucket for hash 129")
|
|
}
|
|
|
|
// ch72 should NOT be named "#wardriving" — it should be the placeholder
|
|
if ch72["name"] == "#wardriving" {
|
|
t.Errorf("hash 72 bucket should NOT be named '#wardriving' (rainbow-table mismatch rejected)")
|
|
}
|
|
if ch72["name"] != "ch72" {
|
|
t.Errorf("expected hash 72 bucket named 'ch72', got %q", ch72["name"])
|
|
}
|
|
|
|
// ch129 should be named "#wardriving"
|
|
if ch129["name"] != "#wardriving" {
|
|
t.Errorf("expected hash 129 bucket named '#wardriving', got %q", ch129["name"])
|
|
}
|
|
}
|
|
|
|
// TestChannelNameMatchesHash verifies the hash validation function.
|
|
func TestChannelNameMatchesHash(t *testing.T) {
|
|
// #wardriving hashes to 129
|
|
if !channelNameMatchesHash("#wardriving", "129") {
|
|
t.Error("expected #wardriving to match hash 129")
|
|
}
|
|
if channelNameMatchesHash("#wardriving", "72") {
|
|
t.Error("expected #wardriving to NOT match hash 72")
|
|
}
|
|
// Without leading # should also work
|
|
if !channelNameMatchesHash("wardriving", "129") {
|
|
t.Error("expected wardriving (without #) to match hash 129")
|
|
}
|
|
}
|
|
|
|
// TestIsPlaceholderName verifies placeholder detection.
|
|
func TestIsPlaceholderName(t *testing.T) {
|
|
if !isPlaceholderName("ch129") {
|
|
t.Error("ch129 should be placeholder")
|
|
}
|
|
if !isPlaceholderName("ch0") {
|
|
t.Error("ch0 should be placeholder")
|
|
}
|
|
if isPlaceholderName("#wardriving") {
|
|
t.Error("#wardriving should NOT be placeholder")
|
|
}
|
|
if isPlaceholderName("Public") {
|
|
t.Error("Public should NOT be placeholder")
|
|
}
|
|
}
|
|
|
|
// TestComputeAnalyticsChannels_PublicChannelPreserved is the regression test for
|
|
// #1729: the firmware-default "Public" channel (channel-hash byte 0x11 = 17,
|
|
// key-derived via SHA256(key)[0], NOT the hashtag scheme's 186) was being
|
|
// discarded by the #978 rainbow-table validation and rendered as
|
|
// "Encrypted (0x11)". The ingestor decrypts Public packets with the builtin
|
|
// well-known key and marks them decryptionStatus:"decrypted"; the server must
|
|
// trust that name instead of re-applying the hashtag hash check.
|
|
func TestComputeAnalyticsChannels_PublicChannelPreserved(t *testing.T) {
|
|
packets := []*StoreTx{
|
|
makeGrpTxWithStatus(17, "Public", "hello net", "alice", "decrypted"),
|
|
makeGrpTxWithStatus(17, "Public", "morning", "bob", "decrypted"),
|
|
}
|
|
|
|
store := newChannelTestStore(packets)
|
|
result := store.computeAnalyticsChannels("", "", TimeWindow{})
|
|
|
|
channels := result["channels"].([]map[string]interface{})
|
|
if len(channels) != 1 {
|
|
t.Fatalf("expected 1 channel bucket, got %d: %+v", len(channels), channels)
|
|
}
|
|
ch := channels[0]
|
|
if ch["name"] != "Public" {
|
|
t.Errorf("expected name 'Public' preserved, got %q (must not be downgraded to ch17)", ch["name"])
|
|
}
|
|
if ch["encrypted"] != false {
|
|
t.Errorf("expected encrypted=false for decrypted Public channel, got %v", ch["encrypted"])
|
|
}
|
|
if ch["hash"] != "17" {
|
|
t.Errorf("expected hash '17', got %v", ch["hash"])
|
|
}
|
|
}
|
|
|
|
// TestComputeAnalyticsChannels_UndecryptedNameStillValidated ensures the #1729
|
|
// fix does not weaken #978: when the ingestor did NOT mark the packet
|
|
// "decrypted" (no_key / decryption_failed / absent), a channel name that fails
|
|
// the hashtag hash check is still discarded to "chNN", even if text/sender
|
|
// happen to be present (e.g. a stale/foreign rainbow-table hit).
|
|
func TestComputeAnalyticsChannels_UndecryptedNameStillValidated(t *testing.T) {
|
|
// Hash 17 is NOT the hashtag hash of #Public (that is 186). Without a
|
|
// "decrypted" status, the name must be rejected.
|
|
packets := []*StoreTx{
|
|
makeGrpTxWithStatus(17, "#Public", "leaked", "eve", ""),
|
|
}
|
|
|
|
store := newChannelTestStore(packets)
|
|
result := store.computeAnalyticsChannels("", "", TimeWindow{})
|
|
|
|
channels := result["channels"].([]map[string]interface{})
|
|
if len(channels) != 1 {
|
|
t.Fatalf("expected 1 channel bucket, got %d: %+v", len(channels), channels)
|
|
}
|
|
ch := channels[0]
|
|
if ch["name"] != "ch17" {
|
|
t.Errorf("expected undecrypted mismatch downgraded to 'ch17', got %q", ch["name"])
|
|
}
|
|
if ch["encrypted"] != true {
|
|
t.Errorf("expected encrypted=true for rejected rainbow-table name, got %v", ch["encrypted"])
|
|
}
|
|
}
|