From 1e7ce40357de0088ff179bf067094df72034f890 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:07:20 +0000 Subject: [PATCH 1/3] test(store): reproduce evicted transmissions left in byPathHop (#115) evict_resolved_pathhop_115_test.go indexes transmissions through the real helpers (addTxToPathHopIndex, indexResolvedPathHops with two observations, so resolved keys hold duplicates) and evicts the older half by retention, by memory and through RunEviction, in both useResolvedPathIndex modes. On master 5 of 7 fail: each evicted transmission is still in byPathHop 6 times (its two resolved keys twice each, plus a key only evicted transmissions used), because eviction removes raw hop keys only; and a raw bucket keeps the evicted pointer in its backing array when the evicted entry is not the last one. The rebuild-resurrection and cache invalidation controls pass. BenchmarkEvictPathHops_115 measures a 1% eviction batch at 20k and 100k transmissions. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/evict_resolved_pathhop_115_test.go | 272 ++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 cmd/server/evict_resolved_pathhop_115_test.go 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..5593ab36d --- /dev/null +++ b/cmd/server/evict_resolved_pathhop_115_test.go @@ -0,0 +1,272 @@ +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 ( + evict115PK1 = "a3f19c2b7d4e5081aa22bb33cc44dd55ee66ff778899000112233445566778899" + evict115PK2 = "b7002211ffeeddccbbaa99887766554433221100aabbccddeeff001122334455" + evict115Only = "c0ffee00112233445566778899aabbccddeeff00112233445566778899aabbcc" // 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() + 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) + } + } + if _, ok := store.byPathHop[evict115Only]; ok { + t.Errorf("bucket %q held only evicted transmissions but was not deleted", evict115Only[:8]) + } + 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) + }) + } +} + +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") + } + assertEvictedGone115(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 a realistic batch (1% of the store, +// the oldest) from a store of n transmissions with 3 raw 1-byte hops and +// 2 resolved hops each (two observations, so 4 resolved entries). +func BenchmarkEvictPathHops_115(b *testing.B) { + prev := log.Writer() + log.SetOutput(io.Discard) + defer log.SetOutput(prev) + for _, n := range []int{20000, 100000} { + b.Run(fmt.Sprintf("txs=%d", n), func(b *testing.B) { + batch := n / 100 + 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 hops drawn +// from 256 one-byte prefixes, 2 resolved keys drawn from 2000 relays. +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++ { + h1, h2, h3 := i%256, (i*7+3)%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) + pks := []string{fmt.Sprintf("%064x", i%2000), fmt.Sprintf("%064x", 5000+(i*31)%2000)} + 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) +} From 1a2f77335e3ad6731ebb92f43073aba64c94ed4e Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:13:43 +0000 Subject: [PATCH 2/3] fix(store): evict transmissions from resolved byPathHop buckets too (#115) Eviction removed a transmission only from the byPathHop keys of its raw wire hops, one occurrence each. indexResolvedPathHops also puts it under resolved full-pubkey keys, once per observation that resolved it, so evicted transmissions stayed in relay counts, transported scopes and memory until the next full rebuild. evictFromPathHopIndex now removes the evicted batch from every bucket in one pass: raw and resolved keys, all duplicates. Empty buckets are deleted and slices.DeleteFunc zeroes each compacted tail, so no backing array keeps an evicted transmission alive. The resolved pubkeys are kept nowhere per transmission (#800) and may come from path_json reconstruction as well as resolved_path, so a sweep is the only complete way; it runs once per batch under the write lock eviction already holds, like compactDistIndex. removeTxFromSlice (raw path change) now removes every occurrence and zeroes the tail as well. The rebuild defence (retainResolvedPathHops) is unchanged and its comment updated; relay readers keep materializing under the read lock. The memory-eviction test evicts only a quarter of the store, so it now skips the "bucket used only by evicted transmissions is deleted" check, which applies when all of them are evicted. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/evict_resolved_pathhop_115_test.go | 15 +++- cmd/server/store.go | 74 +++++++++++++------ 2 files changed, 63 insertions(+), 26 deletions(-) diff --git a/cmd/server/evict_resolved_pathhop_115_test.go b/cmd/server/evict_resolved_pathhop_115_test.go index 5593ab36d..c19286d23 100644 --- a/cmd/server/evict_resolved_pathhop_115_test.go +++ b/cmd/server/evict_resolved_pathhop_115_test.go @@ -70,15 +70,22 @@ func countIn(idx map[string][]*StoreTx, tx *StoreTx) int { } 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) } } - if _, ok := store.byPathHop[evict115Only]; ok { - t.Errorf("bucket %q held only evicted transmissions but was not deleted", evict115Only[:8]) - } 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) @@ -135,7 +142,7 @@ func TestMemoryEvictionRemovesResolvedPathHops_115(t *testing.T) { if n == 0 { t.Fatal("fixture: memory eviction evicted nothing") } - assertEvictedGone115(t, store, before[:n], nil, 0) + 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) diff --git a/cmd/server/store.go b/cmd/server/store.go index 1852d3d54..020304d05 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,17 +4516,45 @@ 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 - } - } - if len(idx[key]) == 0 { + 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 (#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), so one pass over all +// buckets per eviction batch is the only complete way. Empty buckets are +// deleted and the discarded tail of each compacted bucket is zeroed, so no +// backing array keeps an evicted transmission alive. +// +// Cost: O(total byPathHop entries) with one map lookup each, once per +// eviction batch, under the write lock eviction already holds; the same +// shape as compactDistIndex. BenchmarkEvictPathHops_115 measures it. +func evictFromPathHopIndex(idx map[string][]*StoreTx, evicted map[*StoreTx]bool) { + if len(evicted) == 0 { + return + } + for key, list := range idx { + 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 + } } } @@ -4957,6 +4986,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 +5064,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 +5119,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 From 1477f579529512a9cf673ad22d78152ec796b8f2 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 10:09:02 +0000 Subject: [PATCH 3/3] perf(store): sweep only byPathHop buckets under evicted hop prefixes (#115) Eviction runs every minute and usually evicts a handful of transmissions, but the first fix swept every byPathHop bucket per batch. A resolved pubkey always starts with the hop it resolves (the ingestor's prefixIndex maps a hop prefix to the pubkeys that start with it, and byPathHop's resolved keys come only from persisted resolved_path), so evictFromPathHopIndex now collects the hops of every evicted transmission, from tx.PathJSON and every observation's path, and sweeps only buckets whose key starts with one of them (case-insensitively). Still complete: all raw and resolved buckets of the batch, all duplicates. Tests: the fixture's resolved pubkeys now start with their hops as in real data; new case for a key resolved from a non-display observation path; the benchmark uses 4000 prefix-consistent relays and adds a one-minute batch (0.01%). BenchmarkEvictPathHops_115 (min ms, same loaded machine, master is the incomplete raw-only removal): 100k/batch 10: 0.43 -> 2.1; 20k/batch 2: 0.10 -> 0.86; 100k/batch 1000: 5.05 -> 15.4. Relates to #115 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/evict_resolved_pathhop_115_test.go | 52 +++++++++++---- cmd/server/store.go | 63 +++++++++++++++---- 2 files changed, 91 insertions(+), 24 deletions(-) diff --git a/cmd/server/evict_resolved_pathhop_115_test.go b/cmd/server/evict_resolved_pathhop_115_test.go index c19286d23..14683bcdd 100644 --- a/cmd/server/evict_resolved_pathhop_115_test.go +++ b/cmd/server/evict_resolved_pathhop_115_test.go @@ -18,9 +18,11 @@ import ( // counts, transported scopes, retained memory) until the next full rebuild. const ( - evict115PK1 = "a3f19c2b7d4e5081aa22bb33cc44dd55ee66ff778899000112233445566778899" - evict115PK2 = "b7002211ffeeddccbbaa99887766554433221100aabbccddeeff001122334455" - evict115Only = "c0ffee00112233445566778899aabbccddeeff00112233445566778899aabbcc" // only old txs + // 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 @@ -124,6 +126,24 @@ func TestEvictRemovesRawAndResolvedPathHops_115(t *testing.T) { } } +// 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) { @@ -204,16 +224,18 @@ func TestRelayStatsConcurrentWithEviction_115(t *testing.T) { assertEvictedGone115(t, store, old, young, young115Refs) } -// BenchmarkEvictPathHops_115 evicts a realistic batch (1% of the store, -// the oldest) from a store of n transmissions with 3 raw 1-byte hops and -// 2 resolved hops each (two observations, so 4 resolved entries). +// 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 _, n := range []int{20000, 100000} { - b.Run(fmt.Sprintf("txs=%d", n), func(b *testing.B) { - batch := n / 100 + 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) @@ -234,15 +256,18 @@ func mustParseRFC3339(s string) time.Time { return t } -// evict115BenchStore: transmissions one second apart, 3 raw hops drawn -// from 256 one-byte prefixes, 2 resolved keys drawn from 2000 relays. +// 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++ { - h1, h2, h3 := i%256, (i*7+3)%256, (i*13+5)%256 + // 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), @@ -253,7 +278,8 @@ func evict115BenchStore(n int) *PacketStore { store.byHash[tx.Hash] = tx store.byTxID[tx.ID] = tx addTxToPathHopIndex(store.byPathHop, tx) - pks := []string{fmt.Sprintf("%064x", i%2000), fmt.Sprintf("%064x", 5000+(i*31)%2000)} + // 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) } diff --git a/cmd/server/store.go b/cmd/server/store.go index 020304d05..096c2b261 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -4529,23 +4529,53 @@ func removeTxFromSlice(idx map[string][]*StoreTx, key string, tx *StoreTx) { } // evictFromPathHopIndex removes the evicted transmissions from every -// byPathHop bucket (#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), so one pass over all -// buckets per eviction batch is the only complete way. Empty buckets are -// deleted and the discarded tail of each compacted bucket is zeroed, so no -// backing array keeps an evicted transmission alive. +// 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: O(total byPathHop entries) with one map lookup each, once per -// eviction batch, under the write lock eviction already holds; the same -// shape as compactDistIndex. BenchmarkEvictPathHops_115 measures it. +// 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)) + } + } + } + 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 @@ -4558,6 +4588,17 @@ func evictFromPathHopIndex(idx map[string][]*StoreTx, evicted map[*StoreTx]bool) } } +// 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 // transmissions and recomputes them. Builds lookup maps once, amortising the // cost across all changed txs in a single ingest cycle. Must be called with