diff --git a/cmd/server/analytics_cache_wiring_test.go b/cmd/server/analytics_cache_wiring_test.go index be73c5fc2..416114264 100644 --- a/cmd/server/analytics_cache_wiring_test.go +++ b/cmd/server/analytics_cache_wiring_test.go @@ -353,12 +353,19 @@ func TestHandleAnalyticsSubpaths_MaxLenFirmwareCeiling(t *testing.T) { for i := range hops { hops[i] = fmt.Sprintf("%02x", i) } + // A time window counts every subpath length of each transmission (the + // unfiltered list stops at subpathMaxHops, #413), so the 63-hop path is + // asked for through ?window=. + pathJSON, _ := json.Marshal(hops) srv.store.mu.Lock() - srv.store.spIndex[strings.Join(hops, ",")] = 1 + srv.store.packets = append(srv.store.packets, &StoreTx{ + ID: 1 << 30, Hash: "ceiling63", PathJSON: string(pathJSON), + FirstSeen: time.Now().UTC().Format(time.RFC3339), + }) srv.store.mu.Unlock() get := func(maxLen string) []interface{} { - req := httptest.NewRequest(http.MethodGet, "/api/analytics/subpaths?minLen=63&maxLen="+maxLen, nil) + req := httptest.NewRequest(http.MethodGet, "/api/analytics/subpaths?minLen=63&window=24h&maxLen="+maxLen, nil) w := httptest.NewRecorder() router.ServeHTTP(w, req) if w.Code != 200 { diff --git a/cmd/server/channel_analytics_test.go b/cmd/server/channel_analytics_test.go index cc1f47ed2..f35ae5866 100644 --- a/cmd/server/channel_analytics_test.go +++ b/cmd/server/channel_analytics_test.go @@ -27,8 +27,6 @@ func newChannelTestStore(packets []*StoreTx) *PacketStore { chanCache: make(map[string]*cachedResult), distCache: make(map[string]*cachedResult), subpathCache: make(map[string]*cachedResult), - spIndex: make(map[string]int), - spTxIndex: make(map[string][]*StoreTx), advertPubkeys: make(map[string]int), clockSkew: NewClockSkewEngine(), } diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index d0d263932..4068b9a54 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -528,7 +528,6 @@ func (s *PacketStore) LoadChunked(chunkSize int) error { sort.SliceStable(s.packets, func(i, j int) bool { return s.packets[i].FirstSeen < s.packets[j].FirstSeen }) - s.buildSubpathIndex() s.buildPathHopIndex() s.buildDistanceIndex(nodeSnapshot) if s.hotStartupHours > 0 { diff --git a/cmd/server/chunked_load_fixes_test.go b/cmd/server/chunked_load_fixes_test.go index 4a93526ee..031a03fcf 100644 --- a/cmd/server/chunked_load_fixes_test.go +++ b/cmd/server/chunked_load_fixes_test.go @@ -24,7 +24,7 @@ func TestLoadChunked_MarksIndexesReady(t *testing.T) { store := openChunkedTestStore(t, 100) defer store.db.conn.Close() - if store.SubpathIndexReady() || store.PathHopIndexReady() { + if store.PathHopIndexReady() { t.Fatal("indexes must start NOT ready") } @@ -32,9 +32,6 @@ func TestLoadChunked_MarksIndexesReady(t *testing.T) { t.Fatalf("LoadChunked: %v", err) } - if !store.SubpathIndexReady() { - t.Fatal("SubpathIndexReady() must be true after LoadChunked builds the index") - } if !store.PathHopIndexReady() { t.Fatal("PathHopIndexReady() must be true after LoadChunked builds the index") } diff --git a/cmd/server/coverage_test.go b/cmd/server/coverage_test.go index 18a48563d..14b095437 100644 --- a/cmd/server/coverage_test.go +++ b/cmd/server/coverage_test.go @@ -8,6 +8,7 @@ import ( "math" "net/http" "net/http/httptest" + "reflect" "strings" "testing" "time" @@ -2290,26 +2291,15 @@ func TestStoreGetAnalyticsSubpaths(t *testing.T) { }) } -func TestSubpathPrecomputedIndex(t *testing.T) { +func TestSubpathCountsFromPackets(t *testing.T) { db := setupRichTestDB(t) defer db.Close() store := NewPacketStore(db, nil) store.Load() - // #1008: indexes built in background goroutine; wait before reading. - if !store.WaitIndexesReady(5 * time.Second) { - t.Fatal("indexes never became ready") - } - - // After Load(), the precomputed index must be populated. - if len(store.spIndex) == 0 { - t.Fatal("expected spIndex to be populated after Load()") - } - if store.spTotalPaths == 0 { - t.Fatal("expected spTotalPaths > 0 after Load()") - } // The rich test DB has paths ["aa","bb"], ["aabb","ccdd"], and - // ["eeff","0011","2233"]. That yields 5 unique raw subpaths. + // ["eeff","0011","2233"]. That yields 5 unique raw subpaths, counted + // on demand from the packets (#413). expectedRaw := map[string]int{ "aa,bb": 1, "aabb,ccdd": 1, @@ -2317,21 +2307,20 @@ func TestSubpathPrecomputedIndex(t *testing.T) { "0011,2233": 1, "eeff,0011,2233": 1, } - for key, want := range expectedRaw { - got, ok := store.spIndex[key] - if !ok { - t.Errorf("expected spIndex[%q] to exist", key) - } else if got != want { - t.Errorf("spIndex[%q] = %d, want %d", key, got, want) - } + fast := store.GetAnalyticsSubpaths("", 2, 8, 100) + gotRaw := map[string]int{} + for _, sp := range fast["subpaths"].([]map[string]interface{}) { + gotRaw[strings.Join(sp["rawHops"].([]string), ",")] += sp["count"].(int) } - if store.spTotalPaths != 3 { - t.Errorf("spTotalPaths = %d, want 3", store.spTotalPaths) + if !reflect.DeepEqual(gotRaw, expectedRaw) { + t.Errorf("raw subpath counts = %v, want %v", gotRaw, expectedRaw) + } + if tp := fast["totalPaths"]; tp != 3 { + t.Errorf("totalPaths = %v, want 3", tp) } // Fast-path (no region) and slow-path (with region) must return the // same shape. - fast := store.GetAnalyticsSubpaths("", 2, 8, 100) slow := store.GetAnalyticsSubpaths("SJC", 2, 4, 50) for _, r := range []map[string]interface{}{fast, slow} { if _, ok := r["subpaths"]; !ok { @@ -2341,41 +2330,15 @@ func TestSubpathPrecomputedIndex(t *testing.T) { t.Error("missing totalPaths in result") } } - - // Verify fast path totalPaths matches index. - if tp, ok := fast["totalPaths"].(int); ok && tp != store.spTotalPaths { - t.Errorf("fast totalPaths=%d, spTotalPaths=%d", tp, store.spTotalPaths) - } } -func TestSubpathTxIndexPopulated(t *testing.T) { +func TestSubpathDetailMatchCount(t *testing.T) { db := setupRichTestDB(t) defer db.Close() store := NewPacketStore(db, nil) store.Load() - // #1008: indexes built in background goroutine; wait before reading. - if !store.WaitIndexesReady(5 * time.Second) { - t.Fatal("indexes never became ready") - } - - // spTxIndex must be populated alongside spIndex - if len(store.spTxIndex) == 0 { - t.Fatal("expected spTxIndex to be populated after Load()") - } - - // Every key in spIndex must also exist in spTxIndex with matching count - for key, count := range store.spIndex { - txs, ok := store.spTxIndex[key] - if !ok { - t.Errorf("spTxIndex missing key %q that exists in spIndex", key) - continue - } - if len(txs) != count { - t.Errorf("spTxIndex[%q] has %d txs, spIndex count is %d", key, len(txs), count) - } - } - // GetSubpathDetail should return correct match count via indexed lookup + // GetSubpathDetail counts the transmissions holding the subpath detail := store.GetSubpathDetail([]string{"eeff", "0011"}) if matches := detail.TotalMatches; matches != 1 { t.Errorf("totalMatches = %d, want 1", matches) diff --git a/cmd/server/eviction_test.go b/cmd/server/eviction_test.go index 0c8feb323..ed7baeebd 100644 --- a/cmd/server/eviction_test.go +++ b/cmd/server/eviction_test.go @@ -19,7 +19,6 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore byNode: make(map[string][]*StoreTx), nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), - spIndex: make(map[string]int), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), @@ -83,9 +82,6 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore store.distHops = append(store.distHops, distHopRecord{tx: tx, Hash: hash}) store.distPaths = append(store.distPaths, distPathRecord{tx: tx, Hash: hash}) - // Subpath index - addTxToSubpathIndex(store.spIndex, tx) - // Track bytes for self-accounting store.trackedBytes += rechargeTx(tx) for _, obs := range tx.Observations { @@ -261,7 +257,6 @@ func TestEvictStale_CleansResolvedPathNodeIndexes(t *testing.T) { byNode: make(map[string][]*StoreTx), nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), - spIndex: make(map[string]int), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), @@ -565,14 +560,11 @@ func TestEstimateStoreTxBytes(t *testing.T) { PathJSON: `["aa","bb"]`, } est := estimateStoreTxBytes(tx) - // Manual calculation: base + string lengths + index entries + path hops + subpaths + // Manual calculation: base + string lengths + index entries + path hops hops := int64(len(txGetParsedPath(tx))) manualCalc := int64(storeTxBaseBytes) + int64(len(tx.RawHex)+len(tx.Hash)+len(tx.DecodedJSON)+len(tx.PathJSON)) + int64(numIndexesPerTx*indexEntryBytes) manualCalc += int64(decodedCacheFactor * len(tx.DecodedJSON)) manualCalc += hops * perPathHopBytes - if hops > 1 { - manualCalc += (hops * (hops - 1) / 2) * perSubpathEntryBytes - } if est != manualCalc { t.Fatalf("estimateStoreTxBytes = %d, want %d (manual calc)", est, manualCalc) } diff --git a/cmd/server/hash_migrate.go b/cmd/server/hash_migrate.go index 5dbab02f3..8cfe7e352 100644 --- a/cmd/server/hash_migrate.go +++ b/cmd/server/hash_migrate.go @@ -125,9 +125,9 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration continue } - // A merge edits the subpath and path-hop indexes: let their - // background build finish first, or it could index a duplicate that - // was just removed. + // A merge edits the path-hop index: let its background build + // finish first, or it could index a duplicate that was just + // removed. if !indexesReady { store.WaitIndexesReady(30 * time.Second) indexesReady = true @@ -446,9 +446,6 @@ func (s *PacketStore) finishHashMerge(m *hashMerge, metadata pathNodeSnapshot) i s.trackedBytes -= s.txChargedBytes(l) s.untrackAdvertPubkey(l) s.removeFromResolvedPubkeyIndex(l.ID) - if removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, l) { - s.spTotalPaths-- - } delete(s.pathHopResolved, l) delete(s.fallbackByNode, l) if l.PayloadType != nil { diff --git a/cmd/server/hash_migrate_log_race_301_test.go b/cmd/server/hash_migrate_log_race_301_test.go index 8191cd40d..657f24f52 100644 --- a/cmd/server/hash_migrate_log_race_301_test.go +++ b/cmd/server/hash_migrate_log_race_301_test.go @@ -8,28 +8,28 @@ import ( "time" ) -// hm301InstallSubpathGate holds the subpath index build until the returned +// hm301InstallPathHopGate holds the path-hop index build until the returned // release runs. release is idempotent, so the caller can defer it right after // the gate goes in and still release it on purpose later: without that // deferred net a t.Fatal between the two (store.Load returning an error, say) // would leave the build goroutine blocked on the channel for the rest of the // run (#310, nit 3 of the #307 review). -func hm301InstallSubpathGate(store *PacketStore) (release func()) { +func hm301InstallPathHopGate(store *PacketStore) (release func()) { gate := make(chan struct{}) - store.subpathBuildGate = func() { <-gate } + store.pathHopBuildGate = func() { <-gate } return sync.OnceFunc(func() { close(gate) }) } // The deferred net and the deliberate release are the same func, so releasing // twice must be a no-op rather than a "close of closed channel" panic. -func TestHashMigrateSubpathGateReleaseIsIdempotent_310(t *testing.T) { +func TestHashMigrateBuildGateReleaseIsIdempotent_310(t *testing.T) { store := &PacketStore{} - release := hm301InstallSubpathGate(store) + release := hm301InstallPathHopGate(store) blocked := make(chan struct{}) go func() { defer close(blocked) - store.subpathBuildGate() + store.pathHopBuildGate() }() release() @@ -42,12 +42,12 @@ func TestHashMigrateSubpathGateReleaseIsIdempotent_310(t *testing.T) { } } -// Issue #301: Load() starts the two background index builds, and each builder -// logs one more line ("index build complete") AFTER it has flipped its ready -// flag. WaitIndexesReady returning therefore does not mean the builders are -// done logging, and a builder of an earlier store can log at any later time. +// Issue #301: Load() starts the background index build, and the builder logs +// one more line ("index build complete") AFTER it has flipped its ready flag. +// WaitIndexesReady returning therefore does not mean the builder is done +// logging, and a builder of an earlier store can log at any later time. // The log capture of the hash-migrate harness must take those writes without a -// data race: here the subpath builder is released inside the capture, so its +// data race: here the path-hop builder is released inside the capture, so its // last line lands in the captured output while nothing orders it before the // harness's reads. Red under -race with an unsynchronized buffer. func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) { @@ -59,13 +59,13 @@ func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) { } t.Cleanup(func() { db.conn.Close() }) store := NewPacketStore(db, &PacketStoreConfig{}) - release := hm301InstallSubpathGate(store) + release := hm301InstallPathHopGate(store) defer release() // a t.Fatal below must not strand the build goroutine if err := store.Load(); err != nil { t.Fatal(err) } - const want = "index build complete: subpath" + const want = "index build complete: pathHop" out := captureLogRead(func(read func() string) { release() if !store.WaitIndexesReady(30 * time.Second) { @@ -82,13 +82,13 @@ func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) { deadline := time.Now().Add(30 * time.Second) for !strings.Contains(read(), want) { if time.Now().After(deadline) { - t.Error("the subpath builder's last line did not arrive") + t.Error("the path-hop builder's last line did not arrive") return } runtime.Gosched() } }) if !strings.Contains(out, want) { - t.Fatalf("the subpath builder's last line was not captured:\n%s", out) + t.Fatalf("the path-hop builder's last line was not captured:\n%s", out) } } diff --git a/cmd/server/hash_migrate_readonly_215_test.go b/cmd/server/hash_migrate_readonly_215_test.go index 0e1e8c046..f0504061e 100644 --- a/cmd/server/hash_migrate_readonly_215_test.go +++ b/cmd/server/hash_migrate_readonly_215_test.go @@ -163,8 +163,6 @@ func hm215OpenIdx(t testing.TB, d *hm215DB, resolvedIndex bool) *PacketStore { } // hm215Snapshot is the part of the store a leftover would show up in. -// spTotalPaths is left out on purpose: eviction never decrements it (master -// behaviour, unrelated to this change), so it cannot return to a baseline. type hm215Snapshot struct { Packets int TrackedBytes int64 @@ -181,8 +179,6 @@ type hm215Snapshot struct { TotalObs int AdvertKeys map[string]int ByPathHop map[string]int - SpIndex map[string]int - SpTxIndex map[string]int } func hm215Snap(s *PacketStore) hm215Snapshot { @@ -196,7 +192,7 @@ func hm215Snap(s *PacketStore) hm215Snapshot { PathHopRecs: len(s.pathHopResolved), FallbackRecs: len(s.fallbackByNode), ResolvedIdx: len(s.resolvedPubkeyIndex), TotalObs: s.totalObs, AdvertKeys: map[string]int{}, - ByPathHop: map[string]int{}, SpIndex: map[string]int{}, SpTxIndex: map[string]int{}, + ByPathHop: map[string]int{}, } for pk, list := range s.byNode { for _, tx := range list { @@ -222,12 +218,6 @@ func hm215Snap(s *PacketStore) hm215Snapshot { for k, list := range s.byPathHop { sn.ByPathHop[k] = len(list) } - for k, n := range s.spIndex { - sn.SpIndex[k] = n - } - for k, list := range s.spTxIndex { - sn.SpTxIndex[k] = len(list) - } return sn } @@ -461,17 +451,6 @@ func TestHashMigrate_MergesDuplicatesInMemory_215(t *testing.T) { } } } - // The subpath total follows the merge: it counts transmissions with a - // path of 2+ hops, and a merged-away duplicate no longer counts. - multiHop := 0 - for _, tx := range s.packets { - if len(txGetParsedPath(tx)) >= 2 { - multiHop++ - } - } - if s.spTotalPaths != multiHop { - t.Errorf("spTotalPaths = %d, want %d (transmissions with 2+ hop paths)", s.spTotalPaths, multiHop) - } // Every observation of the DB survives, except the 4 identical // (obs0, ["aa"]) ones that merge into 1 per group. if want := 2 /*ballast*/ + 2*(1+3+3) + 2; s.totalObs != want { diff --git a/cmd/server/index_ready_1008.go b/cmd/server/index_ready_1008.go index c075a62bc..83ca36f2a 100644 --- a/cmd/server/index_ready_1008.go +++ b/cmd/server/index_ready_1008.go @@ -1,23 +1,23 @@ -// Issue #1008: background-deferred subpath + pathHop index builds. +// Issue #1008: background-deferred pathHop index build. // // Pattern mirrors the distance index (#1011) — but where distance is -// fully lazy (built on first request), these two indexes are kicked off +// fully lazy (built on first request), this index is kicked off // eagerly by Load() in a background goroutine so HTTP becomes ready -// immediately while the indexes finish populating. +// immediately while the index finishes populating. Route Patterns +// (subpaths) had a second such index until #413; they are now counted +// on demand and need no build or gate. // // 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 + +// - pathHopReady is an atomic.Bool flag written exactly once by the +// background builder (false → true) and never reset thereafter. +// Handlers read it via PathHopIndexReady() before touching +// s.byPathHop. While it is false, a gated 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 +// buildPathHopIndex() method. It replaces s.byPathHop with a +// freshly-allocated map 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 @@ -29,8 +29,8 @@ // 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 +// pruning / hash-collision paths continue to write 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 @@ -44,11 +44,9 @@ // // 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. +// - HARD-GATED with 503 + Retry-After: 5 — endpoints whose entire +// response is the index lookup. Empty data would be visibly broken. +// See routes.go: /api/nodes/{pubkey}/paths. // // - BEST-EFFORT (not gated) — endpoints where the index drives // enrichment fields that callers already treat as optional. During @@ -65,7 +63,7 @@ // 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 +// (#1262) refreshes them every 5min once the index flips ready // (prewarm guarded by WaitIndexesReady — see review M1). // // This is documented rather than gated so operators do not see @@ -89,13 +87,6 @@ func writeIndexLoading503(w http.ResponseWriter) { _, _ = 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. @@ -103,8 +94,8 @@ 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 +// indexReadyCh returns the channel that is closed when the path-hop +// index has 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{} { @@ -112,22 +103,21 @@ func (s *PacketStore) indexReadyCh() <-chan struct{} { 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), + // If it is already ready (e.g. background chunk loader + // flipped it synchronously before any waiter showed up), // close immediately so the channel is usable as a one-shot. - if s.subpathReady.Load() && s.pathHopReady.Load() { + if 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. +// maybeCloseIndexReadyCh closes the ready channel iff the flag is set. +// Idempotent and safe to call from the builder goroutine on the +// green-path transition, as well as from markIndexesReadySync. func (s *PacketStore) maybeCloseIndexReadyCh() { - if !(s.subpathReady.Load() && s.pathHopReady.Load()) { + if !s.pathHopReady.Load() { return } s.indexReadyChMu.Lock() @@ -148,38 +138,24 @@ func (s *PacketStore) maybeCloseIndexReadyCh() { } // 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. +// to populate the path-hop index off the critical path (#1008). It +// returns immediately; the work runs in a background goroutine that +// acquires s.mu.Lock(), installs the map, then sets the 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. +// readiness ~60s inside Load() under s.mu. func (s *PacketStore) startBackgroundIndexBuilds() { - gate := s.subpathBuildGate // read before the goroutine starts + gate := s.pathHopBuildGate // read before the goroutine starts go func() { if gate != nil { gate() } - 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() + // Atomic.Store happens-after s.mu.Unlock; readers that + // observe Ready()==true synchronize-with this store. s.pathHopReady.Store(true) s.maybeCloseIndexReadyCh() log.Printf("[startup] index build complete: pathHop (%s)", @@ -189,26 +165,25 @@ func (s *PacketStore) startBackgroundIndexBuilds() { // 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 +// loader rebuilds the index under s.mu.Lock(); after the Unlock it +// calls this to flip the ready flag 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. +// WaitIndexesReady blocks until the background index built by +// startBackgroundIndexBuilds() reports ready, or the deadline expires. +// Returns true if it flipped in time. Intended for tests that read +// s.byPathHop directly after Load(), and for background work that +// needs it (recomputer prewarm, hash migration); request paths check +// PathHopIndexReady() 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() { + if s.PathHopIndexReady() { return true } ch := s.indexReadyCh() @@ -216,7 +191,7 @@ func (s *PacketStore) WaitIndexesReady(timeout time.Duration) bool { case <-ch: return true case <-time.After(timeout): - return s.SubpathIndexReady() && s.PathHopIndexReady() + return s.PathHopIndexReady() } } diff --git a/cmd/server/index_ready_1008_test.go b/cmd/server/index_ready_1008_test.go index f5d8027a8..0dbd184a1 100644 --- a/cmd/server/index_ready_1008_test.go +++ b/cmd/server/index_ready_1008_test.go @@ -1,13 +1,16 @@ -// Issue #1008: subpath + pathHop index builds must move off the -// synchronous Load() critical path into a background goroutine. +// Issue #1008: the pathHop index build must move off the synchronous +// Load() critical path into a background goroutine. // // Contract: -// 1. Immediately after Load() returns, SubpathIndexReady() and -// PathHopIndexReady() report false (the goroutine has not finished). -// 2. Analytics handlers that depend on those indices respond 503 with -// Retry-After: 5 until the corresponding ready flag flips true. +// 1. Immediately after Load() returns, PathHopIndexReady() reports +// false (the goroutine has not finished). +// 2. Handlers that depend on the index respond 503 with Retry-After: 5 +// until the ready flag flips true. // 3. After the background build completes (waitable via a helper), -// both flags flip true and handlers respond 200. +// the flag flips true and handlers respond 200. +// +// Route Patterns (subpaths) had their own index and gate until #413; they +// are counted on demand and answer as soon as Load() returns. package main import ( @@ -17,26 +20,13 @@ import ( "sync" "testing" "time" -) -// TestIssue1008_SubpathIndexReadyFalseImmediatelyAfterLoad asserts the -// subpath ready flag is false the instant Load() returns. Red commit: the -// stub returns true → assertion fires. Green commit: the flag is owned by -// the background goroutine, which has not yet run, so the assertion holds. -func TestIssue1008_SubpathIndexReadyFalseImmediatelyAfterLoad(t *testing.T) { - db := setupRichTestDB(t) - defer db.Close() - store := NewPacketStore(db, nil) - if err := store.Load(); err != nil { - t.Fatalf("Load() error: %v", err) - } - if store.SubpathIndexReady() { - t.Fatal("expected SubpathIndexReady()==false immediately after Load(); want background-deferred build (#1008)") - } -} + "github.com/gorilla/mux" +) -// TestIssue1008_PathHopIndexReadyFalseImmediatelyAfterLoad: same contract -// for the path-hop index. +// TestIssue1008_PathHopIndexReadyFalseImmediatelyAfterLoad asserts the +// path-hop ready flag is false the instant Load() returns: the flag is owned +// by the background goroutine, which has not yet run. func TestIssue1008_PathHopIndexReadyFalseImmediatelyAfterLoad(t *testing.T) { db := setupRichTestDB(t) defer db.Close() @@ -49,42 +39,50 @@ func TestIssue1008_PathHopIndexReadyFalseImmediatelyAfterLoad(t *testing.T) { } } -// TestIssue1008_HandlerReturns503WhileSubpathIndexLoading asserts the -// analytics/subpaths handler returns 503 + Retry-After: 5 + a JSON body -// matching the triage spec while the subpath index is still building, -// and 200 once it is built. +// TestIssue1008_HandlerReturns503WhilePathHopIndexLoading asserts that +// /api/nodes/{pubkey}/paths returns 503 + Retry-After: 5 + a JSON body +// matching the triage spec while the path-hop index is still building, and +// 200 once it is built — and that Route Patterns, which keep no index +// (#413), are served with their counts throughout. // // The background build of the small fixture used to finish before the // request under CPU load, so the 503 was a race the test could lose (#227). -// subpathBuildGate now holds the build until the 503 has been asserted. -func TestIssue1008_HandlerReturns503WhileSubpathIndexLoading(t *testing.T) { - db := setupRichTestDB(t) +// pathHopBuildGate now holds the build until the 503 has been asserted. +func TestIssue1008_HandlerReturns503WhilePathHopIndexLoading(t *testing.T) { + db := setupTestDB(t) defer db.Close() + seedTestData(t, db) store := NewPacketStore(db, nil) release := make(chan struct{}) var releaseOnce sync.Once releaseBuild := func() { releaseOnce.Do(func() { close(release) }) } t.Cleanup(releaseBuild) // a failed assertion must not strand the builder building := make(chan struct{}) - store.subpathBuildGate = func() { close(building); <-release } + store.pathHopBuildGate = func() { close(building); <-release } if err := store.Load(); err != nil { t.Fatalf("Load() error: %v", err) } select { case <-building: // the build has started and is held case <-time.After(5 * time.Second): - t.Fatal("background subpath index build never started") + t.Fatal("background path-hop index build never started") } cfg := &Config{} cfg.applyListLimitsDefaults() - srv := &Server{store: store, cfg: cfg} - - req := httptest.NewRequest("GET", "/api/analytics/subpaths?minLen=2&maxLen=4&limit=10", nil) - rec := httptest.NewRecorder() - srv.handleAnalyticsSubpaths(rec, req) + srv := NewServer(db, cfg, NewHub()) + srv.store = store + router := mux.NewRouter() + srv.RegisterRoutes(router) + get := func(url string) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest("GET", url, nil)) + return rec + } + const nodePaths = "/api/nodes/aabbccdd11223344/paths" + rec := get(nodePaths) if rec.Code != http.StatusServiceUnavailable { - t.Fatalf("status = %d, want 503 (subpath index loading, #1008)", rec.Code) + t.Fatalf("status = %d, want 503 (path-hop index loading, #1008)", rec.Code) } if got := rec.Header().Get("Retry-After"); got != "5" { t.Errorf("Retry-After header = %q, want %q", got, "5") @@ -97,14 +95,30 @@ func TestIssue1008_HandlerReturns503WhileSubpathIndexLoading(t *testing.T) { t.Errorf(`body["error"] = %v, want "index loading"`, body["error"]) } + // Route Patterns do not wait for any build (#413). + for _, url := range []string{ + "/api/analytics/subpaths?minLen=2&maxLen=4&limit=10", + "/api/analytics/subpaths-bulk?groups=2-2:10", + "/api/analytics/subpath-detail?hops=aa,bb", + } { + if rec := get(url); rec.Code != http.StatusOK { + t.Errorf("%s while the path-hop index builds: status %d, want 200", url, rec.Code) + } + } + var subpaths SubpathsResponse + if err := json.Unmarshal(get("/api/analytics/subpaths?minLen=2&maxLen=4&limit=10").Body.Bytes(), &subpaths); err != nil { + t.Fatal(err) + } + if subpaths.TotalPaths == 0 || len(subpaths.Subpaths) == 0 { + t.Errorf("subpaths while the path-hop index builds = %+v, want the counted paths", subpaths) + } + // Released, the build completes and the same request is served. releaseBuild() if !store.WaitIndexesReady(5 * time.Second) { - t.Fatal("indexes not ready within 5s of releasing the build") + t.Fatal("index not ready within 5s of releasing the build") } - rec = httptest.NewRecorder() - srv.handleAnalyticsSubpaths(rec, httptest.NewRequest("GET", "/api/analytics/subpaths?minLen=2&maxLen=4&limit=10", nil)) - if rec.Code != http.StatusOK { + if rec := get(nodePaths); rec.Code != http.StatusOK { t.Fatalf("status after the build = %d, want 200", rec.Code) } } @@ -119,18 +133,15 @@ func TestIssue1008_HandlerRecoversAfterIndexReady(t *testing.T) { t.Fatalf("Load() error: %v", err) } - // Wait up to 5s for both background builds to finish on this small + // Wait up to 5s for the background build to finish on this small // fixture (rich test DB has ~3 packets; build is sub-millisecond). deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { - if store.SubpathIndexReady() && store.PathHopIndexReady() { + if store.PathHopIndexReady() { break } time.Sleep(10 * time.Millisecond) } - if !store.SubpathIndexReady() { - t.Fatal("SubpathIndexReady() never flipped true within 5s") - } if !store.PathHopIndexReady() { t.Fatal("PathHopIndexReady() never flipped true within 5s") } @@ -146,11 +157,11 @@ func TestIssue1008_HandlerRecoversAfterIndexReady(t *testing.T) { } } -// TestIssue1008_m7_BothFlagsSetAfterParallelStart verifies that the -// parallel two-goroutine version of startBackgroundIndexBuilds (review -// m7) sets BOTH ready flags after a bounded wait, regardless of which -// goroutine wins the race to s.mu.Lock(). Sanity check that breaking -// the two builds apart didn't drop the pathHop flag flip. +// TestIssue1008_m7_BothFlagsSetAfterParallelStart verifies that +// startBackgroundIndexBuilds sets the ready flag after a bounded wait. +// Review m7 split the subpath and pathHop builds into two goroutines; +// #413 removed the subpath build, and this is the sanity check that the +// pathHop flag flip survived both changes. func TestIssue1008_m7_BothFlagsSetAfterParallelStart(t *testing.T) { db := setupRichTestDB(t) defer db.Close() @@ -161,9 +172,6 @@ func TestIssue1008_m7_BothFlagsSetAfterParallelStart(t *testing.T) { if !store.WaitIndexesReady(5 * time.Second) { t.Fatal("indexes never ready after parallel start (#1008 m7)") } - if !store.SubpathIndexReady() { - t.Error("subpath flag not set after WaitIndexesReady returned true") - } if !store.PathHopIndexReady() { t.Error("pathHop flag not set after WaitIndexesReady returned true") } diff --git a/cmd/server/neighbor_graph_test.go b/cmd/server/neighbor_graph_test.go index 2bf999da9..f16633db9 100644 --- a/cmd/server/neighbor_graph_test.go +++ b/cmd/server/neighbor_graph_test.go @@ -33,7 +33,6 @@ func ngTestStore(nodes []nodeInfo, packets []*StoreTx) *PacketStore { chanCache: make(map[string]*cachedResult), distCache: make(map[string]*cachedResult), subpathCache: make(map[string]*cachedResult), - spIndex: make(map[string]int), } ps.nodeCache = nodes ps.nodePM = buildPrefixMap(nodes) diff --git a/cmd/server/regions_observed_test.go b/cmd/server/regions_observed_test.go index 942365704..e4cf8c251 100644 --- a/cmd/server/regions_observed_test.go +++ b/cmd/server/regions_observed_test.go @@ -27,7 +27,6 @@ func regionsObservedFixture(t testing.TB) (*Server, time.Time, string) { store.packets = []*StoreTx{{FirstSeen: store.oldestLoaded}} store.backgroundLoadDone.Store(true) store.pathHopReady.Store(true) - store.subpathReady.Store(true) srv := NewServer(db, &Config{}, NewHub()) srv.store = store return srv, now, key diff --git a/cmd/server/relay_airtime_share_test.go b/cmd/server/relay_airtime_share_test.go index a1b41d4eb..7423b857a 100644 --- a/cmd/server/relay_airtime_share_test.go +++ b/cmd/server/relay_airtime_share_test.go @@ -29,8 +29,6 @@ func newRelayAirtimeShareTestStore(packets []*StoreTx) *PacketStore { chanCache: make(map[string]*cachedResult), distCache: make(map[string]*cachedResult), subpathCache: make(map[string]*cachedResult), - spIndex: make(map[string]int), - spTxIndex: make(map[string][]*StoreTx), advertPubkeys: make(map[string]int), } ps.useResolvedPathIndex = true diff --git a/cmd/server/repeater_enrich_recomputer_1008_test.go b/cmd/server/repeater_enrich_recomputer_1008_test.go index 0227e41ab..26b131bb2 100644 --- a/cmd/server/repeater_enrich_recomputer_1008_test.go +++ b/cmd/server/repeater_enrich_recomputer_1008_test.go @@ -36,7 +36,6 @@ func TestIssue1008_M1_PrewarmWaitsForIndexes(t *testing.T) { // WaitIndexesReady would return immediately on the channel select // (correct for production semantics where flags never reset, // wrong for this synthetic test). - store.subpathReady.Store(false) store.pathHopReady.Store(false) store.indexReadyChMu.Lock() store.indexReadyChan = nil diff --git a/cmd/server/resolved_index_test.go b/cmd/server/resolved_index_test.go index 66350cd64..0023b56cb 100644 --- a/cmd/server/resolved_index_test.go +++ b/cmd/server/resolved_index_test.go @@ -288,7 +288,6 @@ func TestEviction_ByNodeCleanup_OnDemandSQL(t *testing.T) { byNode: make(map[string][]*StoreTx), nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), - spIndex: make(map[string]int), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), @@ -363,7 +362,6 @@ func TestRunEviction_NoLockDuringSQLFetch(t *testing.T) { nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), byPathHop: make(map[string][]*StoreTx), - spIndex: make(map[string]int), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), @@ -1150,8 +1148,6 @@ func TestResolvedPubkeyIndex_BoundedByEviction(t *testing.T) { nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), byPathHop: make(map[string][]*StoreTx), - spIndex: make(map[string]int), - spTxIndex: make(map[string][]*StoreTx), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), @@ -1276,8 +1272,6 @@ func TestRunEviction_TriggersCompaction(t *testing.T) { nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), byPathHop: make(map[string][]*StoreTx), - spIndex: make(map[string]int), - spTxIndex: make(map[string][]*StoreTx), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), @@ -1364,8 +1358,6 @@ func TestResolvedPubkeyIndex_MemoryStableThroughEvictionCycles(t *testing.T) { nodeHashes: make(map[string]map[string]bool), byPayloadType: make(map[int][]*StoreTx), byPathHop: make(map[string][]*StoreTx), - spIndex: make(map[string]int), - spTxIndex: make(map[string][]*StoreTx), distHops: make([]distHopRecord, 0), distPaths: make([]distPathRecord, 0), rfCache: make(map[string]*cachedResult), diff --git a/cmd/server/routes.go b/cmd/server/routes.go index de6850110..1e1c0964d 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -3095,10 +3095,6 @@ func (s *Server) handleAnalyticsHashCollisions(w http.ResponseWriter, r *http.Re func (s *Server) handleAnalyticsSubpaths(w http.ResponseWriter, r *http.Request) { if s.store != nil { - if !s.store.SubpathIndexReady() { - writeIndexLoading503(w) - return - } region := r.URL.Query().Get("region") if s.rejectAnalyticsFilters(w, region, "") { return @@ -3140,10 +3136,6 @@ func (s *Server) handleAnalyticsSubpaths(w http.ResponseWriter, r *http.Request) // // ?groups=2-2:50,3-3:30,4-4:20,5-8:15 (minLen-maxLen:limit per group) func (s *Server) handleAnalyticsSubpathsBulk(w http.ResponseWriter, r *http.Request) { - if s.store != nil && !s.store.SubpathIndexReady() { - writeIndexLoading503(w) - return - } region := r.URL.Query().Get("region") if s.rejectAnalyticsFilters(w, region, "") { return @@ -3228,10 +3220,6 @@ func (s *Server) handleAnalyticsSubpathDetail(w http.ResponseWriter, r *http.Req // resolves it to; a hidden one is answered like a hop that matches no // node (#410). if s.store != nil { - if !s.store.SubpathIndexReady() { - writeIndexLoading503(w) - return - } parts := s.store.SubpathDetail(rawHops) vis, err := s.subpathVisibility(r.Context(), parts.pubkeys()) if err != nil { diff --git a/cmd/server/store.go b/cmd/server/store.go index 7515b4dca..ac38e8cc8 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -1,6 +1,7 @@ package main import ( + "container/heap" "crypto/sha256" "database/sql" "encoding/json" @@ -315,29 +316,24 @@ type PacketStore struct { // Per-store dedupe set for one-shot schema-degradation warnings. Field // (not package-level) so each test gets a fresh state — see #1199 item 5. schemaDegradationLogged sync.Map - // Precomputed subpath index: raw comma-joined hops → occurrence count. - // Built during Load(), incrementally updated on ingest. Avoids full - // packet iteration at query time (O(unique_subpaths) vs O(total_packets)). - spIndex map[string]int // "hop1,hop2" → count - spTxIndex map[string][]*StoreTx // "hop1,hop2" → transmissions containing this subpath - spTotalPaths int // transmissions with paths >= 2 hops - // Background-build ready gates for spIndex/spTxIndex and byPathHop - // (#1008). Flipped from false→true exactly once by the goroutine - // kicked off in Load() (or synchronously by the background chunk - // loader). Handlers gate reads via SubpathIndexReady() / - // PathHopIndexReady(); while false, they respond 503 + Retry-After. - subpathReady atomic.Bool + // Route Patterns (subpaths) keep no index: they are counted on demand + // from s.packets (#413). + // + // Background-build ready gate for byPathHop (#1008). Flipped from + // false→true exactly once by the goroutine kicked off in Load() (or + // synchronously by the background chunk loader). Readers check it via + // PathHopIndexReady(). pathHopReady atomic.Bool - // indexReadyChan is closed exactly once when BOTH subpathReady - // and pathHopReady are true (#1008 review m6). Replaces the - // previous 2ms poll in WaitIndexesReady. Lazily allocated by - // indexReadyCh / maybeCloseIndexReadyCh in index_ready_1008.go. + // indexReadyChan is closed exactly once when pathHopReady is true + // (#1008 review m6). Replaces the previous 2ms poll in + // WaitIndexesReady. Lazily allocated by indexReadyCh / + // maybeCloseIndexReadyCh in index_ready_1008.go. indexReadyChMu sync.Mutex indexReadyChan chan struct{} - // subpathBuildGate, nil in production, runs at the start of the - // background subpath index build that Load() starts. A test sets it + // pathHopBuildGate, nil in production, runs at the start of the + // background path-hop index build that Load() starts. A test sets it // before Load() to hold the not-ready window open (#227). - subpathBuildGate func() + pathHopBuildGate func() // Precomputed distance analytics: hop distances and path totals. // Built LAZILY on first /api/analytics/distance request (#1011) — // previously eager in Load() at startup, which was O(n²) work for @@ -639,6 +635,12 @@ type PacketStore struct { // Test-only hook called with "obs-miss", "obs-scan" and "obs-write" as a // cache miss moves through loadObsCounts. Nil in production. statsHook func(stage string) + // Concurrent subpathCache misses for one key share one count (#413); + // see shareSubpathCount. + subpathSF singleflight.Group + // Test-only hook called with "miss" and "count" as a subpathCache miss + // moves through shareSubpathCount. Nil in production. + subpathHook func(stage string) // Test-only hook fired at the very start of loadBackgroundChunks // (after the #1809 invariant check). Nil in production. Used by @@ -848,8 +850,6 @@ func NewPacketStore(db *DB, cfg *PacketStoreConfig, cacheTTLs ...map[string]inte rfCacheTTL: 60 * time.Second, collisionCacheTTL: 3600 * time.Second, invCooldown: 300 * time.Second, - spIndex: make(map[string]int, 4096), - spTxIndex: make(map[string][]*StoreTx, 4096), advertPubkeys: make(map[string]int), pathHopResolved: make(map[*StoreTx][]uint64), fallbackByNode: make(map[*StoreTx][]string), @@ -1182,13 +1182,6 @@ func (s *PacketStore) Load() error { s.indexByNode(tx) } - // Build precomputed subpath index for O(1) analytics queries - // — DEFERRED to a background goroutine (#1008). Same rationale - // as the distance index (#1011): synchronous build under s.mu - // blocks HTTP readiness ~60s at Cascadia scale. The goroutine is - // started AFTER s.loaded = true below. - // s.buildSubpathIndex() - // Build path-hop index for O(1) node path lookups // — DEFERRED to a background goroutine (#1008). // s.buildPathHopIndex() @@ -1226,11 +1219,10 @@ func (s *PacketStore) Load() error { log.Printf("[store] cold load: skipped %d ambiguous-prefix relay hops (unique_prefix gate, PR #1643 R1)", coldLoadAmbiguousHopsSkipped) } - // Kick off background subpath + path-hop index builds (#1008). - // The goroutine acquires s.mu.Lock() and so will block until Load's - // deferred Unlock fires when this function returns. HTTP handlers - // gate reads behind SubpathIndexReady() / PathHopIndexReady() and - // respond 503 + Retry-After: 5 until the builds finish. + // Kick off the background path-hop index build (#1008). The + // goroutine acquires s.mu.Lock() and so will block until Load's + // deferred Unlock fires when this function returns. Readers check + // PathHopIndexReady() until the build finishes. s.startBackgroundIndexBuilds() return nil } @@ -1243,8 +1235,8 @@ func (s *PacketStore) Load() error { // last_seen but can overlap it by first_seen. localPackets are merged into // s.packets by first_seen (mergeByFirstSeen, #114). // -// byPayloadType is updated here incrementally. byPathHop, spIndex, and -// distHops are NOT updated here — the caller (loadBackgroundChunks) rebuilds +// byPayloadType is updated here incrementally. byPathHop and distHops are +// NOT updated here — the caller (loadBackgroundChunks) rebuilds // those once after all chunks are merged. // accumulateDedup appends pks to byTx[txID], deduplicating per-tx via the @@ -1840,7 +1832,6 @@ func (s *PacketStore) loadBackgroundChunks() { // Rebuild analytics indexes once after all chunks are merged. s.mu.Lock() - s.buildSubpathIndex() s.buildPathHopIndex() // Distance index is now lazy (#1011) — built on first // /api/analytics/distance request, not at background-load @@ -1852,10 +1843,10 @@ func (s *PacketStore) loadBackgroundChunks() { // it unnecessary. s.distDataGen.Add(1) s.mu.Unlock() - // #1008 review m3: flip the ready flags after the synchronous + // #1008 review m3: flip the ready flag after the synchronous // rebuild for symmetry with startBackgroundIndexBuilds. Safe // today because the chunk loader runs after Load() has already - // kicked the goroutines that set these to true; this is a + // kicked the goroutine that sets it to true; this is a // belt-and-suspenders against a future reorder where the chunk // loader could be the first writer. s.markIndexesReadySync() @@ -1961,7 +1952,7 @@ func setTxPath(tx *StoreTx, pathJSON string) { // rechargeTx re-estimates tx and returns the change since it was last // charged, for the caller to add to trackedBytes (#113). A tx is first // charged when it is created, before its observations are merged, so its -// path costs (byPathHop, spTxIndex) are unknown then: call it again once +// path costs (byPathHop) are unknown then: call it again once // pickBestObservation has set the path. Eviction subtracts what was charged // (chargedBytes, see txChargedBytes) and never re-estimates, so the total // stays exact however often the path changed in between. @@ -3526,11 +3517,8 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac s.trackedBytes += rechargeTx(tx) } - // Incrementally update precomputed subpath index with new transmissions + // Incrementally update the path-hop index with new transmissions for _, tx := range broadcastTxs { - if addTxToSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { - s.spTotalPaths++ - } addTxToPathHopIndex(s.byPathHop, tx) } if len(broadcastTxs) > 0 { @@ -3965,8 +3953,8 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] broadcastMaps = append(broadcastMaps, broadcastMap) } - // Re-pick best observation for updated transmissions and update subpath index - // if the path changed. + // Re-pick best observation for updated transmissions and update the + // path-hop index if the path changed. oldPaths := make(map[int]string, len(updatedTxs)) for txID, tx := range updatedTxs { oldPaths[txID] = tx.PathJSON @@ -4022,22 +4010,12 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] return broadcastMaps } -// reindexTxPath moves tx in the subpath and path-hop indexes after its best +// reindexTxPath moves tx in the path-hop index after its best // path changed from oldPath to tx.PathJSON (pickBestObservation, which also // parsed it into tx.parsedPath). The caller invalidates the relay-stats cache once for // a batch. Must hold s.mu for writing. func (s *PacketStore) reindexTxPath(tx *StoreTx, oldPath string) { - // Path changed — remove old subpaths, add new ones. oldHops := parsePathJSON(oldPath) - if len(oldHops) >= 2 { - // Temporarily set parsedPath to old hops for removal. - saved, savedFlag := tx.parsedPath, tx.pathParsed - tx.parsedPath, tx.pathParsed = oldHops, true - if removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { - s.spTotalPaths-- - } - tx.parsedPath, tx.pathParsed = saved, savedFlag - } // Remove old path-hop index entries using old hops. // Resolved pubkey entries are managed via resolvedPubkeyIndex, not byPathHop. if len(oldHops) > 0 { @@ -4047,9 +4025,6 @@ func (s *PacketStore) reindexTxPath(tx *StoreTx, oldPath string) { tx.parsedPath, tx.pathParsed = saved, savedFlag } // pickBestObservation already parsed the new path into tx.parsedPath. - if addTxToSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { - s.spTotalPaths++ - } addTxToPathHopIndex(s.byPathHop, tx) } @@ -4828,87 +4803,6 @@ func txGetParsedPath(tx *StoreTx) []string { return tx.parsedPath } -// addTxToSubpathIndex extracts all raw subpaths (lengths 2–8) from tx and -// increments their counts in the index. Returns true if the tx contributed -// (path had ≥ 2 hops). -func addTxToSubpathIndex(idx map[string]int, tx *StoreTx) bool { - return addTxToSubpathIndexFull(idx, nil, tx) -} - -// addTxToSubpathIndexFull is like addTxToSubpathIndex but also appends -// tx to txIdx for each subpath key (if txIdx is non-nil). -func addTxToSubpathIndexFull(idx map[string]int, txIdx map[string][]*StoreTx, tx *StoreTx) bool { - hops := txGetParsedPath(tx) - if len(hops) < 2 { - return false - } - maxL := min(8, len(hops)) - for l := 2; l <= maxL; l++ { - for start := 0; start <= len(hops)-l; start++ { - key := strings.ToLower(strings.Join(hops[start:start+l], ",")) - idx[key]++ - if txIdx != nil { - txIdx[key] = append(txIdx[key], tx) - } - } - } - return true -} - -// removeTxFromSubpathIndex is the inverse of addTxToSubpathIndex — it -// decrements counts for all raw subpaths of tx. Returns true if the tx -// had a path. -func removeTxFromSubpathIndex(idx map[string]int, tx *StoreTx) bool { - return removeTxFromSubpathIndexFull(idx, nil, tx) -} - -// removeTxFromSubpathIndexFull is like removeTxFromSubpathIndex but also -// removes tx from txIdx for each subpath key (if txIdx is non-nil). -func removeTxFromSubpathIndexFull(idx map[string]int, txIdx map[string][]*StoreTx, tx *StoreTx) bool { - hops := txGetParsedPath(tx) - if len(hops) < 2 { - return false - } - maxL := min(8, len(hops)) - for l := 2; l <= maxL; l++ { - for start := 0; start <= len(hops)-l; start++ { - key := strings.ToLower(strings.Join(hops[start:start+l], ",")) - idx[key]-- - if idx[key] <= 0 { - delete(idx, key) - } - if txIdx != nil { - txs := txIdx[key] - for i, t := range txs { - if t == tx { - txIdx[key] = append(txs[:i], txs[i+1:]...) - break - } - } - if len(txIdx[key]) == 0 { - delete(txIdx, key) - } - } - } - } - return true -} - -// buildSubpathIndex scans all packets and populates spIndex + spTotalPaths. -// Must be called with s.mu held. -func (s *PacketStore) buildSubpathIndex() { - s.spIndex = make(map[string]int, 4096) - s.spTxIndex = make(map[string][]*StoreTx, 4096) - s.spTotalPaths = 0 - for _, tx := range s.packets { - if addTxToSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { - s.spTotalPaths++ - } - } - log.Printf("[store] Built subpath index: %d unique raw subpaths from %d paths", - len(s.spIndex), s.spTotalPaths) -} - // buildPathHopIndex rebuilds byPathHop: raw wire hops from every packet's // path_json, plus the resolved full-pubkey hops carried over from the // previous index (see retainResolvedPathHops). @@ -5599,7 +5493,7 @@ func (s *PacketStore) buildDistanceIndex(snapshots ...pathNodeSnapshot) { // usage and independent of GC state. // // Issue #743: Previous estimates missed major per-packet allocations: -// - spTxIndex: O(path²) entries per tx (50-150MB at scale) +// - spTxIndex: O(path²) entries per tx (50-150MB at scale; removed by #413) // - Per-tx maps: obsKeys, observerSet (~11MB at scale; only busy tx since #411) // - byPathHop index entries (20-40MB at scale) // Note: ResolvedPath per-obs overhead eliminated by #800 refactor. @@ -5614,9 +5508,6 @@ const ( // Per path hop: byPathHop index entry (pointer + map bucket) perPathHopBytes = 50 - // Per subpath entry in spTxIndex: string key + slice append + pointer - perSubpathEntryBytes = 40 - // ParsedDecoded caches json.Unmarshal of DecodedJSON as a // map[string]interface{}, about 4x the JSON length. Analytics touch // every tx, so it is charged up front rather than when the cache fills @@ -5650,8 +5541,8 @@ const ( ) // estimateStoreTxBytes returns the estimated memory cost of a StoreTx (excluding observations). -// Includes byPathHop entries and spTxIndex subpath entries; the dedup maps of a -// busy tx are charged separately (obsIndexBytes). +// Includes byPathHop entries; the dedup maps of a busy tx are charged +// separately (obsIndexBytes). func estimateStoreTxBytes(tx *StoreTx) int64 { base := int64(storeTxBaseBytes) base += int64(len(tx.RawHex) + len(tx.Hash) + len(tx.DecodedJSON) + len(tx.PathJSON)) @@ -5662,12 +5553,6 @@ func estimateStoreTxBytes(tx *StoreTx) int64 { hops := int64(len(txGetParsedPath(tx))) base += hops * perPathHopBytes - // spTxIndex: O(path²) subpath combinations - if hops > 1 { - subpaths := hops * (hops - 1) / 2 - base += subpaths * perSubpathEntryBytes - } - return base } @@ -5681,7 +5566,6 @@ func estimateStoreTxBytesTypical(numObs int) int64 { base += decodedCacheFactor * 200 hops := int64(3) base += hops * perPathHopBytes - base += (hops * (hops - 1) / 2) * perSubpathEntryBytes base += resolvedRelayBytes(typicalResolvedRelays) // Add observation costs obsBase := int64(storeObsBaseBytes) + 30 + 30 + 60 + 25 // observer ID + name + path + timestamp @@ -5951,9 +5835,6 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { // Remove from resolved pubkey index s.removeFromResolvedPubkeyIndex(tx.ID) - - // Remove from subpath index - removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, tx) } // Remove from the path-hop index: raw AND resolved keys, all duplicates, // in one pass per batch (#115). See evictFromPathHopIndex. @@ -6185,7 +6066,7 @@ func hasUpperASCII(s string) bool { // loops that resolve hops outside any per-tx scope (subpath/topology // aggregations). Caller passes the slice of txs to consider; we union the // per-tx contexts with de-dup. Used by call sites that read from precomputed -// indices (s.spIndex, s.spTxIndex) or that resolve user-supplied hops. +// indices or that resolve user-supplied hops. // // Result is order-independent in semantics; iteration order is deterministic // only modulo Go's map iteration (acceptable — the resolver's tier-2 averages @@ -11062,90 +10943,42 @@ func (s *PacketStore) GetNodeHopAnalytics(pubkey string, days int) (*NodeHopAnal // ignored the Time window filter because callers always reached the unbounded // path. With a zero TimeWindow this is byte-equivalent to GetAnalyticsSubpaths. // -// For non-zero windows we iterate the packet list and filter on -// `tx.FirstSeen`. We deliberately do not consult the precomputed `spIndex` -// (which has no per-tx timestamp), so the windowed path is O(N_tx · path²); -// this matches the slow region-filtered path and keeps the fast unbounded -// hot path untouched. Results are cached by (region|area|window) so repeated -// renders of the same window don't re-scan. +// For non-zero windows the transmissions whose `tx.FirstSeen` falls inside +// the window are counted one by one, like a region-filtered query, so the +// windowed path is O(N_tx · path²). Results are cached by +// (region|area|window) so repeated renders of the same window don't re-scan. func (s *PacketStore) GetAnalyticsSubpathsWithWindow(region string, minLen, maxLen, limit int, window TimeWindow) map[string]interface{} { if window.IsZero() { return s.GetAnalyticsSubpaths(region, minLen, maxLen, limit) } cacheKey := fmt.Sprintf("%s|%d|%d|%d|w=%s", region, minLen, maxLen, limit, window.CacheKey()) - s.cacheMu.Lock() - if cached, ok := s.subpathCache[cacheKey]; ok && time.Now().Before(cached.expiresAt) { - s.cacheHits++ - s.cacheMu.Unlock() - return cached.data - } - s.cacheMisses++ - s.cacheMu.Unlock() - - result := s.computeAnalyticsSubpathsWindowed(region, minLen, maxLen, limit, window) - - s.cacheMu.Lock() - putCachedResultLocked(s.subpathCache, cacheKey, &cachedResult{data: result, expiresAt: time.Now().Add(s.rfCacheTTL)}) - s.cacheMu.Unlock() - - return result + return s.loadSubpaths(cacheKey, func() *cachedResult { + return &cachedResult{data: s.computeAnalyticsSubpathsWindowed(region, minLen, maxLen, limit, window)} + }).data } -// computeAnalyticsSubpathsWindowed iterates s.packets and bounds the result -// to transmissions whose FirstSeen falls inside `window`. Optionally also -// region-filters. Mirrors computeSubpathsSlow but with the window predicate -// inlined and without requiring a non-empty region. +// computeAnalyticsSubpathsWindowed counts the subpaths of the transmissions +// whose FirstSeen falls inside `window` (a zero window holds them all), +// optionally region-filtered: per transmission, by resolved hop name, every +// length from minLen to maxLen, rawHops as the path stores them. func (s *PacketStore) computeAnalyticsSubpathsWindowed(region string, minLen, maxLen, limit int, window TimeWindow) map[string]interface{} { _, pm := s.getCachedNodesAndPM() var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) } - s.mu.RLock() - defer s.mu.RUnlock() - - contextPubkeys := buildAggregateHopContextPubkeys(s.packets, pm) - hopCache := make(map[string]*nodeInfo) - graph := s.graph.Load() - resolveHop := func(hop string) string { - if cached, ok := hopCache[hop]; ok { - if cached != nil { - return cached.Name - } - return hop - } - r, _, _ := pm.resolveWithContext(hop, contextPubkeys, graph) - hopCache[hop] = r - if r != nil { - return r.Name - } - return hop - } + paths, contextPubkeys := s.snapshotSubpathPaths(regionObs, pm) + resolveHop := subpathHopResolver(pm, contextPubkeys, s.graph.Load()) subpathCounts := make(map[string]*subpathAccum) totalPaths := 0 - for _, tx := range s.packets { - if !window.Includes(tx.FirstSeen) { - continue - } - hops := txGetParsedPath(tx) - if len(hops) < 2 { + for _, p := range paths { + if !window.IsZero() && !window.Includes(p.firstSeen) { continue } - if regionObs != nil { - match := false - for _, obs := range tx.Observations { - if regionObs[obs.ObserverID] { - match = true - break - } - } - if !match { - continue - } - } + hops := p.hops totalPaths++ named := make([]string, len(hops)) @@ -11172,23 +11005,70 @@ func (s *PacketStore) computeAnalyticsSubpathsWindowed(region string, minLen, ma func (s *PacketStore) GetAnalyticsSubpaths(region string, minLen, maxLen, limit int) map[string]interface{} { cacheKey := fmt.Sprintf("%s|%d|%d|%d", region, minLen, maxLen, limit) + return s.loadSubpaths(cacheKey, func() *cachedResult { + return &cachedResult{data: s.computeAnalyticsSubpaths(region, minLen, maxLen, limit)} + }).data +} +// loadSubpaths returns the subpathCache entry for cacheKey or, on a miss, +// the entry of one count shared by every concurrent caller of cacheKey, +// which is stored with the TTL set. +func (s *PacketStore) loadSubpaths(cacheKey string, count func() *cachedResult) *cachedResult { s.cacheMu.Lock() if cached, ok := s.subpathCache[cacheKey]; ok && time.Now().Before(cached.expiresAt) { s.cacheHits++ s.cacheMu.Unlock() - return cached.data + return cached } s.cacheMisses++ s.cacheMu.Unlock() - result := s.computeAnalyticsSubpaths(region, minLen, maxLen, limit) + return shareSubpathCount(s, "1:"+cacheKey, func() (*cachedResult, bool) { + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + return s.subpathCachedLocked(cacheKey) + }, func() *cachedResult { + entry := count() + s.cacheMu.Lock() + entry.expiresAt = time.Now().Add(s.rfCacheTTL) + putCachedResultLocked(s.subpathCache, cacheKey, entry) + s.cacheMu.Unlock() + return entry + }) +} - s.cacheMu.Lock() - putCachedResultLocked(s.subpathCache, cacheKey, &cachedResult{data: result, expiresAt: time.Now().Add(s.rfCacheTTL)}) - s.cacheMu.Unlock() +// subpathCachedLocked returns the unexpired subpathCache entry for key. +// Caller holds s.cacheMu. +func (s *PacketStore) subpathCachedLocked(key string) (*cachedResult, bool) { + cached, ok := s.subpathCache[key] + if !ok || !time.Now().Before(cached.expiresAt) { + return nil, false + } + return cached, true +} - return result +// shareSubpathCount runs count for a subpathCache miss, once for every +// concurrent caller of flightKey (#413). A count scans the whole store and +// allocates about as much as the removed index held, so N overlapping misses +// must not run N counts (PR #529 review F1). cached checks the cache again +// inside the flight, so a caller that missed just before an earlier count +// stored its result uses that result instead of counting again. +func shareSubpathCount[T any](s *PacketStore, flightKey string, cached func() (T, bool), count func() T) T { + s.subpathHookAt("miss") + v, _, _ := s.subpathSF.Do(flightKey, func() (interface{}, error) { + if v, ok := cached(); ok { + return v, nil + } + s.subpathHookAt("count") + return count(), nil + }) + return v.(T) +} + +func (s *PacketStore) subpathHookAt(stage string) { + if s.subpathHook != nil { + s.subpathHook(stage) + } } // GetAnalyticsSubpathsBulkWithWindow is the window-aware variant. For a zero @@ -11207,7 +11087,7 @@ func (s *PacketStore) GetAnalyticsSubpathsBulkWithWindow(region string, groups [ } // GetAnalyticsSubpathsBulk returns multiple length-range buckets from a single -// scan of the subpath index, avoiding repeated iterations. +// count of the subpaths, avoiding repeated scans. func (s *PacketStore) GetAnalyticsSubpathsBulk(region string, groups []subpathGroup) []map[string]interface{} { // For region queries or when there are few groups, fall back to individual calls // which benefit from per-key caching. @@ -11219,57 +11099,80 @@ func (s *PacketStore) GetAnalyticsSubpathsBulk(region string, groups []subpathGr return results } - // Check if all groups are cached. - allCached := true - cachedResults := make([]map[string]interface{}, len(groups)) - s.cacheMu.Lock() + cacheKeys := make([]string, len(groups)) for i, g := range groups { - cacheKey := fmt.Sprintf("|%d|%d|%d", g.MinLen, g.MaxLen, g.Limit) - if cached, ok := s.subpathCache[cacheKey]; ok && time.Now().Before(cached.expiresAt) { - cachedResults[i] = cached.data - } else { - allCached = false - break + cacheKeys[i] = fmt.Sprintf("|%d|%d|%d", g.MinLen, g.MaxLen, g.Limit) + } + // allCached returns the entries when every group is cached. + allCached := func() ([]*cachedResult, bool) { + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + entries := make([]*cachedResult, len(groups)) + for i, key := range cacheKeys { + cached, ok := s.subpathCachedLocked(key) + if !ok { + return nil, false + } + entries[i] = cached } + return entries, true } - if allCached { + entries, ok := allCached() + if ok { + s.cacheMu.Lock() s.cacheHits += int64(len(groups)) s.cacheMu.Unlock() - return cachedResults + } else { + entries = shareSubpathCount(s, "n:"+strings.Join(cacheKeys, ";"), allCached, func() []*cachedResult { + perGroup, totalPaths := s.countSubpathGroups(groups) + entries := make([]*cachedResult, len(groups)) + for i, g := range groups { + entries[i] = &cachedResult{data: s.rankSubpaths(perGroup[i], totalPaths, g.Limit)} + } + + // Cache individual results for future single-key lookups too. + s.cacheMu.Lock() + for i, e := range entries { + e.expiresAt = time.Now().Add(s.rfCacheTTL) + putCachedResultLocked(s.subpathCache, cacheKeys[i], e) + } + s.cacheMu.Unlock() + return entries + }) } - s.cacheMu.Unlock() + results := make([]map[string]interface{}, len(groups)) + for i, e := range entries { + results[i] = e.data + } + return results +} - // Refresh DB-backed metadata before the store-only scan. +// subpathMaxHops is the longest subpath the unfiltered Route Patterns count, +// the cap the former precomputed index had (#413). Window and region +// queries count every length up to maxLen. +const subpathMaxHops = 8 + +// countSubpathGroups counts the subpaths of every transmission once for all +// groups: by raw hops, lower-cased and comma-joined, 2 to subpathMaxHops +// long, each occurrence counted. Those keys are then resolved to node names +// per group, raw keys with the same names merged. totalPaths is the number +// of transmissions with a path of 2+ hops. +func (s *PacketStore) countSubpathGroups(groups []subpathGroup) (perGroup []map[string]*subpathAccum, totalPaths int) { _, pm := s.getCachedNodesAndPM() - s.mu.RLock() - // Aggregate hop-disambiguation context across all packets so the - // resolver's tiers 1 and 2 light up even on this bulk-aggregate path - // (the index iterates raw subpath strings, not per-tx). See #1197. - contextPubkeys := buildAggregateHopContextPubkeys(s.packets, pm) - hopCache := make(map[string]*nodeInfo) - graph := s.graph.Load() // hoist out of resolver closure (PR #1208 carmack #1) - resolveHop := func(hop string) string { - if cached, ok := hopCache[hop]; ok { - if cached != nil { - return cached.Name - } - return hop - } - r, _, _ := pm.resolveWithContext(hop, contextPubkeys, graph) - hopCache[hop] = r - if r != nil { - return r.Name - } - return hop + paths, contextPubkeys := s.snapshotSubpathPaths(nil, pm) + maxLen := 0 + for _, g := range groups { + maxLen = max(maxLen, g.MaxLen) } + counts := countRawSubpaths(paths, min(maxLen, subpathMaxHops)) + resolveHop := subpathHopResolver(pm, contextPubkeys, s.graph.Load()) - perGroup := make([]map[string]*subpathAccum, len(groups)) + perGroup = make([]map[string]*subpathAccum, len(groups)) for i := range groups { perGroup[i] = make(map[string]*subpathAccum) } - - for rawKey, count := range s.spIndex { - hops := strings.Split(rawKey, ",") + for _, c := range counts { + hops := strings.Split(c.key, ",") hopLen := len(hops) // Resolve hop names once, reuse across groups. @@ -11294,57 +11197,145 @@ func (s *PacketStore) GetAnalyticsSubpathsBulk(region string, groups []subpathGr entry = &subpathAccum{} perGroup[gi][namedKey] = entry } - entry.add(rawKey, count) + entry.add(c.key, c.count) } } - totalPaths := s.spTotalPaths - s.mu.RUnlock() + return perGroup, len(paths) +} - results := make([]map[string]interface{}, len(groups)) - for i, g := range groups { - results[i] = s.rankSubpaths(perGroup[i], totalPaths, g.Limit) - } +// subpathPath is a transmission's parsed hop list and FirstSeen, copied by +// snapshotSubpathPaths. hops is shared with the transmission, not copied: a +// path change installs a new slice and never edits one in place. +type subpathPath struct { + hops []string + firstSeen string +} - // Cache individual results for future single-key lookups too. - s.cacheMu.Lock() - for i, g := range groups { - cacheKey := fmt.Sprintf("|%d|%d|%d", g.MinLen, g.MaxLen, g.Limit) - putCachedResultLocked(s.subpathCache, cacheKey, &cachedResult{data: results[i], expiresAt: time.Now().Add(s.rfCacheTTL)}) +// snapshotSubpathPaths copies, under s.mu, what Route Patterns count: every +// transmission with a path of 2+ hops (with an observer in regionObs when +// regionObs is non-nil), and the hop-disambiguation context aggregated over +// all packets (#1197). Enumerating, resolving and ranking run on the copy +// after the lock is released (#413). +func (s *PacketStore) snapshotSubpathPaths(regionObs map[string]bool, pm *prefixMap) ([]subpathPath, []string) { + s.mu.RLock() + defer s.mu.RUnlock() + paths := make([]subpathPath, 0, len(s.packets)) + for _, tx := range s.packets { + hops := txGetParsedPath(tx) + if len(hops) < 2 { + continue + } + if regionObs != nil { + match := false + for _, obs := range tx.Observations { + if regionObs[obs.ObserverID] { + match = true + break + } + } + if !match { + continue + } + } + paths = append(paths, subpathPath{hops: hops, firstSeen: tx.FirstSeen}) } - s.cacheMu.Unlock() - - return results + return paths, buildAggregateHopContextPubkeys(s.packets, pm) } -// subpathAccum holds a running count for a single named subpath. -type subpathAccum struct { +// rawSubpathCount is one distinct raw subpath key and its occurrences. +type rawSubpathCount struct { + key string count int - raw string // smallest raw-hop key merged into this named subpath (rawHops in the API response) } -// add merges n occurrences of raw into the accumulator. Several raw keys can -// resolve to the same display name (two nodes sharing a name), so the one -// reported as rawHops is the lexicographically smallest: deterministic across -// map iteration orders, which the visibility filter (#410) relies on. -func (a *subpathAccum) add(raw string, n int) { - if a.count == 0 || raw < a.raw { - a.raw = raw +// countRawSubpaths counts every subpath of 2 to maxHops hops of paths, each +// occurrence once, by its lower-cased, comma-joined hops. The key is built in +// a reused buffer and allocated only for a subpath not seen before. +func countRawSubpaths(paths []subpathPath, maxHops int) []rawSubpathCount { + var counts []rawSubpathCount + idx := make(map[string]int) + var buf []byte + for _, p := range paths { + hops := p.hops + for start := 0; start+2 <= len(hops); start++ { + buf = appendLowerHop(buf[:0], hops[start]) + for l := 2; l <= maxHops && start+l <= len(hops); l++ { + buf = appendLowerHop(append(buf, ','), hops[start+l-1]) + if i, ok := idx[string(buf)]; ok { + counts[i].count++ + continue + } + idx[string(buf)] = len(counts) + counts = append(counts, rawSubpathCount{key: string(buf), count: 1}) + } + } } - a.count += n + return counts } -func (s *PacketStore) computeAnalyticsSubpaths(region string, minLen, maxLen, limit int) map[string]interface{} { - _, pm := s.getCachedNodesAndPM() - regionObs := s.resolveRegionObservers(region) - s.mu.RLock() - defer s.mu.RUnlock() +// subpathOccurrences is how often key, a lower-cased comma-joined subpath, +// occurs in hops as a run of 2 to subpathMaxHops hops (each run compared +// lower-cased): the count countRawSubpaths gives key for this path. +func subpathOccurrences(hops []string, key string) int { + n := 0 + for start := 0; start+2 <= len(hops); start++ { + pos, ok := matchLowerHop(key, 0, hops[start]) + for l := 2; ok && l <= subpathMaxHops && start+l <= len(hops); l++ { + if pos >= len(key) || key[pos] != ',' { + break + } + if pos, ok = matchLowerHop(key, pos+1, hops[start+l-1]); ok && pos == len(key) { + n++ + break + } + } + } + return n +} + +// matchLowerHop reports whether key continues at pos with +// strings.ToLower(hop), and where that ends. It compares in place for an +// ASCII hop (the hex prefixes every path holds). +func matchLowerHop(key string, pos int, hop string) (int, bool) { + for i := 0; i < len(hop); i++ { + c := hop[i] + if c >= utf8.RuneSelf { + lower := strings.ToLower(hop) + return pos + len(lower), strings.HasPrefix(key[pos:], lower) + } + if 'A' <= c && c <= 'Z' { + c += 'a' - 'A' + } + if pos+i >= len(key) || key[pos+i] != c { + return pos, false + } + } + return pos + len(hop), true +} + +// appendLowerHop appends strings.ToLower(hop) to buf, without allocating for +// an ASCII hop (the hex prefixes every path holds). +func appendLowerHop(buf []byte, hop string) []byte { + for i := 0; i < len(hop); i++ { + if hop[i] >= utf8.RuneSelf { + return append(buf, strings.ToLower(hop)...) + } + } + for i := 0; i < len(hop); i++ { + c := hop[i] + if 'A' <= c && c <= 'Z' { + c += 'a' - 'A' + } + buf = append(buf, c) + } + return buf +} - // Aggregate hop-disambiguation context across all packets — bulk - // aggregator over s.spIndex / per-tx fallback both need it. See #1197. - contextPubkeys := buildAggregateHopContextPubkeys(s.packets, pm) +// subpathHopResolver resolves a hop to the name of the node it stands for, or +// to the hop itself when it matches no node, once per hop per request. +func subpathHopResolver(pm *prefixMap, contextPubkeys []string, graph *NeighborGraph) func(string) string { hopCache := make(map[string]*nodeInfo) - graph := s.graph.Load() // hoist out of resolver closure (PR #1208 carmack #1) - resolveHop := func(hop string) string { + return func(hop string) string { if cached, ok := hopCache[hop]; ok { if cached != nil { return cached.Name @@ -11358,109 +11349,83 @@ func (s *PacketStore) computeAnalyticsSubpaths(region string, minLen, maxLen, li } return hop } +} - // For region queries fall back to packet iteration (region filtering - // requires per-transmission observer checks). - if region != "" { - return s.computeSubpathsSlow(regionObs, minLen, maxLen, limit, resolveHop) - } +// subpathAccum holds a running count for a single named subpath. +type subpathAccum struct { + count int + raw string // smallest raw-hop key merged into this named subpath (rawHops in the API response) +} - // Fast path: read from precomputed raw-hop subpath index. - // Resolve raw hop prefixes to names and merge counts. - namedCounts := make(map[string]*subpathAccum, len(s.spIndex)) - for rawKey, count := range s.spIndex { - hops := strings.Split(rawKey, ",") - hopLen := len(hops) - if hopLen < minLen || hopLen > maxLen { - continue - } - named := make([]string, hopLen) - for i, h := range hops { - named[i] = resolveHop(h) - } - namedKey := strings.Join(named, " → ") - entry := namedCounts[namedKey] - if entry == nil { - entry = &subpathAccum{} - namedCounts[namedKey] = entry - } - entry.add(rawKey, count) +// add merges n occurrences of raw into the accumulator. Several raw keys can +// resolve to the same display name (two nodes sharing a name), so the one +// reported as rawHops is the lexicographically smallest: deterministic across +// map iteration orders, which the visibility filter (#410) relies on. +func (a *subpathAccum) add(raw string, n int) { + if a.count == 0 || raw < a.raw { + a.raw = raw } - - return s.rankSubpaths(namedCounts, s.spTotalPaths, limit) + a.count += n } -// computeSubpathsSlow is the original O(N) packet-iteration path, used only -// for region-filtered queries where we must check per-transmission observers. -func (s *PacketStore) computeSubpathsSlow(regionObs map[string]bool, minLen, maxLen, limit int, resolveHop func(string) string) map[string]interface{} { - - subpathCounts := make(map[string]*subpathAccum) - totalPaths := 0 - - for _, tx := range s.packets { - hops := txGetParsedPath(tx) - if len(hops) < 2 { - continue - } - if regionObs != nil { - match := false - for _, obs := range tx.Observations { - if regionObs[obs.ObserverID] { - match = true - break - } - } - if !match { - continue - } - } - totalPaths++ +func (s *PacketStore) computeAnalyticsSubpaths(region string, minLen, maxLen, limit int) map[string]interface{} { + // Region filtering needs each transmission's observers, so a region + // query is counted per transmission like a windowed one. + if region != "" { + return s.computeAnalyticsSubpathsWindowed(region, minLen, maxLen, limit, TimeWindow{}) + } + perGroup, totalPaths := s.countSubpathGroups([]subpathGroup{{minLen, maxLen, limit}}) + return s.rankSubpaths(perGroup[0], totalPaths, limit) +} - named := make([]string, len(hops)) - for i, h := range hops { - named[i] = resolveHop(h) - } +// subpathEntry is one named subpath of a ranking. +type subpathEntry struct { + path string + count int + raw string +} - for l := minLen; l <= maxLen && l <= len(named); l++ { - for start := 0; start <= len(named)-l; start++ { - sub := strings.Join(named[start:start+l], " → ") - raw := strings.Join(hops[start:start+l], ",") - entry := subpathCounts[sub] - if entry == nil { - entry = &subpathAccum{} - subpathCounts[sub] = entry - } - entry.add(raw, 1) - } - } +// subpathRanksBefore is the Route Patterns order: the most frequent first and +// #256 ties broken on the path, so the order and the limit cut are stable. +func subpathRanksBefore(a, b subpathEntry) bool { + if a.count != b.count { + return a.count > b.count } + return a.path < b.path +} - return s.rankSubpaths(subpathCounts, totalPaths, limit) +// subpathTopN is a heap of the best entries seen so far, the worst on top. +type subpathTopN []subpathEntry + +func (h subpathTopN) Len() int { return len(h) } +func (h subpathTopN) Less(i, j int) bool { return subpathRanksBefore(h[j], h[i]) } +func (h subpathTopN) Swap(i, j int) { h[i], h[j] = h[j], h[i] } +func (h *subpathTopN) Push(x any) { *h = append(*h, x.(subpathEntry)) } +func (h *subpathTopN) Pop() any { + old := *h + e := old[len(old)-1] + *h = old[:len(old)-1] + return e } -// rankSubpaths sorts accumulated subpath counts by frequency, truncates to -// limit, and builds the API response map. +// rankSubpaths keeps the `limit` highest-ranked accumulated subpaths and +// builds the API response map. Paths are distinct map keys, so the order is +// total and the bounded heap keeps exactly the entries a full sort cut at +// limit would, without sorting every distinct subpath (#413). func (s *PacketStore) rankSubpaths(counts map[string]*subpathAccum, totalPaths, limit int) map[string]interface{} { - type subpathEntry struct { - path string - count int - raw string - } - ranked := make([]subpathEntry, 0, len(counts)) + limit = max(limit, 0) + top := make(subpathTopN, 0, min(limit, len(counts))) for path, data := range counts { - ranked = append(ranked, subpathEntry{path, data.count, data.raw}) - } - // #256: counts iteration order is random; ties break on the path so the - // order and the limit cut are stable. - sort.Slice(ranked, func(i, j int) bool { - if ranked[i].count != ranked[j].count { - return ranked[i].count > ranked[j].count + e := subpathEntry{path, data.count, data.raw} + if len(top) < limit { + heap.Push(&top, e) + } else if limit > 0 && subpathRanksBefore(e, top[0]) { + top[0] = e + heap.Fix(&top, 0) } - return ranked[i].path < ranked[j].path - }) - if len(ranked) > limit { - ranked = ranked[:limit] } + ranked := []subpathEntry(top) + sort.Slice(ranked, func(i, j int) bool { return subpathRanksBefore(ranked[i], ranked[j]) }) subpaths := make([]map[string]interface{}, 0, len(ranked)) for _, e := range ranked { @@ -11534,22 +11499,92 @@ type subpathDetailObserver struct { count int } +// subpathMatch is what a subpath detail reads of one matched transmission, +// copied under s.mu so the detail is built after the lock is released. +type subpathMatch struct { + firstSeen string + snr, rssi float64 + hasSNR, hasRSSI bool + observerID, observerName string + hops []string // shared with the transmission, see subpathPath + hopContext []string +} + +// subpathTxHops is a transmission and its hop list, copied under s.mu for +// SubpathDetail to match outside it: tx is only read under s.mu, hops is +// shared with tx (see subpathPath). +type subpathTxHops struct { + tx *StoreTx + hops []string +} + // SubpathDetail gathers the parts of a subpath detail. func (s *PacketStore) SubpathDetail(rawHops []string) *subpathDetailParts { _, pm := s.getCachedNodesAndPM() - s.mu.RLock() - defer s.mu.RUnlock() - // Build the subpath key the same way the index does (lowercase, comma-joined) + // The subpath key as Route Patterns counts it (lowercase, comma-joined) spKey := strings.ToLower(strings.Join(rawHops, ",")) - // Direct lookup instead of scanning all packets - matchedTxs := s.spTxIndex[spKey] + // Under the lock only the hop lists are copied. Finding the + // transmissions that hold the subpath, once per occurrence as Route + // Patterns counts them, runs after it is released, so a writer waits for + // a pointer copy per packet rather than a comparison of every path (#413, + // PR #529 review F2). + s.mu.RLock() + n := len(s.packets) + s.mu.RUnlock() + all := make([]subpathTxHops, 0, n) // allocated outside the lock + s.mu.RLock() + for _, tx := range s.packets { + if hops := txGetParsedPath(tx); len(hops) >= 2 { + all = append(all, subpathTxHops{tx: tx, hops: hops}) + } + } + s.mu.RUnlock() + + s.subpathHookAt("match") + var hits []subpathTxHops + for _, c := range all { + for n := subpathOccurrences(c.hops, spKey); n > 0; n-- { + hits = append(hits, c) + } + } + all = nil + // Under the lock again, O(matches): copy what the detail reads of the + // matched transmissions. Which transmissions matched, and on which path, + // is as of the first copy. + s.mu.RLock() + matchedTxs := make([]*StoreTx, len(hits)) + for i, c := range hits { + matchedTxs[i] = c.tx + } // Hop-disambiguation context: union over the matched txs that produced // this subpath. This is the right scope — those are the packets that // witnessed the requested hop sequence. See #1197. contextPubkeys := buildAggregateHopContextPubkeys(matchedTxs, pm) + matched := make([]subpathMatch, len(hits)) + for i, c := range hits { + tx, m := c.tx, &matched[i] + m.firstSeen, m.observerID, m.observerName = tx.FirstSeen, tx.ObserverID, tx.ObserverName + if tx.SNR != nil { + m.snr, m.hasSNR = *tx.SNR, true + } + if tx.RSSI != nil { + m.rssi, m.hasRSSI = *tx.RSSI, true + } + m.hops = c.hops + // Per-tx context so the resolver picks the right candidate when + // prefixes are ambiguous (#1197); a tx's occurrences are adjacent + // and share it. + if i > 0 && hits[i-1].tx == tx { + m.hopContext = matched[i-1].hopContext + } else { + m.hopContext = buildHopContextPubkeys(tx, pm) + } + } + s.mu.RUnlock() + // Hoist atomic graph.Load() once for the whole request (PR #1208 // carmack #1) — used by the rawHops loop AND the matched-tx loop below. graph := s.graph.Load() @@ -11560,10 +11595,10 @@ func (s *PacketStore) SubpathDetail(rawHops []string) *subpathDetailParts { pm: pm, graph: graph, hopContext: contextPubkeys, - matchCount: len(matchedTxs), + matchCount: len(matched), hourBuckets: make([]int, 24), } - // Resolve the requested hops (copies: the parts outlive the lock) + // Resolve the requested hops (copies: the parts outlive the request) for i, hop := range rawHops { if r, _, _ := pm.resolveWithContext(hop, contextPubkeys, graph); r != nil { n := *r @@ -11573,8 +11608,8 @@ func (s *PacketStore) SubpathDetail(rawHops []string) *subpathDetailParts { parents := map[string]*subpathDetailParent{} observers := map[[2]string]int{} // keyed by observer id and name - for _, tx := range matchedTxs { - ts := tx.FirstSeen + for _, m := range matched { + ts := m.firstSeen if ts != "" { if p.firstSeen == "" || ts < p.firstSeen { p.firstSeen = ts @@ -11590,27 +11625,24 @@ func (s *PacketStore) SubpathDetail(rawHops []string) *subpathDetailParts { p.hourBuckets[t.Hour()]++ } } - if tx.SNR != nil { - p.snrSum += *tx.SNR + if m.hasSNR { + p.snrSum += m.snr p.snrCount++ } - if tx.RSSI != nil { - p.rssiSum += *tx.RSSI + if m.hasRSSI { + p.rssiSum += m.rssi p.rssiCount++ } - if tx.ObserverName != "" { - observers[[2]string{tx.ObserverID, tx.ObserverName}]++ + if m.observerName != "" { + observers[[2]string{m.observerID, m.observerName}]++ } - // Full parent path (resolved). Per-tx context so the resolver picks - // the right candidate when prefixes are ambiguous. See #1197. - txCtx := buildHopContextPubkeys(tx, pm) - hops := txGetParsedPath(tx) - pathHops := make([]subpathDetailHop, len(hops)) - key := make([]string, 0, 2*len(hops)) - for i, h := range hops { + // Full parent path (resolved), with the tx's own hop context. + pathHops := make([]subpathDetailHop, len(m.hops)) + key := make([]string, 0, 2*len(m.hops)) + for i, h := range m.hops { pathHops[i].raw = h - if r, _, _ := pm.resolveWithContext(h, txCtx, graph); r != nil { + if r, _, _ := pm.resolveWithContext(h, m.hopContext, graph); r != nil { pathHops[i].pubkey, pathHops[i].name = r.PublicKey, r.Name } key = append(key, h, pathHops[i].pubkey) diff --git a/cmd/server/subpath_casefold_groups_413_test.go b/cmd/server/subpath_casefold_groups_413_test.go new file mode 100644 index 000000000..b71e65376 --- /dev/null +++ b/cmd/server/subpath_casefold_groups_413_test.go @@ -0,0 +1,79 @@ +package main + +import ( + "fmt" + "reflect" + "strings" + "testing" + "time" +) + +// Hops are hex prefixes in practice, but the counting folds case like +// strings.ToLower for any hop, as the removed index did: a non-ASCII hop +// takes the fallback in appendLowerHop and matchLowerHop (PR #529 review N1). +// "ẞ" lower-cases to "ß", one byte shorter, so the match position must +// follow the folded length, not the hop's. +func TestSubpathNonASCIIHopsFoldCase_413(t *testing.T) { + for _, hop := range []string{"ÄÖ", "äö", "ẞx", "aÄ", "AB", ""} { + want := strings.ToLower(hop) + if got := string(appendLowerHop([]byte("p,"), hop)); got != "p,"+want { + t.Errorf("appendLowerHop(%q) = %q, want %q", hop, got, "p,"+want) + } + key := "p," + want + ",10" + if end, ok := matchLowerHop(key, 2, hop); !ok || end != 2+len(want) { + t.Errorf("matchLowerHop(%q, 2, %q) = %d, %v, want %d, true", key, hop, end, ok, 2+len(want)) + } + } + if _, ok := matchLowerHop("öx,10", 0, "ÄX"); ok { + t.Error(`matchLowerHop matched "ÄX" against "öx"`) + } + + pt := PayloadTXT_MSG + var packets []*StoreTx + for i, path := range []string{`["ÄÖ","10"]`, `["äö","10"]`, `["Äö","10"]`, `["ẞX","10"]`, `["ßx","10"]`} { + packets = append(packets, &StoreTx{ + ID: i + 1, Hash: fmt.Sprintf("cf%d", i), PayloadType: &pt, + FirstSeen: time.Now().UTC().Format(time.RFC3339), PathJSON: path, + }) + } + s := newSubpath413Store([]nodeInfo{{PublicKey: "ee99000000000000", Name: "E9", Role: "repeater"}}, packets) + + counts := map[string]int{} + for _, row := range s.GetAnalyticsSubpaths("", 2, 2, 10)["subpaths"].([]map[string]interface{}) { + counts[row["path"].(string)] = row["count"].(int) + } + if want := map[string]int{"äö → 10": 3, "ßx → 10": 2}; !reflect.DeepEqual(counts, want) { + t.Errorf("subpaths = %v, want %v", counts, want) + } + for hops, want := range map[string]int{"ÄÖ,10": 3, "äÖ,10": 3, "ẞx,10": 2, "ßX,10": 2} { + if got := s.GetSubpathDetail(strings.Split(hops, ",")).TotalMatches; got != want { + t.Errorf("subpath-detail %s: totalMatches = %d, want %d", hops, got, want) + } + } +} + +// The bulk count runs once to the longest group's maxLen, whatever order the +// caller lists the groups in (PR #529 review N2): every group must equal the +// same group asked for on its own. +func TestSubpathsBulkGroupOrderDoesNotCapLength_413(t *testing.T) { + nodes, packets := subpath413Data(1500, 150, 7) + s := newSubpath413Store(nodes, packets) + for _, groups := range [][]subpathGroup{ + {{5, 8, 15}, {4, 4, 20}, {3, 3, 30}, {2, 2, 50}}, + {{2, 2, 50}, {5, 8, 15}, {3, 3, 30}}, + {{3, 3, 30}, {2, 8, 40}, {2, 2, 10}}, + } { + clearSubpathCache413(s) + bulk := s.GetAnalyticsSubpathsBulk("", groups) + for i, g := range groups { + clearSubpathCache413(s) + want := s.GetAnalyticsSubpaths("", g.MinLen, g.MaxLen, g.Limit) + if len(want["subpaths"].([]map[string]interface{})) == 0 { + t.Fatalf("setup: group %v has no subpaths", g) + } + if !reflect.DeepEqual(bulk[i], want) { + t.Errorf("groups %v: group %v differs from the same group on its own", groups, g) + } + } + } +} diff --git a/cmd/server/subpath_detail_tie_order_273_test.go b/cmd/server/subpath_detail_tie_order_273_test.go index c6f55411b..5469e88ce 100644 --- a/cmd/server/subpath_detail_tie_order_273_test.go +++ b/cmd/server/subpath_detail_tie_order_273_test.go @@ -13,15 +13,10 @@ import ( // at N it was random which tied entries made the cut. These tests compute the // detail many times and require the same, fully specified order every time. -// newSubpathDetailStore builds a tie-order store and feeds every packet through -// the same subpath index that production uses, so GetSubpathDetail resolves its -// matched transmissions exactly as it would at runtime. +// newSubpathDetailStore builds a tie-order store; GetSubpathDetail finds its +// matched transmissions in the packets exactly as it would at runtime. func newSubpathDetailStore(packets []*StoreTx) *PacketStore { - s := newTieOrderStore(packets) - for _, tx := range packets { - addTxToSubpathIndexFull(s.spIndex, s.spTxIndex, tx) - } - return s + return newTieOrderStore(packets) } func TestSubpathDetailParentPathsTieOrder_273(t *testing.T) { diff --git a/cmd/server/subpath_fixture_413_test.go b/cmd/server/subpath_fixture_413_test.go new file mode 100644 index 000000000..7e40cdce2 --- /dev/null +++ b/cmd/server/subpath_fixture_413_test.go @@ -0,0 +1,143 @@ +package main + +import ( + "encoding/json" + "fmt" + "math/rand" + "sort" + "strings" + "time" +) + +// Synthetic Route Patterns data for #413: numNodes nodes, a quarter of them +// repeaters wired into a sparse neighbour mesh, and numTx transmissions whose +// paths are random walks over that mesh. Path lengths follow the hop-count +// distribution of test-fixtures/e2e-fixture.db (median 7, tail to 37), hops +// are upper-case prefixes like the decoder writes, 1-byte for most packets and +// 2-byte for the rest, so prefixes collide the way they do on a real mesh. +// Seeded, so every run (and the before/after benchmark) sees the same data. + +// subpath413PathLens is the fixture's path-length histogram: weight per length. +var subpath413PathLens = map[int]int{ + 0: 101, 1: 28, 2: 9, 3: 2, 4: 16, 5: 49, 6: 65, 7: 63, 8: 57, 9: 33, + 10: 28, 11: 13, 12: 9, 13: 9, 14: 4, 15: 3, 16: 1, 17: 1, 18: 1, 19: 3, + 20: 1, 22: 2, 31: 1, 37: 1, +} + +func subpath413Data(numTx, numNodes int, seed int64) ([]nodeInfo, []*StoreTx) { + rng := rand.New(rand.NewSource(seed)) + nodes := make([]nodeInfo, numNodes) + for i := range nodes { + pk := make([]byte, 32) + rng.Read(pk) + role := "companion" + if i%4 == 0 { + role = "repeater" + } + nodes[i] = nodeInfo{ + PublicKey: fmt.Sprintf("%x", pk), Name: fmt.Sprintf("Node %04d", i), Role: role, + Lat: 55 + rng.Float64(), Lon: 12 + rng.Float64(), HasGPS: true, + ObservationCount: rng.Intn(500), + } + } + var repeaters []int + for i, n := range nodes { + if n.Role == "repeater" { + repeaters = append(repeaters, i) + } + } + // Each repeater has 3-8 neighbours. + neigh := make(map[int][]int, len(repeaters)) + for _, r := range repeaters { + for k := 3 + rng.Intn(6); k > 0; k-- { + o := repeaters[rng.Intn(len(repeaters))] + if o != r { + neigh[r] = append(neigh[r], o) + neigh[o] = append(neigh[o], r) + } + } + } + + var lens []int + for l, w := range subpath413PathLens { + for ; w > 0; w-- { + lens = append(lens, l) + } + } + // Map iteration above is random; sort so the seed alone fixes the data. + sort.Ints(lens) + + t0 := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) + packets := make([]*StoreTx, numTx) + for i := range packets { + hopBytes := 1 + if rng.Intn(5) == 0 { + hopBytes = 2 + } + n := lens[rng.Intn(len(lens))] + hops := make([]string, n) + cur := repeaters[rng.Intn(len(repeaters))] + for h := range hops { + hops[h] = strings.ToUpper(nodes[cur].PublicKey[:2*hopBytes]) + if nb := neigh[cur]; len(nb) > 0 { + cur = nb[rng.Intn(len(nb))] + } + } + path, _ := json.Marshal(hops) + sender := nodes[rng.Intn(numNodes)] + observer := nodes[repeaters[rng.Intn(len(repeaters))]] + pt := PayloadTXT_MSG + if rng.Intn(4) == 0 { + pt = PayloadADVERT + } + snr := float64(rng.Intn(40)-20) / 2 + rssi := float64(-60 - rng.Intn(60)) + packets[i] = &StoreTx{ + ID: i + 1, Hash: fmt.Sprintf("%016x", rng.Uint64()), + FirstSeen: t0.Add(time.Duration(i) * 20 * time.Second).Format(time.RFC3339), + PayloadType: &pt, PathJSON: string(path), + DecodedJSON: fmt.Sprintf(`{"pubKey":%q}`, sender.PublicKey), + ObserverID: observer.PublicKey, ObserverName: observer.Name, + SNR: &snr, RSSI: &rssi, + } + } + return nodes, packets +} + +// newSubpath413Store is a DB-less store holding packets, with a node cache +// that does not expire during a long benchmark. +func newSubpath413Store(nodes []nodeInfo, packets []*StoreTx) *PacketStore { + s := newTieOrderStore(packets) + for _, tx := range packets { + s.byHash[tx.Hash] = tx + s.byTxID[tx.ID] = tx + } + s.nodeCache = nodes + s.nodePM = buildPrefixMap(nodes) + s.nodeCacheTime = time.Now().Add(time.Hour) + s.rfCacheTTL = time.Minute + return s +} + +// subpath413DetailKeys picks two subpath-detail queries from the data: the +// most frequent 2-hop subpath (a hot repeater pair, ties to the smallest key) +// and a 5-hop subpath taken from the first packet long enough to hold one. +func subpath413DetailKeys(packets []*StoreTx) (hot, cold []string) { + counts := map[string]int{} + for _, tx := range packets { + hops := txGetParsedPath(tx) + for i := 0; i+2 <= len(hops); i++ { + counts[strings.ToLower(hops[i]+","+hops[i+1])]++ + } + if cold == nil && len(hops) >= 5 { + cold = append([]string(nil), hops[:5]...) + } + } + best := "" + for k, n := range counts { + if n > counts[best] || (n == counts[best] && k < best) { + best = k + } + } + return strings.Split(best, ","), cold +} diff --git a/cmd/server/subpath_on_demand_413_bench_test.go b/cmd/server/subpath_on_demand_413_bench_test.go new file mode 100644 index 000000000..0bc7a9146 --- /dev/null +++ b/cmd/server/subpath_on_demand_413_bench_test.go @@ -0,0 +1,253 @@ +package main + +import ( + "fmt" + "io" + "log" + "os" + "runtime" + "runtime/debug" + "runtime/metrics" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Benchmarks for #413 on 30,000 transmissions over 2,000 nodes +// (subpath413Data). The on-demand benchmarks time what a request does now; +// the legacy ones time what the removed index cost on every ingest, eviction +// and rebuild, and how much heap it held, using its frozen copy +// (legacy413Index). Run with: +// +// go test -run '^$' -bench '_413' -benchtime 2s . + +const ( + bench413Tx = 30000 + bench413Nodes = 2000 +) + +func bench413Store(b *testing.B) *PacketStore { + b.Helper() + log.SetOutput(io.Discard) + b.Cleanup(func() { log.SetOutput(os.Stderr) }) + nodes, packets := subpath413Data(bench413Tx, bench413Nodes, 413) + s := newSubpath413Store(nodes, packets) + for _, tx := range packets { + txGetParsedPath(tx) + tx.ParsedDecoded() + } + return s +} + +// The Route Patterns tab's four length groups. +var bench413Groups = []subpathGroup{{2, 2, 50}, {3, 3, 30}, {4, 4, 20}, {5, 8, 15}} + +func BenchmarkSubpathsOnDemand_413(b *testing.B) { + s := bench413Store(b) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.computeAnalyticsSubpaths("", 2, 8, 100) + } +} + +func BenchmarkSubpathsBulkOnDemand_413(b *testing.B) { + s := bench413Store(b) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.GetAnalyticsSubpathsBulk("", bench413Groups) + clearSubpathCache413(s) + } +} + +// The part of a subpaths request that holds PacketStore.mu. +func BenchmarkSubpathsLockedSnapshot_413(b *testing.B) { + s := bench413Store(b) + _, pm := s.getCachedNodesAndPM() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.snapshotSubpathPaths(nil, pm) + } +} + +func BenchmarkSubpathDetailOnDemand_413(b *testing.B) { + s := bench413Store(b) + hot, cold := subpath413DetailKeys(s.packets) + for _, c := range []struct { + name string + hops []string + }{{"hot", hot}, {"cold", cold}} { + b.Run(c.name, func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + s.GetSubpathDetail(c.hops) + } + }) + } +} + +// BenchmarkLegacySubpathIndex_413 times the removed index's upkeep: adding one +// transmission (ingest), removing one (eviction, a linear scan of each of its +// subpaths' transmission lists) and the full rebuild of Load and the chunk +// loader. heap-MB is the heap the built index held. +func BenchmarkLegacySubpathIndex_413(b *testing.B) { + s := bench413Store(b) + b.Run("ingest", func(b *testing.B) { + x := newLegacy413Index(s.packets) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + tx := s.packets[i%len(s.packets)] + b.StopTimer() + x.remove(tx, txGetParsedPath(tx)) + b.StartTimer() + x.add(tx, txGetParsedPath(tx)) + } + }) + b.Run("evict", func(b *testing.B) { + x := newLegacy413Index(s.packets) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + tx := s.packets[i%len(s.packets)] + x.remove(tx, txGetParsedPath(tx)) + b.StopTimer() + x.add(tx, txGetParsedPath(tx)) + b.StartTimer() + } + }) + b.Run("build", func(b *testing.B) { + var before, after runtime.MemStats + var x *legacy413Index + b.ReportAllocs() + for i := 0; i < b.N; i++ { + x = nil + b.StopTimer() + runtime.GC() + runtime.ReadMemStats(&before) + b.StartTimer() + x = newLegacy413Index(s.packets) + } + b.StopTimer() + runtime.GC() + runtime.ReadMemStats(&after) + entries := 0 + for _, l := range x.txs { + entries += len(l) + } + b.ReportMetric(float64(after.HeapAlloc-before.HeapAlloc)/(1<<20), "heap-MB") + b.ReportMetric(float64(len(x.counts)), "keys") + b.ReportMetric(float64(entries), "tx-refs") + }) +} + +// BenchmarkSubpathsColdConcurrent_413 sends n concurrent cold requests for +// the same key (PR #529 review F1). peak-heap-MB is the highest heap held by +// live and unswept objects while they run (sampled every 200 µs), counts the +// number of store counts they ran. Run with -benchtime 1x -count 3. +func BenchmarkSubpathsColdConcurrent_413(b *testing.B) { + s := bench413Store(b) + sample := []metrics.Sample{{Name: "/memory/classes/heap/objects:bytes"}} + for _, n := range []int{1, 4, 8} { + b.Run(fmt.Sprintf("n=%d", n), func(b *testing.B) { + var counts atomic.Int32 + s.subpathHook = func(stage string) { + if stage == "count" { + counts.Add(1) + } + } + defer func() { s.subpathHook = nil }() + var peak uint64 + for i := 0; i < b.N; i++ { + b.StopTimer() + clearSubpathCache413(s) + runtime.GC() + stop := make(chan struct{}) + sampled := make(chan uint64) + go func() { + var max uint64 + tick := time.NewTicker(200 * time.Microsecond) + defer tick.Stop() + for { + metrics.Read(sample) + if v := sample[0].Value.Uint64(); v > max { + max = v + } + select { + case <-stop: + sampled <- max + return + case <-tick.C: + } + } + }() + start := make(chan struct{}) + var wg sync.WaitGroup + for j := 0; j < n; j++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + s.GetAnalyticsSubpaths("", 2, 8, 100) + }() + } + b.StartTimer() + close(start) + wg.Wait() + b.StopTimer() + close(stop) + if p := <-sampled; p > peak { + peak = p + } + } + b.ReportMetric(float64(peak)/(1<<20), "peak-heap-MB") + b.ReportMetric(float64(counts.Load())/float64(b.N), "counts/op") + }) + } +} + +// BenchmarkSubpathDetailWriterWait_413 runs a store writer that takes +// PacketStore.mu in a loop while subpath-detail requests run (PR #529 review +// F2). max-writer-wait-µs is the longest it waited for the lock. GC is off +// while it runs, so a collection's pause does not count as lock time. +func BenchmarkSubpathDetailWriterWait_413(b *testing.B) { + s := bench413Store(b) + defer debug.SetGCPercent(debug.SetGCPercent(-1)) + hot, cold := subpath413DetailKeys(s.packets) + for _, c := range []struct { + name string + hops []string + }{{"hot", hot}, {"cold", cold}} { + b.Run(c.name, func(b *testing.B) { + stop := make(chan struct{}) + waited := make(chan time.Duration) + go func() { + var max time.Duration + for { + select { + case <-stop: + waited <- max + return + default: + } + t0 := time.Now() + s.mu.Lock() + if w := time.Since(t0); w > max { + max = w + } + s.mu.Unlock() + runtime.Gosched() + } + }() + for i := 0; i < b.N; i++ { + s.GetSubpathDetail(c.hops) + } + b.StopTimer() + close(stop) + b.ReportMetric(float64((<-waited).Microseconds()), "max-writer-wait-µs") + }) + } +} diff --git a/cmd/server/subpath_on_demand_413_test.go b/cmd/server/subpath_on_demand_413_test.go new file mode 100644 index 000000000..c38e5b5cf --- /dev/null +++ b/cmd/server/subpath_on_demand_413_test.go @@ -0,0 +1,771 @@ +package main + +// Issue #413: Route Patterns counts subpaths on demand from the packets +// instead of keeping the permanent spIndex/spTxIndex up to date on every +// ingest, path change, merge and eviction. +// +// The tests compare the on-demand endpoints with legacy413Index, a frozen copy +// of the index the store kept up to master 0350322e (addTxToSubpathIndexFull, +// removeTxFromSubpathIndexFull, the unbounded aggregation of +// GetAnalyticsSubpathsBulk, the sort of rankSubpaths and SubpathDetail), +// maintained through the same ingest, path-change and eviction steps. Counts, +// rawHops, ranking, tie-breaks, the 8-hop cap, case folding, totalPaths and +// the detail must come out identical. +// +// The legacy copy differs from master in two places, each on purpose: +// - total is decremented on eviction. Master's eviction removed the +// subpaths but never decremented spTotalPaths, so totalPaths (and every +// pct) drifted upward on a long-running server; TestSubpathTotalPaths +// FollowsEviction_413 pins the fix. +// - the detail takes the matched transmissions in packet order. The index +// held them in insertion order, which a path change rearranged; nothing +// the detail reports depends on that order except the float summation +// order of the signal averages. + +import ( + "encoding/json" + "fmt" + "math/rand" + "reflect" + "runtime" + "sort" + "strings" + "sync" + "testing" + "time" +) + +// ─── Legacy reference (frozen copy of the removed index) ──────────────────── + +type legacy413Index struct { + counts map[string]int + txs map[string][]*StoreTx + total int +} + +func newLegacy413Index(packets []*StoreTx) *legacy413Index { + x := &legacy413Index{counts: map[string]int{}, txs: map[string][]*StoreTx{}} + for _, tx := range packets { + x.add(tx, txGetParsedPath(tx)) + } + return x +} + +// add is addTxToSubpathIndexFull plus the spTotalPaths++ of its callers. +func (x *legacy413Index) add(tx *StoreTx, hops []string) { + if len(hops) < 2 { + return + } + maxL := min(8, len(hops)) + for l := 2; l <= maxL; l++ { + for start := 0; start <= len(hops)-l; start++ { + key := strings.ToLower(strings.Join(hops[start:start+l], ",")) + x.counts[key]++ + x.txs[key] = append(x.txs[key], tx) + } + } + x.total++ +} + +// remove is removeTxFromSubpathIndexFull plus spTotalPaths--. +func (x *legacy413Index) remove(tx *StoreTx, hops []string) { + if len(hops) < 2 { + return + } + maxL := min(8, len(hops)) + for l := 2; l <= maxL; l++ { + for start := 0; start <= len(hops)-l; start++ { + key := strings.ToLower(strings.Join(hops[start:start+l], ",")) + x.counts[key]-- + if x.counts[key] <= 0 { + delete(x.counts, key) + } + txs := x.txs[key] + for i, t := range txs { + if t == tx { + x.txs[key] = append(txs[:i], txs[i+1:]...) + break + } + } + if len(x.txs[key]) == 0 { + delete(x.txs, key) + } + } + } + x.total-- +} + +func legacy413Resolver(s *PacketStore, ctxTxs []*StoreTx) (func(string) string, *prefixMap) { + _, pm := s.getCachedNodesAndPM() + contextPubkeys := buildAggregateHopContextPubkeys(ctxTxs, pm) + hopCache := make(map[string]*nodeInfo) + graph := s.graph.Load() + return func(hop string) string { + if cached, ok := hopCache[hop]; ok { + if cached != nil { + return cached.Name + } + return hop + } + r, _, _ := pm.resolveWithContext(hop, contextPubkeys, graph) + hopCache[hop] = r + if r != nil { + return r.Name + } + return hop + }, pm +} + +// legacy413Bulk is the unbounded aggregation over spIndex +// (GetAnalyticsSubpathsBulk, and computeAnalyticsSubpaths for one group). +func legacy413Bulk(s *PacketStore, x *legacy413Index, groups []subpathGroup) []map[string]interface{} { + resolveHop, _ := legacy413Resolver(s, s.packets) + perGroup := make([]map[string]*subpathAccum, len(groups)) + for i := range groups { + perGroup[i] = make(map[string]*subpathAccum) + } + for rawKey, count := range x.counts { + hops := strings.Split(rawKey, ",") + var namedKey string + resolved := false + for gi, g := range groups { + if len(hops) < g.MinLen || len(hops) > g.MaxLen { + continue + } + if !resolved { + named := make([]string, len(hops)) + for i, h := range hops { + named[i] = resolveHop(h) + } + namedKey = strings.Join(named, " → ") + resolved = true + } + entry := perGroup[gi][namedKey] + if entry == nil { + entry = &subpathAccum{} + perGroup[gi][namedKey] = entry + } + entry.add(rawKey, count) + } + } + results := make([]map[string]interface{}, len(groups)) + for i, g := range groups { + results[i] = legacy413Rank(perGroup[i], x.total, g.Limit) + } + return results +} + +// legacy413Windowed is computeAnalyticsSubpathsWindowed/computeSubpathsSlow: +// per transmission, named keys, original-case rawHops, no 8-hop cap. +func legacy413Windowed(s *PacketStore, regionObs map[string]bool, minLen, maxLen, limit int, window TimeWindow) map[string]interface{} { + resolveHop, _ := legacy413Resolver(s, s.packets) + subpathCounts := make(map[string]*subpathAccum) + totalPaths := 0 + for _, tx := range s.packets { + if !window.Includes(tx.FirstSeen) { + continue + } + hops := txGetParsedPath(tx) + if len(hops) < 2 { + continue + } + if regionObs != nil { + match := false + for _, obs := range tx.Observations { + if regionObs[obs.ObserverID] { + match = true + break + } + } + if !match { + continue + } + } + totalPaths++ + named := make([]string, len(hops)) + for i, h := range hops { + named[i] = resolveHop(h) + } + for l := minLen; l <= maxLen && l <= len(named); l++ { + for start := 0; start <= len(named)-l; start++ { + sub := strings.Join(named[start:start+l], " → ") + raw := strings.Join(hops[start:start+l], ",") + entry := subpathCounts[sub] + if entry == nil { + entry = &subpathAccum{} + subpathCounts[sub] = entry + } + entry.add(raw, 1) + } + } + } + return legacy413Rank(subpathCounts, totalPaths, limit) +} + +// legacy413Rank is rankSubpaths with its full sort. +func legacy413Rank(counts map[string]*subpathAccum, totalPaths, limit int) map[string]interface{} { + type subpathEntry struct { + path string + count int + raw string + } + ranked := make([]subpathEntry, 0, len(counts)) + for path, data := range counts { + ranked = append(ranked, subpathEntry{path, data.count, data.raw}) + } + sort.Slice(ranked, func(i, j int) bool { + if ranked[i].count != ranked[j].count { + return ranked[i].count > ranked[j].count + } + return ranked[i].path < ranked[j].path + }) + if len(ranked) > limit { + ranked = ranked[:limit] + } + subpaths := make([]map[string]interface{}, 0, len(ranked)) + for _, e := range ranked { + subpaths = append(subpaths, map[string]interface{}{ + "path": e.path, + "rawHops": strings.Split(e.raw, ","), + "count": e.count, + "hops": len(strings.Split(e.path, " → ")), + "pct": subpathPct(e.count, totalPaths), + }) + } + return map[string]interface{}{"subpaths": subpaths, "totalPaths": totalPaths} +} + +// legacy413Detail is SubpathDetail over spTxIndex, matched transmissions in +// packet order (see the file comment). +func legacy413Detail(s *PacketStore, x *legacy413Index, rawHops []string) SubpathDetailResponse { + pos := make(map[*StoreTx]int, len(s.packets)) + for i, tx := range s.packets { + pos[tx] = i + } + matchedTxs := append([]*StoreTx(nil), x.txs[strings.ToLower(strings.Join(rawHops, ","))]...) + sort.SliceStable(matchedTxs, func(i, j int) bool { return pos[matchedTxs[i]] < pos[matchedTxs[j]] }) + + _, pm := s.getCachedNodesAndPM() + contextPubkeys := buildAggregateHopContextPubkeys(matchedTxs, pm) + graph := s.graph.Load() + p := &subpathDetailParts{ + rawHops: rawHops, nodes: make([]*nodeInfo, len(rawHops)), pm: pm, graph: graph, + hopContext: contextPubkeys, matchCount: len(matchedTxs), hourBuckets: make([]int, 24), + } + for i, hop := range rawHops { + if r, _, _ := pm.resolveWithContext(hop, contextPubkeys, graph); r != nil { + n := *r + p.nodes[i] = &n + } + } + parents := map[string]*subpathDetailParent{} + observers := map[[2]string]int{} + for _, tx := range matchedTxs { + ts := tx.FirstSeen + if ts != "" { + if p.firstSeen == "" || ts < p.firstSeen { + p.firstSeen = ts + } + if p.lastSeen == "" || ts > p.lastSeen { + p.lastSeen = ts + } + t, err := time.Parse(time.RFC3339, ts) + if err != nil { + t, err = time.Parse("2006-01-02 15:04:05", ts) + } + if err == nil { + p.hourBuckets[t.Hour()]++ + } + } + if tx.SNR != nil { + p.snrSum += *tx.SNR + p.snrCount++ + } + if tx.RSSI != nil { + p.rssiSum += *tx.RSSI + p.rssiCount++ + } + if tx.ObserverName != "" { + observers[[2]string{tx.ObserverID, tx.ObserverName}]++ + } + txCtx := buildHopContextPubkeys(tx, pm) + hops := txGetParsedPath(tx) + pathHops := make([]subpathDetailHop, len(hops)) + key := make([]string, 0, 2*len(hops)) + for i, h := range hops { + pathHops[i].raw = h + if r, _, _ := pm.resolveWithContext(h, txCtx, graph); r != nil { + pathHops[i].pubkey, pathHops[i].name = r.PublicKey, r.Name + } + key = append(key, h, pathHops[i].pubkey) + } + k := strings.Join(key, "\x00") + if pp := parents[k]; pp != nil { + pp.count++ + } else { + parents[k] = &subpathDetailParent{hops: pathHops, count: 1} + } + } + for _, pp := range parents { + p.parents = append(p.parents, *pp) + } + for o, count := range observers { + p.observers = append(p.observers, subpathDetailObserver{id: o[0], name: o[1], count: count}) + } + return p.response(nil) +} + +// ─── Shared data ───────────────────────────────────────────────────────────── + +// subpath413Mesh is subpath413Data plus the cases the index treated +// specially: lower- and mixed-case hops of the same prefix (the index folded +// case), a path that repeats a subpath (each occurrence counts), paths far +// beyond 8 hops (the index stopped at 8), two nodes sharing a name (several +// raw keys behind one path, rawHops the smallest), hops that resolve to no +// node, one observation per packet for the region filter, and a few packets +// without a path. +func subpath413Mesh(numTx, numNodes int, seed int64) ([]nodeInfo, []*StoreTx) { + nodes, packets := subpath413Data(numTx, numNodes, seed) + rng := rand.New(rand.NewSource(seed * 7919)) + nodes[4].Name = nodes[8].Name // two repeaters, one name + for i, tx := range packets { + hops := parsePathJSON(tx.PathJSON) + switch i % 11 { + case 1: + for h := range hops { + hops[h] = strings.ToLower(hops[h]) + } + case 2: + if len(hops) > 0 { + hops[0] = strings.ToLower(hops[0][:1]) + hops[0][1:] + } + case 3: + if len(hops) >= 2 { + hops = append(hops, hops[0], hops[1], hops[0], hops[1]) + } + case 4: + for len(hops) < 40 { + hops = append(hops, hops...) + if len(hops) == 0 { + hops = []string{"7F", "80"} + } + } + case 5: + if len(hops) >= 2 { + hops[1] = fmt.Sprintf("Z%d", rng.Intn(3)) // matches no node + } + } + pj, _ := json.Marshal(hops) + tx.PathJSON = string(pj) + tx.parsedPath, tx.pathParsed = nil, false + tx.Observations = []*StoreObs{{ObserverID: tx.ObserverID, ObserverName: tx.ObserverName}} + } + return nodes, packets +} + +// subpath413Region gives region "R1" every other repeater as its observers. +func subpath413Region(s *PacketStore, nodes []nodeInfo) map[string]bool { + obs := map[string]bool{} + for i, n := range nodes { + if n.Role == "repeater" && i%8 == 0 { + obs[n.PublicKey] = true + } + } + s.regionObsMu.Lock() + s.regionObsCache = map[string]map[string]bool{"R1": obs} + s.regionObsCacheTime = time.Now().Add(time.Hour) + s.regionObsMu.Unlock() + return obs +} + +func clearSubpathCache413(s *PacketStore) { + s.cacheMu.Lock() + s.subpathCache = make(map[string]*cachedResult) + s.cacheMu.Unlock() +} + +var subpath413Groups = []subpathGroup{ + {2, 2, 50}, {3, 3, 30}, {4, 4, 20}, {5, 8, 15}, // the Route Patterns tab + {2, 8, 100}, {2, 63, 1000}, {3, 3, 7}, {9, 63, 100}, {8, 8, 1}, +} + +// ─── Equivalence: on-demand == legacy index ───────────────────────────────── + +func TestSubpathsOnDemandMatchLegacyIndex_413(t *testing.T) { + for _, seed := range []int64{1, 2} { + t.Run(fmt.Sprint("seed", seed), func(t *testing.T) { + nodes, all := subpath413Mesh(1200, 320, seed) + rng := rand.New(rand.NewSource(seed)) + s := newSubpath413Store(nodes, append([]*StoreTx(nil), all[:800]...)) + region := subpath413Region(s, nodes) + x := newLegacy413Index(s.packets) + check413(t, "loaded", s, x, region) + + // Ingest. + for _, tx := range all[800:] { + s.packets = append(s.packets, tx) + x.add(tx, txGetParsedPath(tx)) + } + // Best path changes (reindexTxPath): another packet's path. + for i := 0; i < 100; i++ { + tx := s.packets[rng.Intn(len(s.packets))] + x.remove(tx, txGetParsedPath(tx)) + tx.PathJSON = s.packets[rng.Intn(len(s.packets))].PathJSON + tx.parsedPath, tx.pathParsed = parsePathJSON(tx.PathJSON), true + x.add(tx, txGetParsedPath(tx)) + } + // Eviction of the oldest. + for _, tx := range s.packets[:300] { + x.remove(tx, txGetParsedPath(tx)) + } + s.packets = s.packets[300:] + check413(t, "after ingest, path changes and eviction", s, x, region) + }) + } +} + +func check413(t *testing.T, stage string, s *PacketStore, x *legacy413Index, region map[string]bool) { + t.Helper() + t.Run("counts and ranking", func(t *testing.T) { + for _, g := range subpath413Groups { + clearSubpathCache413(s) + got := s.GetAnalyticsSubpaths("", g.MinLen, g.MaxLen, g.Limit) + want := legacy413Bulk(s, x, []subpathGroup{g})[0] + if !reflect.DeepEqual(got, want) { + t.Errorf("%s: group %v differs from the index\n%s", stage, g, diff413(got, want)) + } + } + }) + t.Run("bulk", func(t *testing.T) { + clearSubpathCache413(s) + got := s.GetAnalyticsSubpathsBulk("", subpath413Groups) + want := legacy413Bulk(s, x, subpath413Groups) + for i := range want { + if !reflect.DeepEqual(got[i], want[i]) { + t.Errorf("%s: bulk group %v differs from the index\n%s", stage, subpath413Groups[i], diff413(got[i], want[i])) + } + } + }) + t.Run("window", func(t *testing.T) { + since := s.packets[len(s.packets)/2].FirstSeen + w := TimeWindow{Since: since, Label: "custom"} + for _, g := range []subpathGroup{{2, 8, 100}, {3, 40, 25}} { + clearSubpathCache413(s) + got := s.GetAnalyticsSubpathsWithWindow("", g.MinLen, g.MaxLen, g.Limit, w) + want := legacy413Windowed(s, nil, g.MinLen, g.MaxLen, g.Limit, w) + if !reflect.DeepEqual(got, want) { + t.Errorf("%s: window %v group %v differs\n%s", stage, w, g, diff413(got, want)) + } + } + }) + t.Run("region", func(t *testing.T) { + clearSubpathCache413(s) + got := s.GetAnalyticsSubpaths("R1", 2, 12, 60) + want := legacy413Windowed(s, region, 2, 12, 60, TimeWindow{}) + if !reflect.DeepEqual(got, want) { + t.Errorf("%s: region differs\n%s", stage, diff413(got, want)) + } + }) + t.Run("detail", func(t *testing.T) { + hot, cold := subpath413DetailKeys(s.packets) + var long []string + for _, tx := range s.packets { + if hops := txGetParsedPath(tx); len(hops) >= 9 { + long = hops[:9] // beyond the 8-hop cap: no match, as with the index + break + } + } + upper := make([]string, len(hot)) + for i, h := range hot { + upper[i] = strings.ToUpper(h) + } + for _, q := range [][]string{hot, cold, upper, long, long[:8], hot[:1], {"zz", "yy"}} { + got := s.GetSubpathDetail(q) + want := legacy413Detail(s, x, q) + if !reflect.DeepEqual(got, want) { + t.Errorf("%s: detail %v differs\n got %+v\n want %+v", stage, q, got, want) + } + } + }) +} + +func diff413(got, want map[string]interface{}) string { + if got["totalPaths"] != want["totalPaths"] { + return fmt.Sprintf("totalPaths got %v want %v", got["totalPaths"], want["totalPaths"]) + } + g, _ := got["subpaths"].([]map[string]interface{}) + w, _ := want["subpaths"].([]map[string]interface{}) + if len(g) != len(w) { + return fmt.Sprintf("len got %d want %d", len(g), len(w)) + } + for i := range w { + if !reflect.DeepEqual(g[i], w[i]) { + return fmt.Sprintf("entry %d:\n got %v\n want %v", i, g[i], w[i]) + } + } + return "types differ" +} + +// ─── totalPaths after eviction ────────────────────────────────────────────── + +// The index's eviction removed a packet's subpaths but left spTotalPaths +// alone, so every pct shrank as the server ran. Counted on demand, totalPaths +// is the packets with a 2+ hop path that are still held. +func TestSubpathTotalPathsFollowsEviction_413(t *testing.T) { + db := setupSubpathWindowDB(t) + defer db.Close() + // A recent single-hop packet: held, but not a path Route Patterns count. + recent := time.Now().UTC().Add(-time.Hour) + db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES ('03', 'recent_single_hop_003', ?, 1, 4, '{}')`, recent.Format(time.RFC3339)) + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (3, 1, 10.0, -90, '["ee"]', ?)`, recent.Unix()) + store := NewPacketStore(db, nil) + if err := store.Load(); err != nil { + t.Fatal(err) + } + if !store.WaitIndexesReady(5 * time.Second) { + t.Fatal("indexes not ready") + } + if n := len(store.packets); n != 3 { + t.Fatalf("setup: %d packets loaded, want 3", n) + } + store.retentionHours = 7 * 24 + store.mu.Lock() + evicted := store.EvictStale() + store.mu.Unlock() + if evicted != 1 { + t.Fatalf("evicted %d, want the 30-day-old packet", evicted) + } + + got := store.GetAnalyticsSubpaths("", 2, 8, 100) + if tp := got["totalPaths"]; tp != 1 { + t.Errorf("totalPaths = %v after evicting one of two paths, want 1", tp) + } + subs := got["subpaths"].([]map[string]interface{}) + if len(subs) != 1 || subs[0]["path"] != "aa → bb" || subs[0]["pct"] != 100.0 { + t.Errorf("subpaths = %v, want only aa → bb at 100%%", subs) + } +} + +// ─── No permanent index ───────────────────────────────────────────────────── + +func TestSubpathCountsKeepNoPermanentIndex_413(t *testing.T) { + typ := reflect.TypeOf(PacketStore{}) + for _, f := range []string{"spIndex", "spTxIndex", "spTotalPaths"} { + if _, ok := typ.FieldByName(f); ok { + t.Errorf("PacketStore still has %s: Route Patterns must count on demand (#413)", f) + } + } + + // The memory budget charged O(hops²) per transmission for the index. With + // the index gone a transmission's path costs the same per hop at any length. + tx := func(n int) *StoreTx { + hops := make([]string, n) + for i := range hops { + hops[i] = fmt.Sprintf("%02X", i) + } + pj, _ := json.Marshal(hops) + return &StoreTx{Hash: "h", PathJSON: string(pj)} + } + short, long := tx(4), tx(8) + want := int64(len(long.PathJSON)-len(short.PathJSON)) + 4*perPathHopBytes + if got := estimateStoreTxBytes(long) - estimateStoreTxBytes(short); got != want { + t.Errorf("8-hop minus 4-hop estimate = %d bytes, want %d (no per-subpath charge)", got, want) + } +} + +// ─── Ranking: bounded top-N == full sort ──────────────────────────────────── + +// rankSubpaths keeps the limit best entries in a bounded heap instead of +// sorting all of them. Ties break on the path BEFORE the cut, so the heap must +// select exactly the entries a full sort cut at limit selects, in the same +// order, on every run (#256). +func TestSubpathRankTopNMatchesFullSort_413(t *testing.T) { + s := &PacketStore{} + rng := rand.New(rand.NewSource(413)) + for round := 0; round < 12; round++ { + counts := map[string]*subpathAccum{} + n := 1 + rng.Intn(300) + for i := 0; i < n; i++ { + a := &subpathAccum{} + for k := 1 + rng.Intn(2); k > 0; k-- { + a.add(fmt.Sprintf("%02x,%02x", rng.Intn(256), rng.Intn(256)), 1+rng.Intn(3)) + } + counts[fmt.Sprintf("N%d → N%d", rng.Intn(60), rng.Intn(60))] = a + } + total := rng.Intn(1000) + for _, limit := range []int{0, 1, 2, 7, 50, n - 1, n, n + 5} { + if limit < 0 { + continue + } + want := legacy413Rank(counts, total, limit) + assertSameEveryRun(t, fmt.Sprintf("round %d limit %d", round, limit), want, func() interface{} { + return s.rankSubpaths(counts, total, limit) + }) + } + } +} + +// ─── Lock: count outside PacketStore.mu ───────────────────────────────────── + +// Only the copy of the hop lists (and the hop context) happens under +// PacketStore.mu; enumerating, resolving and ranking happen after it is +// released. The test parks the request inside hop resolution (the resolver +// reads the neighbour graph, whose lock the test holds) and requires a store +// writer to get the lock meanwhile. With the counting under the read lock the +// writer waits until the request is done. +func TestSubpathCountingReleasesStoreLock_413(t *testing.T) { + nodes := []nodeInfo{ + {PublicKey: "aa11000000000000", Name: "A1", Role: "repeater"}, + {PublicKey: "aa22000000000000", Name: "A2", Role: "repeater"}, + {PublicKey: "bb33000000000000", Name: "B3", Role: "repeater"}, + {PublicKey: "cc44000000000000", Name: "C4", Role: "companion"}, + } + pt := PayloadTXT_MSG + var packets []*StoreTx + for i := 0; i < 3; i++ { + packets = append(packets, &StoreTx{ + ID: i + 1, Hash: fmt.Sprintf("lk%d", i), PayloadType: &pt, + FirstSeen: time.Now().UTC().Format(time.RFC3339), PathJSON: `["AA","BB"]`, + DecodedJSON: `{"pubKey":"cc44000000000000"}`, ObserverID: "obs1", ObserverName: "Obs", + Observations: []*StoreObs{{ObserverID: "obs1"}}, + }) + } + since := TimeWindow{Since: time.Now().Add(-time.Hour).UTC().Format(time.RFC3339), Label: "1h"} + cases := map[string]func(s *PacketStore){ + "subpaths": func(s *PacketStore) { s.GetAnalyticsSubpaths("", 2, 8, 10) }, + "subpaths-bulk": func(s *PacketStore) { s.GetAnalyticsSubpathsBulk("", []subpathGroup{{2, 2, 5}}) }, + "window": func(s *PacketStore) { s.GetAnalyticsSubpathsWithWindow("", 2, 8, 10, since) }, + "region": func(s *PacketStore) { s.GetAnalyticsSubpaths("R1", 2, 8, 10) }, + "subpath-detail": func(s *PacketStore) { s.GetSubpathDetail([]string{"aa", "bb"}) }, + } + for name, call := range cases { + t.Run(name, func(t *testing.T) { + s := newSubpath413Store(nodes, packets) + s.regionObsCache = map[string]map[string]bool{"R1": {"obs1": true}} + s.regionObsCacheTime = time.Now().Add(time.Hour) + g := NewNeighborGraph() + s.graph.Store(g) + + g.mu.Lock() + graphLocked := true + unlockGraph := func() { + if graphLocked { + graphLocked = false + g.mu.Unlock() + } + } + defer unlockGraph() + done := make(chan struct{}) + go func() { defer close(done); call(s) }() + defer func() { unlockGraph(); <-done }() + + // Wait until the request is parked on the graph lock. + deadline := time.Now().Add(10 * time.Second) + for !strings.Contains(allStacks413(), "(*NeighborGraph).Neighbors") { + if time.Now().After(deadline) { + t.Fatal("the request never reached hop resolution") + } + runtime.Gosched() + } + deadline = time.Now().Add(2 * time.Second) + for !s.mu.TryLock() { + if time.Now().After(deadline) { + t.Fatal("a store writer could not take PacketStore.mu while the request resolved hops: counting runs under the lock") + } + runtime.Gosched() + } + s.mu.Unlock() + }) + } + + // The detail matches the subpath against every packet; that scan must + // not hold the lock either (PR #529 review F2), only the copy of the hop + // lists before it and of the matched transmissions after it. + t.Run("subpath-detail-match", func(t *testing.T) { + s := newSubpath413Store(nodes, packets) + parked, release := make(chan struct{}), make(chan struct{}) + s.subpathHook = func(stage string) { + if stage == "match" { + close(parked) + <-release + } + } + done := make(chan struct{}) + go func() { defer close(done); s.GetSubpathDetail([]string{"aa", "bb"}) }() + defer func() { <-done }() + defer close(release) + select { + case <-parked: + case <-time.After(10 * time.Second): + t.Fatal("the detail never reached matching") + } + deadline := time.Now().Add(2 * time.Second) + for !s.mu.TryLock() { + if time.Now().After(deadline) { + t.Fatal("a store writer could not take PacketStore.mu while the detail matched: matching runs under the lock") + } + runtime.Gosched() + } + s.mu.Unlock() + }) +} + +func allStacks413() string { + buf := make([]byte, 1<<20) + return string(buf[:runtime.Stack(buf, true)]) +} + +// ─── Concurrency: snapshot reads vs. writers (run with -race) ─────────────── + +// Writers change paths, signal and observers in place, append and evict +// under PacketStore.mu while requests run, as ingest and eviction do. Under +// -race any read of a transmission after the lock is released is reported. +func TestSubpathOnDemandConcurrentWriters_413(t *testing.T) { + nodes, all := subpath413Mesh(900, 200, 9) + s := newSubpath413Store(nodes, append([]*StoreTx(nil), all[:600]...)) + subpath413Region(s, nodes) + hot, _ := subpath413DetailKeys(s.packets) + since := TimeWindow{Since: s.packets[300].FirstSeen, Label: "custom"} + + stop := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + rng := rand.New(rand.NewSource(1)) + next := 600 + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + s.mu.Lock() + tx := s.packets[rng.Intn(len(s.packets))] + tx.PathJSON = s.packets[rng.Intn(len(s.packets))].PathJSON + tx.parsedPath, tx.pathParsed = parsePathJSON(tx.PathJSON), true + snr := float64(i % 20) + tx.SNR, tx.ObserverName, tx.FirstSeen = &snr, fmt.Sprint("obs", i%5), tx.FirstSeen+"" + if next < len(all) { + txGetParsedPath(all[next]) + s.packets = append(s.packets, all[next]) + next++ + s.packets = s.packets[1:] + } + s.mu.Unlock() + runtime.Gosched() + } + }() + for i := 0; i < 15; i++ { + clearSubpathCache413(s) + s.GetAnalyticsSubpaths("", 2, 8, 50) + s.GetAnalyticsSubpathsBulk("", subpath413Groups[:4]) + s.GetAnalyticsSubpathsWithWindow("", 2, 8, 50, since) + s.GetAnalyticsSubpaths("R1", 2, 8, 50) + s.GetSubpathDetail(hot) + } + close(stop) + wg.Wait() +} diff --git a/cmd/server/subpath_singleflight_413_test.go b/cmd/server/subpath_singleflight_413_test.go new file mode 100644 index 000000000..9fa3644c8 --- /dev/null +++ b/cmd/server/subpath_singleflight_413_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +// An uncached Route Patterns request scans the whole store and allocates +// about as much as the removed index held (#413). N concurrent misses for +// one key must share one count, or the heap the index saved is spent again +// as soon as a few overlap (review F1 on PR #529). +// +// The first count waits until all n callers have missed, so without a +// shared flight every one of them counts; with it the other n-1 join the +// first, and a caller that reaches the flight after it stored its result +// finds that result when the cache is checked again inside the flight. +func TestSubpathColdMissesShareOneCount_413(t *testing.T) { + const n = 8 + window := TimeWindow{Since: "2000-01-01T00:00:00Z"} + cases := []struct { + name string + call func(s *PacketStore) interface{} + }{ + {"list", func(s *PacketStore) interface{} { return s.GetAnalyticsSubpaths("", 2, 8, 100) }}, + {"window", func(s *PacketStore) interface{} { + return s.GetAnalyticsSubpathsWithWindow("", 2, 5, 50, window) + }}, + {"bulk", func(s *PacketStore) interface{} { return s.GetAnalyticsSubpathsBulk("", bench413Groups) }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + nodes, packets := subpath413Data(1500, 150, 7) + s := newSubpath413Store(nodes, packets) + + var missed sync.WaitGroup + missed.Add(n) + var counts atomic.Int32 + s.subpathHook = func(stage string) { + switch stage { + case "miss": + missed.Done() + case "count": + if counts.Add(1) != 1 { + return + } + done := make(chan struct{}) + go func() { missed.Wait(); close(done) }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Error("not every caller missed the cache") + } + } + } + + results := make([]interface{}, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + results[i] = c.call(s) + }(i) + } + wg.Wait() + + if got := counts.Load(); got != 1 { + t.Fatalf("%d concurrent misses ran %d counts, want 1", n, got) + } + for i := 1; i < n; i++ { + if !reflect.DeepEqual(results[i], results[0]) { + t.Fatalf("caller %d got a different result", i) + } + } + if again := c.call(s); !reflect.DeepEqual(again, results[0]) || counts.Load() != 1 { + t.Fatalf("the shared result was not cached (counts=%d)", counts.Load()) + } + }) + } +} diff --git a/cmd/server/subpaths_window_test.go b/cmd/server/subpaths_window_test.go index 933dbff64..b0469dfb8 100644 --- a/cmd/server/subpaths_window_test.go +++ b/cmd/server/subpaths_window_test.go @@ -62,8 +62,8 @@ func TestSubpathsHonorsTimeWindow_StoreLevel(t *testing.T) { if err := store.Load(); err != nil { t.Fatalf("store.Load failed: %v", err) } - // #1008: indexes build in the background after Load(); tests that - // read s.spIndex / s.spTxIndex must wait for the ready flag. + // #1008: the path-hop index builds in the background after Load(); + // wait so it does not race the assertions below. if !store.WaitIndexesReady(5 * time.Second) { t.Fatalf("indexes not ready after 5s") } @@ -106,9 +106,9 @@ func TestSubpathsHandlerHonorsTimeWindow(t *testing.T) { if err := store.Load(); err != nil { t.Fatalf("store.Load failed: %v", err) } - // #1008: handler is hard-gated behind SubpathIndexReady() and returns - // 503 until the background build completes. Wait for ready before - // hitting the route. + // #1008: the path-hop index builds in the background after Load(); + // wait so it does not race the requests below. Route Patterns + // themselves need no index since #413. if !store.WaitIndexesReady(5 * time.Second) { t.Fatalf("indexes not ready after 5s") } diff --git a/cmd/server/tracked_bytes_accounting_113_test.go b/cmd/server/tracked_bytes_accounting_113_test.go index a01955e5e..e2f11fd84 100644 --- a/cmd/server/tracked_bytes_accounting_113_test.go +++ b/cmd/server/tracked_bytes_accounting_113_test.go @@ -481,26 +481,46 @@ func TestTrackedBytes_MixedSequenceDoesNotDrift_113(t *testing.T) { } } +// acct113LoadTracking loads a store of acct113CreateDB transmissions that +// tracks at least minBytes, sizing the transmission count from a first load +// so the test does not depend on the current per-tx estimate. +func acct113LoadTracking(t *testing.T, minBytes int64) *PacketStore { + t.Helper() + n := 3000 + for attempt := 0; attempt < 3; attempt++ { + dbPath := filepath.Join(t.TempDir(), fmt.Sprintf("acct%d.db", n)) + acct113CreateDB(t, dbPath, n, 0) + store := acct113Load(t, dbPath) + if store.trackedBytes >= minBytes { + return store + } + n = int(float64(n)*float64(minBytes)/float64(store.trackedBytes)*1.05) + 1 + } + t.Fatalf("setup: no store reached %d tracked bytes", minBytes) + return nil +} + // A memory-triggered pass must start above the high watermark and finish at // or below the low one, without taking much more than needed. func TestMemoryEviction_CrossesHighAndReachesLow_113(t *testing.T) { - dbPath := filepath.Join(t.TempDir(), "acct.db") - acct113CreateDB(t, dbPath, 3000, 0) - store := acct113Load(t, dbPath) + // maxMemoryMB is whole MiB just below the tracked size T, so one pass + // must free 1-0.85*floor(T)/T of the store. That is under the 25% safety + // cap only while T is large against 1 MiB: at T=7.9 MiB it was 25.03% + // once #411 and #413 had each removed a per-tx term from the estimate. + // From 16 MiB on it stays between 15% and 20.4% whatever the estimate. + store := acct113LoadTracking(t, 16*1048576) acct113Check(t, store, "after Load") tracked := store.trackedBytes - // Just below the tracked size: one pass needs about 15-24% of the store, - // under the 25% safety cap. store.maxMemoryMB = int(tracked / 1048576) - if store.maxMemoryMB < 4 { - t.Fatalf("setup: only %d bytes tracked; the store is too small for the watermark arithmetic", tracked) - } high := int64(store.maxMemoryMB) * 1048576 low := int64(float64(high) * 0.85) if tracked <= high { t.Fatalf("setup: trackedBytes %d does not exceed the high watermark %d", tracked, high) } + if share := float64(tracked-low) / float64(tracked); share >= 0.21 { + t.Fatalf("setup: one pass must free %.2f%% of the store, too close to the 25%% cap", share*100) + } before := len(store.packets) store.RunEviction() if store.trackedBytes > low { diff --git a/cmd/server/tracked_bytes_measure_113_test.go b/cmd/server/tracked_bytes_measure_113_test.go index 0aaeb3d4d..4e2874a83 100644 --- a/cmd/server/tracked_bytes_measure_113_test.go +++ b/cmd/server/tracked_bytes_measure_113_test.go @@ -121,8 +121,8 @@ func acct113Load(tb testing.TB, dbPath string) *PacketStore { if err := store.Load(); err != nil { tb.Fatal(err) } - // Load starts the path-hop and subpath index builds in the background; - // they read every tx, so let them finish before a test touches one. + // Load starts the path-hop index build in the background; it reads + // every tx, so let it finish before a test touches one. if !store.WaitIndexesReady(30 * time.Second) { tb.Fatal("background index builds did not finish") } diff --git a/cmd/server/tracked_bytes_test.go b/cmd/server/tracked_bytes_test.go index d02d8855d..7d2bd85d6 100644 --- a/cmd/server/tracked_bytes_test.go +++ b/cmd/server/tracked_bytes_test.go @@ -18,7 +18,7 @@ func TestEstimateStoreTxBytes_ReasonableValues(t *testing.T) { } got := estimateStoreTxBytes(tx) - // Should be at least base (384) + maps (200) + indexes + path/subpath costs + // Should be at least base (384) + maps (200) + indexes + path costs if got < 700 { t.Errorf("estimate too low for 3-hop tx: %d", got) } @@ -28,7 +28,8 @@ func TestEstimateStoreTxBytes_ReasonableValues(t *testing.T) { } // TestEstimateStoreTxBytes_ManyHopsSubpaths verifies that packets with many -// hops estimate significantly more due to O(path²) subpath index entries. +// hops estimate more by their path-hop index entries only: the O(path²) +// subpath index they were also charged for is gone (#413). func TestEstimateStoreTxBytes_ManyHopsSubpaths(t *testing.T) { tx2 := &StoreTx{ Hash: "aabb", @@ -43,12 +44,8 @@ func TestEstimateStoreTxBytes_ManyHopsSubpaths(t *testing.T) { est2 := estimateStoreTxBytes(tx2) est10 := estimateStoreTxBytes(tx10) - // 10 hops → 45 subpath combos × 40 = 1800 bytes just for subpaths - if est10 <= est2 { - t.Errorf("10-hop (%d) should estimate more than 2-hop (%d)", est10, est2) - } - if est10 < est2+1500 { - t.Errorf("10-hop (%d) should estimate at least 1500 more than 2-hop (%d)", est10, est2) + if want := est2 + 8*perPathHopBytes; est10 != want { + t.Errorf("10-hop (%d) should estimate 8 path hops more than 2-hop (%d), %d", est10, est2, want) } } diff --git a/docs/go-migration.md b/docs/go-migration.md index 3a1d7483e..e83ab287a 100644 --- a/docs/go-migration.md +++ b/docs/go-migration.md @@ -88,7 +88,7 @@ Both engines take a memory budget for the in-memory packet store from `packetSto Unset (the default) means no limit: every packet within retention is loaded and none is evicted for memory. An old Node-era value left in a config file is therefore **not** inert — it caps the Go store too. Check it before upgrading, and read `/api/perf` (`packetStore.trackedMB` against `maxMB`) afterwards. -**What the budget counts.** The store tracks what it holds per transmission and per observation, and evicts exactly what it charged. The estimate covers the stored rows, the index entries, the path indexes (`byPathHop` and the subpath index, both of which depend on the path chosen after the transmission was created), the decoded-packet cache (charged up front for every transmission, whether or not analytics have parsed it yet), the per-observation dedup key and strings, and the resolved relay hops (#164). Since #113 and #164 the estimate matches the real heap of the store within a few percent on synthetic data; before, it covered about two thirds of it, and less once relays were resolved. A budget that used to hold the whole store may therefore now evict, and the cold load admits correspondingly fewer packets at the same `maxMemoryMB`. +**What the budget counts.** The store tracks what it holds per transmission and per observation, and evicts exactly what it charged. The estimate covers the stored rows, the index entries, the path index (`byPathHop`, which depends on the path chosen after the transmission was created; Route Patterns keep no index since #413), the decoded-packet cache (charged up front for every transmission, whether or not analytics have parsed it yet), the per-observation dedup key and strings, and the resolved relay hops (#164). Since #113 and #164 the estimate matches the real heap of the store within a few percent on synthetic data; before, it covered about two thirds of it, and less once relays were resolved. A budget that used to hold the whole store may therefore now evict, and the cold load admits correspondingly fewer packets at the same `maxMemoryMB`. ### `channelKeys` / `channel-rainbow.json` (compatible) diff --git a/docs/specs/startup-performance.md b/docs/specs/startup-performance.md index f3f9493db..2f701aa3b 100644 --- a/docs/specs/startup-performance.md +++ b/docs/specs/startup-performance.md @@ -9,7 +9,7 @@ CoreScope takes 30–45 minutes to start on large databases (325K transmissions, | Phase | Time | Blocking? | |---|---|---| | `Load()` — read SQLite → memory | ~90s | Yes | -| Build subpath index | ~20s | Yes | +| Build subpath index (removed by #413: Route Patterns now count on demand) | ~20s | Yes | | Build distance index | ~15s | Yes | | Build path-hop index | <1s | Yes | | Load neighbor edges from SQLite | <1s | Yes | diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 9e9126068..62413da5f 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -235,7 +235,7 @@ Lower values = fresher data but more server load. | `packetStore.estimatedPacketBytes` | `450` | Estimated bytes per packet (for memory budgeting) | | `packetStore.retentionHours` | `0` | Only load packets younger than N hours on startup and keep them in memory. **Set this on any instance with a large DB.** `0` = unlimited (loads full DB history — causes OOM on cold start when the DB has hundreds of thousands of paths). Recommended: same as `retention.packetDays × 24` (e.g. `168` for 7 days). | -> **Warning:** Leaving `retentionHours` at `0` on a large database will cause the server to OOM-kill itself on every cold start. The full packet history is loaded into the subpath index at startup; a DB with ~280K paths produces ~13M index entries before the process is killed. +> **Warning:** Leaving `retentionHours` at `0` on a large database will cause the server to OOM-kill itself on every cold start. The full packet history is loaded into memory at startup, with its per-packet indexes. ## Timestamps