diff --git a/cmd/server/node_liveness_regression_test.go b/cmd/server/node_liveness_regression_test.go index aef934f67..d5291f3eb 100644 --- a/cmd/server/node_liveness_regression_test.go +++ b/cmd/server/node_liveness_regression_test.go @@ -451,12 +451,32 @@ func TestNodeActivity_AlternateObservationRelaySurvivesRestart(t *testing.T) { if tx == nil || len(tx.Observations) != 2 || tx.PathJSON != `["D4","E7"]` { t.Fatalf("fixture must load two observations with the longer display path: %+v", tx) } - for _, bucket := range []string{relay, "a1"} { + // Index shape after a restart. The raw-hop pass never produces + // these buckets: the display path is ["D4","E7"], so "a1" and the + // full relay pubkey can only come from a RESOLVED path. + // + // Before #1904 they were absent in both cases, because + // buildPathHopIndex rebuilt from raw hops and discarded whatever + // indexResolvedPathHops had added. Since that fix, a restart that + // has resolved_path persisted keeps the relay bucket — the index + // is built more than once during Load, and the later build + // retains what the earlier population established. Without a + // persisted resolved_path there is still nothing to retain. + indexedIn := func(bucket string) bool { for _, indexed := range loaded.byPathHop[bucket] { if indexed == tx { - t.Fatalf("fixture must reproduce the restart index shape; tx found in byPathHop[%s]", bucket) + return true } } + return false + } + if got := indexedIn(relay); got != persisted { + t.Fatalf("byPathHop[relay] contains tx = %v, want %v (persisted resolved_path = %v)", got, persisted, persisted) + } + // "a1" is a raw wire hop of the SHORTER observation, which is not + // the display path, so it is never indexed either way. + if indexedIn("a1") { + t.Fatal("tx must not be indexed under a non-display raw hop") } // Live: same rows ingested after startup. diff --git a/cmd/server/pathhop_rebuild_1904_test.go b/cmd/server/pathhop_rebuild_1904_test.go new file mode 100644 index 000000000..b9704912b --- /dev/null +++ b/cmd/server/pathhop_rebuild_1904_test.go @@ -0,0 +1,160 @@ +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) + } +} + +// A bucket holding BOTH a live and an evicted transmission is the realistic +// shape — several relays share a resolved pubkey key, and eviction removes +// them one at a time. Testing only single-tenant keys let a mutation that +// drops the whole key when ANY entry is non-live pass unnoticed. +func TestBuildPathHopIndex_MixedLiveAndEvictedBucket_1904(t *testing.T) { + live := pathHopTx(1, hop1904Raw) + evicted := pathHopTx(2, hop1904Raw) + + store := newPathHopStore([]*StoreTx{live}, map[string][]*StoreTx{ + hop1904Raw: {live}, + // One key, both transmissions: the evicted one must be pruned and + // the live one must survive. + hop1904Resolved: {evicted, live}, + }) + + store.buildPathHopIndex() + + if !hopKeyHas(store.byPathHop, hop1904Resolved, live) { + t.Fatal("live transmission dropped from a bucket it shared with an evicted one") + } + if hopKeyHas(store.byPathHop, hop1904Resolved, evicted) { + t.Fatal("evicted transmission retained from a mixed bucket") + } + if got := len(store.byPathHop[hop1904Resolved]); got != 1 { + t.Fatalf("mixed bucket has %d entries after rebuild, want exactly the live one", got) + } +} + +// indexResolvedPathHops dedups within a call but not across the several +// observations of one transmission, so a pre-rebuild bucket legitimately +// holds the same *StoreTx more than once. The retain pass must collapse +// those: without it the bucket grows on every rebuild, and the repeater +// usefulness readers count ENTRIES, not distinct transmissions. +func TestBuildPathHopIndex_CollapsesRepeatedTxInPrevBucket_1904(t *testing.T) { + tx := pathHopTx(1, hop1904Raw) + + store := newPathHopStore([]*StoreTx{tx}, map[string][]*StoreTx{ + hop1904Raw: {tx}, + // Three observations of one transmission → three appends. + hop1904Resolved: {tx, tx, tx}, + }) + + store.buildPathHopIndex() + + if got := len(store.byPathHop[hop1904Resolved]); got != 1 { + t.Fatalf("resolved bucket has %d entries after rebuild, want 1 — repeated appends of one tx must collapse", got) + } + + // And it must stay collapsed rather than regrowing across rebuilds. + store.buildPathHopIndex() + if got := len(store.byPathHop[hop1904Resolved]); got != 1 { + t.Fatalf("resolved bucket regrew to %d entries on a second rebuild", got) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 3de247a8a..dbfb4d14e 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -4153,14 +4153,90 @@ 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) + // addResolvedPubkeysToPathHopIndex states the contract: mutating + // byPathHop must be paired with invalidateRelayStatsCache. A rebuild + // replaces the whole map, and now also restores relay attribution, so + // without this the 300s batch cache can pin the pre-rebuild empty relay + // stats for minutes AFTER the index was fixed — hiding exactly the data + // this retention restores. GetRepeaterNodeStatsBatchCached is not gated + // on PathHopIndexReady, and HTTP binds before the load completes, so a + // request landing just before the rebuild is enough to trigger it. + s.invalidateRelayStatsCache() + log.Printf("[store] Built path-hop index: %d unique keys (%d entries carried over from the previous index)", + 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.