diff --git a/cmd/server/pathhop_rebuild_1904_test.go b/cmd/server/pathhop_rebuild_1904_test.go new file mode 100644 index 00000000..7848cb6a --- /dev/null +++ b/cmd/server/pathhop_rebuild_1904_test.go @@ -0,0 +1,105 @@ +package main + +import ( + "sync" + "testing" +) + +// Issue #1904: buildPathHopIndex() rebuilds byPathHop from raw path_json +// hops and opens by discarding the map. But byPathHop also holds resolved +// full-pubkey keys fed by indexResolvedPathHops, and those pubkey strings +// are retained nowhere (#800 dropped the per-StoreTx ResolvedPath field in +// favour of a hash-only index), so a plain rebuild silently drops every +// resolved attribution. On a cold start that leaves relay_count_*, +// last_relayed and transported_scopes empty for every node until live +// ingestion refills the index. +// +// The rebuild must therefore RETAIN resolved entries — but only for +// transmissions still in s.packets. Eviction's removeTxFromPathHopIndex +// only strips raw hops (it derives them from txGetParsedPath), so evicted +// transmissions linger under their resolved keys; the rebuild is where +// they get dropped. + +const ( + hop1904Raw = "a3" + hop1904Resolved = "a3f19c2b7d4e5081aa22bb33cc44dd55ee66ff778899000112233445566778899" + hop1904Evicted = "b7002211ffeeddccbbaa99887766554433221100aabbccddeeff001122334455" +) + +// pathHopTx builds a transmission whose wire path is a single raw hop. +func pathHopTx(id int, path string) *StoreTx { + return &StoreTx{ID: id, Hash: "pathhop-tx-" + path, PathJSON: `["` + hop1904Raw + `"]`} +} + +func newPathHopStore(packets []*StoreTx, idx map[string][]*StoreTx) *PacketStore { + return &PacketStore{packets: packets, byPathHop: idx, mu: sync.RWMutex{}} +} + +func hopKeyHas(idx map[string][]*StoreTx, key string, tx *StoreTx) bool { + for _, t := range idx[key] { + if t == tx { + return true + } + } + return false +} + +func TestBuildPathHopIndex_RetainsResolvedHops_1904(t *testing.T) { + tx := pathHopTx(1, hop1904Raw) + // Shape produced by a chunk scan: raw hop from path_json PLUS the + // resolved full pubkey that indexResolvedPathHops added. + store := newPathHopStore([]*StoreTx{tx}, map[string][]*StoreTx{ + hop1904Raw: {tx}, + hop1904Resolved: {tx}, + }) + + store.buildPathHopIndex() + + if !hopKeyHas(store.byPathHop, hop1904Raw, tx) { + t.Fatalf("raw hop %q lost by rebuild", hop1904Raw) + } + if !hopKeyHas(store.byPathHop, hop1904Resolved, tx) { + t.Fatalf("resolved pubkey hop %q dropped by rebuild — #1904", hop1904Resolved) + } +} + +func TestBuildPathHopIndex_DropsResolvedHopsOfEvictedTx_1904(t *testing.T) { + live := pathHopTx(1, hop1904Raw) + evicted := pathHopTx(2, hop1904Raw) // NOT in s.packets any more + + store := newPathHopStore([]*StoreTx{live}, map[string][]*StoreTx{ + hop1904Raw: {live}, + hop1904Resolved: {live}, + hop1904Evicted: {evicted}, + }) + + store.buildPathHopIndex() + + if hopKeyHas(store.byPathHop, hop1904Evicted, evicted) { + t.Fatal("evicted transmission retained under its resolved key — the rebuild must not turn eviction's known gap into a permanent leak") + } + if _, ok := store.byPathHop[hop1904Evicted]; ok { + t.Fatalf("empty key %q left behind after dropping its only entry", hop1904Evicted) + } + if !hopKeyHas(store.byPathHop, hop1904Resolved, live) { + t.Fatal("live resolved hop dropped while pruning evicted ones") + } +} + +func TestBuildPathHopIndex_NoDuplicateOnRepeatedBuild_1904(t *testing.T) { + tx := pathHopTx(1, hop1904Raw) + store := newPathHopStore([]*StoreTx{tx}, map[string][]*StoreTx{ + hop1904Raw: {tx}, + hop1904Resolved: {tx}, + }) + + store.buildPathHopIndex() + store.buildPathHopIndex() + + if got := len(store.byPathHop[hop1904Raw]); got != 1 { + t.Fatalf("raw hop bucket has %d entries after two builds, want 1", got) + } + if got := len(store.byPathHop[hop1904Resolved]); got != 1 { + t.Fatalf("resolved hop bucket has %d entries after two builds, want 1", got) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 8ca5bc60..6c36499d 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -4028,14 +4028,81 @@ func (s *PacketStore) buildSubpathIndex() { len(s.spIndex), s.spTotalPaths) } -// buildPathHopIndex scans all packets and populates byPathHop. +// 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). // Must be called with s.mu held. func (s *PacketStore) buildPathHopIndex() { - s.byPathHop = make(map[string][]*StoreTx, 4096) + prev := s.byPathHop + s.byPathHop = make(map[string][]*StoreTx, max(4096, len(prev))) for _, tx := range s.packets { addTxToPathHopIndex(s.byPathHop, tx) } - log.Printf("[store] Built path-hop index: %d unique keys", len(s.byPathHop)) + retained := s.retainResolvedPathHops(prev) + log.Printf("[store] Built path-hop index: %d unique keys (%d resolved-hop entries retained)", + len(s.byPathHop), retained) +} + +// retainResolvedPathHops re-merges the entries of a pre-rebuild byPathHop +// that the raw-hop pass above cannot reproduce: the resolved full-pubkey +// keys fed by indexResolvedPathHops. Their pubkey strings are retained +// nowhere — #800 dropped the per-StoreTx ResolvedPath field in favour of a +// hash-only membership index — so a plain rebuild silently discarded every +// resolved relay attribution, leaving relay counts and transported scopes +// empty after each cold load until live ingestion refilled them (#1904). +// +// Only transmissions still in s.packets are carried over. This matters: +// eviction's removeTxFromPathHopIndex strips raw hops only (it derives them +// from txGetParsedPath), so evicted transmissions linger in prev under their +// resolved keys. Filtering them here is what keeps the index bounded by the +// eviction policy instead of turning that gap into a permanent leak. +// +// Cost is O(entries in prev) with one reused scratch map, and it runs only +// where buildPathHopIndex already runs — cold load and background-fill +// completion — never on an ingest or request path. +// +// Returns the number of entries carried over. Must be called with s.mu held, +// after s.byPathHop has been rebuilt from raw hops. +func (s *PacketStore) retainResolvedPathHops(prev map[string][]*StoreTx) int { + if len(prev) == 0 { + return 0 + } + live := make(map[*StoreTx]struct{}, len(s.packets)) + for _, tx := range s.packets { + live[tx] = struct{}{} + } + + // Reused across keys (cleared per key) so a large index does not churn + // one map allocation per key. Guards against both a key that the raw + // pass already produced and repeated appends of the same tx in prev — + // indexResolvedPathHops dedups within a call, not across the several + // observations of one transmission. + seen := make(map[*StoreTx]struct{}, 16) + retained := 0 + for key, list := range prev { + if len(list) == 0 { + continue + } + clear(seen) + for _, tx := range s.byPathHop[key] { + seen[tx] = struct{}{} + } + for _, tx := range list { + if tx == nil { + continue + } + if _, ok := live[tx]; !ok { + continue + } + if _, dup := seen[tx]; dup { + continue + } + seen[tx] = struct{}{} + s.byPathHop[key] = append(s.byPathHop[key], tx) + retained++ + } + } + return retained } // addTxToPathHopIndex indexes a transmission under each unique raw hop key.