diff --git a/cmd/server/db_test.go b/cmd/server/db_test.go index 19cddd7b2..77ea6f88a 100644 --- a/cmd/server/db_test.go +++ b/cmd/server/db_test.go @@ -13,7 +13,7 @@ import ( ) // setupTestDB creates an in-memory SQLite database with the v3 schema. -func setupTestDB(t *testing.T) *DB { +func setupTestDB(t testing.TB) *DB { t.Helper() conn, err := sql.Open("sqlite", ":memory:") if err != nil { diff --git a/cmd/server/evict_resolved_pathhop_115_test.go b/cmd/server/evict_resolved_pathhop_115_test.go index 14683bcdd..e04b41d4a 100644 --- a/cmd/server/evict_resolved_pathhop_115_test.go +++ b/cmd/server/evict_resolved_pathhop_115_test.go @@ -11,10 +11,9 @@ import ( // Issue #115: eviction must remove a transmission from every byPathHop // bucket it is in. indexResolvedPathHops puts it under resolved full-pubkey -// keys as well as its raw wire hops, and several observations of one -// transmission legitimately add it to the same resolved key more than once. -// removeTxFromPathHopIndex derived its keys from the raw path only and -// removed one occurrence, so evicted transmissions stayed reachable (relay +// keys as well as its raw wire hops (once per key since #158, however many +// observations resolve it). removeTxFromPathHopIndex derived its keys from +// the raw path only, so evicted transmissions stayed reachable (relay // counts, transported scopes, retained memory) until the next full rebuild. const ( @@ -27,7 +26,7 @@ const ( // evict115Store builds count transmissions (raw path aa,bb,cc), half of them // older than the 24h retention, indexed through the real byPathHop helpers: -// raw hops, and resolved hops from two observations (duplicates). +// raw hops, and resolved hops from two observations. func evict115Store(t testing.TB, count int, resolvedIndex bool) (*PacketStore, []*StoreTx, []*StoreTx) { t.Helper() now := time.Now().UTC() @@ -52,7 +51,7 @@ func evict115Store(t testing.TB, count int, resolvedIndex bool) (*PacketStore, [ if i < count/2 { pks = append(pks, evict115Only) } - for obs := 0; obs < 2; obs++ { // two observations -> duplicate entries + for obs := 0; obs < 2; obs++ { // two observations, one entry per key (#158) store.indexResolvedPathHops(tx, pks, hopsSeen) } } @@ -108,15 +107,15 @@ func assertEvictedGone115Partial(t *testing.T, store *PacketStore, old, young [] } } -// Survivors: 3 raw keys once each, 2 resolved keys twice each. -const young115Refs = 3 + 2*2 +// Survivors: 3 raw keys and 2 resolved keys, once each (#158). +const young115Refs = 3 + 2 func TestEvictRemovesRawAndResolvedPathHops_115(t *testing.T) { for _, mode := range []bool{true, false} { t.Run(fmt.Sprintf("useResolvedPathIndex=%v", mode), func(t *testing.T) { store, old, young := evict115Store(t, 40, mode) - if n := countIn(store.byPathHop, old[0]); n != 3+3*2 { - t.Fatalf("fixture: old tx indexed %d times, want %d", n, 3+3*2) + if n := countIn(store.byPathHop, old[0]); n != 3+3 { + t.Fatalf("fixture: old tx indexed %d times, want %d", n, 3+3) } if got := store.EvictStale(); got != len(old) { t.Fatalf("evicted %d, want %d", got, len(old)) @@ -134,7 +133,7 @@ func TestEvictRemovesResolvedKeysOfOtherObservationPaths_115(t *testing.T) { tx := old[0] tx.Observations = append(tx.Observations, &StoreObs{ID: 900001, TransmissionID: tx.ID, PathJSON: `["dd"]`}) store.indexResolvedPathHops(tx, []string{otherPK}, map[string]bool{}) - if countIn(store.byPathHop, tx) != 3+3*2+1 { + if countIn(store.byPathHop, tx) != 3+3+1 { t.Fatal("fixture: other-path resolved key not indexed") } store.EvictStale() @@ -144,6 +143,27 @@ func TestEvictRemovesResolvedKeysOfOtherObservationPaths_115(t *testing.T) { } } +// A byPathHop index built before #158 can hold the same transmission several +// times in one resolved bucket (once per observation). Eviction must drop +// every occurrence, and delete a bucket that ends up empty. +func TestEvictRemovesDuplicateResolvedEntries_115(t *testing.T) { + store, old, young := evict115Store(t, 40, true) + dup := old[0] + before := countIn(store.byPathHop, dup) + for _, key := range []string{evict115Only, evict115PK1} { // one emptied, one shared with survivors + for i := 0; i < 2; i++ { + store.byPathHop[key] = append(store.byPathHop[key], dup) + } + } + if n := countIn(store.byPathHop, dup); n != before+4 { + t.Fatalf("fixture: legacy duplicates not in place, tx indexed %d times", n) + } + if got := store.EvictStale(); got != len(old) { + t.Fatalf("evicted %d, want %d", got, len(old)) + } + assertEvictedGone115(t, store, old, young, young115Refs) +} + func TestRunEvictionRemovesResolvedPathHops_115(t *testing.T) { store, old, young := evict115Store(t, 40, true) if got := store.RunEviction(); got != len(old) { @@ -163,9 +183,9 @@ func TestMemoryEvictionRemovesResolvedPathHops_115(t *testing.T) { t.Fatal("fixture: memory eviction evicted nothing") } assertEvictedGone115Partial(t, store, before[:n], nil, 0) - for _, tx := range before[n : len(before)/2] { // older half not evicted: all 9 refs - if got := countIn(store.byPathHop, tx); got != 3+3*2 { - t.Fatalf("surviving tx %d has %d entries, want 9", tx.ID, got) + for _, tx := range before[n : len(before)/2] { // older half not evicted: all 6 refs + if got := countIn(store.byPathHop, tx); got != 3+3 { + t.Fatalf("surviving tx %d has %d entries, want 6", tx.ID, got) } } } @@ -226,7 +246,7 @@ func TestRelayStatsConcurrentWithEviction_115(t *testing.T) { // BenchmarkEvictPathHops_115 evicts the oldest transmissions from a store // of n with 3 raw 1-byte hops and 2 resolved hops each (two observations, -// so 4 resolved entries): a large batch (1% of the store) and a typical +// 2 resolved entries since #158): a large batch (1% of the store) and a typical // one-minute batch (0.01%; eviction runs every minute over 168 h). func BenchmarkEvictPathHops_115(b *testing.B) { prev := log.Writer() diff --git a/cmd/server/pathhop_dedupe_158_test.go b/cmd/server/pathhop_dedupe_158_test.go new file mode 100644 index 000000000..625afa7a3 --- /dev/null +++ b/cmd/server/pathhop_dedupe_158_test.go @@ -0,0 +1,685 @@ +package main + +import ( + "fmt" + "io" + "log" + "math" + "reflect" + "slices" + "strings" + "testing" + "time" +) + +// Issue #158: 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. Load and the background chunk load rebuild the +// index (retainResolvedPathHops dedups by *StoreTx), so values looked sane +// right after a restart and 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 relays158 = []string{ + "a1" + strings.Repeat("11", 31), + "b2" + strings.Repeat("22", 31), + "c3" + strings.Repeat("33", 31), + "d4" + strings.Repeat("44", 31), +} + +// paths158 is the fixed transmission set: indexes into relays158. The +// longest path has 3 hops, so the sum of all relays' share is at most 3. +var paths158 = [][]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 ( + maxPathLen158 = 3 + lateObs158 = 10 // extra observations per transmission (acceptance N) + observerCount158 = 1 + lateObs158 +) + +func rawHopsJSON158(path []int) string { + hops := make([]string, len(path)) + for i, r := range path { + hops[i] = `"` + strings.ToUpper(relays158[r][:2]) + `"` + } + return "[" + strings.Join(hops, ",") + "]" +} + +func resolvedJSON158(path []int) string { + pks := make([]string, len(path)) + for i, r := range path { + pks[i] = `"` + relays158[r] + `"` + } + return "[" + strings.Join(pks, ",") + "]" +} + +// seed158 creates the relays (as repeater nodes) and the observers. +func seed158(t testing.TB, db *DB, observers int) { + t.Helper() + ts := time.Now().UTC().Add(-time.Hour).Format(time.RFC3339) + for i, pk := range relays158 { + exec158(t, db, `INSERT INTO nodes (public_key, name, role, last_seen, first_seen, advert_count) VALUES (?, ?, 'repeater', ?, '2026-01-01', 1)`, + pk, fmt.Sprintf("Relay158-%d", i), ts) + } + for i := 1; i <= observers; i++ { + exec158(t, db, `INSERT INTO observers (rowid, id, name, iata) VALUES (?, ?, ?, 'TST')`, + i, fmt.Sprintf("obs158-%02d", i), fmt.Sprintf("Observer %d", i)) + } +} + +func exec158(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) + } +} + +// insertTx158 inserts transmission i of paths158 (non-advert TXT_MSG, flood). +func insertTx158(t testing.TB, db *DB, i int, firstSeen time.Time) int { + t.Helper() + id := i + 1 + exec158(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (?, 'CAFE', ?, ?, 1, 2, '{"type":"TXT_MSG"}')`, id, fmt.Sprintf("tx158-%03d", i), firstSeen.Format(time.RFC3339)) + return id +} + +// insertObs158 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 insertObs158(t testing.TB, db *DB, i, obsIdx int, ts time.Time, persisted bool) { + t.Helper() + var rp any + if persisted { + rp = resolvedJSON158(paths158[i]) + } + exec158(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) + VALUES (?, ?, 5, -90, ?, ?, ?)`, i+1, obsIdx, rawHopsJSON158(paths158[i]), ts.Unix(), rp) +} + +func maxObsID158(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 loadedStore158(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 countTxIn158(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 assertOncePerRelay158(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 != observerCount158 { + t.Fatalf("fixture: tx %d has %d observations, want %d", tx.ID, got, observerCount158) + } + for _, r := range paths158[i] { + if n := countTxIn158(store, relays158[r], tx); n != 1 { + t.Errorf("tx %d is in byPathHop[%s…] %d times, want exactly 1", tx.ID, relays158[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_158(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + seed158(t, db, observerCount158) + store := loadedStore158(t, db) + + now := time.Now().UTC().Add(-10 * time.Minute) + insertTx158(t, db, 0, now) + insertObs158(t, db, 0, 1, now, false) + store.IngestNewFromDB(0, 100) + for _, r := range paths158[0] { + if n := countTxIn158(store, relays158[r], store.byTxID[1]); n != 1 { + t.Fatalf("fixture: first observation must index tx once under %s…, got %d", relays158[r][:8], n) + } + } + + for k := 1; k <= lateObs158; k++ { + since := maxObsID158(t, db) + insertObs158(t, db, 0, 1+k, now.Add(time.Duration(k)*time.Second), false) + store.IngestNewObservations(since, 100) + } + assertOncePerRelay158(t, store, 0) +} + +// Live-ingest path: the transmission arrives together with all of its +// observations in one IngestNewFromDB batch. +func TestPathHopIndexOncePerTx_LiveIngestBatch_158(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + seed158(t, db, observerCount158) + store := loadedStore158(t, db) + + now := time.Now().UTC().Add(-10 * time.Minute) + insertTx158(t, db, 0, now) + for o := 1; o <= observerCount158; o++ { + insertObs158(t, db, 0, o, now.Add(time.Duration(o)*time.Second), false) + } + store.IngestNewFromDB(0, 100) + assertOncePerRelay158(t, store, 0) +} + +// shares158 returns every relay's traffic share as seen by the three score +// functions, failing if they disagree. +func shares158(t *testing.T, store *PacketStore) map[string]float64 { + t.Helper() + bulk := store.computeRepeaterUsefulnessScoreMap() + batch := store.GetRepeaterNodeStatsBatch(relays158, 24) + out := make(map[string]float64, len(relays158)) + for _, pk := range relays158 { + 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 +} + +// expectedShares158 is the definition applied to the fixture: the fraction +// of the (all non-advert) transmissions whose path contains the relay. +func expectedShares158() map[string]float64 { + out := make(map[string]float64, len(relays158)) + for _, path := range paths158 { + for _, r := range path { + out[relays158[r]] += 1 / float64(len(paths158)) + } + } + return out +} + +func assertShares158(t *testing.T, label string, got map[string]float64) { + t.Helper() + want := expectedShares158() + sum := 0.0 + for _, pk := range relays158 { + 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 > maxPathLen158+1e-9 { + t.Errorf("%s: sum of all relays' traffic share = %.3f, exceeds the maximum path length %d", label, sum, maxPathLen158) + } +} + +// 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_158(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() + seed158(t, loadDB, observerCount158) + for i := range paths158 { + insertTx158(t, loadDB, i, base.Add(time.Duration(i)*time.Second)) + for o := 1; o <= observerCount158; o++ { + insertObs158(t, loadDB, i, o, base.Add(time.Duration(i*100+o)*time.Second), true) + } + } + loaded := loadedStore158(t, loadDB) + afterLoad := shares158(t, loaded) + assertShares158(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() + seed158(t, liveDB, observerCount158) + live := loadedStore158(t, liveDB) + for i := range paths158 { + insertTx158(t, liveDB, i, base.Add(time.Duration(i)*time.Second)) + insertObs158(t, liveDB, i, 1, base.Add(time.Duration(i*100+1)*time.Second), false) + } + live.IngestNewFromDB(0, 1000) + afterIngest := shares158(t, live) + assertShares158(t, "after live ingest", afterIngest) + + for round := 0; round < 2; round++ { + since := maxObsID158(t, liveDB) + for i := range paths158 { + for k := 0; k < lateObs158/2; k++ { + o := 2 + round*(lateObs158/2) + k + insertObs158(t, liveDB, i, o, base.Add(time.Duration(i*100+o)*time.Second), false) + } + } + live.IngestNewObservations(since, 10000) + got := shares158(t, live) + label := fmt.Sprintf("after late-observation round %d", round+1) + assertShares158(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 paths158 { + assertOncePerRelay158(t, live, i) + } + if !reflect.DeepEqual(shares158(t, live), afterLoad) { + t.Errorf("live shares differ from a fresh load of the same data:\nlive %v\nload %v", shares158(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_158(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + seed158(t, db, observerCount158) + store := loadedStore158(t, db) + base := time.Now().UTC().Add(-30 * time.Minute) + for i := range paths158 { + insertTx158(t, db, i, base.Add(time.Duration(i)*time.Second)) + insertObs158(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 <= observerCount158; o++ { + since := maxObsID158(t, db) + for i := range paths158 { + insertObs158(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) + } + } +} + +// Defence in depth: whatever the index holds, the score counts distinct +// transmissions, not bucket entries. +func TestTrafficShareCountsDistinctTransmissions_158(t *testing.T) { + store := makeTestStore(0, time.Now().UTC(), 0) + store.byPathHop = make(map[string][]*StoreTx) + pt, advert := 2, payloadTypeAdvert + relay := relays158[0] + var txs []*StoreTx + for i := 0; i < 10; i++ { + tx := &StoreTx{ID: i + 1, Hash: fmt.Sprintf("d158-%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: "d158-advert", PayloadType: &advert} + store.byPayloadType[advert] = append(store.byPayloadType[advert], ad) + // Two of the ten transmissions, each present 6 times (the pre-#158 index + // shape 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) + } +} + +// The per-transmission record behind the dedupe holds one hash per relay key +// of a live transmission: it does not grow with observations, and eviction +// removes it together with the transmission's byPathHop entries (#115). +func TestPathHopResolvedRecordBounded_158(t *testing.T) { + store, old, young := evict115Store(t, 40, true) + hopsSeen := map[string]bool{} + for _, tx := range young { + for k := 0; k < lateObs158; k++ { + store.indexResolvedPathHops(tx, []string{evict115PK1, evict115PK2}, 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 %d observations, want 2", tx.ID, got, 2+lateObs158) + } + } + store.EvictStale() + for _, tx := range old { + if _, ok := store.pathHopResolved[tx]; ok { + t.Fatalf("evicted tx %d is still in the indexed-key record", tx.ID) + } + } + if got := len(store.pathHopResolved); got != len(young) { + t.Fatalf("record holds %d transmissions after eviction, want %d", got, len(young)) + } + assertEvictedGone115(t, store, old, young, young115Refs) +} + +// 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_158(t *testing.T) { + store, _, young := evict115Store(t, 40, true) + gone := store.packets[0] + store.packets = store.packets[1:] + delete(store.byTxID, gone.ID) + delete(store.byHash, gone.Hash) + store.buildPathHopIndex() + if _, ok := store.pathHopResolved[gone]; ok { + t.Fatal("rebuild kept the record of a transmission no longer in the store") + } + hopsSeen := map[string]bool{} + for _, tx := range young { + store.indexResolvedPathHops(tx, []string{evict115PK1, evict115PK2}, hopsSeen) + if n := countIn(store.byPathHop, tx); n != young115Refs { + t.Fatalf("tx %d has %d entries after a rebuild and another observation, want %d", tx.ID, n, young115Refs) + } + } +} + +// Restart: Load indexes every persisted observation and rebuilds the index; +// a live observation after that must not add the transmission again. +func TestPathHopIndexOncePerTx_AfterLoad_158(t *testing.T) { + db := setupTestDB(t) + defer db.conn.Close() + seed158(t, db, observerCount158) + base := time.Now().UTC().Add(-30 * time.Minute) + insertTx158(t, db, 0, base) + for o := 1; o < observerCount158; o++ { + insertObs158(t, db, 0, o, base.Add(time.Duration(o)*time.Second), true) + } + store := loadedStore158(t, db) + since := maxObsID158(t, db) + insertObs158(t, db, 0, observerCount158, base.Add(time.Minute), false) + store.IngestNewObservations(since, 100) + assertOncePerRelay158(t, store, 0) +} + +func TestCountDistinctNonAdvert_158(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} + 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}, + } + 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) + } + } +} + +// duplicateFullKeyEntries158 reproduces the pre-#158 index shape: every +// transmission in a full-pubkey bucket is present `extra` more times. +func duplicateFullKeyEntries158(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 #158 there. +func TestPathHopConsumersIgnoreDuplicateEntries_158(t *testing.T) { + base := time.Now().UTC().Add(-30 * time.Minute) + db := setupTestDB(t) + defer db.conn.Close() + seed158(t, db, observerCount158) + for i := range paths158 { + insertTx158(t, db, i, base.Add(time.Duration(i)*time.Second)) + for o := 1; o <= 2; o++ { + insertObs158(t, db, i, o, base.Add(time.Duration(i*100+o)*time.Second), true) + } + } + store := loadedStore158(t, db) + + type snapshot struct { + hops map[string][]HopAnalyticsPacket + info map[string]RepeaterRelayInfo + multi []MultiByteCapEntry + } + take := func() snapshot { + s := snapshot{hops: map[string][]HopAnalyticsPacket{}, info: map[string]RepeaterRelayInfo{}} + batch := store.GetRepeaterNodeStatsBatch(relays158, 24) + for _, pk := range relays158 { + resp, err := store.GetNodeHopAnalytics(pk, 7) + if err != nil || resp == nil { + t.Fatalf("GetNodeHopAnalytics(%s…): %v, %v", pk[:8], resp, err) + } + s.hops[pk] = resp.Packets + s.info[pk] = batch[pk].Info + } + s.multi = store.computeMultiByteCapability(nil) + return s + } + before := take() + if len(before.hops[relays158[0]]) == 0 || before.info[relays158[0]].RelayCount24h == 0 { + t.Fatalf("fixture: relay 0 must have hop analytics and relay counts: %+v / %+v", before.hops[relays158[0]], before.info[relays158[0]]) + } + duplicateFullKeyEntries158(store, lateObs158) + after := take() + for _, pk := range relays158 { + if !reflect.DeepEqual(before.hops[pk], after.hops[pk]) { + t.Errorf("GetNodeHopAnalytics(%s…) changed with duplicate entries:\nbefore %+v\n after %+v", pk[:8], before.hops[pk], after.hops[pk]) + } + 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.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_158(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 < lateObs158; 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) + } +} + +// lateObsBenchStore158: n transmissions one second apart, each with 3 raw +// one-byte hops of which 2 resolve to one of 4000 relays (the shape of +// evict115BenchStore), indexed once as by their first observation. +func lateObsBenchStore158(n int) (*PacketStore, [][]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{} + pks := make([][]string, n) + for i := 0; i < n; i++ { + r1, r2 := i%2000, 2000+(5000+i*31)%2000 + h1, h2, h3 := r1%256, r2%256, (i*13+5)%256 + 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"]`, h1, h2, h3), + 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) + pks[i] = []string{fmt.Sprintf("%02x%062x", h1, r1), fmt.Sprintf("%02x%062x", h2, r2)} + store.indexResolvedPathHops(tx, pks[i], hopsSeen) + } + return store, pks +} + +// BenchmarkLateObservationIndex_158 measures the store side of +// IngestNewObservations for one late observation (indexResolvedPathHops +// under the write lock, as the ingest loop calls it) on stores of realistic +// size, and reports how many byPathHop entries a transmission ends up with. +// Before #158 that number grows with every observation; after, it stays at +// 3 raw + 2 resolved. +func BenchmarkLateObservationIndex_158(b *testing.B) { + for _, n := range []int{20000, 100000} { + b.Run(fmt.Sprintf("txs=%d", n), func(b *testing.B) { + store, pks := lateObsBenchStore158(n) + hopsSeen := map[string]bool{} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + j := (i * 7919) % n // spread late observations over the store + tx := store.packets[j] + store.mu.Lock() + store.indexResolvedPathHops(tx, pks[j], hopsSeen) + store.mu.Unlock() + } + b.StopTimer() + entries := 0 + for _, list := range store.byPathHop { + entries += len(list) + } + b.ReportMetric(float64(entries)/float64(n), "pathhop-entries/tx") + }) + } +} + +// BenchmarkIngestNewObservations_158 drives the real late-observation ingest +// (SQL scan + index) for a batch of one new observation per transmission. +func BenchmarkIngestNewObservations_158(b *testing.B) { + prev := log.Writer() + log.SetOutput(io.Discard) + defer log.SetOutput(prev) + db := setupTestDB(b) + defer db.conn.Close() + const observers = 512 + seed158(b, db, observers) + base := time.Now().UTC().Add(-30 * time.Minute) + for i := range paths158 { + insertTx158(b, db, i, base.Add(time.Duration(i)*time.Second)) + insertObs158(b, db, i, 1, base.Add(time.Duration(i*1000+1)*time.Second), false) + } + store := loadedStore158(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 := maxObsID158(b, db) + for i := range paths158 { + insertObs158(b, db, i, o, base.Add(time.Duration(i*1000+o)*time.Second), false) + } + b.StartTimer() + store.IngestNewObservations(since, 10000) + } +} + +// BenchmarkTrafficShareScoreMap_158 measures the bulk traffic-share pass +// (cache-miss path of GetRepeaterUsefulnessScoreMap) on realistic sizes. +// order=reversed reverses every bucket, as when background chunks load +// older transmissions after newer ones: no bucket is in ascending ID order. +func BenchmarkTrafficShareScoreMap_158(b *testing.B) { + for _, c := range []struct { + n int + reversed bool + }{{20000, false}, {100000, false}, {20000, true}, {100000, true}} { + order := "ingest" + if c.reversed { + order = "reversed" + } + b.Run(fmt.Sprintf("txs=%d/order=%s", c.n, order), func(b *testing.B) { + store, _ := lateObsBenchStore158(c.n) + if c.reversed { + for _, list := range store.byPathHop { + slices.Reverse(list) + } + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + store.computeRepeaterUsefulnessScoreMap() + } + }) + } +} diff --git a/cmd/server/repeater_enrich_bulk.go b/cmd/server/repeater_enrich_bulk.go index c3794c6f2..bfca1afd6 100644 --- a/cmd/server/repeater_enrich_bulk.go +++ b/cmd/server/repeater_enrich_bulk.go @@ -235,16 +235,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 (#158). 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_158). + 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 a7476bb72..76324c45d 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,10 +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, #158). 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 @@ -45,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 { @@ -67,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 (#158); 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_158). +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 @@ -87,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() @@ -100,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 8383d866d..5fcd95244 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -418,6 +418,14 @@ type PacketStore struct { usefulnessAxesRecompMu sync.Mutex usefulnessAxesRecompStarted bool + // pathHopResolved records, per transmission, the hashes of the resolved + // pubkey keys addResolvedPubkeysToPathHopIndex has put it under in + // byPathHop, so further observations through the same relays do not + // append it again (#158). At most one hash per relay key per live + // transmission: evictStaleInternal deletes evicted entries and + // retainResolvedPathHops drops anything no longer in s.packets. + pathHopResolved map[*StoreTx][]uint64 + // 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 @@ -750,6 +758,7 @@ 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][]uint64), clockSkew: NewClockSkewEngine(), useResolvedPathIndex: true, areaNodeCache: make(map[string]map[string]bool), @@ -1870,6 +1879,10 @@ func pathFirstHop(pathJSON string) string { // 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 (#158). func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen map[string]bool) { if len(pks) == 0 { return @@ -4406,18 +4419,26 @@ 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) 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: entries of transmissions that are gone are not (#158). + 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 #158 made the resolved append idempotent). seen := make(map[*StoreTx]struct{}, 16) retained := 0 for key, list := range prev { @@ -4560,11 +4581,20 @@ 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 (#158). 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: one hash per pubkey and a scan of the transmission's own +// record (one hash per distinct relay key, a handful); a repeat observation +// through known relays allocates nothing and leaves byPathHop, and so the +// relay-stats cache, untouched. // // Must be called with s.mu held. func (s *PacketStore) addResolvedPubkeysToPathHopIndex(tx *StoreTx, pubkeys []string, hopsSeen map[string]bool) bool { @@ -4575,19 +4605,44 @@ 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][]uint64) + } + 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 + h := pathHopKeyHash(pk) + if slices.Contains(indexed, h) { + continue } + indexed = append(indexed, h) + s.byPathHop[pk] = append(s.byPathHop[pk], tx) } + if len(indexed) == before { + return false + } + s.pathHopResolved[tx] = indexed // Mutating byPathHop invalidates the batch relay-stats cache (#1164). - if mutated { - s.invalidateRelayStatsCache() + s.invalidateRelayStatsCache() + return true +} + +// pathHopKeyHash is 64-bit FNV-1a over a byPathHop key, without the +// allocations of hash/fnv. It identifies a key within one transmission's +// pathHopResolved record only: a collision would need two distinct relay +// pubkeys of the same transmission to hash alike, and would at worst leave +// that transmission out of one bucket — it can never inflate a count. +func pathHopKeyHash(key string) uint64 { + h := uint64(14695981039346656037) + for i := 0; i < len(key); i++ { + h ^= uint64(key[i]) + h *= 1099511628211 } - return mutated + return h } // invalidateRelayStatsCache drops the cached batch relay-stats result so @@ -4616,13 +4671,13 @@ func removeTxFromSlice(idx map[string][]*StoreTx, key string, tx *StoreTx) { // evictFromPathHopIndex removes the evicted transmissions from every // byPathHop bucket they can be in (#115). A transmission is in its raw hop // keys and, via indexResolvedPathHops, in resolved full-pubkey keys, once -// per observation that resolved it. The resolved pubkeys are kept nowhere -// per transmission (#800 keeps only a hash-only membership index, and they -// can come from path_json reconstruction as well as resolved_path), but -// each one resolves a hop of one of the transmission's observed paths, so -// it starts with that hop's prefix. Only buckets whose key starts -// (case-insensitively) with a hop of an evicted transmission are swept; -// every duplicate there is removed. Empty buckets are deleted and the +// each (#158). The resolved pubkey strings are kept nowhere per +// transmission (#800 keeps only a hash-only membership index, and +// pathHopResolved only hashes), but each one resolves a hop of one of the +// transmission's observed paths, so it starts with that hop's prefix. Only +// buckets whose key starts (case-insensitively) with a hop of an evicted +// transmission are swept; any duplicate there (an index built before #158) +// is removed too. Empty buckets are deleted and the // discarded tail of each compacted bucket is zeroed, so no backing array // keeps an evicted transmission alive. // @@ -5255,6 +5310,11 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { // Remove from the path-hop index: raw AND resolved keys, all duplicates, // in one pass per batch (#115). See evictFromPathHopIndex. evictFromPathHopIndex(s.byPathHop, evictedTxSet) + // ...and their record of indexed resolved keys (#158), so it stays + // bounded by the live transmissions. + for tx := range evictedTxSet { + delete(s.pathHopResolved, tx) + } s.invalidateRelayStatsCache() // Batch-remove from byObserver: single pass per affected observer slice