From 7309bdb5a910eefb025cc63d4b7f4acd04ae4c7d Mon Sep 17 00:00:00 2001 From: dborup <3627142+dborup@users.noreply.github.com> Date: Thu, 8 Oct 2026 14:34:31 +0200 Subject: [PATCH] fix(store): index a transmission once per relay key in byPathHop (#2117) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Relates to #2108 ## Problem `traffic_share_score` grows with server uptime until it is far above reality, and many relays end up clamped at 1.0. The score is the number of non-advert entries in `byPathHop[pubkey]` divided by the number of non-advert transmissions. `indexResolvedPathHops` runs once per **observation**, on the live-ingest path (`IngestNewFromDB`) and on the late-observation path (`IngestNewObservations`). `addResolvedPubkeysToPathHopIndex` only de-duplicates within one call, so every further observation through the same relay appends the transmission to that relay's bucket again. The denominator counts each transmission once. After a restart the values look right, because `buildPathHopIndex` → `retainResolvedPathHops` de-duplicates by `*StoreTx`. They then drift upwards as live observations arrive. None of the affected functions has changed on `master` since the diagnosis against `415362c`. This branch is based on `000d9ab`. ## Fix 1. **Idempotent insert, per (transmission, relay key).** `addResolvedPubkeysToPathHopIndex` keeps a side map `pathHopResolved map[*StoreTx][]string`. It holds the resolved keys each transmission is already indexed under and skips those. - Keys are compared **exactly**, so no collision can ever drop an entry. - A later observation through a **new** relay still adds that relay, once. - `StoreTx` is unchanged. 2. **Interned keys.** `pathHopKeys map[string]string` keeps one shared copy of each resolved key. The record then holds a 16-byte string header per entry, and not the string each observation's resolve allocated: a fresh `json.Unmarshal` string per persisted observation on `Load`, and a `strings.ToLower` result on the live paths. The `byPathHop` key is the same shared copy. 3. **Eviction and rebuild.** - `evictStaleInternal` drops the record of evicted transmissions, plus any interned key whose bucket it deletes. - `retainResolvedPathHops` keeps the records of live transmissions only and drops interned keys whose bucket was not carried over. When a rebuild starts from an empty index, it clears both. 4. **Defence in depth.** The single (`GetRepeaterUsefulnessScore`), batch (`GetRepeaterNodeStatsBatch`) and bulk (`computeRepeaterUsefulnessScoreMap`) scores count **distinct** non-advert transmissions per bucket, via `countDistinctNonAdvert`. - IDs are collected in a reused slice. - A bucket in ascending ID order is already distinct; only out-of-order buckets are sorted. - The bulk pass does this for full-pubkey keys only. Raw-hop buckets get one entry per transmission from `addTxToPathHopIndex`. 5. **Cache side effect.** A repeated observation through known relays no longer mutates `byPathHop`, so it no longer drops the batch relay-stats cache, as the cache contract from `1164` intends. An observation through a new relay still drops it. The other `byPathHop` consumers already de-duplicate by `tx.ID` or read only raw prefix keys: `GetNodeHopAnalytics`, `computeMultiByteCapability`, `handleNodePaths`, `computeRepeaterRelayInfoMap` and the relay info in `GetRepeaterNodeStatsBatch`. Two tests pin that they ignore duplicate entries. ## Exact keys vs. a 64-bit fingerprint The first version of this fix stored a 64-bit FNV-1a hash per key. As requested, I measured both, plus exact keys without interning, on the real `addResolvedPubkeysToPathHopIndex`. **Keys per transmission.** In the e2e fixture, the union of resolved relay keys over all observations of one transmission has a mean of 5.4 and a maximum of 17. The protocol limit is 64 path bytes (`MAX_PATH_SIZE`), i.e. 64 / hash size hops. **Memory** retained by the record, per transmission. `BenchmarkPathHopRecordMemory_2108`: 100K transmissions, keys from 4,000 relays, each transmission heard twice with freshly allocated keys. The record map and the interned keys are included. | keys per tx | exact, interned (this PR) | 64-bit fingerprint | exact, not interned | |---|---|---|---| | 2 | 88 B | 68 B | 207 B | | 5 | 136 B | 101 B | 447 B | | 17 | 344 B | 197 B | 1,423 B | - At realistic key counts, exact keys cost 20–35 B per transmission more than the fingerprint. That is about 3.5 MB per 100K transmissions, and small next to what the store already charges per transmission (`storeTxBaseBytes` alone is 384 B). - Without interning, exact keys would cost 3–4× more, and most of that would be held during every `Load`. Interning is what makes exact keys affordable. **Time.** I ran the variants interleaved, 3 rounds each, median ns/op on an Apple M4. The record step is not measurable inside the full call. The isolated record step was faster with exact keys in a separate micro-benchmark, because comparing a handful of 64-char strings is cheaper than hashing each one. | benchmark | exact, interned | fingerprint | exact, not interned | |---|---|---|---| | late observation, 20K txs | 312 | 358 | 331 | | late observation, 100K txs | 760 | 836 | 758 | | `IngestNewObservations` (SQL + resolve + index) | 531 µs | 497 µs | 527 µs | **Collision probability of the fingerprint.** For k distinct keys in one transmission it is about k(k−1)/2 / 2^64: | k | per transmission | per 10^9 transmissions | |---|---|---| | 2 | 5.4e-20 | 5.4e-11 | | 5 | 5.4e-19 | 5.4e-10 | | 17 | 7.4e-18 | 7.4e-9 | | 64 | 1.1e-16 | 1.1e-7 | For random keys this is negligible. FNV-1a is unkeyed, though, so two relay keys that collide could be found deliberately (a birthday search over about 2^32 key pairs). A collision would leave a transmission out of one relay's bucket. **Decision.** Exact keys are cheap enough once interned: the same speed within noise, and a few tens of bytes per transmission. They remove the question of collision-driven omissions entirely, so this PR uses exact keys and drops the fingerprint. The two new maps (`pathHopResolved`, `pathHopKeys`) are not added to `trackedBytes`, so the `maxMemoryMB` trigger undercounts by roughly the per-transmission figures above. Both are bounded by live state (eviction and rebuild prune them), so this is a steady proportional undercount rather than a leak. ## Tests **New: `cmd/server/pathhop_dedupe_2108_test.go`.** Behaviour tests that use only existing API. Each one runs against the real SQLite schema through `Load`, `IngestNewFromDB` and `IngestNewObservations` where noted. | test | covers | on `master` | |---|---|---| | `TestPathHopIndexOncePerTx_LateObservations_2108` | 1 + 10 late observations through the same relays: one entry per relay | **fails** (11 entries) | | `TestPathHopIndexOncePerTx_LiveIngestBatch_2108` | 11 observations in one `IngestNewFromDB` batch | **fails** (11) | | `TestPathHopIndexOncePerTx_AfterLoad_2108` | `Load` + rebuild, then one live observation | **fails** (2) | | `TestPathHopIndexAddsNewRelayFromLaterObservation_2108` | a later observation through a **new** relay adds it exactly once, and its share becomes correct | **fails** (2) | | `TestTrafficShareStableAcrossLateObservations_2108` | single, batch and bulk scores agree, equal the definition, stay put over rounds of late observations, and equal a fresh `Load` of the same data | **fails** (all four relays at 1.0, want 0.4–0.5) | | `TestPathHopIndexSizeBoundedByTransmissions_2108` | the index size does not grow with observations | **fails** | | `TestRelayStatsCacheAcrossRepeatedObservations_2108` | a repeated observation keeps the relay-stats cache, and the kept cache equals a fresh compute; a new relay drops it, and the next read sees the relay | **fails** | | `TestAddResolvedPubkeysToPathHopIndex_PerRelayIdempotent_2108` | the helper's return value and cache invalidation per (tx, relay), in any key order | **fails** | | `TestTrafficShareCountsDistinctTransmissions_2108` | all three scores count distinct transmissions in a duplicated bucket | **fails** | | `TestPathHopConsumersIgnoreDuplicateEntries_2108` | pin of the consumer audit | passes (by design) | | `TestMultiByteCapabilityIgnoresDuplicateEntries_2108` | pin of the consumer audit | passes (by design) | **New: `cmd/server/pathhop_record_2108_test.go`.** These tests use the new symbols, so they do not compile against `master`. - `TestCountDistinctNonAdvert_2108`: ascending, descending, interleaved duplicates, adverts, nils, untyped. - `TestPathHopResolvedRecordBoundedAndEvicted_2108`: the record stays at 2 keys after 11 observations. Eviction removes the record and every entry. A survivor's next observation adds nothing. Evicting everything leaves no record, no interned key and no bucket. - `TestPathHopResolvedRecordAcrossRebuild_2108`: a rebuild keeps the records of live transmissions, so the next observation adds nothing. It drops the record of a removed transmission and the interned key only that transmission used. - `TestPathHopResolvedRecordClearedWithEmptyIndex_2108`: a rebuild from an empty index clears the record and the interned keys, and the next observation puts the transmission back. - `TestPathHopResolvedRecordInternsKeys_2108`: record entries and the `byPathHop` key share one copy, even though every observation passes freshly allocated keys. **Changed: `pathhop_eviction_1908_test.go`.** It built its duplicate entries by repeating `indexResolvedPathHops`, which no longer duplicates. It now seeds the duplicates directly, so the `1908` sweep is still tested against buckets that hold one transmission several times. The expected buckets are unchanged. **Changed: `db_test.go`.** `setupTestDB` takes `testing.TB`, so the SQL benchmark can use it. **Mutants.** I applied each mutant on its own to this branch. All 14 are killed: | mutant | killed by | |---|---| | record never consulted | the OncePerTx, stable-share, size, cache and record tests | | dedupe per transmission instead of per relay | `AddsNewRelayFromLaterObservation`, `RelayStatsCacheAcrossRepeatedObservations`, `PerRelayIdempotent` | | eviction keeps the record | `RecordBoundedAndEvicted` | | rebuild keeps records of removed transmissions | `RecordAcrossRebuild` | | rebuild from an empty index keeps the record | `RecordClearedWithEmptyIndex` | | cache dropped on every call | `RelayStatsCacheAcrossRepeatedObservations`, `PerRelayIdempotent`, existing `NoMutation_PreservesCache` | | cache kept although `byPathHop` changed | `RelayStatsCacheAcrossRepeatedObservations`, `PerRelayIdempotent`, existing `InvalidatesRelayStatsCache` | | single score counts entries | `TrafficShareCountsDistinctTransmissions` | | batch score counts entries | `TrafficShareCountsDistinctTransmissions` | | bulk score counts entries | `TrafficShareCountsDistinctTransmissions` | | distinct count trusts any bucket order | `TrafficShareCountsDistinctTransmissions`, `CountDistinctNonAdvert` | | keys not interned | `RecordInternsKeys` | | eviction keeps interned keys | `RecordBoundedAndEvicted` | | rebuild keeps interned keys of dropped buckets | `RecordAcrossRebuild` | **Commands run:** - `gofmt -l` on all tracked Go files: clean. - `go vet ./...` in all 14 modules: clean. - `cd cmd/server && go test -race ./...`: pass. - `cd cmd/ingestor && go test ./...`: pass. - `sh test-all.sh`: 184 of 186 suites pass locally. The other two, `test-issue-1956-release-routing.js` and `test-preflight-xss-gate.js`, shell out to scripts that need bash ≥ 4 (`mapfile`). They fail under macOS's bash 3.2 regardless of this change. This PR touches no frontend or script files. ## Benchmark: `master` vs. this branch I ran `master`'s sources (`000d9ab`) and this branch interleaved, 5 rounds, with the same benchmark files. Medians on an Apple M4. | benchmark | `master` | this PR | change | |---|---|---|---| | late observation through known relays, 20K txs | 446 ns, 38 B/op | 441 ns, 0 B/op | within noise | | late observation through known relays, 100K txs | 710 ns, 69 B/op | 755 ns, 0 B/op | within noise (runs overlap) | | … index size afterwards, entries/tx (20K / 100K) | 24.99 / 8.99, still growing | 4.99 / 4.99 | bounded | | `IngestNewObservations`, one observation for each of 20 txs | 532 µs | 513 µs | within noise | | bulk score pass, clean index, 20K, ingest order | 246 µs | 316 µs | +28 % | | bulk score pass, clean index, 100K, ingest order | 3.03 ms | 4.30 ms | +42 % | | bulk score pass, clean index, 20K, reversed buckets | 257 µs | 467 µs | +82 % | | bulk score pass, clean index, 100K, reversed buckets | 3.38 ms | 4.79 ms | +42 % | | bulk score pass after 11 observations per tx, 20K | 663 µs | 284 µs | −57 % | | bulk score pass after 11 observations per tx, 100K | 4.72 ms | 3.19 ms | −32 % | - On an index that is clean on both builds, the distinct count makes the bulk pass slower. That pass runs on the cache-miss path, which the background recomputer refreshes every 5 minutes by default. - On the index a live server actually holds without this fix, `master` walks every accumulated duplicate, so the real-world pass is faster after the fix. The gap grows with uptime. - The late-observation step no longer allocates. On `master` its index grows with every observation. Benchmarks: `BenchmarkLateObservationIndex_2108`, `BenchmarkIngestNewObservations_2108`, `BenchmarkTrafficShareScoreMap_2108` and `BenchmarkPathHopRecordMemory_2108`. ## Production motivation We have run this fix on two production instances. Before the fix, the summed `traffic_share_score` grew past 180 and many relays sat at the 1.0 clamp; with the fix no node reaches `≥ 0.999`. A controlled 12 h A/B run makes the drift explicit. Two servers read one identical database — one on this `master` base, one with the fix — alongside a reference server that freshly loads the same database (a fresh load is correct, because the rebuild dedupes). The unfixed server drifted to 9 relays at the 1.0 clamp and up to 0.95 absolute error per node against the reference; the fixed server stayed within 0.05 of the reference for every node, with no relay at the clamp. A smaller residual rise remains and is a separate cause (startup-vs-live hop resolution); it is deliberately left for a follow-up so its effect stays measurable. ## Out of scope That remaining slow rise comes from a separate cause: live ingest resolves some hops that the startup load does not. This PR deliberately leaves that drift alone, so its effect stays measurable. The fix will follow in a separate PR. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- cmd/server/pathhop_dedupe_2108_bench_test.go | 162 +++++ cmd/server/pathhop_dedupe_2108_test.go | 599 +++++++++++++++++++ cmd/server/pathhop_eviction_1908_test.go | 8 +- cmd/server/pathhop_record_2108_test.go | 294 +++++++++ cmd/server/repeater_enrich_bulk.go | 21 +- cmd/server/repeater_usefulness.go | 85 ++- cmd/server/store.go | 105 +++- 7 files changed, 1220 insertions(+), 54 deletions(-) create mode 100644 cmd/server/pathhop_dedupe_2108_bench_test.go create mode 100644 cmd/server/pathhop_dedupe_2108_test.go create mode 100644 cmd/server/pathhop_record_2108_test.go diff --git a/cmd/server/pathhop_dedupe_2108_bench_test.go b/cmd/server/pathhop_dedupe_2108_bench_test.go new file mode 100644 index 00000000..bd53037a --- /dev/null +++ b/cmd/server/pathhop_dedupe_2108_bench_test.go @@ -0,0 +1,162 @@ +package main + +import ( + "fmt" + "io" + "log" + "slices" + "testing" + "time" +) + +// phdBenchStore: n non-advert transmissions one second apart, each with 3 +// raw one-byte hops of which 2 resolve to one of 4000 relays (the shape of +// a relay-heavy store), indexed once as by their first observation. +// resolve returns the keys a further observation of transmission j resolves +// to, freshly allocated as every ingest path allocates them. +func phdBenchStore(n int) (store *PacketStore, resolve func(j int) []string) { + 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{} + relays := func(i int) (int, int) { return i % 2000, 2000 + (5000+i*31)%2000 } + resolve = func(j int) []string { + r1, r2 := relays(j) + return []string{fmt.Sprintf("%02x%062x", r1%256, r1), fmt.Sprintf("%02x%062x", r2%256, r2)} + } + for i := 0; i < n; i++ { + r1, r2 := relays(i) + pt := 2 + tx := &StoreTx{ + ID: i + 1, + Hash: fmt.Sprintf("late%07d", i), + FirstSeen: start.Add(time.Duration(i) * time.Second).Format(time.RFC3339), + PathJSON: fmt.Sprintf(`["%02x","%02x","%02x"]`, r1%256, r2%256, (i*13+5)%256), + PayloadType: &pt, + } + store.packets = append(store.packets, tx) + store.byHash[tx.Hash] = tx + store.byTxID[tx.ID] = tx + store.byPayloadType[pt] = append(store.byPayloadType[pt], tx) + addTxToPathHopIndex(store.byPathHop, tx) + store.indexResolvedPathHops(tx, resolve(i), hopsSeen) + } + return store, resolve +} + +func phdEntriesPerTx(store *PacketStore) float64 { + entries := 0 + for _, list := range store.byPathHop { + entries += len(list) + } + return float64(entries) / float64(len(store.packets)) +} + +// BenchmarkLateObservationIndex_2108 measures the store side of +// IngestNewObservations for one late observation through the same relays +// (indexResolvedPathHops under the write lock, as the ingest loop calls it; +// the keys are resolved, so allocated, outside the timer as the resolver +// allocates them before this call) and reports how many byPathHop entries +// a transmission ends up with. Without the dedupe that number grows with +// every observation; with it, it stays at 3 raw + 2 resolved. +func BenchmarkLateObservationIndex_2108(b *testing.B) { + for _, n := range []int{20000, 100000} { + b.Run(fmt.Sprintf("txs=%d", n), func(b *testing.B) { + store, resolve := phdBenchStore(n) + hopsSeen := map[string]bool{} + const batch = 1024 + pks := make([][]string, batch) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + j := (i * 7919) % n // spread late observations over the store + if i%batch == 0 { + b.StopTimer() + for k := range pks { + pks[k] = resolve(((i + k) * 7919) % n) + } + b.StartTimer() + } + store.mu.Lock() + store.indexResolvedPathHops(store.packets[j], pks[i%batch], hopsSeen) + store.mu.Unlock() + } + b.StopTimer() + b.ReportMetric(phdEntriesPerTx(store), "pathhop-entries/tx") + }) + } +} + +// BenchmarkIngestNewObservations_2108 drives the real late-observation +// ingest (SQL scan + resolve + index) for a batch of one new observation per +// transmission through the same relays. +func BenchmarkIngestNewObservations_2108(b *testing.B) { + prev := log.Writer() + log.SetOutput(io.Discard) + defer log.SetOutput(prev) + db := setupTestDB(b) + defer db.conn.Close() + const observers = 512 + phdSeed(b, db, observers) + base := time.Now().UTC().Add(-30 * time.Minute) + for i := range phdPaths { + phdInsertTx(b, db, i, base.Add(time.Duration(i)*time.Second)) + phdInsertObs(b, db, i, 1, base.Add(time.Duration(i*1000+1)*time.Second), false) + } + store := phdLoadedStore(b, db) + store.IngestNewFromDB(0, 1000) + b.ReportAllocs() + b.ResetTimer() + for it := 0; it < b.N; it++ { + b.StopTimer() + o := 2 + it%(observers-1) + since := phdMaxObsID(b, db) + for i := range phdPaths { + phdInsertObs(b, db, i, o, base.Add(time.Duration(i*1000+o)*time.Second), false) + } + b.StartTimer() + store.IngestNewObservations(since, 10000) + } +} + +// BenchmarkTrafficShareScoreMap_2108 measures the bulk traffic-share pass +// (cache-miss path of GetRepeaterUsefulnessScoreMap) on realistic sizes. +// obs is how many observations of each transmission were indexed through +// the real indexResolvedPathHops: with obs=1 the index is clean on any +// build; with obs=11 it is what a store holds after 11 observers heard each +// transmission (clean with the dedupe, 10 extra entries per resolved key +// without it). order=reversed reverses every bucket, as when background +// chunks load older transmissions after newer ones: no bucket is in +// ascending ID order, so every full-key bucket takes the sort path. +func BenchmarkTrafficShareScoreMap_2108(b *testing.B) { + for _, c := range []struct { + n, obs int + reversed bool + }{{20000, 1, false}, {100000, 1, false}, {20000, 1, true}, {100000, 1, true}, {20000, 11, false}, {100000, 11, false}} { + order := "ingest" + if c.reversed { + order = "reversed" + } + b.Run(fmt.Sprintf("txs=%d/obs=%d/order=%s", c.n, c.obs, order), func(b *testing.B) { + store, resolve := phdBenchStore(c.n) + hopsSeen := map[string]bool{} + for o := 1; o < c.obs; o++ { + for j, tx := range store.packets { + store.indexResolvedPathHops(tx, resolve(j), hopsSeen) + } + } + if c.reversed { + for _, list := range store.byPathHop { + slices.Reverse(list) + } + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + store.computeRepeaterUsefulnessScoreMap() + } + b.StopTimer() + b.ReportMetric(phdEntriesPerTx(store), "pathhop-entries/tx") + }) + } +} diff --git a/cmd/server/pathhop_dedupe_2108_test.go b/cmd/server/pathhop_dedupe_2108_test.go new file mode 100644 index 00000000..b77b2fe5 --- /dev/null +++ b/cmd/server/pathhop_dedupe_2108_test.go @@ -0,0 +1,599 @@ +package main + +import ( + "fmt" + "math" + "reflect" + "strings" + "testing" + "time" +) + +// Issue #2108: traffic_share_score grew with server uptime. The score is +// |non-advert txs in byPathHop[pubkey]| / |non-advert txs in byPayloadType|, +// but every observation of a transmission that resolved to the same relays +// appended the transmission to the relay's byPathHop bucket again, while the +// denominator counts it once. A rebuild of the index (retainResolvedPathHops +// dedups by *StoreTx) made values look sane right after a restart; they then +// drifted upwards with live observations. + +// Four relays with distinct first bytes, so a raw 1-byte hop resolves to +// exactly one of them (unique prefix) on every ingest path. +var phdRelays = []string{ + "a1" + strings.Repeat("11", 31), + "b2" + strings.Repeat("22", 31), + "c3" + strings.Repeat("33", 31), + "d4" + strings.Repeat("44", 31), +} + +// phdPaths is the fixed transmission set: indexes into phdRelays. The +// longest path has 3 hops, so the sum of all relays' share is at most 3. +var phdPaths = [][]int{ + {0, 1, 2}, {1, 2}, {3}, {0, 3}, {2}, + {0, 1}, {1}, {2, 3}, {0}, {1, 3}, + {0, 1, 2}, {3}, {2, 1}, {0, 2}, {1}, + {3, 2, 0}, {2}, {0, 1}, {1, 2, 3}, {3}, +} + +const ( + phdMaxPathLen = 3 + phdLateObs = 10 // extra observations per transmission + phdObserverCount = 1 + phdLateObs +) + +func phdRawHopsJSON(path []int) string { + hops := make([]string, len(path)) + for i, r := range path { + hops[i] = `"` + strings.ToUpper(phdRelays[r][:2]) + `"` + } + return "[" + strings.Join(hops, ",") + "]" +} + +func phdResolvedJSON(path []int) string { + pks := make([]string, len(path)) + for i, r := range path { + pks[i] = `"` + phdRelays[r] + `"` + } + return "[" + strings.Join(pks, ",") + "]" +} + +// phdSeed creates the relays (as repeater nodes) and the observers. +func phdSeed(t testing.TB, db *DB, observers int) { + t.Helper() + ts := time.Now().UTC().Add(-time.Hour).Format(time.RFC3339) + for i, pk := range phdRelays { + phdExec(t, db, `INSERT INTO nodes (public_key, name, role, last_seen, first_seen, advert_count) VALUES (?, ?, 'repeater', ?, '2026-01-01', 1)`, + pk, fmt.Sprintf("phdRelay-%d", i), ts) + } + for i := 1; i <= observers; i++ { + phdExec(t, db, `INSERT INTO observers (rowid, id, name, iata) VALUES (?, ?, ?, 'TST')`, + i, fmt.Sprintf("phdObs-%02d", i), fmt.Sprintf("Observer %d", i)) + } +} + +func phdExec(t testing.TB, db *DB, q string, args ...any) { + t.Helper() + if _, err := db.conn.Exec(q, args...); err != nil { + t.Fatalf("%s: %v", q, err) + } +} + +// phdInsertTx inserts transmission i of phdPaths (non-advert TXT_MSG, flood). +func phdInsertTx(t testing.TB, db *DB, i int, firstSeen time.Time) { + t.Helper() + phdExec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (?, 'CAFE', ?, ?, 1, 2, '{"type":"TXT_MSG"}')`, i+1, fmt.Sprintf("phdTx-%03d", i), firstSeen.Format(time.RFC3339)) +} + +// phdInsertObs inserts one observation of transmission i through its path, +// heard by observer obsIdx. persisted controls resolved_path (the ingestor +// path) vs NULL (resolved fresh by the server on ingest). +func phdInsertObs(t testing.TB, db *DB, i, obsIdx int, ts time.Time, persisted bool) { + t.Helper() + phdInsertObsVia(t, db, i, phdPaths[i], obsIdx, ts, persisted) +} + +// phdInsertObsVia inserts one observation of transmission i heard through +// path (indexes into phdRelays), which need not be the path of phdPaths[i]. +func phdInsertObsVia(t testing.TB, db *DB, i int, path []int, obsIdx int, ts time.Time, persisted bool) { + t.Helper() + var rp any + if persisted { + rp = phdResolvedJSON(path) + } + phdExec(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) + VALUES (?, ?, 5, -90, ?, ?, ?)`, i+1, obsIdx, phdRawHopsJSON(path), ts.Unix(), rp) +} + +func phdMaxObsID(t testing.TB, db *DB) int { + t.Helper() + var id int + if err := db.conn.QueryRow(`SELECT COALESCE(MAX(id), 0) FROM observations`).Scan(&id); err != nil { + t.Fatal(err) + } + return id +} + +func phdLoadedStore(t testing.TB, db *DB) *PacketStore { + t.Helper() + store := NewPacketStore(db, nil) + if err := store.Load(); err != nil { + t.Fatal(err) + } + if !store.WaitIndexesReady(5 * time.Second) { + t.Fatal("indexes not ready") + } + return store +} + +func phdCountTxIn(store *PacketStore, key string, tx *StoreTx) int { + store.mu.RLock() + defer store.mu.RUnlock() + n := 0 + for _, t := range store.byPathHop[key] { + if t == tx { + n++ + } + } + return n +} + +func phdAssertOncePerRelay(t *testing.T, store *PacketStore, i int) { + t.Helper() + tx := store.byTxID[i+1] + if tx == nil { + t.Fatalf("tx %d not in store", i+1) + } + if got := len(tx.Observations); got != phdObserverCount { + t.Fatalf("fixture: tx %d has %d observations, want %d", tx.ID, got, phdObserverCount) + } + for _, r := range phdPaths[i] { + if n := phdCountTxIn(store, phdRelays[r], tx); n != 1 { + t.Errorf("tx %d is in byPathHop[%s…] %d times, want exactly 1", tx.ID, phdRelays[r][:8], n) + } + } +} + +// Late-observation path: the transmission is ingested with its first +// observation, then 10 more observers hear it through the same relays. +func TestPathHopIndexOncePerTx_LateObservations_2108(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, phdObserverCount) + store := phdLoadedStore(t, db) + + now := time.Now().UTC().Add(-10 * time.Minute) + phdInsertTx(t, db, 0, now) + phdInsertObs(t, db, 0, 1, now, false) + store.IngestNewFromDB(0, 100) + for _, r := range phdPaths[0] { + if n := phdCountTxIn(store, phdRelays[r], store.byTxID[1]); n != 1 { + t.Fatalf("fixture: first observation must index tx once under %s…, got %d", phdRelays[r][:8], n) + } + } + + for k := 1; k <= phdLateObs; k++ { + since := phdMaxObsID(t, db) + phdInsertObs(t, db, 0, 1+k, now.Add(time.Duration(k)*time.Second), false) + store.IngestNewObservations(since, 100) + } + phdAssertOncePerRelay(t, store, 0) +} + +// Live-ingest path: the transmission arrives together with all of its +// observations in one IngestNewFromDB batch. +func TestPathHopIndexOncePerTx_LiveIngestBatch_2108(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, phdObserverCount) + store := phdLoadedStore(t, db) + + now := time.Now().UTC().Add(-10 * time.Minute) + phdInsertTx(t, db, 0, now) + for o := 1; o <= phdObserverCount; o++ { + phdInsertObs(t, db, 0, o, now.Add(time.Duration(o)*time.Second), false) + } + store.IngestNewFromDB(0, 100) + phdAssertOncePerRelay(t, store, 0) +} + +// Restart: Load indexes every persisted observation and rebuilds the index; +// a live observation after that must not add the transmission again. +func TestPathHopIndexOncePerTx_AfterLoad_2108(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, phdObserverCount) + base := time.Now().UTC().Add(-30 * time.Minute) + phdInsertTx(t, db, 0, base) + for o := 1; o < phdObserverCount; o++ { + phdInsertObs(t, db, 0, o, base.Add(time.Duration(o)*time.Second), true) + } + store := phdLoadedStore(t, db) + since := phdMaxObsID(t, db) + phdInsertObs(t, db, 0, phdObserverCount, base.Add(time.Minute), false) + store.IngestNewObservations(since, 100) + phdAssertOncePerRelay(t, store, 0) +} + +// The dedupe is per (transmission, relay), not per transmission: a later +// observation that was heard through a relay none of the earlier ones went +// through must still add the transmission under that relay, exactly once. +func TestPathHopIndexAddsNewRelayFromLaterObservation_2108(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, 4) + store := phdLoadedStore(t, db) + + now := time.Now().UTC().Add(-10 * time.Minute) + phdInsertTx(t, db, 0, now) // phdPaths[0] = relays 0, 1, 2 + phdInsertObs(t, db, 0, 1, now, false) + store.IngestNewFromDB(0, 100) + tx := store.byTxID[1] + newRelay := phdRelays[3] + if n := phdCountTxIn(store, newRelay, tx); n != 0 { + t.Fatalf("fixture: tx already under the new relay (%d entries)", n) + } + before := store.GetRepeaterUsefulnessScore(newRelay) + + // Two further observers hear it through relays 0, 1 and the new relay 3. + for o := 2; o <= 3; o++ { + since := phdMaxObsID(t, db) + phdInsertObsVia(t, db, 0, []int{0, 1, 3}, o, now.Add(time.Duration(o)*time.Second), false) + store.IngestNewObservations(since, 100) + if n := phdCountTxIn(store, newRelay, tx); n != 1 { + t.Fatalf("after observation %d through the new relay: tx is in its bucket %d times, want 1", o, n) + } + } + for _, r := range []int{0, 1, 2} { + if n := phdCountTxIn(store, phdRelays[r], tx); n != 1 { + t.Errorf("tx is in byPathHop[%s…] %d times, want 1", phdRelays[r][:8], n) + } + } + if got, want := store.GetRepeaterUsefulnessScore(newRelay), 1.0; before != 0 || got != want { + t.Errorf("new relay's traffic share = %v (was %v), want %v (the only transmission)", got, before, want) + } +} + +// phdShares returns every relay's traffic share as seen by the three score +// functions, failing if they disagree. +func phdShares(t *testing.T, store *PacketStore) map[string]float64 { + t.Helper() + bulk := store.computeRepeaterUsefulnessScoreMap() + batch := store.GetRepeaterNodeStatsBatch(phdRelays, 24) + out := make(map[string]float64, len(phdRelays)) + for _, pk := range phdRelays { + single := store.GetRepeaterUsefulnessScore(pk) + if bulk[pk] != single || batch[pk].Score != single { + t.Fatalf("score functions disagree for %s…: bulk=%v single=%v batch=%v", pk[:8], bulk[pk], single, batch[pk].Score) + } + out[pk] = single + } + return out +} + +// phdExpectedShares is the definition applied to the fixture: the fraction +// of the (all non-advert) transmissions whose path contains the relay. +func phdExpectedShares() map[string]float64 { + out := make(map[string]float64, len(phdRelays)) + for _, path := range phdPaths { + for _, r := range path { + out[phdRelays[r]] += 1 / float64(len(phdPaths)) + } + } + return out +} + +func phdAssertShares(t *testing.T, label string, got map[string]float64) { + t.Helper() + want := phdExpectedShares() + sum := 0.0 + for _, pk := range phdRelays { + sum += got[pk] + if math.Abs(got[pk]-want[pk]) > 1e-9 { + t.Errorf("%s: traffic share of %s… = %.4f, want %.4f", label, pk[:8], got[pk], want[pk]) + } + if got[pk] >= 1 { + t.Errorf("%s: traffic share of %s… reached the 1.0 clamp", label, pk[:8]) + } + } + if sum > phdMaxPathLen+1e-9 { + t.Errorf("%s: sum of all relays' traffic share = %.3f, exceeds the maximum path length %d", label, sum, phdMaxPathLen) + } +} + +// The traffic share of a fixed set of transmissions is the same right after +// a load and after any number of further observations of them. +func TestTrafficShareStableAcrossLateObservations_2108(t *testing.T) { + base := time.Now().UTC().Add(-30 * time.Minute) + + // Restart: every observation is already persisted (resolved_path set by + // the ingestor) and comes in through Load. + loadDB := setupTestDB(t) + defer loadDB.conn.Close() + phdSeed(t, loadDB, phdObserverCount) + for i := range phdPaths { + phdInsertTx(t, loadDB, i, base.Add(time.Duration(i)*time.Second)) + for o := 1; o <= phdObserverCount; o++ { + phdInsertObs(t, loadDB, i, o, base.Add(time.Duration(i*100+o)*time.Second), true) + } + } + loaded := phdLoadedStore(t, loadDB) + afterLoad := phdShares(t, loaded) + phdAssertShares(t, "after load", afterLoad) + + // Live: the same transmissions arrive with one observation each, then + // the late observations arrive in rounds. + liveDB := setupTestDB(t) + defer liveDB.conn.Close() + phdSeed(t, liveDB, phdObserverCount) + live := phdLoadedStore(t, liveDB) + for i := range phdPaths { + phdInsertTx(t, liveDB, i, base.Add(time.Duration(i)*time.Second)) + phdInsertObs(t, liveDB, i, 1, base.Add(time.Duration(i*100+1)*time.Second), false) + } + live.IngestNewFromDB(0, 1000) + afterIngest := phdShares(t, live) + phdAssertShares(t, "after live ingest", afterIngest) + + for round := 0; round < 2; round++ { + since := phdMaxObsID(t, liveDB) + for i := range phdPaths { + for k := 0; k < phdLateObs/2; k++ { + o := 2 + round*(phdLateObs/2) + k + phdInsertObs(t, liveDB, i, o, base.Add(time.Duration(i*100+o)*time.Second), false) + } + } + live.IngestNewObservations(since, 10000) + got := phdShares(t, live) + label := fmt.Sprintf("after late-observation round %d", round+1) + phdAssertShares(t, label, got) + if !reflect.DeepEqual(got, afterIngest) { + t.Errorf("%s: shares changed without new transmissions:\n got %v\nwant %v", label, got, afterIngest) + } + } + for i := range phdPaths { + phdAssertOncePerRelay(t, live, i) + } + if !reflect.DeepEqual(phdShares(t, live), afterLoad) { + t.Errorf("live shares differ from a fresh load of the same data:\nlive %v\nload %v", phdShares(t, live), afterLoad) + } +} + +// Index size is a function of the transmissions, not of how many times they +// were heard: further observations through the same relays add nothing. +func TestPathHopIndexSizeBoundedByTransmissions_2108(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, phdObserverCount) + store := phdLoadedStore(t, db) + base := time.Now().UTC().Add(-30 * time.Minute) + for i := range phdPaths { + phdInsertTx(t, db, i, base.Add(time.Duration(i)*time.Second)) + phdInsertObs(t, db, i, 1, base.Add(time.Duration(i*100+1)*time.Second), false) + } + store.IngestNewFromDB(0, 1000) + entries := func() int { + store.mu.RLock() + defer store.mu.RUnlock() + n := 0 + for _, list := range store.byPathHop { + n += len(list) + } + return n + } + want := entries() + for o := 2; o <= phdObserverCount; o++ { + since := phdMaxObsID(t, db) + for i := range phdPaths { + phdInsertObs(t, db, i, o, base.Add(time.Duration(i*100+o)*time.Second), false) + } + store.IngestNewObservations(since, 10000) + if got := entries(); got != want { + t.Fatalf("after observation %d of each tx: byPathHop holds %d entries, want %d (one per relay key per tx)", o, got, want) + } + } +} + +// A repeated observation through relays the transmission is already indexed +// under changes nothing in byPathHop, so it must leave the cached batch +// relay stats in place (the cache is dropped only when byPathHop changes, +// #1164). An observation through a new relay does change byPathHop and +// must drop the cache, so the next read sees the new relay. +func TestRelayStatsCacheAcrossRepeatedObservations_2108(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, 4) + store := phdLoadedStore(t, db) + + now := time.Now().UTC().Add(-10 * time.Minute) + phdInsertTx(t, db, 0, now) // relays 0, 1, 2 + phdInsertObs(t, db, 0, 1, now, false) + store.IngestNewFromDB(0, 100) + cached := func() map[string]RepeaterNodeStats { + store.relayStatsCacheMu.Lock() + defer store.relayStatsCacheMu.Unlock() + return store.relayStatsCache + } + first := store.GetRepeaterNodeStatsBatchCached(phdRelays, 24) + if cached() == nil || first[phdRelays[0]].Score != 1 || first[phdRelays[3]].Score != 0 { + t.Fatalf("fixture: cache not populated as expected: %+v", first) + } + + for o := 2; o <= 3; o++ { + since := phdMaxObsID(t, db) + phdInsertObs(t, db, 0, o, now.Add(time.Duration(o)*time.Second), false) + store.IngestNewObservations(since, 100) + if cached() == nil { + t.Fatalf("repeated observation %d through known relays dropped the relay-stats cache", o) + } + } + if fresh := store.GetRepeaterNodeStatsBatch(phdRelays, 24); !reflect.DeepEqual(fresh, store.GetRepeaterNodeStatsBatchCached(phdRelays, 24)) { + t.Fatalf("cache kept across repeated observations is stale:\nfresh %+v\ncached %+v", fresh, cached()) + } + + since := phdMaxObsID(t, db) + phdInsertObsVia(t, db, 0, []int{0, 1, 3}, 4, now.Add(4*time.Second), false) + store.IngestNewObservations(since, 100) + if cached() != nil { + t.Fatal("observation through a new relay must drop the relay-stats cache") + } + if got := store.GetRepeaterNodeStatsBatchCached(phdRelays, 24)[phdRelays[3]].Score; got != 1 { + t.Errorf("new relay's cached score = %v, want 1", got) + } +} + +// The helper itself: the return value and the cache follow whether byPathHop +// changed, per (transmission, relay). +func TestAddResolvedPubkeysToPathHopIndex_PerRelayIdempotent_2108(t *testing.T) { + keyA, keyB, keyC := strings.Repeat("a", 64), strings.Repeat("b", 64), strings.Repeat("c", 64) + s := &PacketStore{byPathHop: make(map[string][]*StoreTx)} + tx := &StoreTx{ID: 1, parsedPath: []string{"aa", "bb"}, pathParsed: true} + hopsSeen := map[string]bool{} + seed := func() { s.relayStatsCache = map[string]RepeaterNodeStats{"sentinel": {}} } + step := func(label string, pks []string, wantMutated bool) { + t.Helper() + seed() + if got := s.addResolvedPubkeysToPathHopIndex(tx, pks, hopsSeen); got != wantMutated { + t.Fatalf("%s: mutated = %v, want %v", label, got, wantMutated) + } + if dropped := s.relayStatsCache == nil; dropped != wantMutated { + t.Fatalf("%s: relay-stats cache dropped = %v, want %v", label, dropped, wantMutated) + } + } + step("first observation", []string{keyA, keyB}, true) + step("same relays again", []string{keyA, keyB}, false) + step("same relays, other order", []string{keyB, keyA}, false) + step("one new relay", []string{keyA, keyC}, true) + step("all known", []string{keyC, keyB, keyA}, false) + for _, k := range []string{keyA, keyB, keyC} { + if n := len(s.byPathHop[k]); n != 1 { + t.Errorf("byPathHop[%s…] holds %d entries, want 1", k[:4], n) + } + } +} + +// Defence in depth: whatever the index holds, the score counts distinct +// transmissions, not bucket entries. +func TestTrafficShareCountsDistinctTransmissions_2108(t *testing.T) { + store := makeTestStore(0, time.Now().UTC(), 0) + store.byPathHop = make(map[string][]*StoreTx) + pt, advert := 2, payloadTypeAdvert + relay := phdRelays[0] + var txs []*StoreTx + for i := 0; i < 10; i++ { + tx := &StoreTx{ID: i + 1, Hash: fmt.Sprintf("phd-%d", i), PayloadType: &pt} + txs = append(txs, tx) + store.packets = append(store.packets, tx) + store.byTxID[tx.ID] = tx + store.byPayloadType[pt] = append(store.byPayloadType[pt], tx) + } + ad := &StoreTx{ID: 99, Hash: "phd-advert", PayloadType: &advert} + store.byPayloadType[advert] = append(store.byPayloadType[advert], ad) + // Two of the ten transmissions, each present 6 times (the index shape the + // bug produced after 6 observations), plus a duplicated advert. + for k := 0; k < 6; k++ { + store.byPathHop[relay] = append(store.byPathHop[relay], txs[0], txs[1], ad) + } + + const want = 2.0 / 10 + if got := store.GetRepeaterUsefulnessScore(relay); got != want { + t.Errorf("GetRepeaterUsefulnessScore = %v, want %v", got, want) + } + if got := store.computeRepeaterUsefulnessScoreMap()[relay]; got != want { + t.Errorf("computeRepeaterUsefulnessScoreMap = %v, want %v", got, want) + } + if got := store.GetRepeaterNodeStatsBatch([]string{relay}, 24)[relay].Score; got != want { + t.Errorf("GetRepeaterNodeStatsBatch score = %v, want %v", got, want) + } +} + +// phdDuplicateFullKeyEntries reproduces the index shape the bug produced: every +// transmission in a full-pubkey bucket is present `extra` more times. +func phdDuplicateFullKeyEntries(store *PacketStore, extra int) { + store.mu.Lock() + defer store.mu.Unlock() + for key, list := range store.byPathHop { + if len(key) != 64 { + continue + } + orig := append([]*StoreTx(nil), list...) + for k := 0; k < extra; k++ { + store.byPathHop[key] = append(store.byPathHop[key], orig...) + } + } + store.invalidateRelayStatsCache() +} + +// The other byPathHop consumers already deduplicate by transmission (or only +// read raw prefix keys and take a maximum): duplicate entries do not change +// their results. This pins that, so a refactor cannot reintroduce the +// inflation there. +func TestPathHopConsumersIgnoreDuplicateEntries_2108(t *testing.T) { + base := time.Now().UTC().Add(-30 * time.Minute) + db := setupTestDB(t) + defer db.conn.Close() + phdSeed(t, db, phdObserverCount) + for i := range phdPaths { + phdInsertTx(t, db, i, base.Add(time.Duration(i)*time.Second)) + for o := 1; o <= 2; o++ { + phdInsertObs(t, db, i, o, base.Add(time.Duration(i*100+o)*time.Second), true) + } + } + store := phdLoadedStore(t, db) + + type snapshot struct { + info map[string]RepeaterRelayInfo + bulk map[string]RepeaterRelayInfo + multi []MultiByteCapEntry + } + take := func() snapshot { + s := snapshot{info: map[string]RepeaterRelayInfo{}} + batch := store.GetRepeaterNodeStatsBatch(phdRelays, 24) + for _, pk := range phdRelays { + s.info[pk] = batch[pk].Info + } + s.bulk = store.computeRepeaterRelayInfoMap(24) + s.multi = store.computeMultiByteCapability(nil) + return s + } + before := take() + if before.info[phdRelays[0]].RelayCount24h == 0 || before.bulk[phdRelays[0]].RelayCount24h == 0 { + t.Fatalf("fixture: relay 0 must have relay counts: %+v / %+v", before.info[phdRelays[0]], before.bulk[phdRelays[0]]) + } + phdDuplicateFullKeyEntries(store, phdLateObs) + after := take() + for _, pk := range phdRelays { + if !reflect.DeepEqual(before.info[pk], after.info[pk]) { + t.Errorf("relay info of %s… changed with duplicate entries:\nbefore %+v\n after %+v", pk[:8], before.info[pk], after.info[pk]) + } + if !reflect.DeepEqual(before.bulk[pk], after.bulk[pk]) { + t.Errorf("bulk relay info of %s… changed with duplicate entries:\nbefore %+v\n after %+v", pk[:8], before.bulk[pk], after.bulk[pk]) + } + } + if !reflect.DeepEqual(before.multi, after.multi) { + t.Errorf("computeMultiByteCapability changed with duplicate entries:\nbefore %+v\n after %+v", before.multi, after.multi) + } +} + +// computeMultiByteCapability reads raw prefix buckets only and keeps a +// maximum hash size per node, so duplicates there cannot inflate it either. +func TestMultiByteCapabilityIgnoresDuplicateEntries_2108(t *testing.T) { + db := setupCapabilityTestDB(t) + defer db.conn.Close() + db.conn.Exec("INSERT INTO nodes (public_key, name, role, last_seen) VALUES (?, ?, ?, ?)", + "aabbccdd11223344", "RepB", "repeater", recentTS(48)) + store := NewPacketStore(db, nil) + pt := 1 + pkt := &StoreTx{RawHex: "01" + buildPathByte(2, 1) + "aabb", PayloadType: &pt, PathJSON: `["aabb"]`, FirstSeen: recentTS(48)} + addTestPacket(store, pkt) + before := store.computeMultiByteCapability(nil) + store.mu.Lock() + for k := 0; k < phdLateObs; k++ { + store.byPathHop["aabb"] = append(store.byPathHop["aabb"], pkt) + } + store.mu.Unlock() + after := store.computeMultiByteCapability(nil) + if len(before) != 1 || !reflect.DeepEqual(before, after) { + t.Fatalf("duplicates changed multi-byte capability:\nbefore %+v\n after %+v", before, after) + } +} diff --git a/cmd/server/pathhop_eviction_1908_test.go b/cmd/server/pathhop_eviction_1908_test.go index f4b81264..33b6d3c7 100644 --- a/cmd/server/pathhop_eviction_1908_test.go +++ b/cmd/server/pathhop_eviction_1908_test.go @@ -44,9 +44,11 @@ func TestPathHopEviction1908(t *testing.T) { addTxToPathHopIndex(store.byPathHop, tx) store.indexResolvedPathHops(tx, []string{shared, pk}, hopsSeen) } - // Repeated observations can append the same resolved association again. - store.indexResolvedPathHops(store.packets[0], []string{shared, expiredOnly}, hopsSeen) - store.indexResolvedPathHops(store.packets[7], []string{shared}, hopsSeen) + // Repeated observations no longer append the same resolved + // association again (#2108), but eviction must still drop every + // occurrence from a bucket that holds one, so seed duplicates directly. + store.byPathHop[shared] = append(store.byPathHop[shared], store.packets[0], store.packets[7]) + store.byPathHop[expiredOnly] = append(store.byPathHop[expiredOnly], store.packets[0]) addTxToPathHopIndex(store.byPathHop, store.packets[0]) // Keep the original backing arrays observable: a shortened slice must diff --git a/cmd/server/pathhop_record_2108_test.go b/cmd/server/pathhop_record_2108_test.go new file mode 100644 index 00000000..b10e549f --- /dev/null +++ b/cmd/server/pathhop_record_2108_test.go @@ -0,0 +1,294 @@ +package main + +import ( + "fmt" + "runtime" + "strings" + "testing" + "time" + "unsafe" +) + +func TestCountDistinctNonAdvert_2108(t *testing.T) { + pt, advert := 2, payloadTypeAdvert + tx := func(id int) *StoreTx { return &StoreTx{ID: id, PayloadType: &pt} } + t1, t2, t3 := tx(1), tx(2), tx(3) + ad := &StoreTx{ID: 4, PayloadType: &advert} + untyped := &StoreTx{ID: 5} + cases := []struct { + name string + list []*StoreTx + want int + }{ + {"empty", nil, 0}, + {"ascending", []*StoreTx{t1, t2, t3}, 3}, + {"descending (chunk load order)", []*StoreTx{t3, t2, t1}, 3}, + {"adjacent duplicate", []*StoreTx{t1, t1}, 1}, + {"ascending then duplicate", []*StoreTx{t1, t2, t3, t3}, 3}, + {"interleaved duplicates", []*StoreTx{t1, t2, t1, t3, t2}, 3}, + {"adverts and nils skipped", []*StoreTx{nil, ad, t2, ad, nil, t2}, 1}, + {"untyped counted", []*StoreTx{untyped, t1, untyped}, 2}, + } + var ids []int + for _, c := range cases { + var got int + got, ids = countDistinctNonAdvert(c.list, ids) + if got != c.want { + t.Errorf("%s: got %d, want %d", c.name, got, c.want) + } + } +} + +// --- The per-transmission record behind the dedupe (pathHopResolved) --- +// +// It holds the resolved relay keys a live transmission is indexed under. +// These tests pin its lifecycle: bounded by relays (not observations), +// removed on eviction, kept for live transmissions only across an index +// rebuild, and emptied when a rebuild starts from an empty index. + +var ( + phdRecordKey1 = strings.Repeat("a", 64) + phdRecordKey2 = strings.Repeat("b", 64) +) + +// Raw hops of every transmission built by phdRecordStore (makeTestStore's +// PathJSON), plus the two resolved keys above. +const phdRecordRefsPerTx = 3 + 2 + +// phdRecordStore builds count non-advert transmissions, the older half past +// the 1 h retention, each indexed under its raw hops and two resolved keys +// through the real byPathHop helpers. +func phdRecordStore(count int) (store *PacketStore, old, young []*StoreTx) { + now := time.Now().UTC() + store = makeTestStore(count, now.Add(-30*time.Minute), 0) + store.byPayloadType[5] = store.byPayloadType[4] + delete(store.byPayloadType, 4) + store.byPathHop = make(map[string][]*StoreTx) + store.useResolvedPathIndex = true + store.initResolvedPathIndex() + store.retentionHours = 1 + hopsSeen := make(map[string]bool) + for i, tx := range store.packets { + *tx.PayloadType = 5 + if i < count/2 { + tx.FirstSeen = now.Add(-2 * time.Hour).Format(time.RFC3339) + old = append(old, tx) + } else { + young = append(young, tx) + } + addTxToPathHopIndex(store.byPathHop, tx) + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + } + return store, old, young +} + +// phdFreshKeys returns the two resolved keys as newly allocated strings, as +// every ingest path produces them (json.Unmarshal / strings.ToLower). +func phdFreshKeys() []string { + return []string{strings.Clone(phdRecordKey1), strings.Clone(phdRecordKey2)} +} + +func phdEntriesOf(store *PacketStore, tx *StoreTx) int { + n := 0 + for _, list := range store.byPathHop { + for _, t := range list { + if t == tx { + n++ + } + } + } + return n +} + +// The record does not grow with observations, and eviction removes it +// together with the transmission's byPathHop entries. +func TestPathHopResolvedRecordBoundedAndEvicted_2108(t *testing.T) { + store, old, young := phdRecordStore(40) + hopsSeen := map[string]bool{} + for _, tx := range store.packets { + for k := 0; k < 10; k++ { + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + } + } + if got := len(store.pathHopResolved); got != len(old)+len(young) { + t.Fatalf("record holds %d transmissions, want %d", got, len(old)+len(young)) + } + for _, tx := range young { + if got := len(store.pathHopResolved[tx]); got != 2 { + t.Fatalf("tx %d record holds %d keys after 11 observations, want 2", tx.ID, got) + } + if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { + t.Fatalf("tx %d has %d byPathHop entries after 11 observations, want %d", tx.ID, n, phdRecordRefsPerTx) + } + } + + store.mu.Lock() + evicted := store.EvictStale() + store.mu.Unlock() + if evicted != len(old) { + t.Fatalf("evicted %d, want %d", evicted, len(old)) + } + for _, tx := range old { + if _, ok := store.pathHopResolved[tx]; ok { + t.Fatalf("evicted tx %d is still in the record", tx.ID) + } + if n := phdEntriesOf(store, tx); n != 0 { + t.Fatalf("evicted tx %d is still in %d byPathHop buckets", tx.ID, n) + } + } + if got := len(store.pathHopResolved); got != len(young) { + t.Fatalf("record holds %d transmissions after eviction, want %d", got, len(young)) + } + // A survivor's next observation still adds nothing. + for _, tx := range young { + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { + t.Fatalf("tx %d has %d entries after eviction and another observation, want %d", tx.ID, n, phdRecordRefsPerTx) + } + } + + // Evicting the rest leaves no record and no interned key behind. + for _, tx := range young { + tx.FirstSeen = time.Now().UTC().Add(-2 * time.Hour).Format(time.RFC3339) + } + store.mu.Lock() + store.EvictStale() + store.mu.Unlock() + if len(store.pathHopResolved) != 0 || len(store.pathHopKeys) != 0 || len(store.byPathHop) != 0 { + t.Fatalf("after evicting every transmission: record %d, interned keys %d, byPathHop keys %d; want 0/0/0", + len(store.pathHopResolved), len(store.pathHopKeys), len(store.byPathHop)) + } +} + +// A rebuild keeps the record of live transmissions (their resolved entries +// are carried over), so observations after it still add nothing; it drops +// the record of transmissions that are no longer in the store. +func TestPathHopResolvedRecordAcrossRebuild_2108(t *testing.T) { + store, _, young := phdRecordStore(40) + gone := store.packets[0] + goneOnly := strings.Repeat("e", 64) // a relay only the removed transmission went through + store.indexResolvedPathHops(gone, []string{goneOnly}, map[string]bool{}) + store.packets = store.packets[1:] + delete(store.byTxID, gone.ID) + delete(store.byHash, gone.Hash) + + store.mu.Lock() + store.buildPathHopIndex() + store.mu.Unlock() + + if _, ok := store.pathHopResolved[gone]; ok { + t.Fatal("rebuild kept the record of a transmission no longer in the store") + } + if _, ok := store.byPathHop[goneOnly]; ok { + t.Fatal("fixture: rebuild carried over a bucket of a removed transmission") + } + if _, ok := store.pathHopKeys[goneOnly]; ok || len(store.pathHopKeys) != 2 { + t.Fatalf("rebuild kept interned keys without a bucket: %d interned, want 2", len(store.pathHopKeys)) + } + hopsSeen := map[string]bool{} + for _, tx := range young { + if _, ok := store.pathHopResolved[tx]; !ok { + t.Fatalf("rebuild dropped the record of live tx %d", tx.ID) + } + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { + t.Fatalf("tx %d has %d entries after a rebuild and another observation, want %d", tx.ID, n, phdRecordRefsPerTx) + } + } +} + +// When the index being rebuilt holds no resolved entries at all, nothing is +// indexed under any resolved key any more, so the record must not claim +// otherwise: the next observation has to put the transmission back. +func TestPathHopResolvedRecordClearedWithEmptyIndex_2108(t *testing.T) { + store, _, young := phdRecordStore(40) + store.byPathHop = make(map[string][]*StoreTx) // nothing carried over + + store.mu.Lock() + store.buildPathHopIndex() + store.mu.Unlock() + + if got := len(store.pathHopResolved); got != 0 { + t.Fatalf("record still holds %d transmissions after a rebuild from an empty index", got) + } + if got := len(store.pathHopKeys); got != 0 { + t.Fatalf("%d interned keys survive a rebuild from an empty index", got) + } + hopsSeen := map[string]bool{} + for _, tx := range young { + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { + t.Fatalf("tx %d has %d entries after the rebuild and another observation, want %d", tx.ID, n, phdRecordRefsPerTx) + } + } +} + +// The record compares keys exactly, and keeps one shared copy of each key +// rather than the per-observation string it was handed: every record entry +// and the byPathHop key point at the same bytes. +func TestPathHopResolvedRecordInternsKeys_2108(t *testing.T) { + store, _, young := phdRecordStore(10) + canon, ok := store.pathHopKeys[phdRecordKey1] + if !ok || len(store.pathHopKeys) != 2 { + t.Fatalf("interned keys = %d (key1 present %v), want 2", len(store.pathHopKeys), ok) + } + for _, tx := range young { + rec := store.pathHopResolved[tx] + if len(rec) != 2 || unsafe.StringData(rec[0]) != unsafe.StringData(canon) { + t.Fatalf("tx %d record %v does not share the interned key", tx.ID, rec) + } + } + for k := range store.byPathHop { + if k == phdRecordKey1 && unsafe.StringData(k) != unsafe.StringData(canon) { + t.Fatal("byPathHop key is not the interned copy") + } + } +} + +// BenchmarkPathHopRecordMemory_2108 reports what the record costs: heap +// retained by pathHopResolved and pathHopKeys (including any key string +// only they keep alive) per transmission, for n transmissions with k +// distinct resolved relay keys each from 4000 relays, every transmission +// heard twice with freshly allocated keys as the ingest paths produce them. +// The fixture's union of resolved keys per transmission over all of its +// observations has a mean of 5 and a maximum of 17. +func BenchmarkPathHopRecordMemory_2108(b *testing.B) { + heap := func() uint64 { + var m runtime.MemStats + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m) + return m.HeapAlloc + } + const relays = 4000 + for _, n := range []int{100000} { + for _, k := range []int{2, 5, 17} { + b.Run(fmt.Sprintf("txs=%d/keys=%d", n, k), func(b *testing.B) { + for it := 0; it < b.N; it++ { + b.StopTimer() + store := makeTestStore(0, time.Now().UTC(), 0) + store.byPathHop = make(map[string][]*StoreTx) + hopsSeen := map[string]bool{} + pks := make([]string, k) + for i := 0; i < n; i++ { + tx := &StoreTx{ID: i + 1, parsedPath: []string{"zz"}, pathParsed: true} + store.packets = append(store.packets, tx) + for obs := 0; obs < 2; obs++ { + for j := range pks { + r := (i*31 + j*997) % relays + pks[j] = fmt.Sprintf("%02x%062x", r%256, r) + } + store.indexResolvedPathHops(tx, pks, hopsSeen) + } + } + with := heap() + store.pathHopResolved, store.pathHopKeys = nil, nil + without := heap() + runtime.KeepAlive(store) + b.ReportMetric((float64(with)-float64(without))/float64(n), "record-B/tx") + b.StartTimer() + } + }) + } + } +} diff --git a/cmd/server/repeater_enrich_bulk.go b/cmd/server/repeater_enrich_bulk.go index b26c5669..4fb6cb69 100644 --- a/cmd/server/repeater_enrich_bulk.go +++ b/cmd/server/repeater_enrich_bulk.go @@ -248,16 +248,19 @@ func (s *PacketStore) computeRepeaterUsefulnessScoreMap() map[string]float64 { return out } denom := float64(totalNonAdvert) + var ids []int for key, list := range s.byPathHop { - relayed := 0 - for _, tx := range list { - if tx == nil { - continue - } - if tx.PayloadType != nil && *tx.PayloadType == payloadTypeAdvert { - continue - } - relayed++ + // Distinct transmissions, matching the denominator (#2108). Only + // full-pubkey keys (the resolved hops, and the only keys a + // repeater's score is read under) can hold a transmission more + // than once; raw hop buckets get one entry per transmission from + // addTxToPathHopIndex and are counted directly, which keeps this + // pass near its previous cost (BenchmarkTrafficShareScoreMap_2108). + var relayed int + if len(key) == 64 { + relayed, ids = countDistinctNonAdvert(list, ids) + } else { + relayed = countNonAdvert(list) } if relayed == 0 { continue diff --git a/cmd/server/repeater_usefulness.go b/cmd/server/repeater_usefulness.go index 49b8f86c..2e96ccc6 100644 --- a/cmd/server/repeater_usefulness.go +++ b/cmd/server/repeater_usefulness.go @@ -1,6 +1,7 @@ package main import ( + "slices" "sort" "strings" "time" @@ -11,13 +12,10 @@ import ( // repeater as a relay hop. Issue #672 (Traffic axis only — bridge, // coverage, and redundancy axes are deferred to follow-up work). // -// Numerator: count of non-advert StoreTx entries indexed under -// -// pubkey in byPathHop. -// -// Denominator: total non-advert StoreTx entries in the store -// -// (sum of byPayloadType for all keys != payloadTypeAdvert). +// Numerator: the number of distinct non-advert transmissions indexed under +// pubkey in byPathHop (countDistinctNonAdvert, #2108). Denominator: total +// non-advert transmissions in the store (sum of byPayloadType for all keys +// != payloadTypeAdvert). // // Returns 0 when there is no non-advert traffic, the pubkey is empty, // or the repeater never appears as a relay hop. Scores are clamped to @@ -48,17 +46,8 @@ func (s *PacketStore) GetRepeaterUsefulnessScore(pubkey string) float64 { return 0 } - // Numerator: this repeater's non-advert hop appearances. - relayed := 0 - for _, tx := range s.byPathHop[key] { - if tx == nil { - continue - } - if tx.PayloadType != nil && *tx.PayloadType == payloadTypeAdvert { - continue - } - relayed++ - } + // Numerator: the distinct non-advert transmissions this repeater relayed. + relayed, _ := countDistinctNonAdvert(s.byPathHop[key], nil) score := float64(relayed) / float64(totalNonAdvert) if score < 0 { @@ -70,6 +59,55 @@ func (s *PacketStore) GetRepeaterUsefulnessScore(pubkey string) float64 { return score } +// countDistinctNonAdvert returns the number of distinct non-advert +// transmissions in list, a byPathHop bucket. The index holds a transmission +// once per key (#2108); counting distinct transmissions keeps traffic share +// a fraction of transmissions even if a bucket ever held duplicates, since +// the denominator counts each transmission once. ids is caller-owned +// scratch (may be nil), returned for reuse. +// +// Cost: O(len(list)) with no allocation once ids has grown. Buckets are +// appended in ingest order, so their IDs are usually strictly increasing, +// which already proves them distinct; only otherwise (background chunks +// load older transmissions after newer ones) are the IDs sorted, +// O(n log n) on a flat slice. A map per bucket was measured ~10x slower +// on the bulk pass (BenchmarkTrafficShareScoreMap_2108). +func countDistinctNonAdvert(list []*StoreTx, ids []int) (int, []int) { + ids = ids[:0] + ascending := true + for _, tx := range list { + if tx == nil || (tx.PayloadType != nil && *tx.PayloadType == payloadTypeAdvert) { + continue + } + if n := len(ids); n > 0 && tx.ID <= ids[n-1] { + ascending = false + } + ids = append(ids, tx.ID) + } + if ascending || len(ids) < 2 { + return len(ids), ids + } + slices.Sort(ids) + n := 1 + for i := 1; i < len(ids); i++ { + if ids[i] != ids[i-1] { + n++ + } + } + return n, ids +} + +// countNonAdvert returns the number of non-advert entries in list. +func countNonAdvert(list []*StoreTx) int { + n := 0 + for _, tx := range list { + if tx != nil && (tx.PayloadType == nil || *tx.PayloadType != payloadTypeAdvert) { + n++ + } + } + return n +} + // RepeaterNodeStats bundles relay-activity and usefulness data for a single node. type RepeaterNodeStats struct { Info RepeaterRelayInfo @@ -90,7 +128,7 @@ func (s *PacketStore) GetRepeaterNodeStatsBatch(pubkeys []string, windowHours fl type nodeSnap struct { entries []relayEntry - relayed int // non-advert count in full-key list only (for usefulness score) + relayed int // distinct non-advert txs in full-key list only (for usefulness score) } s.mu.RLock() @@ -103,15 +141,12 @@ func (s *PacketStore) GetRepeaterNodeStatsBatch(pubkeys []string, windowHours fl } snaps := make(map[string]nodeSnap, len(pubkeys)) + var ids []int for _, pk := range pubkeys { key := strings.ToLower(pk) entries := s.collectRelayEntriesLocked(key) - relayed := 0 - for _, tx := range s.byPathHop[key] { - if tx != nil && (tx.PayloadType == nil || *tx.PayloadType != payloadTypeAdvert) { - relayed++ - } - } + var relayed int + relayed, ids = countDistinctNonAdvert(s.byPathHop[key], ids) snaps[pk] = nodeSnap{entries: entries, relayed: relayed} } diff --git a/cmd/server/store.go b/cmd/server/store.go index a26f72fa..e5919418 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -394,6 +394,21 @@ type PacketStore struct { usefulnessAxesRecompMu sync.Mutex usefulnessAxesRecompStarted bool + // pathHopResolved records, per transmission, the resolved pubkey keys + // addResolvedPubkeysToPathHopIndex has put it under in byPathHop, so + // further observations through the same relays do not append it again + // (#2108). Keys are compared exactly. At most one entry per relay key + // per live transmission: evictStaleInternal deletes evicted + // transmissions and retainResolvedPathHops drops any no longer in + // s.packets. + pathHopResolved map[*StoreTx][]string + // pathHopKeys interns those keys: one shared copy per resolved byPathHop + // key, so the record holds a 16-byte string header per entry instead of + // keeping alive the string each observation's resolve allocated. It + // holds only keys that have a byPathHop bucket; eviction and rebuild + // drop the others. + pathHopKeys map[string]string + // Precomputed distinct advert pubkey count (refcounted for eviction correctness). // Updated incrementally during Load/Ingest/Evict — avoids JSON parsing in GetPerfStoreStats. advertPubkeys map[string]int // pubkey → number of advert packets referencing it @@ -715,6 +730,8 @@ func NewPacketStore(db *DB, cfg *PacketStoreConfig, cacheTTLs ...map[string]inte spIndex: make(map[string]int, 4096), spTxIndex: make(map[string][]*StoreTx, 4096), advertPubkeys: make(map[string]int), + pathHopResolved: make(map[*StoreTx][]string), + pathHopKeys: make(map[string]string), clockSkew: NewClockSkewEngine(), useResolvedPathIndex: true, areaNodeCache: make(map[string]map[string]bool), @@ -1799,6 +1816,10 @@ func pathLen(pathJSON string) int { // empties); the helper is a no-op when pks is empty. // - hopsSeen is a reusable scratch map; addResolvedPubkeysToPathHopIndex // clear()s it on entry. +// - Safe to call once per observation: every index it feeds is +// idempotent per (transmission, pubkey) — byNode via nodeHashes, the +// resolved index via its forward-list check, byPathHop via +// pathHopResolved (#2108). func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen map[string]bool) { if len(pks) == 0 { return @@ -4225,18 +4246,27 @@ func (s *PacketStore) buildPathHopIndex() { // after s.byPathHop has been rebuilt from raw hops. func (s *PacketStore) retainResolvedPathHops(prev map[string][]*StoreTx) int { if len(prev) == 0 { + // No resolved entries anywhere, so nothing is recorded as indexed. + clear(s.pathHopResolved) + clear(s.pathHopKeys) return 0 } live := make(map[*StoreTx]struct{}, len(s.packets)) for _, tx := range s.packets { live[tx] = struct{}{} } + // Keep the per-transmission record in step with what is carried over + // below: transmissions that are gone are not (#2108). + for tx := range s.pathHopResolved { + if _, ok := live[tx]; !ok { + delete(s.pathHopResolved, tx) + } + } // 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. + // pass already produced and repeated appends of the same tx in prev + // (an index built before #2108 made the resolved append idempotent). seen := make(map[*StoreTx]struct{}, 16) retained := 0 for key, list := range prev { @@ -4262,6 +4292,12 @@ func (s *PacketStore) retainResolvedPathHops(prev map[string][]*StoreTx) int { retained++ } } + // Interned keys whose bucket was not carried over have no entries left. + for key := range s.pathHopKeys { + if _, ok := s.byPathHop[key]; !ok { + delete(s.pathHopKeys, key) + } + } return retained } @@ -4378,11 +4414,21 @@ func removeTxFromPathHopIndex(idx map[string][]*StoreTx, tx *StoreTx) { } // addResolvedPubkeysToPathHopIndex appends tx into byPathHop under each -// resolved pubkey key that isn't already present as a raw hop. Mutating -// byPathHop here MUST be paired with invalidateRelayStatsCache so the -// cached batch relay stats don't go stale for up to relayStatsCacheTTL. -// hopsSeen is a scratch map the caller can reuse across calls (it will -// be cleared on entry). +// resolved pubkey key that isn't already present as a raw hop and that tx +// has not been appended under before. It is called once per observation, +// and the observations of one transmission usually resolve to the same +// relays: the per-transmission record in s.pathHopResolved keeps the index +// at one entry per (key, transmission), so per-key counts are distinct +// transmissions (#2108). Mutating byPathHop here MUST be paired with +// invalidateRelayStatsCache so the cached batch relay stats don't go stale +// for up to relayStatsCacheTTL. hopsSeen is a scratch map the caller can +// reuse across calls (it will be cleared on entry). +// +// Cost per call: an exact string compare of each pubkey against the +// transmission's own record (one entry per distinct relay key, a handful). +// A repeat observation through known relays allocates nothing and leaves +// byPathHop, and so the relay-stats cache, untouched. A new key costs one +// pathHopKeys lookup to share the interned copy. // // Must be called with s.mu held. func (s *PacketStore) addResolvedPubkeysToPathHopIndex(tx *StoreTx, pubkeys []string, hopsSeen map[string]bool) bool { @@ -4393,19 +4439,40 @@ func (s *PacketStore) addResolvedPubkeysToPathHopIndex(tx *StoreTx, pubkeys []st for _, hop := range txGetParsedPath(tx) { hopsSeen[strings.ToLower(hop)] = true } - mutated := false + if s.pathHopResolved == nil { + s.pathHopResolved = make(map[*StoreTx][]string) + } + if s.pathHopKeys == nil { + s.pathHopKeys = make(map[string]string) + } + indexed := s.pathHopResolved[tx] + before := len(indexed) for _, pk := range pubkeys { - if !hopsSeen[pk] { - hopsSeen[pk] = true - s.byPathHop[pk] = append(s.byPathHop[pk], tx) - mutated = true + if hopsSeen[pk] { + continue } + hopsSeen[pk] = true + if slices.Contains(indexed, pk) { + continue + } + key, ok := s.pathHopKeys[pk] + if !ok { + key = pk + s.pathHopKeys[key] = key + } + if indexed == nil { + indexed = make([]string, 0, len(pubkeys)) + } + indexed = append(indexed, key) + s.byPathHop[key] = append(s.byPathHop[key], tx) } + if len(indexed) == before { + return false + } + s.pathHopResolved[tx] = indexed // Mutating byPathHop invalidates the batch relay-stats cache (#1164). - if mutated { - s.invalidateRelayStatsCache() - } - return mutated + s.invalidateRelayStatsCache() + return true } // invalidateRelayStatsCache drops the cached batch relay-stats result so @@ -4917,6 +4984,9 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { for _, tx := range evicting { delete(s.byHash, tx.Hash) delete(s.byTxID, tx.ID) + // Its record of indexed resolved keys goes with it (#2108), so the + // record stays bounded by the live transmissions. + delete(s.pathHopResolved, tx) evictedTxIDs[tx.ID] = struct{}{} evictedBytes += tx.accountedBytes @@ -4992,6 +5062,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { }) if len(filtered) == 0 { delete(s.byPathHop, key) + delete(s.pathHopKeys, key) } else { s.byPathHop[key] = filtered }