Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 22 additions & 2 deletions cmd/server/node_liveness_regression_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
160 changes: 160 additions & 0 deletions cmd/server/pathhop_rebuild_1904_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
82 changes: 79 additions & 3 deletions cmd/server/store.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading