fix(store): index a transmission once per relay key in byPathHop (#2117)

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 <noreply@anthropic.com>
This commit is contained in:
dborup
2026-10-08 14:34:31 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent f3ae8ac25a
commit 7309bdb5a9
7 changed files with 1220 additions and 54 deletions
@@ -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")
})
}
}
+599
View File
@@ -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)
}
}
+5 -3
View File
@@ -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
+294
View File
@@ -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()
}
})
}
}
}
+12 -9
View File
@@ -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
+60 -25
View File
@@ -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}
}
+88 -17
View File
@@ -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
}