From 610e70e41612178592d700388d65b0b72144bb94 Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Thu, 1 Oct 2026 10:02:17 +0000
Subject: [PATCH 1/4] test(store): pin one byPathHop entry per relay per
 transmission (#158)

traffic_share_score grows with uptime because every observation of a
transmission appends it to its relays' byPathHop buckets again, while
the denominator counts it once. These tests are red on master:

- a transmission ingested once and then heard by 10 more observers
  through the same relays is in each relay bucket exactly once, via the
  late-observation path (IngestNewObservations) and the live-ingest
  batch path (IngestNewFromDB);
- the traffic share of a fixed transmission set is identical right
  after Load and after any number of extra observations, equal to the
  definition, sums to at most the longest path and never hits the 1.0
  clamp; the three score functions agree;
- byPathHop does not grow with observations of known transmissions;
- the score counts distinct transmissions even if the index held
  duplicates.

Two further tests (green on master) pin that GetNodeHopAnalytics,
GetRepeaterNodeStatsBatch relay info and computeMultiByteCapability
are unaffected by duplicate entries. Benchmarks for the late-observation
index step, the real IngestNewObservations path and the bulk score pass
give before/after numbers. setupTestDB takes testing.TB so the
DB-backed benchmark can reuse it.

Relates to #158

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QykybhB9Hkxz692fbADZMi
---
 cmd/server/db_test.go                 |   2 +-
 cmd/server/pathhop_dedupe_158_test.go | 571 ++++++++++++++++++++++++++
 2 files changed, 572 insertions(+), 1 deletion(-)
 create mode 100644 cmd/server/pathhop_dedupe_158_test.go

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/pathhop_dedupe_158_test.go b/cmd/server/pathhop_dedupe_158_test.go
new file mode 100644
index 000000000..e53544e81
--- /dev/null
+++ b/cmd/server/pathhop_dedupe_158_test.go
@@ -0,0 +1,571 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"math"
+	"reflect"
+	"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)
+	}
+}
+
+// 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.
+func BenchmarkTrafficShareScoreMap_158(b *testing.B) {
+	for _, n := range []int{20000, 100000} {
+		b.Run(fmt.Sprintf("txs=%d", n), func(b *testing.B) {
+			store, _ := lateObsBenchStore158(n)
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				store.computeRepeaterUsefulnessScoreMap()
+			}
+		})
+	}
+}

From b671ee1449d4eed92fda6c9f8c166848b27ba5fa Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Thu, 1 Oct 2026 10:08:18 +0000
Subject: [PATCH 2/4] fix(store): index a transmission once per relay key in
 byPathHop (#158)

indexResolvedPathHops runs once per observation, and every observation
of a transmission that resolved to the same relays appended it to their
byPathHop buckets again. traffic_share_score divides that entry count by
the number of non-advert transmissions, so it grew with uptime until a
restart rebuilt the index (prod: the sum over all repeaters reached
235.8 after 98 h, many clamped at 1.0).

- addResolvedPubkeysToPathHopIndex keeps, per transmission, the hashes
  of the resolved keys it has appended the transmission under
  (PacketStore.pathHopResolved) and skips keys already there. Bounded:
  one 8-byte hash per relay key per live transmission. StoreTx is
  unchanged (its 320-byte layout is pinned by tests).
- Eviction deletes the record of evicted transmissions next to the
  #115/#144 bucket sweep; retainResolvedPathHops drops the record of
  transmissions that are no longer in s.packets, so it stays in step
  with what a rebuild carries over.
- Defence in depth: GetRepeaterUsefulnessScore,
  computeRepeaterUsefulnessScoreMap and GetRepeaterNodeStatsBatch count
  distinct non-advert transmissions (countDistinctNonAdvert), matching
  the denominator.
- GetNodeHopAnalytics, the batch relay info and
  computeMultiByteCapability already deduplicate (or only take a
  maximum over raw prefix buckets) and are unchanged.
- The #115 eviction tests encoded the duplicates (3+3*2 entries per
  transmission); they now expect one entry per key.

A repeat observation through known relays no longer mutates byPathHop,
so it no longer invalidates the batch relay-stats cache either (the
#1164 contract: invalidate when byPathHop changes).

Relates to #158

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QykybhB9Hkxz692fbADZMi
---
 cmd/server/evict_resolved_pathhop_115_test.go |  29 +++--
 cmd/server/pathhop_dedupe_158_test.go         |  71 ++++++++++++
 cmd/server/repeater_enrich_bulk.go            |  14 +--
 cmd/server/repeater_usefulness.go             |  71 ++++++++----
 cmd/server/store.go                           | 106 ++++++++++++++----
 5 files changed, 221 insertions(+), 70 deletions(-)

diff --git a/cmd/server/evict_resolved_pathhop_115_test.go b/cmd/server/evict_resolved_pathhop_115_test.go
index 14683bcdd..47f1e8478 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()
@@ -163,9 +162,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 +225,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
index e53544e81..395f1ef28 100644
--- a/cmd/server/pathhop_dedupe_158_test.go
+++ b/cmd/server/pathhop_dedupe_158_test.go
@@ -369,6 +369,77 @@ func TestTrafficShareCountsDistinctTransmissions_158(t *testing.T) {
 	}
 }
 
+// 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)
+}
+
 // 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) {
diff --git a/cmd/server/repeater_enrich_bulk.go b/cmd/server/repeater_enrich_bulk.go
index c3794c6f2..9cad18c87 100644
--- a/cmd/server/repeater_enrich_bulk.go
+++ b/cmd/server/repeater_enrich_bulk.go
@@ -235,17 +235,11 @@ func (s *PacketStore) computeRepeaterUsefulnessScoreMap() map[string]float64 {
 		return out
 	}
 	denom := float64(totalNonAdvert)
+	seen := make(map[*StoreTx]struct{}, 64)
 	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++
-		}
+		seen = boundedScratch(seen)
+		// Distinct transmissions, matching the denominator (#158).
+		relayed := countDistinctNonAdvert(list, seen)
 		if relayed == 0 {
 			continue
 		}
diff --git a/cmd/server/repeater_usefulness.go b/cmd/server/repeater_usefulness.go
index a7476bb72..f18094991 100644
--- a/cmd/server/repeater_usefulness.go
+++ b/cmd/server/repeater_usefulness.go
@@ -11,10 +11,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 +45,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 +58,45 @@ 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. seen is caller-owned
+// scratch, cleared here; nil allocates one sized for list.
+//
+// Cost: O(len(list)) map operations.
+func countDistinctNonAdvert(list []*StoreTx, seen map[*StoreTx]struct{}) int {
+	if seen == nil {
+		seen = make(map[*StoreTx]struct{}, len(list))
+	} else {
+		clear(seen)
+	}
+	n := 0
+	for _, tx := range list {
+		if tx == nil || (tx.PayloadType != nil && *tx.PayloadType == payloadTypeAdvert) {
+			continue
+		}
+		if _, dup := seen[tx]; dup {
+			continue
+		}
+		seen[tx] = struct{}{}
+		n++
+	}
+	return n
+}
+
+// boundedScratch returns seen for reuse by countDistinctNonAdvert, or a
+// fresh small map once it has grown large: clear() costs the map's
+// capacity, so a scratch map that once held a big bucket must not be
+// cleared again for every small bucket after it.
+func boundedScratch(seen map[*StoreTx]struct{}) map[*StoreTx]struct{} {
+	if len(seen) > 1024 {
+		return make(map[*StoreTx]struct{}, 64)
+	}
+	return seen
+}
+
 // RepeaterNodeStats bundles relay-activity and usefulness data for a single node.
 type RepeaterNodeStats struct {
 	Info  RepeaterRelayInfo
@@ -87,7 +117,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 +130,12 @@ func (s *PacketStore) GetRepeaterNodeStatsBatch(pubkeys []string, windowHours fl
 	}
 
 	snaps := make(map[string]nodeSnap, len(pubkeys))
+	seen := make(map[*StoreTx]struct{})
 	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++
-			}
-		}
+		seen = boundedScratch(seen)
+		relayed := countDistinctNonAdvert(s.byPathHop[key], seen)
 		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

From 682cd623bba4e333a6e567dfbdbdd74231f7219c Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Thu, 1 Oct 2026 10:35:38 +0000
Subject: [PATCH 3/4] perf(store): count distinct transmissions without a map
 per bucket (#158)

The defensive distinct count from the previous commit used a map per
byPathHop bucket. BenchmarkTrafficShareScoreMap_158 showed the bulk
traffic-share pass going from 1.4 ms to 9.5 ms (20k txs) and 8.8 ms to
100 ms (100k txs, 18.7 MB allocated per pass), all under the read lock.

countDistinctNonAdvert now collects IDs into a reused slice; buckets
are appended in ingest order, so strictly increasing IDs already prove
them distinct and only out-of-order buckets (background chunk loads)
are sorted. The bulk pass counts distinct transmissions for full-pubkey
keys only: those are the resolved hops that could hold duplicates and
the only keys a repeater's score is read under; raw hop buckets get one
entry per transmission from addTxToPathHopIndex and are counted
directly, as before.

Bulk pass vs master (benchstat, n=6+8): +26..31% at ingest order,
+25..56% with every bucket reversed (8.8 -> 11.1 ms at 100k txs);
allocations unchanged (213 KiB). A unit test pins the helper on
ascending, descending, adjacent and interleaved duplicates, adverts and
nils; the benchmark gains an order=reversed case.

Relates to #158

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QykybhB9Hkxz692fbADZMi
---
 cmd/server/pathhop_dedupe_158_test.go | 49 ++++++++++++++++++--
 cmd/server/repeater_enrich_bulk.go    | 17 +++++--
 cmd/server/repeater_usefulness.go     | 65 ++++++++++++++++-----------
 3 files changed, 97 insertions(+), 34 deletions(-)

diff --git a/cmd/server/pathhop_dedupe_158_test.go b/cmd/server/pathhop_dedupe_158_test.go
index 395f1ef28..625afa7a3 100644
--- a/cmd/server/pathhop_dedupe_158_test.go
+++ b/cmd/server/pathhop_dedupe_158_test.go
@@ -6,6 +6,7 @@ import (
 	"log"
 	"math"
 	"reflect"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -440,6 +441,34 @@ func TestPathHopIndexOncePerTx_AfterLoad_158(t *testing.T) {
 	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) {
@@ -628,10 +657,24 @@ func BenchmarkIngestNewObservations_158(b *testing.B) {
 
 // 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 _, n := range []int{20000, 100000} {
-		b.Run(fmt.Sprintf("txs=%d", n), func(b *testing.B) {
-			store, _ := lateObsBenchStore158(n)
+	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++ {
diff --git a/cmd/server/repeater_enrich_bulk.go b/cmd/server/repeater_enrich_bulk.go
index 9cad18c87..bfca1afd6 100644
--- a/cmd/server/repeater_enrich_bulk.go
+++ b/cmd/server/repeater_enrich_bulk.go
@@ -235,11 +235,20 @@ func (s *PacketStore) computeRepeaterUsefulnessScoreMap() map[string]float64 {
 		return out
 	}
 	denom := float64(totalNonAdvert)
-	seen := make(map[*StoreTx]struct{}, 64)
+	var ids []int
 	for key, list := range s.byPathHop {
-		seen = boundedScratch(seen)
-		// Distinct transmissions, matching the denominator (#158).
-		relayed := countDistinctNonAdvert(list, seen)
+		// 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 f18094991..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"
@@ -46,7 +47,7 @@ func (s *PacketStore) GetRepeaterUsefulnessScore(pubkey string) float64 {
 	}
 
 	// Numerator: the distinct non-advert transmissions this repeater relayed.
-	relayed := countDistinctNonAdvert(s.byPathHop[key], nil)
+	relayed, _ := countDistinctNonAdvert(s.byPathHop[key], nil)
 
 	score := float64(relayed) / float64(totalNonAdvert)
 	if score < 0 {
@@ -62,39 +63,49 @@ func (s *PacketStore) GetRepeaterUsefulnessScore(pubkey string) float64 {
 // 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. seen is caller-owned
-// scratch, cleared here; nil allocates one sized for list.
+// the denominator counts each transmission once. ids is caller-owned
+// scratch (may be nil), returned for reuse.
 //
-// Cost: O(len(list)) map operations.
-func countDistinctNonAdvert(list []*StoreTx, seen map[*StoreTx]struct{}) int {
-	if seen == nil {
-		seen = make(map[*StoreTx]struct{}, len(list))
-	} else {
-		clear(seen)
-	}
-	n := 0
+// 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 _, dup := seen[tx]; dup {
-			continue
+		if n := len(ids); n > 0 && tx.ID <= ids[n-1] {
+			ascending = false
 		}
-		seen[tx] = struct{}{}
-		n++
+		ids = append(ids, tx.ID)
 	}
-	return n
+	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
 }
 
-// boundedScratch returns seen for reuse by countDistinctNonAdvert, or a
-// fresh small map once it has grown large: clear() costs the map's
-// capacity, so a scratch map that once held a big bucket must not be
-// cleared again for every small bucket after it.
-func boundedScratch(seen map[*StoreTx]struct{}) map[*StoreTx]struct{} {
-	if len(seen) > 1024 {
-		return make(map[*StoreTx]struct{}, 64)
+// 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 seen
+	return n
 }
 
 // RepeaterNodeStats bundles relay-activity and usefulness data for a single node.
@@ -130,12 +141,12 @@ func (s *PacketStore) GetRepeaterNodeStatsBatch(pubkeys []string, windowHours fl
 	}
 
 	snaps := make(map[string]nodeSnap, len(pubkeys))
-	seen := make(map[*StoreTx]struct{})
+	var ids []int
 	for _, pk := range pubkeys {
 		key := strings.ToLower(pk)
 		entries := s.collectRelayEntriesLocked(key)
-		seen = boundedScratch(seen)
-		relayed := countDistinctNonAdvert(s.byPathHop[key], seen)
+		var relayed int
+		relayed, ids = countDistinctNonAdvert(s.byPathHop[key], ids)
 		snaps[pk] = nodeSnap{entries: entries, relayed: relayed}
 	}
 

From baa68cddd674e39113aedb20a09f6f066e05a1ef Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Thu, 1 Oct 2026 12:40:12 +0000
Subject: [PATCH 4/4] test(store): pin eviction of legacy duplicate resolved
 byPathHop entries (#158)

After #158 the #115 fixture holds no duplicates, so a sweep that removed only
the first occurrence of an evicted transmission per bucket survived. Insert
the same transmission twice more into a resolved bucket (the shape of an index
built before #158) and require every occurrence to go and an emptied bucket
to be deleted. Behaviour is unchanged, so the test also passes on master.

Relates to #158

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VRiEwpZQZpYPxKTJRJAvc5
---
 cmd/server/evict_resolved_pathhop_115_test.go | 21 +++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/cmd/server/evict_resolved_pathhop_115_test.go b/cmd/server/evict_resolved_pathhop_115_test.go
index 47f1e8478..e04b41d4a 100644
--- a/cmd/server/evict_resolved_pathhop_115_test.go
+++ b/cmd/server/evict_resolved_pathhop_115_test.go
@@ -143,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) {