diff --git a/cmd/server/evict_resolved_pathhop_115_test.go b/cmd/server/evict_resolved_pathhop_115_test.go new file mode 100644 index 000000000..14683bcdd --- /dev/null +++ b/cmd/server/evict_resolved_pathhop_115_test.go @@ -0,0 +1,305 @@ +package main + +import ( + "fmt" + "io" + "log" + "sync" + "testing" + "time" +) + +// Issue #115: eviction must remove a transmission from every byPathHop +// bucket it is in. indexResolvedPathHops puts it under resolved full-pubkey +// keys as well as its raw wire hops, and several observations of one +// transmission legitimately add it to the same resolved key more than once. +// removeTxFromPathHopIndex derived its keys from the raw path only and +// removed one occurrence, so evicted transmissions stayed reachable (relay +// counts, transported scopes, retained memory) until the next full rebuild. + +const ( + // Resolved pubkeys start with the hop they resolve (the ingestor's + // prefixIndex maps a hop prefix to the pubkeys that start with it). + evict115PK1 = "aaf19c2b7d4e5081aa22bb33cc44dd55ee66ff778899000112233445566778899" + evict115PK2 = "bb002211ffeeddccbbaa99887766554433221100aabbccddeeff001122334455" + evict115Only = "ccffee00112233445566778899aabbccddeeff00112233445566778899aabbcc" // only old txs +) + +// evict115Store builds count transmissions (raw path aa,bb,cc), half of them +// older than the 24h retention, indexed through the real byPathHop helpers: +// raw hops, and resolved hops from two observations (duplicates). +func evict115Store(t testing.TB, count int, resolvedIndex bool) (*PacketStore, []*StoreTx, []*StoreTx) { + t.Helper() + now := time.Now().UTC() + store := makeTestStore(count, now.Add(-48*time.Hour), 0) + store.byPathHop = make(map[string][]*StoreTx) + store.retentionHours = 24 + store.useResolvedPathIndex = resolvedIndex + if resolvedIndex { + store.initResolvedPathIndex() + } + var old, young []*StoreTx + hopsSeen := map[string]bool{} + for i, tx := range store.packets { + if i >= count/2 { + tx.FirstSeen = now.Add(-time.Hour).Format(time.RFC3339) + young = append(young, tx) + } else { + old = append(old, tx) + } + addTxToPathHopIndex(store.byPathHop, tx) + pks := []string{evict115PK1, evict115PK2} + if i < count/2 { + pks = append(pks, evict115Only) + } + for obs := 0; obs < 2; obs++ { // two observations -> duplicate entries + store.indexResolvedPathHops(tx, pks, hopsSeen) + } + } + return store, old, young +} + +func countIn(idx map[string][]*StoreTx, tx *StoreTx) int { + n := 0 + for _, list := range idx { + for _, t := range list { + if t == tx { + n++ + } + } + } + return n +} + +func assertEvictedGone115(t *testing.T, store *PacketStore, old, young []*StoreTx, wantYoungRefs int) { + t.Helper() + assertEvictedGone115Partial(t, store, old, young, wantYoungRefs) + if _, ok := store.byPathHop[evict115Only]; ok { + t.Errorf("bucket %q held only evicted transmissions but was not deleted", evict115Only[:8]) + } +} + +// assertEvictedGone115Partial is assertEvictedGone115 for a pass that +// evicted only some of the transmissions sharing evict115Only. +func assertEvictedGone115Partial(t *testing.T, store *PacketStore, old, young []*StoreTx, wantYoungRefs int) { + t.Helper() + for _, tx := range old { + if n := countIn(store.byPathHop, tx); n != 0 { + t.Fatalf("evicted tx %d is still in byPathHop %d times", tx.ID, n) + } + } + for _, tx := range young { + if n := countIn(store.byPathHop, tx); n != wantYoungRefs { + t.Fatalf("surviving tx %d has %d byPathHop entries, want %d", tx.ID, n, wantYoungRefs) + } + } + // no discarded slot of any bucket's backing array still points at an + // evicted transmission (it would keep it alive for the GC) + evicted := map[*StoreTx]bool{} + for _, tx := range old { + evicted[tx] = true + } + for key, list := range store.byPathHop { + for i, tx := range list[:cap(list)] { + if i >= len(list) && evicted[tx] { + t.Fatalf("bucket %q keeps evicted tx %d in its backing array (slot %d, len %d)", key, tx.ID, i, len(list)) + } + } + } +} + +// Survivors: 3 raw keys once each, 2 resolved keys twice each. +const young115Refs = 3 + 2*2 + +func TestEvictRemovesRawAndResolvedPathHops_115(t *testing.T) { + for _, mode := range []bool{true, false} { + t.Run(fmt.Sprintf("useResolvedPathIndex=%v", mode), func(t *testing.T) { + store, old, young := evict115Store(t, 40, mode) + if n := countIn(store.byPathHop, old[0]); n != 3+3*2 { + t.Fatalf("fixture: old tx indexed %d times, want %d", n, 3+3*2) + } + if got := store.EvictStale(); got != len(old) { + t.Fatalf("evicted %d, want %d", got, len(old)) + } + assertEvictedGone115(t, store, old, young, young115Refs) + }) + } +} + +// A transmission is also indexed under pubkeys resolved from an observation +// whose path differs from the display path (tx.PathJSON); those go too. +func TestEvictRemovesResolvedKeysOfOtherObservationPaths_115(t *testing.T) { + store, old, young := evict115Store(t, 40, true) + const otherPK = "dd00112233445566778899aabbccddeeff00112233445566778899aabbccddee" + tx := old[0] + tx.Observations = append(tx.Observations, &StoreObs{ID: 900001, TransmissionID: tx.ID, PathJSON: `["dd"]`}) + store.indexResolvedPathHops(tx, []string{otherPK}, map[string]bool{}) + if countIn(store.byPathHop, tx) != 3+3*2+1 { + t.Fatal("fixture: other-path resolved key not indexed") + } + store.EvictStale() + assertEvictedGone115(t, store, old, young, young115Refs) + if _, ok := store.byPathHop[otherPK]; ok { + t.Fatal("the bucket of a pubkey resolved from another observation's path survived") + } +} + +func TestRunEvictionRemovesResolvedPathHops_115(t *testing.T) { + store, old, young := evict115Store(t, 40, true) + if got := store.RunEviction(); got != len(old) { + t.Fatalf("evicted %d, want %d", got, len(old)) + } + assertEvictedGone115(t, store, old, young, young115Refs) +} + +func TestMemoryEvictionRemovesResolvedPathHops_115(t *testing.T) { + store, _, _ := evict115Store(t, 40, true) + store.retentionHours = 0 + store.maxMemoryMB = 1 + store.trackedBytes = 2 * 1048576 // over the high watermark: the 25% cap applies + before := append([]*StoreTx(nil), store.packets...) + n := store.EvictStale() + if n == 0 { + t.Fatal("fixture: memory eviction evicted nothing") + } + assertEvictedGone115Partial(t, store, before[:n], nil, 0) + for _, tx := range before[n : len(before)/2] { // older half not evicted: all 9 refs + if got := countIn(store.byPathHop, tx); got != 3+3*2 { + t.Fatalf("surviving tx %d has %d entries, want 9", tx.ID, got) + } + } +} + +// A rebuild after eviction must not bring an evicted transmission back +// (buildPathHopIndex retains resolved keys from the previous index). +func TestRebuildAfterEvictionDoesNotResurrect_115(t *testing.T) { + store, old, young := evict115Store(t, 40, true) + store.EvictStale() + store.buildPathHopIndex() + for _, tx := range old { + if n := countIn(store.byPathHop, tx); n != 0 { + t.Fatalf("rebuild resurrected evicted tx %d (%d entries)", tx.ID, n) + } + } + for _, tx := range young { + if countIn(store.byPathHop, tx) == 0 { + t.Fatalf("rebuild lost surviving tx %d", tx.ID) + } + } +} + +func TestEvictionInvalidatesRelayStatsCache_115(t *testing.T) { + store, _, _ := evict115Store(t, 40, true) + store.relayStatsCache = map[string]RepeaterNodeStats{evict115PK1: {}} + store.EvictStale() + if store.relayStatsCache != nil { + t.Fatal("relay stats cache survived an eviction that changed byPathHop") + } +} + +// Relay readers take the read lock and materialize what they need under it; +// eviction compacts buckets under the write lock. Run both concurrently under +// -race, then check the index. +func TestRelayStatsConcurrentWithEviction_115(t *testing.T) { + store, old, young := evict115Store(t, 400, true) + var wg sync.WaitGroup + stop := make(chan struct{}) + for g := 0; g < 4; g++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + store.GetRepeaterNodeStatsBatch([]string{evict115PK1, evict115PK2, evict115Only}, 72) + } + } + }() + } + store.RunEviction() + close(stop) + wg.Wait() + assertEvictedGone115(t, store, old, young, young115Refs) +} + +// BenchmarkEvictPathHops_115 evicts the oldest transmissions from a store +// of n with 3 raw 1-byte hops and 2 resolved hops each (two observations, +// so 4 resolved entries): a large batch (1% of the store) and a typical +// one-minute batch (0.01%; eviction runs every minute over 168 h). +func BenchmarkEvictPathHops_115(b *testing.B) { + prev := log.Writer() + log.SetOutput(io.Discard) + defer log.SetOutput(prev) + for _, c := range []struct{ n, per10k int }{{20000, 100}, {100000, 100}, {20000, 1}, {100000, 1}} { + n := c.n + b.Run(fmt.Sprintf("txs=%d/batch=%d", n, n*c.per10k/10000), func(b *testing.B) { + batch := n * c.per10k / 10000 + for i := 0; i < b.N; i++ { + b.StopTimer() + store := evict115BenchStore(n) + evictTo := store.packets[batch].FirstSeen + store.retentionHours = time.Since(mustParseRFC3339(evictTo)).Hours() + b.StartTimer() + store.EvictStale() + } + }) + } +} + +func mustParseRFC3339(s string) time.Time { + t, err := time.Parse(time.RFC3339, s) + if err != nil { + panic(err) + } + return t +} + +// evict115BenchStore: transmissions one second apart, 3 raw one-byte hops, +// 2 of them resolved to one of 4000 relays whose pubkey starts with the hop. +func evict115BenchStore(n int) *PacketStore { + start := time.Now().UTC().Add(-time.Duration(n+10) * time.Second) + store := makeTestStore(0, start, 0) + store.byPathHop = make(map[string][]*StoreTx) + hopsSeen := map[string]bool{} + for i := 0; i < n; i++ { + // hops 1 and 2 resolve to relays r1 and r2 (4000 relays in all, each + // starting with its hop prefix), hop 3 stays unresolved. + r1, r2 := i%2000, 2000+(5000+i*31)%2000 + h1, h2, h3 := r1%256, r2%256, (i*13+5)%256 + tx := &StoreTx{ + ID: i + 1, + Hash: fmt.Sprintf("bench%07d", i), + FirstSeen: start.Add(time.Duration(i) * time.Second).Format(time.RFC3339), + PathJSON: fmt.Sprintf(`["%02x","%02x","%02x"]`, h1, h2, h3), + } + store.packets = append(store.packets, tx) + store.byHash[tx.Hash] = tx + store.byTxID[tx.ID] = tx + addTxToPathHopIndex(store.byPathHop, tx) + // resolved pubkeys start with the hop they resolve + pks := []string{fmt.Sprintf("%02x%062x", h1, r1), fmt.Sprintf("%02x%062x", h2, r2)} + store.indexResolvedPathHops(tx, pks, hopsSeen) + store.indexResolvedPathHops(tx, pks, hopsSeen) + } + return store +} + +// Background chunks load older transmissions after newer ones, so an evicted +// transmission is often NOT at the front of a raw bucket. Removing it must +// not leave its pointer in the bucket's backing array. +func TestEvictDoesNotRetainPointerInRawBucketTail_115(t *testing.T) { + now := time.Now().UTC() + store := makeTestStore(2, now.Add(-time.Hour), 0) + store.byPathHop = make(map[string][]*StoreTx) + store.retentionHours = 24 + young, old := store.packets[1], store.packets[0] + old.FirstSeen = now.Add(-48 * time.Hour).Format(time.RFC3339) + addTxToPathHopIndex(store.byPathHop, young) // indexed first + addTxToPathHopIndex(store.byPathHop, old) // e.g. from a later, older chunk + if got := store.EvictStale(); got != 1 { + t.Fatalf("evicted %d, want 1", got) + } + assertEvictedGone115(t, store, []*StoreTx{old}, []*StoreTx{young}, 3) +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 1852d3d54..096c2b261 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -8,6 +8,7 @@ import ( "log" "math" "runtime" + "slices" "sort" "strconv" "strings" @@ -4307,11 +4308,10 @@ func (s *PacketStore) buildPathHopIndex() { // 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. +// Only transmissions still in s.packets are carried over. Eviction removes +// raw and resolved keys itself (evictFromPathHopIndex, #115); this check is +// the defence that keeps a rebuild from reintroducing anything a previous +// index still held for a transmission that is gone. // // Cost is O(entries in prev) with one reused scratch map, and it runs only // where buildPathHopIndex already runs — cold load and background-fill @@ -4456,8 +4456,9 @@ func relayMetrics(times []int64, now int64) (count1h, count24h int, lastRelayed return } -// removeTxFromPathHopIndex removes a transmission from all its raw path-hop index entries. -// Resolved pubkey entries are cleaned up via removeFromResolvedPubkeyIndex. +// removeTxFromPathHopIndex removes a transmission from the keys of its raw +// path hops. Used when a transmission's best raw path changes; its resolved +// keys stay. Eviction removes both in one batch (evictFromPathHopIndex). func removeTxFromPathHopIndex(idx map[string][]*StoreTx, tx *StoreTx) { hops := txGetParsedPath(tx) if len(hops) == 0 { @@ -4515,18 +4516,87 @@ func (s *PacketStore) invalidateRelayStatsCache() { s.relayStatsCacheMu.Unlock() } -// removeTxFromSlice removes tx from idx[key] by ID, deleting the key if empty. +// removeTxFromSlice removes every occurrence of tx (by ID) from idx[key], +// deleting the key if empty. slices.DeleteFunc zeroes the discarded tail, so +// the backing array does not keep the removed transmission alive. func removeTxFromSlice(idx map[string][]*StoreTx, key string, tx *StoreTx) { - list := idx[key] - for i, t := range list { - if t.ID == tx.ID { - idx[key] = append(list[:i], list[i+1:]...) - break + list := slices.DeleteFunc(idx[key], func(t *StoreTx) bool { return t.ID == tx.ID }) + if len(list) == 0 { + delete(idx, key) + return + } + idx[key] = list +} + +// evictFromPathHopIndex removes the evicted transmissions from every +// byPathHop bucket they can be in (#115). A transmission is in its raw hop +// keys and, via indexResolvedPathHops, in resolved full-pubkey keys, once +// per observation that resolved it. The resolved pubkeys are kept nowhere +// per transmission (#800 keeps only a hash-only membership index, and they +// can come from path_json reconstruction as well as resolved_path), but +// each one resolves a hop of one of the transmission's observed paths, so +// it starts with that hop's prefix. Only buckets whose key starts +// (case-insensitively) with a hop of an evicted transmission are swept; +// every duplicate there is removed. Empty buckets are deleted and the +// discarded tail of each compacted bucket is zeroed, so no backing array +// keeps an evicted transmission alive. +// +// Cost, under the write lock eviction already holds: one short-prefix +// lookup per key, plus a sweep of the candidate buckets only. A minute's +// eviction batch touches few prefixes; a batch spanning every 1-byte +// prefix sweeps everything, the same shape as compactDistIndex. +// BenchmarkEvictPathHops_115 measures both. +func evictFromPathHopIndex(idx map[string][]*StoreTx, evicted map[*StoreTx]bool) { + if len(evicted) == 0 { + return + } + prefixes := make(map[string]bool) + var lens []int + addHops := func(hops []string) { + for _, h := range hops { + h = strings.ToLower(h) + if h == "" || prefixes[h] { + continue + } + prefixes[h] = true + if !slices.Contains(lens, len(h)) { + lens = append(lens, len(h)) + } } } - if len(idx[key]) == 0 { - delete(idx, key) + for tx := range evicted { + addHops(txGetParsedPath(tx)) + for _, obs := range tx.Observations { + if obs != nil && obs.PathJSON != "" && obs.PathJSON != tx.PathJSON { + addHops(parsePathJSON(obs.PathJSON)) + } + } + } + for key, list := range idx { + if !hasEvictedHopPrefix(key, prefixes, lens) { + continue + } + kept := slices.DeleteFunc(list, func(tx *StoreTx) bool { return evicted[tx] }) + if len(kept) == len(list) { + continue + } + if len(kept) == 0 { + delete(idx, key) + } else { + idx[key] = kept + } + } +} + +// hasEvictedHopPrefix reports whether key starts, case-insensitively, with +// one of prefixes (all lowercase, of the lengths in lens). +func hasEvictedHopPrefix(key string, prefixes map[string]bool, lens []int) bool { + for _, n := range lens { + if len(key) >= n && prefixes[strings.ToLower(key[:n])] { + return true + } } + return false } // updateDistanceIndexForTxs removes old distance records for the given @@ -4957,6 +5027,10 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { // Build sets of evicted IDs for batch removal from secondary indexes evictedTxIDs := make(map[int]struct{}, cutoffIdx) + evictedTxSet := make(map[*StoreTx]bool, cutoffIdx) + for _, tx := range evicting { + evictedTxSet[tx] = true + } evictedObsIDs := make(map[int]struct{}, cutoffIdx*2) // Track which observer IDs and payload types need filtering affectedObservers := make(map[string]struct{}) @@ -5031,9 +5105,10 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { // Remove from subpath index removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, tx) - // Remove from path-hop index - removeTxFromPathHopIndex(s.byPathHop, tx) } + // Remove from the path-hop index: raw AND resolved keys, all duplicates, + // in one pass per batch (#115). See evictFromPathHopIndex. + evictFromPathHopIndex(s.byPathHop, evictedTxSet) s.invalidateRelayStatsCache() // Batch-remove from byObserver: single pass per affected observer slice @@ -5085,10 +5160,6 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { } // Remove from distance indexes — filter out records referencing evicted txs - evictedTxSet := make(map[*StoreTx]bool, cutoffIdx) - for _, tx := range evicting { - evictedTxSet[tx] = true - } s.compactDistIndex(evictedTxSet) // Trim packets slice