From e7d1fe1de854a064911f6f734ca28d0b1d0a7c01 Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 06:56:39 +0000
Subject: [PATCH 1/7] test(store): reproduce traffic share differing between
 live ingest and load (#158)

After #162 the sum of traffic_share_score on staging still rose with
uptime (14.3 at 2.6 h, 20.7 at 18 h). Load indexes the resolved_path the
ingestor persisted, which leaves an ambiguous 1-byte hop null; live
ingest indexes the server's own resolution, which always picks a
candidate and can pick a different one per observer.

The fixture has two repeaters sharing prefix E5 and an observer near
each. On baa68cdd live ingest of the same rows gives each E5 node a 0.5
share (0 after load), doubles the full-key sum and indexes a 2-hop
transmission under 3 relays. A NULL resolved_path is also indexed
differently. Adds BenchmarkIngestNewFromDB_158.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBJURW9dzgaWhWyyp4f9qw
---
 cmd/server/sharegrowth_158_test.go | 272 +++++++++++++++++++++++++++++
 1 file changed, 272 insertions(+)
 create mode 100644 cmd/server/sharegrowth_158_test.go

diff --git a/cmd/server/sharegrowth_158_test.go b/cmd/server/sharegrowth_158_test.go
new file mode 100644
index 000000000..d2de3a9d9
--- /dev/null
+++ b/cmd/server/sharegrowth_158_test.go
@@ -0,0 +1,272 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"math"
+	"strings"
+	"testing"
+	"time"
+)
+
+// Issue #158 follow-up: after #162 the traffic share no longer grew from
+// duplicate byPathHop entries, but its sum still rose with uptime on
+// staging (14.3 at 2.6 h, 20.7 at 18 h). Load and the background chunk load
+// index the resolved_path the ingestor persisted (#1547), which leaves a hop
+// unresolved when its prefix is ambiguous (#1560). Live ingest
+// re-resolved every observation with the server's resolver instead, which
+// always picks a candidate (affinity, geo proximity or observation count)
+// and can pick a different one per observer. So the same transmissions
+// counted for more relays when they arrived live than after a restart, and
+// the sum climbed as live traffic replaced the loaded part of the store.
+
+// Two repeaters share the 1-byte prefix E5, one in Jutland and one on
+// Zealand; the ingestor cannot tell them apart and persists null for that
+// hop. A1 and B2 are unique. The two observers are repeaters near each E5
+// node, so the server's geo-proximity tier resolves E5 differently per
+// observer.
+var (
+	shareA1  = relays158[0]
+	shareB2  = relays158[1]
+	shareE5W = "e5aa" + strings.Repeat("55", 30)
+	shareE5E = "e5bb" + strings.Repeat("66", 30)
+	shareObW = "f1" + strings.Repeat("77", 31)
+	shareObE = "f2" + strings.Repeat("88", 31)
+)
+
+const shareTxs = 20 // even: through A1,E5; odd: through B2 only
+
+func seedShareGrowth(t testing.TB, db *DB) {
+	t.Helper()
+	ts := time.Now().UTC().Add(-time.Hour).Format(time.RFC3339)
+	for _, n := range []struct {
+		pk       string
+		lat, lon any
+	}{
+		{shareA1, nil, nil},
+		{shareB2, nil, nil},
+		{shareE5W, 56.15, 9.55},
+		{shareE5E, 55.65, 12.10},
+		{shareObW, 56.10, 9.50},
+		{shareObE, 55.60, 12.20},
+	} {
+		exec158(t, db, `INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) VALUES (?, ?, 'repeater', ?, ?, ?, '2026-01-01', 1)`,
+			n.pk, "Share-"+n.pk[:4], n.lat, n.lon, ts)
+	}
+	for i, pk := range []string{shareObW, shareObE} {
+		exec158(t, db, `INSERT INTO observers (rowid, id, name, iata) VALUES (?, ?, ?, 'TST')`, i+1, pk, "Obs-"+pk[:4])
+	}
+}
+
+// insertShareObs inserts observation obsIdx (1 or 2) of transmission i the
+// way the ingestor writes it: path_json and, in the same row, its own
+// resolved_path, with null for the ambiguous E5 hop.
+func insertShareObs(t testing.TB, db *DB, i, obsIdx int, ts time.Time) {
+	t.Helper()
+	path, rp := `["B2"]`, `["`+shareB2+`"]`
+	if i%2 == 0 {
+		path, rp = `["A1","E5"]`, `["`+shareA1+`",null]`
+	}
+	exec158(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path)
+		VALUES (?, ?, 5, -90, ?, ?, ?)`, i+1, obsIdx, path, ts.Unix(), rp)
+}
+
+func insertShareTx(t testing.TB, db *DB, i int, ts time.Time) {
+	t.Helper()
+	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"}')`, i+1, fmt.Sprintf("share158-%03d", i), ts.Format(time.RFC3339))
+}
+
+var shareKeys = []string{shareA1, shareB2, shareE5W, shareE5E}
+
+func shareSnapshot(store *PacketStore) (map[string]float64, float64) {
+	bulk := store.computeRepeaterUsefulnessScoreMap()
+	out := make(map[string]float64, len(shareKeys))
+	sum := 0.0
+	for _, pk := range shareKeys {
+		out[pk] = bulk[pk]
+	}
+	for key, v := range bulk {
+		if len(key) == 64 { // full pubkeys only, as /api/nodes reads them
+			sum += v
+		}
+	}
+	return out, sum
+}
+
+// The traffic share of a fixed set of transmissions must not depend on
+// whether they were loaded at startup or ingested live: otherwise the sum
+// drifts with uptime as the retention window turns over.
+func TestTrafficShareLiveMatchesLoad_AmbiguousPrefix_158(t *testing.T) {
+	base := time.Now().UTC().Add(-30 * time.Minute)
+
+	// Restart: everything comes in through Load.
+	loadDB := setupTestDB(t)
+	defer loadDB.conn.Close()
+	seedShareGrowth(t, loadDB)
+	for i := 0; i < shareTxs; i++ {
+		insertShareTx(t, loadDB, i, base.Add(time.Duration(i)*time.Second))
+		insertShareObs(t, loadDB, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+		insertShareObs(t, loadDB, i, 2, base.Add(time.Duration(i*10+2)*time.Second))
+	}
+	loaded := loadedStore158(t, loadDB)
+	loadShares, loadSum := shareSnapshot(loaded)
+	if math.Abs(loadShares[shareA1]-0.5) > 1e-9 || math.Abs(loadShares[shareB2]-0.5) > 1e-9 {
+		t.Fatalf("fixture: after load A1=%v B2=%v, want 0.5 each", loadShares[shareA1], loadShares[shareB2])
+	}
+
+	cases := []struct {
+		name   string
+		ingest func(t *testing.T, db *DB, store *PacketStore)
+	}{
+		{"live batch", func(t *testing.T, db *DB, store *PacketStore) {
+			for i := 0; i < shareTxs; i++ {
+				insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+				insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+				insertShareObs(t, db, i, 2, base.Add(time.Duration(i*10+2)*time.Second))
+			}
+			store.IngestNewFromDB(0, 1000)
+		}},
+		{"late observation", func(t *testing.T, db *DB, store *PacketStore) {
+			for i := 0; i < shareTxs; i++ {
+				insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+				insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+			}
+			store.IngestNewFromDB(0, 1000)
+			since := maxObsID158(t, db)
+			for i := 0; i < shareTxs; i++ {
+				insertShareObs(t, db, i, 2, base.Add(time.Duration(i*10+2)*time.Second))
+			}
+			store.IngestNewObservations(since, 1000)
+		}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			db := setupTestDB(t)
+			defer db.conn.Close()
+			seedShareGrowth(t, db)
+			store := loadedStore158(t, db)
+			tc.ingest(t, db, store)
+
+			got, sum := shareSnapshot(store)
+			for _, pk := range shareKeys {
+				if math.Abs(got[pk]-loadShares[pk]) > 1e-9 {
+					t.Errorf("traffic share of %s… = %.3f live, %.3f after load of the same rows", pk[:6], got[pk], loadShares[pk])
+				}
+			}
+			if math.Abs(sum-loadSum) > 1e-9 {
+				t.Errorf("sum of traffic shares = %.3f live, %.3f after load of the same rows", sum, loadSum)
+			}
+			// One hop is one relay: no transmission may sit under more
+			// resolved full keys than it has distinct hops.
+			store.mu.RLock()
+			defer store.mu.RUnlock()
+			for i := 0; i < shareTxs; i += 2 {
+				tx := store.byTxID[i+1]
+				n := 0
+				for _, pk := range shareKeys {
+					if countTxInLocked(store.byPathHop[pk], tx) > 0 {
+						n++
+					}
+				}
+				if n > 2 {
+					t.Errorf("tx %d (2 hops) is indexed under %d relays", tx.ID, n)
+					break
+				}
+			}
+		})
+	}
+}
+
+func countTxInLocked(list []*StoreTx, tx *StoreTx) int {
+	n := 0
+	for _, t := range list {
+		if t == tx {
+			n++
+		}
+	}
+	return n
+}
+
+// An observation the ingestor could not resolve at all is stored with a
+// NULL resolved_path. Load then re-resolves path_json on unique prefixes
+// into byNode only (#1352, PR #1643); live ingest must index it the same
+// way rather than resolve it itself.
+func TestLiveIngestIndexesLikeLoad_NullResolvedPath_158(t *testing.T) {
+	base := time.Now().UTC().Add(-30 * time.Minute)
+	insert := func(t *testing.T, db *DB) {
+		insertShareTx(t, db, 0, base)
+		exec158(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path)
+			VALUES (1, 1, 5, -90, '["A1","E5"]', ?, NULL)`, base.Unix())
+	}
+	type view struct{ pathHopA1, pathHopE5, byNodeA1, byNodeE5 bool }
+	snapshot := func(store *PacketStore) view {
+		store.mu.RLock()
+		defer store.mu.RUnlock()
+		tx := store.byTxID[1]
+		if tx == nil {
+			t.Fatal("tx 1 not in store")
+		}
+		e5 := countTxInLocked(store.byPathHop[shareE5W], tx)+countTxInLocked(store.byPathHop[shareE5E], tx) > 0
+		e5n := countTxInLocked(store.byNode[shareE5W], tx)+countTxInLocked(store.byNode[shareE5E], tx) > 0
+		return view{
+			pathHopA1: countTxInLocked(store.byPathHop[shareA1], tx) > 0,
+			pathHopE5: e5,
+			byNodeA1:  countTxInLocked(store.byNode[shareA1], tx) > 0,
+			byNodeE5:  e5n,
+		}
+	}
+
+	loadDB := setupTestDB(t)
+	defer loadDB.conn.Close()
+	seedShareGrowth(t, loadDB)
+	insert(t, loadDB)
+	want := snapshot(loadedStore158(t, loadDB))
+	if want != (view{byNodeA1: true}) {
+		t.Fatalf("fixture: after load %+v, want only byNode under the unique A1", want)
+	}
+
+	liveDB := setupTestDB(t)
+	defer liveDB.conn.Close()
+	seedShareGrowth(t, liveDB)
+	live := loadedStore158(t, liveDB)
+	insert(t, liveDB)
+	live.IngestNewFromDB(0, 100)
+	if got := snapshot(live); got != want {
+		t.Errorf("live ingest indexed %+v, load of the same row %+v", got, want)
+	}
+}
+
+// BenchmarkIngestNewFromDB_158 drives the real live ingest (SQL scan +
+// index + broadcast build) of a batch of 20 new transmissions with 11
+// observations each, as the ingestor writes them (resolved_path set).
+func BenchmarkIngestNewFromDB_158(b *testing.B) {
+	prev := log.Writer()
+	log.SetOutput(io.Discard)
+	defer log.SetOutput(prev)
+	db := setupTestDB(b)
+	defer db.conn.Close()
+	seed158(b, db, observerCount158)
+	store := loadedStore158(b, db)
+	base := time.Now().UTC().Add(-30 * time.Minute)
+	nextID := 1
+	b.ReportAllocs()
+	b.ResetTimer()
+	for it := 0; it < b.N; it++ {
+		b.StopTimer()
+		since := nextID - 1
+		for i := range paths158 {
+			ts := base.Add(time.Duration(nextID) * time.Millisecond)
+			exec158(b, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json)
+				VALUES (?, 'CAFE', ?, ?, 1, 2, '{"type":"TXT_MSG"}')`, nextID, fmt.Sprintf("bench158-%07d", nextID), ts.Format(time.RFC3339))
+			for o := 1; o <= observerCount158; o++ {
+				exec158(b, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path)
+					VALUES (?, ?, 5, -90, ?, ?, ?)`, nextID, o, rawHopsJSON158(paths158[i]), ts.Unix()+int64(o), resolvedJSON158(paths158[i]))
+			}
+			nextID++
+		}
+		b.StartTimer()
+		store.IngestNewFromDB(since, 1000)
+	}
+}

From 9e3c5192bc1f49752e6757812e9cc6741474e180 Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 06:56:39 +0000
Subject: [PATCH 2/7] fix(store): index live observations from the persisted
 resolved_path, as Load does (#158)

IngestNewFromDB deliberately did not select resolved_path, on the premise
that Load reads back what live ingest computed (review item #7). That
held while the server wrote resolved_path itself. Since #1287/#1547 the
ingestor writes it, with its own conservative resolver (#1560), so a
transmission was indexed under more relays when it arrived live than
after a restart. traffic_share_score mixes both halves of the store, so
its sum rose with uptime until retention turned the store over.

indexObservationRelayHops is now the one place that feeds the relay-hop
indexes from an observation, used by Load, scanAndMergeChunk,
IngestNewFromDB and IngestNewObservations: the persisted resolved_path
goes to indexResolvedPathHops; a NULL one is re-resolved on unique
prefixes into byNode only, as Load already did. The live broadcast still
carries the server's own resolution.

The #158 fixture inserted live observations with a NULL resolved_path
("resolved fresh by the server"); it now writes resolved_path in the
same row, as the ingestor does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBJURW9dzgaWhWyyp4f9qw
---
 cmd/server/chunked_load.go            |  24 +-----
 cmd/server/pathhop_dedupe_158_test.go |  39 ++++-----
 cmd/server/store.go                   | 114 ++++++++++++++++----------
 3 files changed, 92 insertions(+), 85 deletions(-)

diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go
index 314481508..4003bb870 100644
--- a/cmd/server/chunked_load.go
+++ b/cmd/server/chunked_load.go
@@ -652,26 +652,10 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 				Timestamp: normalizeTimestamp(nullStrVal(obsTimestamp)),
 			}
 
-			rpStr := nullStrVal(resolvedPathStr)
-			if rpStr != "" {
-				rp := unmarshalResolvedPath(rpStr)
-				pks := extractResolvedPubkeys(rp)
-				s.indexResolvedPathHops(tx, pks, hopsSeen)
-			} else if relayPM != nil && obsPJ != "" && obsPJ != "[]" {
-				// resolved_path is NULL on live (since #1287 relay data is
-				// persisted as neighbor_edges, not per-observation). Re-resolve
-				// relay-hop attribution from path_json so relay nodes keep their
-				// analytics history across a restart instead of rebuilding only
-				// from post-restart live traffic. relayPM is passed in from
-				// LoadChunked (fetched before any chunk cursor opened).
-				// byNode ONLY — see the Load() counterpart for why the
-				// resolved_path/path-hop indexes must NOT be populated here.
-				// PR #1643 R1 munger #1: unique_prefix-only gate.
-				rp := resolvePathForObsColdLoad(obsPJ, obsIDStr, tx, relayPM, coldLoadAmbiguousHopsSkipped)
-				for _, pk := range extractResolvedPubkeys(rp) {
-					s.addToByNode(tx, pk)
-				}
-			}
+			// Same relay-hop indexing as Load and live ingest (see
+			// indexObservationRelayHops). relayPM is passed in from
+			// LoadChunked (fetched before any chunk cursor opened).
+			s.indexObservationRelayHops(tx, nullStrVal(resolvedPathStr), obsPJ, obsIDStr, relayPM, hopsSeen, coldLoadAmbiguousHopsSkipped)
 
 			tx.mergeObservedPathHashSize(obsPJ)
 			tx.Observations = append(tx.Observations, obs)
diff --git a/cmd/server/pathhop_dedupe_158_test.go b/cmd/server/pathhop_dedupe_158_test.go
index 625afa7a3..f2bda2328 100644
--- a/cmd/server/pathhop_dedupe_158_test.go
+++ b/cmd/server/pathhop_dedupe_158_test.go
@@ -91,16 +91,13 @@ func insertTx158(t testing.TB, db *DB, i int, firstSeen time.Time) int {
 }
 
 // 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) {
+// heard by observer obsIdx, the way the ingestor writes it: with its
+// resolved_path in the same row (#1547). Load and live ingest both index
+// that resolved_path (indexObservationRelayHops).
+func insertObs158(t testing.TB, db *DB, i, obsIdx int, ts time.Time) {
 	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)
+		VALUES (?, ?, 5, -90, ?, ?, ?)`, i+1, obsIdx, rawHopsJSON158(paths158[i]), ts.Unix(), resolvedJSON158(paths158[i]))
 }
 
 func maxObsID158(t testing.TB, db *DB) int {
@@ -162,7 +159,7 @@ func TestPathHopIndexOncePerTx_LateObservations_158(t *testing.T) {
 
 	now := time.Now().UTC().Add(-10 * time.Minute)
 	insertTx158(t, db, 0, now)
-	insertObs158(t, db, 0, 1, now, false)
+	insertObs158(t, db, 0, 1, now)
 	store.IngestNewFromDB(0, 100)
 	for _, r := range paths158[0] {
 		if n := countTxIn158(store, relays158[r], store.byTxID[1]); n != 1 {
@@ -172,7 +169,7 @@ func TestPathHopIndexOncePerTx_LateObservations_158(t *testing.T) {
 
 	for k := 1; k <= lateObs158; k++ {
 		since := maxObsID158(t, db)
-		insertObs158(t, db, 0, 1+k, now.Add(time.Duration(k)*time.Second), false)
+		insertObs158(t, db, 0, 1+k, now.Add(time.Duration(k)*time.Second))
 		store.IngestNewObservations(since, 100)
 	}
 	assertOncePerRelay158(t, store, 0)
@@ -189,7 +186,7 @@ func TestPathHopIndexOncePerTx_LiveIngestBatch_158(t *testing.T) {
 	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)
+		insertObs158(t, db, 0, o, now.Add(time.Duration(o)*time.Second))
 	}
 	store.IngestNewFromDB(0, 100)
 	assertOncePerRelay158(t, store, 0)
@@ -255,7 +252,7 @@ func TestTrafficShareStableAcrossLateObservations_158(t *testing.T) {
 	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)
+			insertObs158(t, loadDB, i, o, base.Add(time.Duration(i*100+o)*time.Second))
 		}
 	}
 	loaded := loadedStore158(t, loadDB)
@@ -270,7 +267,7 @@ func TestTrafficShareStableAcrossLateObservations_158(t *testing.T) {
 	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)
+		insertObs158(t, liveDB, i, 1, base.Add(time.Duration(i*100+1)*time.Second))
 	}
 	live.IngestNewFromDB(0, 1000)
 	afterIngest := shares158(t, live)
@@ -281,7 +278,7 @@ func TestTrafficShareStableAcrossLateObservations_158(t *testing.T) {
 		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)
+				insertObs158(t, liveDB, i, o, base.Add(time.Duration(i*100+o)*time.Second))
 			}
 		}
 		live.IngestNewObservations(since, 10000)
@@ -310,7 +307,7 @@ func TestPathHopIndexSizeBoundedByTransmissions_158(t *testing.T) {
 	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)
+		insertObs158(t, db, i, 1, base.Add(time.Duration(i*100+1)*time.Second))
 	}
 	store.IngestNewFromDB(0, 1000)
 	entries := func() int {
@@ -326,7 +323,7 @@ func TestPathHopIndexSizeBoundedByTransmissions_158(t *testing.T) {
 	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)
+			insertObs158(t, db, i, o, base.Add(time.Duration(i*100+o)*time.Second))
 		}
 		store.IngestNewObservations(since, 10000)
 		if got := entries(); got != want {
@@ -432,11 +429,11 @@ func TestPathHopIndexOncePerTx_AfterLoad_158(t *testing.T) {
 	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)
+		insertObs158(t, db, 0, o, base.Add(time.Duration(o)*time.Second))
 	}
 	store := loadedStore158(t, db)
 	since := maxObsID158(t, db)
-	insertObs158(t, db, 0, observerCount158, base.Add(time.Minute), false)
+	insertObs158(t, db, 0, observerCount158, base.Add(time.Minute))
 	store.IngestNewObservations(since, 100)
 	assertOncePerRelay158(t, store, 0)
 }
@@ -497,7 +494,7 @@ func TestPathHopConsumersIgnoreDuplicateEntries_158(t *testing.T) {
 	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)
+			insertObs158(t, db, i, o, base.Add(time.Duration(i*100+o)*time.Second))
 		}
 	}
 	store := loadedStore158(t, db)
@@ -637,7 +634,7 @@ func BenchmarkIngestNewObservations_158(b *testing.B) {
 	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)
+		insertObs158(b, db, i, 1, base.Add(time.Duration(i*1000+1)*time.Second))
 	}
 	store := loadedStore158(b, db)
 	store.IngestNewFromDB(0, 1000)
@@ -648,7 +645,7 @@ func BenchmarkIngestNewObservations_158(b *testing.B) {
 		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)
+			insertObs158(b, db, i, o, base.Add(time.Duration(i*1000+o)*time.Second))
 		}
 		b.StartTimer()
 		store.IngestNewObservations(since, 10000)
diff --git a/cmd/server/store.go b/cmd/server/store.go
index 5fcd95244..8f0f8aaef 100644
--- a/cmd/server/store.go
+++ b/cmd/server/store.go
@@ -1037,29 +1037,8 @@ func (s *PacketStore) Load() error {
 				Timestamp:      normalizeTimestamp(nullStrVal(obsTimestamp)),
 			}
 
-			// Decode-window: extract resolved pubkeys for index, don't store on struct.
-			rpStr := nullStrVal(resolvedPathStr)
-			if rpStr != "" {
-				rp := unmarshalResolvedPath(rpStr)
-				pks := extractResolvedPubkeys(rp)
-				// Single point of truth — see indexResolvedPathHops doc + #1558.
-				s.indexResolvedPathHops(tx, pks, hopsSeen)
-			} else if relayPM != nil && obsPJ != "" && obsPJ != "[]" {
-				// resolved_path not persisted — reconstruct relay hops from
-				// path_json so relay-node analytics history survives a restart.
-				// Index into byNode ONLY: the resolved_path / path-hop indexes
-				// (indexResolvedPathHops) are cross-checked by handleNodePaths
-				// against the persisted resolved_path column, which is NULL
-				// here — populating them would make that SQL confirmation fail
-				// and wrongly drop the tx from paths-through (#1352). byNode is
-				// what the node-analytics activity timeline reads.
-				// PR #1643 R1 munger #1: gate on unique_prefix only — ambiguous
-				// hops are silently dropped (skipped counter logged at end).
-				rp := resolvePathForObsColdLoad(obsPJ, obsIDStr, tx, relayPM, &coldLoadAmbiguousHopsSkipped)
-				for _, pk := range extractResolvedPubkeys(rp) {
-					s.addToByNode(tx, pk)
-				}
-			}
+			// Decode-window: feed the relay-hop indexes, don't store on struct.
+			s.indexObservationRelayHops(tx, nullStrVal(resolvedPathStr), obsPJ, obsIDStr, relayPM, hopsSeen, &coldLoadAmbiguousHopsSkipped)
 
 			tx.mergeObservedPathHashSize(obsPJ)
 			tx.Observations = append(tx.Observations, obs)
@@ -1894,6 +1873,35 @@ func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen
 	s.addToResolvedPubkeyIndex(tx.ID, pks)
 }
 
+// indexObservationRelayHops feeds the relay-hop indexes from one stored
+// observation. Every path that materializes an observation (Load,
+// scanAndMergeChunk, IngestNewFromDB, IngestNewObservations) goes through
+// it, so a transmission is indexed under the same relays whether it was
+// loaded at startup or ingested live. The ingestor persists resolved_path
+// in the same row it inserts (#1547) and leaves a hop null when the prefix
+// is ambiguous (#1560); those pubkeys go to indexResolvedPathHops. Without
+// a persisted resolved_path, path_json is re-resolved on unique prefixes
+// only and fed to byNode only (#1352, PR #1643).
+//
+// Live ingest used to index the server's own resolution instead, which
+// always picks a candidate and can pick a different one per observer, so
+// traffic share counted live transmissions for more relays than loaded
+// ones and its sum rose with uptime (#158 follow-up).
+//
+// skipped counts ambiguous cold-load hops (may be nil). Must hold s.mu.
+func (s *PacketStore) indexObservationRelayHops(tx *StoreTx, resolvedPath, pathJSON, observerID string, relayPM *prefixMap, hopsSeen map[string]bool, skipped *int) {
+	if resolvedPath != "" {
+		s.indexResolvedPathHops(tx, extractResolvedPubkeys(unmarshalResolvedPath(resolvedPath)), hopsSeen)
+		return
+	}
+	if relayPM == nil || pathJSON == "" || pathJSON == "[]" {
+		return
+	}
+	for _, pk := range extractResolvedPubkeys(resolvePathForObsColdLoad(pathJSON, observerID, tx, relayPM, skipped)) {
+		s.addToByNode(tx, pk)
+	}
+}
+
 // indexByNode extracts pubkeys from decoded_json and indexes the transmission.
 // indexByNode indexes a transmission under all pubkeys found in its decoded
 // JSON. Resolved path pubkeys are handled separately via the decode-window.
@@ -2803,14 +2811,19 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		limit = 100
 	}
 
-	// NOTE: The SQL query intentionally does NOT select resolved_path from the DB.
-	// New ingests always resolve fresh using the current prefix map and neighbor graph.
-	// On restart, Load() handles reading persisted resolved_path values. (review item #7)
+	// resolved_path is selected so the relay-hop indexes are fed from the
+	// ingestor's persisted resolution, exactly as Load feeds them
+	// (indexObservationRelayHops, #158 follow-up). The server still resolves
+	// each path itself for the live broadcast.
 	var querySQL string
 	obsRHCol := ""
 	if s.db.hasObsRawHex() {
 		obsRHCol = ", o.raw_hex"
 	}
+	rpCol := ""
+	if s.db.hasResolvedPath() {
+		rpCol = ", o.resolved_path"
+	}
 	// #1751: scope_name is on the transmission row; append as the last column.
 	scopeNameCol := ""
 	if s.db.hasScopeName() {
@@ -2825,7 +2838,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		querySQL = `SELECT t.id, t.raw_hex, t.hash, t.first_seen, t.route_type,
 				t.payload_type, t.payload_version, t.decoded_json,
 				o.id, obs.id, obs.name, COALESCE(obs.iata, ''), o.direction,
-				o.snr, o.rssi, o.score, o.path_json, strftime('%Y-%m-%dT%H:%M:%fZ', o.timestamp, 'unixepoch')` + obsRHCol + scopeNameCol + routeMaskCol + `
+				o.snr, o.rssi, o.score, o.path_json, strftime('%Y-%m-%dT%H:%M:%fZ', o.timestamp, 'unixepoch')` + obsRHCol + rpCol + scopeNameCol + routeMaskCol + `
 			FROM transmissions t
 			LEFT JOIN observations o ON o.transmission_id = t.id
 			LEFT JOIN observers obs ON obs.rowid = o.observer_idx
@@ -2835,7 +2848,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		querySQL = `SELECT t.id, t.raw_hex, t.hash, t.first_seen, t.route_type,
 				t.payload_type, t.payload_version, t.decoded_json,
 				o.id, o.observer_id, o.observer_name, COALESCE(obs.iata, ''), o.direction,
-				o.snr, o.rssi, o.score, o.path_json, o.timestamp` + obsRHCol + scopeNameCol + routeMaskCol + `
+				o.snr, o.rssi, o.score, o.path_json, o.timestamp` + obsRHCol + rpCol + scopeNameCol + routeMaskCol + `
 			FROM transmissions t
 			LEFT JOIN observations o ON o.transmission_id = t.id
 			LEFT JOIN observers obs ON obs.id = o.observer_id
@@ -2858,6 +2871,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		obsID                                                              *int
 		observerID, observerName, observerIATA, direction, pathJSON, obsTS string
 		obsRawHex                                                          string
+		resolvedPath                                                       string
 		scopeName                                                          string
 		routeMask                                                          sql.NullInt64
 		snr, rssi                                                          *float64
@@ -2877,6 +2891,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		var snrVal, rssiVal sql.NullFloat64
 		var scoreVal sql.NullInt64
 		var obsRawHex sql.NullString
+		var resolvedPath sql.NullString
 		var scopeName sql.NullString
 		var routeMask sql.NullInt64
 
@@ -2887,6 +2902,9 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		if s.db.hasObsRawHex() {
 			scanArgs2 = append(scanArgs2, &obsRawHex)
 		}
+		if s.db.hasResolvedPath() {
+			scanArgs2 = append(scanArgs2, &resolvedPath)
+		}
 		if s.db.hasScopeName() {
 			scanArgs2 = append(scanArgs2, &scopeName)
 		}
@@ -2920,6 +2938,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 			pathJSON:     nullStrVal(pathJSON),
 			obsTS:        nullStrVal(obsTimestamp),
 			obsRawHex:    nullStrVal(obsRawHex),
+			resolvedPath: nullStrVal(resolvedPath),
 			scopeName:    nullStrVal(scopeName),
 			routeMask:    routeMask,
 			snr:          nullFloatPtr(snrVal),
@@ -3032,15 +3051,14 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 				Timestamp:      normalizeTimestamp(r.obsTS),
 			}
 
-			// Resolve path at ingest time using neighbor graph — decode-window discipline:
-			// decode once, feed consumers, never store on struct.
-			var resolvedPubkeys []string
+			// Relay-hop indexes: same input as Load (#158 follow-up).
+			// Decode-window discipline: decode once, feed consumers, never
+			// store on struct.
+			s.indexObservationRelayHops(tx, r.resolvedPath, r.pathJSON, r.observerID, cachedPM, hopsSeen, nil)
+			// The live broadcast still shows the server's own resolution.
 			var rpForBroadcast []*string
 			if r.pathJSON != "" && r.pathJSON != "[]" && cachedPM != nil {
 				rpForBroadcast = resolvePathForObs(r.pathJSON, r.observerID, tx, cachedPM, cachedGraph)
-				resolvedPubkeys = extractResolvedPubkeys(rpForBroadcast)
-				// Single point of truth — see indexResolvedPathHops doc + #1558.
-				s.indexResolvedPathHops(tx, resolvedPubkeys, hopsSeen)
 			}
 			// Stash rpForBroadcast for later broadcast/persist (keyed by obs ID)
 			if rpForBroadcast != nil {
@@ -3251,6 +3269,11 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 	if s.db.hasObsRawHex() {
 		obsRHCol2 = ", o.raw_hex"
 	}
+	// resolved_path feeds the relay-hop indexes as in Load (#158 follow-up).
+	rpCol2 := ""
+	if s.db.hasResolvedPath() {
+		rpCol2 = ", o.resolved_path"
+	}
 	// #89: the transmission's current route_mask rides along with each new
 	// observation so the live view converges with a cold load.
 	routeMaskCol2, routeMaskJoin2 := "", ""
@@ -3260,7 +3283,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 	}
 	if s.db.isV3() {
 		querySQL = `SELECT o.id, o.transmission_id, obs.id, obs.name, COALESCE(obs.iata, ''), o.direction,
-				o.snr, o.rssi, o.score, o.path_json, strftime('%Y-%m-%dT%H:%M:%fZ', o.timestamp, 'unixepoch')` + obsRHCol2 + routeMaskCol2 + `
+				o.snr, o.rssi, o.score, o.path_json, strftime('%Y-%m-%dT%H:%M:%fZ', o.timestamp, 'unixepoch')` + obsRHCol2 + rpCol2 + routeMaskCol2 + `
 			FROM observations o
 			LEFT JOIN observers obs ON obs.rowid = o.observer_idx` + routeMaskJoin2 + `
 			WHERE o.id > ?
@@ -3268,7 +3291,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 			LIMIT ?`
 	} else {
 		querySQL = `SELECT o.id, o.transmission_id, o.observer_id, o.observer_name, COALESCE(obs.iata, ''), o.direction,
-				o.snr, o.rssi, o.score, o.path_json, o.timestamp` + obsRHCol2 + routeMaskCol2 + `
+				o.snr, o.rssi, o.score, o.path_json, o.timestamp` + obsRHCol2 + rpCol2 + routeMaskCol2 + `
 			FROM observations o
 			LEFT JOIN observers obs ON obs.id = o.observer_id` + routeMaskJoin2 + `
 			WHERE o.id > ?
@@ -3294,6 +3317,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		score        *int
 		pathJSON     string
 		rawHex       string
+		resolvedPath string
 		timestamp    string
 		routeMask    sql.NullInt64
 	}
@@ -3305,6 +3329,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		var snr, rssi sql.NullFloat64
 		var score sql.NullInt64
 		var obsRawHex sql.NullString
+		var resolvedPath sql.NullString
 		var routeMask sql.NullInt64
 
 		scanArgs3 := []interface{}{&oid, &txID, &observerID, &observerName, &observerIATA, &direction,
@@ -3312,6 +3337,9 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		if s.db.hasObsRawHex() {
 			scanArgs3 = append(scanArgs3, &obsRawHex)
 		}
+		if s.db.hasResolvedPath() {
+			scanArgs3 = append(scanArgs3, &resolvedPath)
+		}
 		if s.db.hasRouteMask() {
 			scanArgs3 = append(scanArgs3, &routeMask)
 		}
@@ -3331,6 +3359,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 			score:        nullIntPtr(score),
 			pathJSON:     nullStrVal(pathJSON),
 			rawHex:       nullStrVal(obsRawHex),
+			resolvedPath: nullStrVal(resolvedPath),
 			timestamp:    nullStrVal(ts),
 			routeMask:    routeMask,
 		})
@@ -3395,16 +3424,13 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 			Timestamp: normalizeTimestamp(r.timestamp),
 		}
 
-		// Resolve path at ingest time for late-arriving observations (review item #2).
+		// Relay-hop indexes: same input as Load (#158 follow-up).
 		// Decode-window discipline: decode, feed consumers, don't store on struct.
+		s.indexObservationRelayHops(tx, r.resolvedPath, r.pathJSON, r.observerID, pm, hopsSeen, nil)
+		// The live broadcast still shows the server's own resolution (review item #2).
 		var obsResolvedPath []*string
-		if r.pathJSON != "" && r.pathJSON != "[]" {
-			if pm != nil {
-				obsResolvedPath = resolvePathForObs(r.pathJSON, r.observerID, tx, pm, graphRef)
-				pks := extractResolvedPubkeys(obsResolvedPath)
-				// Single point of truth — see indexResolvedPathHops doc + #1558.
-				s.indexResolvedPathHops(tx, pks, hopsSeen)
-			}
+		if r.pathJSON != "" && r.pathJSON != "[]" && pm != nil {
+			obsResolvedPath = resolvePathForObs(r.pathJSON, r.observerID, tx, pm, graphRef)
 		}
 		// Stash for broadcast/persist
 		if obsResolvedPath != nil {

From f6321b89f3162e41c84eb17451825041b3d3937b Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 08:12:23 +0000
Subject: [PATCH 3/7] test(store): pin late-observation relay indexing from
 resolved_path (#158)

Review of the live-ingest fix found that a mutant of IngestNewObservations
which ignores the persisted resolved_path (indexing "" instead) survived:
the late observation in the existing late-observation case reuses the
first observation's path, so there is nothing new to index.

- TestTrafficShareLiveMatchesLoad_LateObserverNewRelay_158: a late
  observation from another observer whose resolved_path names a relay
  the first observation did not have (B2 for even transmissions, A1 for
  odd ones). Live byPathHop and traffic share must equal a Load of the
  same rows. Red under the mutant (share 0.5 vs 1.0, sum 1.0 vs 2.0,
  10 vs 20 transmissions per bucket).
- TestLateObservationIndexesLikeLoad_NullResolvedPath_158: the NULL
  resolved_path case of TestLiveIngestIndexesLikeLoad_NullResolvedPath_158
  for IngestNewObservations: byNode only, never byPathHop.

Relates to #158

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

diff --git a/cmd/server/sharegrowth_158_test.go b/cmd/server/sharegrowth_158_test.go
index d2de3a9d9..8d3e4d7ac 100644
--- a/cmd/server/sharegrowth_158_test.go
+++ b/cmd/server/sharegrowth_158_test.go
@@ -238,6 +238,140 @@ func TestLiveIngestIndexesLikeLoad_NullResolvedPath_158(t *testing.T) {
 	}
 }
 
+// insertShareObsPath inserts one observation with an explicit path and
+// persisted resolved_path (rp == "" writes NULL), the shape of an
+// observation another observer reports later through different relays.
+func insertShareObsPath(t testing.TB, db *DB, i, obsIdx int, path, rp string, ts time.Time) {
+	t.Helper()
+	var rpArg any
+	if rp != "" {
+		rpArg = rp
+	}
+	exec158(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path)
+		VALUES (?, ?, 5, -90, ?, ?, ?)`, i+1, obsIdx, path, ts.Unix(), rpArg)
+}
+
+// A late observation from ANOTHER observer can carry a persisted
+// resolved_path with relays the first observation did not have. The first
+// observation of the late-observation case above reuses the same path, so
+// there is nothing new to index; here the late observation brings a relay
+// (B2 for even transmissions, A1 for odd ones) that only its resolved_path
+// names. IngestNewObservations must index it exactly as Load does, or live
+// traffic share drifts from the loaded one (mutant: ignore resolved_path
+// and index "" instead).
+func TestTrafficShareLiveMatchesLoad_LateObserverNewRelay_158(t *testing.T) {
+	base := time.Now().UTC().Add(-30 * time.Minute)
+	first := func(t *testing.T, db *DB, i int) {
+		insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+	}
+	late := func(t *testing.T, db *DB, i int) {
+		// Observer 2 heard it through A1 and B2: whichever relay the first
+		// observation lacked is new here.
+		insertShareObsPath(t, db, i, 2, `["A1","B2"]`, `["`+shareA1+`","`+shareB2+`"]`, base.Add(time.Duration(i*10+2)*time.Second))
+	}
+
+	loadDB := setupTestDB(t)
+	defer loadDB.conn.Close()
+	seedShareGrowth(t, loadDB)
+	for i := 0; i < shareTxs; i++ {
+		insertShareTx(t, loadDB, i, base.Add(time.Duration(i)*time.Second))
+		first(t, loadDB, i)
+		late(t, loadDB, i)
+	}
+	loaded := loadedStore158(t, loadDB)
+	loadShares, loadSum := shareSnapshot(loaded)
+	if math.Abs(loadShares[shareA1]-1.0) > 1e-9 || math.Abs(loadShares[shareB2]-1.0) > 1e-9 {
+		t.Fatalf("fixture: after load A1=%v B2=%v, want 1.0 each (every tx passes both through some observation)", loadShares[shareA1], loadShares[shareB2])
+	}
+
+	db := setupTestDB(t)
+	defer db.conn.Close()
+	seedShareGrowth(t, db)
+	store := loadedStore158(t, db)
+	for i := 0; i < shareTxs; i++ {
+		insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+		first(t, db, i)
+	}
+	store.IngestNewFromDB(0, 1000)
+	since := maxObsID158(t, db)
+	for i := 0; i < shareTxs; i++ {
+		late(t, db, i)
+	}
+	store.IngestNewObservations(since, 1000)
+
+	got, sum := shareSnapshot(store)
+	for _, pk := range shareKeys {
+		if math.Abs(got[pk]-loadShares[pk]) > 1e-9 {
+			t.Errorf("traffic share of %s… = %.3f live, %.3f after load of the same rows", pk[:6], got[pk], loadShares[pk])
+		}
+	}
+	if math.Abs(sum-loadSum) > 1e-9 {
+		t.Errorf("sum of traffic shares = %.3f live, %.3f after load of the same rows", sum, loadSum)
+	}
+	store.mu.RLock()
+	defer store.mu.RUnlock()
+	for _, pk := range shareKeys {
+		if g, w := len(store.byPathHop[pk]), len(loaded.byPathHop[pk]); g != w {
+			t.Errorf("byPathHop[%s…] holds %d transmissions live, %d after load", pk[:6], g, w)
+		}
+	}
+}
+
+// The same NULL case as TestLiveIngestIndexesLikeLoad_NullResolvedPath_158,
+// for a late observation of a transmission already in the store: an
+// observation the ingestor could not resolve must be re-resolved on unique
+// prefixes into byNode only, never into byPathHop, exactly as Load does.
+func TestLateObservationIndexesLikeLoad_NullResolvedPath_158(t *testing.T) {
+	base := time.Now().UTC().Add(-30 * time.Minute)
+	type view struct{ pathHopA1, pathHopB2, byNodeA1, byNodeB2 bool }
+	snapshot := func(store *PacketStore) view {
+		store.mu.RLock()
+		defer store.mu.RUnlock()
+		tx := store.byTxID[1]
+		if tx == nil {
+			t.Fatal("tx 1 not in store")
+		}
+		return view{
+			pathHopA1: countTxInLocked(store.byPathHop[shareA1], tx) > 0,
+			pathHopB2: countTxInLocked(store.byPathHop[shareB2], tx) > 0,
+			byNodeA1:  countTxInLocked(store.byNode[shareA1], tx) > 0,
+			byNodeB2:  countTxInLocked(store.byNode[shareB2], tx) > 0,
+		}
+	}
+	// First observation: resolved by the ingestor (A1). Late observation
+	// from the other observer: NULL, path through the unique B2.
+	insertFirst := func(t *testing.T, db *DB) {
+		insertShareTx(t, db, 0, base)
+		insertShareObsPath(t, db, 0, 1, `["A1"]`, `["`+shareA1+`"]`, base.Add(time.Second))
+	}
+	insertLate := func(t *testing.T, db *DB) {
+		insertShareObsPath(t, db, 0, 2, `["B2"]`, "", base.Add(2*time.Second))
+	}
+
+	loadDB := setupTestDB(t)
+	defer loadDB.conn.Close()
+	seedShareGrowth(t, loadDB)
+	insertFirst(t, loadDB)
+	insertLate(t, loadDB)
+	want := snapshot(loadedStore158(t, loadDB))
+	if want != (view{pathHopA1: true, byNodeA1: true, byNodeB2: true}) {
+		t.Fatalf("fixture: after load %+v, want byPathHop only under A1 and byNode under A1 and B2", want)
+	}
+
+	liveDB := setupTestDB(t)
+	defer liveDB.conn.Close()
+	seedShareGrowth(t, liveDB)
+	live := loadedStore158(t, liveDB)
+	insertFirst(t, liveDB)
+	live.IngestNewFromDB(0, 100)
+	since := maxObsID158(t, liveDB)
+	insertLate(t, liveDB)
+	live.IngestNewObservations(since, 100)
+	if got := snapshot(live); got != want {
+		t.Errorf("late observation indexed %+v, load of the same rows %+v", got, want)
+	}
+}
+
 // BenchmarkIngestNewFromDB_158 drives the real live ingest (SQL scan +
 // index + broadcast build) of a batch of 20 new transmissions with 11
 // observations each, as the ingestor writes them (resolved_path set).

From 9ac2b9b06cc12324fef86e8dc26ffb601b51bf5f Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 08:13:59 +0000
Subject: [PATCH 4/7] fix(store): one schema-flag snapshot per ingest call for
 query and Scan (#158)

IngestNewFromDB and IngestNewObservations read hasObsRawHex,
hasResolvedPath, hasScopeName and hasRouteMask once to build the SELECT
and again to build the Scan destinations. The flags are atomics the
schema healer (or main.go's forceTrue) can latch at any time, so a flip
in between gave Scan the wrong destination count; the error is swallowed
and the row was silently skipped. The #158 follow-up added resolved_path
to both queries and so widened that window.

DB.ingestCols takes one snapshot per call, used for both. The test flips
each flag between query and scan through a test-only hook
(PacketStore.ingestAfterQueryHook), in both directions, in both
functions: 14 subcases, all red before this change (rows skipped), all
green after.

Load and LoadChunked read their flags twice in the same way; they are
not changed here (startup schema detection is #184).

Relates to #158

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017nop1P1pX4SrfRMHAernzQ
---
 cmd/server/db.go                       |  20 ++++
 cmd/server/schema_snapshot_158_test.go | 127 +++++++++++++++++++++++++
 cmd/server/store.go                    |  43 ++++++---
 3 files changed, 176 insertions(+), 14 deletions(-)
 create mode 100644 cmd/server/schema_snapshot_158_test.go

diff --git a/cmd/server/db.go b/cmd/server/db.go
index 5485fae9c..2ce71b41f 100644
--- a/cmd/server/db.go
+++ b/cmd/server/db.go
@@ -152,6 +152,26 @@ func (db *DB) hasMultibyteSupCols() bool        { return db.hasMultibyteSupColsF
 func (db *DB) hasLastSeen() bool                { return db.hasLastSeenFlag.get() }
 func (db *DB) hasRouteMask() bool               { return db.hasRouteMaskFlag.get() }
 
+// ingestCols is one reading of the optional-column flags an ingest query
+// depends on. A caller takes ONE snapshot and uses it for both the SELECT
+// list and the Scan destinations: the flags are atomics that the schema
+// healer (or main.go's forceTrue) may latch between two reads, and a query
+// built with one answer scanned with the other fails Scan with the wrong
+// destination count -- an error the ingest loops swallow, silently dropping
+// the row (#158 follow-up).
+type ingestCols struct {
+	obsRawHex, resolvedPath, scopeName, routeMask bool
+}
+
+func (db *DB) ingestCols() ingestCols {
+	return ingestCols{
+		obsRawHex:    db.hasObsRawHex(),
+		resolvedPath: db.hasResolvedPath(),
+		scopeName:    db.hasScopeName(),
+		routeMask:    db.hasRouteMask(),
+	}
+}
+
 // OpenDB opens a read-only SQLite connection with WAL mode.
 func OpenDB(path string) (*DB, error) {
 	dsn := fmt.Sprintf("file:%s?mode=ro&_journal_mode=WAL&_busy_timeout=5000", path)
diff --git a/cmd/server/schema_snapshot_158_test.go b/cmd/server/schema_snapshot_158_test.go
new file mode 100644
index 000000000..a3e4ee6d3
--- /dev/null
+++ b/cmd/server/schema_snapshot_158_test.go
@@ -0,0 +1,127 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+// IngestNewFromDB and IngestNewObservations build their SELECT from the
+// optional-column schema flags (raw_hex, resolved_path, scope_name,
+// route_mask) and then append one Scan destination per flag. The flags are
+// atomics that the background healer (and main.go's forceTrue) can flip at
+// any time, so reading them once for the query and again for the Scan can
+// disagree: Scan gets the wrong number of destinations, the error is
+// swallowed and the row is silently skipped. One snapshot per call keeps
+// query and Scan consistent.
+//
+// The hook flips the flag after the query ran and before the rows are
+// scanned, which is the window the bug lives in.
+
+func TestIngestSchemaFlagFlipBetweenQueryAndScan_158(t *testing.T) {
+	type flagCase struct {
+		name, table string
+		flag        func(*DB) *schemaFlag
+	}
+	cases := []flagCase{
+		{"resolved_path", "observations", func(d *DB) *schemaFlag { return &d.hasResolvedPathFlag }},
+		{"raw_hex", "observations", func(d *DB) *schemaFlag { return &d.hasObsRawHexFlag }},
+		{"route_mask", "transmissions", func(d *DB) *schemaFlag { return &d.hasRouteMaskFlag }},
+		{"scope_name", "transmissions", func(d *DB) *schemaFlag { return &d.hasScopeNameFlag }},
+	}
+	base := time.Now().UTC().Add(-30 * time.Minute)
+
+	for _, c := range cases {
+		for _, start := range []bool{true, false} {
+			dir := "true_to_false"
+			if !start {
+				dir = "false_to_true"
+			}
+
+			t.Run("IngestNewFromDB/"+c.name+"/"+dir, func(t *testing.T) {
+				db := setupTestDB(t)
+				defer db.conn.Close()
+				seedShareGrowth(t, db)
+				ensureColumn158(t, db, c.table, c.name)
+				store := loadedStore158(t, db)
+				for i := 0; i < 3; i++ {
+					insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+					insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+				}
+				f := c.flag(db)
+				f.v.Store(start)
+				store.ingestAfterQueryHook = func() { f.v.Store(!start) }
+
+				store.IngestNewFromDB(0, 100)
+
+				store.mu.RLock()
+				defer store.mu.RUnlock()
+				for id := 1; id <= 3; id++ {
+					if store.byTxID[id] == nil {
+						t.Errorf("tx %d skipped: %s flag flipped %s between query and scan", id, c.name, dir)
+					}
+				}
+			})
+
+			if c.name == "scope_name" {
+				continue // IngestNewObservations does not select it
+			}
+			t.Run("IngestNewObservations/"+c.name+"/"+dir, func(t *testing.T) {
+				db := setupTestDB(t)
+				defer db.conn.Close()
+				seedShareGrowth(t, db)
+				ensureColumn158(t, db, c.table, c.name)
+				for i := 0; i < 3; i++ {
+					insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+					insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+				}
+				store := loadedStore158(t, db)
+				since := maxObsID158(t, db)
+				for i := 0; i < 3; i++ {
+					insertShareObsPath(t, db, i, 2, `["A1","B2"]`, `["`+shareA1+`","`+shareB2+`"]`, base.Add(time.Duration(i*10+2)*time.Second))
+				}
+				f := c.flag(db)
+				f.v.Store(start)
+				store.ingestAfterQueryHook = func() { f.v.Store(!start) }
+
+				store.IngestNewObservations(since, 100)
+
+				store.mu.RLock()
+				defer store.mu.RUnlock()
+				for id := 1; id <= 3; id++ {
+					if n := len(store.byTxID[id].Observations); n != 2 {
+						t.Errorf("tx %d has %d observations, want 2: row skipped when %s flag flipped %s between query and scan", id, n, c.name, dir)
+					}
+				}
+			})
+		}
+	}
+}
+
+// ensureColumn158 adds an optional column the in-memory fixture lacks, so a
+// schema flag can be latched true without the query failing on a missing
+// column (which would mask the scan mismatch under test).
+func ensureColumn158(t testing.TB, db *DB, table, col string) {
+	t.Helper()
+	rows, err := db.conn.Query("PRAGMA table_info(" + table + ")")
+	if err != nil {
+		t.Fatal(err)
+	}
+	have := false
+	for rows.Next() {
+		var cid int
+		var name, typ string
+		var notnull, pk int
+		var dflt any
+		if err := rows.Scan(&cid, &name, &typ, &notnull, &dflt, &pk); err != nil {
+			rows.Close()
+			t.Fatal(err)
+		}
+		if name == col {
+			have = true
+		}
+	}
+	rows.Close()
+	if !have {
+		exec158(t, db, "ALTER TABLE "+table+" ADD COLUMN "+col+" TEXT")
+	}
+}
diff --git a/cmd/server/store.go b/cmd/server/store.go
index 8f0f8aaef..691450f70 100644
--- a/cmd/server/store.go
+++ b/cmd/server/store.go
@@ -583,6 +583,11 @@ type PacketStore struct {
 	// (e.g. RefreshRouteMaskChanges). Test-only (#89 route_mask_changes
 	// watermark placement).
 	loadScannedRowHook func()
+	// ingestAfterQueryHook, if non-nil, runs in IngestNewFromDB and
+	// IngestNewObservations right after the SQL query succeeded and before
+	// its rows are scanned. Test-only: flips an optional-column schema flag
+	// between query construction and Scan (#158 follow-up).
+	ingestAfterQueryHook func()
 }
 
 // Precomputed distance records for fast analytics aggregation.
@@ -2815,23 +2820,25 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 	// ingestor's persisted resolution, exactly as Load feeds them
 	// (indexObservationRelayHops, #158 follow-up). The server still resolves
 	// each path itself for the live broadcast.
+	// One snapshot of the optional-column flags for query AND Scan (ingestCols).
+	cols := s.db.ingestCols()
 	var querySQL string
 	obsRHCol := ""
-	if s.db.hasObsRawHex() {
+	if cols.obsRawHex {
 		obsRHCol = ", o.raw_hex"
 	}
 	rpCol := ""
-	if s.db.hasResolvedPath() {
+	if cols.resolvedPath {
 		rpCol = ", o.resolved_path"
 	}
 	// #1751: scope_name is on the transmission row; append as the last column.
 	scopeNameCol := ""
-	if s.db.hasScopeName() {
+	if cols.scopeName {
 		scopeNameCol = ", t.scope_name"
 	}
 	// #89: route_mask follows scope_name as the last optional column.
 	routeMaskCol := ""
-	if s.db.hasRouteMask() {
+	if cols.routeMask {
 		routeMaskCol = ", t.route_mask"
 	}
 	if s.db.isV3() {
@@ -2862,6 +2869,9 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		return nil, sinceID
 	}
 	defer rows.Close()
+	if s.ingestAfterQueryHook != nil {
+		s.ingestAfterQueryHook()
+	}
 
 	// Scan into temp structures
 	type tempRow struct {
@@ -2899,16 +2909,16 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 			&payloadVersion, &decodedJSON,
 			&obsIDVal, &observerID, &observerName, &observerIATA, &direction,
 			&snrVal, &rssiVal, &scoreVal, &pathJSON, &obsTimestamp}
-		if s.db.hasObsRawHex() {
+		if cols.obsRawHex {
 			scanArgs2 = append(scanArgs2, &obsRawHex)
 		}
-		if s.db.hasResolvedPath() {
+		if cols.resolvedPath {
 			scanArgs2 = append(scanArgs2, &resolvedPath)
 		}
-		if s.db.hasScopeName() {
+		if cols.scopeName {
 			scanArgs2 = append(scanArgs2, &scopeName)
 		}
-		if s.db.hasRouteMask() {
+		if cols.routeMask {
 			scanArgs2 = append(scanArgs2, &routeMask)
 		}
 		if err := rows.Scan(scanArgs2...); err != nil {
@@ -3264,20 +3274,22 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		limit = 500
 	}
 
+	// One snapshot of the optional-column flags for query AND Scan (ingestCols).
+	cols := s.db.ingestCols()
 	var querySQL string
 	obsRHCol2 := ""
-	if s.db.hasObsRawHex() {
+	if cols.obsRawHex {
 		obsRHCol2 = ", o.raw_hex"
 	}
 	// resolved_path feeds the relay-hop indexes as in Load (#158 follow-up).
 	rpCol2 := ""
-	if s.db.hasResolvedPath() {
+	if cols.resolvedPath {
 		rpCol2 = ", o.resolved_path"
 	}
 	// #89: the transmission's current route_mask rides along with each new
 	// observation so the live view converges with a cold load.
 	routeMaskCol2, routeMaskJoin2 := "", ""
-	if s.db.hasRouteMask() {
+	if cols.routeMask {
 		routeMaskCol2 = ", t.route_mask"
 		routeMaskJoin2 = "\n\t\t\tLEFT JOIN transmissions t ON t.id = o.transmission_id"
 	}
@@ -3305,6 +3317,9 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		return nil
 	}
 	defer rows.Close()
+	if s.ingestAfterQueryHook != nil {
+		s.ingestAfterQueryHook()
+	}
 
 	type obsRow struct {
 		obsID        int
@@ -3334,13 +3349,13 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 
 		scanArgs3 := []interface{}{&oid, &txID, &observerID, &observerName, &observerIATA, &direction,
 			&snr, &rssi, &score, &pathJSON, &ts}
-		if s.db.hasObsRawHex() {
+		if cols.obsRawHex {
 			scanArgs3 = append(scanArgs3, &obsRawHex)
 		}
-		if s.db.hasResolvedPath() {
+		if cols.resolvedPath {
 			scanArgs3 = append(scanArgs3, &resolvedPath)
 		}
-		if s.db.hasRouteMask() {
+		if cols.routeMask {
 			scanArgs3 = append(scanArgs3, &routeMask)
 		}
 		if err := rows.Scan(scanArgs3...); err != nil {

From 5947bd6576f39965bedf9b2203d75610d0df3f8e Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 08:26:34 +0000
Subject: [PATCH 5/7] perf(store): decode resolved_path before taking s.mu on
 live ingest (#158)

IngestNewFromDB and IngestNewObservations unmarshalled resolved_path once
per observation inside indexObservationRelayHops, i.e. while holding
s.mu. The rows are already read into tempRows/obsRows before the lock, so
the column is now decoded there (decodePersistedRelayPath) and the index
step receives the decoded pubkeys. Load and scanAndMergeChunk call the
same helper at the same place as before.

The decode itself is cheaper: it unmarshals straight into []string and
compacts in place instead of []*string plus extractResolvedPubkeys. A
test pins it against that old pair, including NULL, all-null, corrupt and
non-string input.

Measured (n=10, interleaved, same test files): total time per call is
unchanged (the work moved, it did not go away: IngestNewFromDB 4.58 ->
4.55 ms, p=0.74), allocations -1.8 % (IngestNewFromDB, p<0.001). Time
spent holding s.mu, measured in an instrumented copy: IngestNewFromDB
2.28 -> 1.90 ms (-16.6 %, p<0.001); IngestNewObservations 1.83 -> 1.72 ms
(not significant, +-24 %).

Relates to #158

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017nop1P1pX4SrfRMHAernzQ
---
 cmd/server/chunked_load.go         |  2 +-
 cmd/server/sharegrowth_158_test.go | 30 +++++++++++++++
 cmd/server/store.go                | 59 +++++++++++++++++++++++++-----
 3 files changed, 81 insertions(+), 10 deletions(-)

diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go
index 4003bb870..3d8343db7 100644
--- a/cmd/server/chunked_load.go
+++ b/cmd/server/chunked_load.go
@@ -655,7 +655,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 			// Same relay-hop indexing as Load and live ingest (see
 			// indexObservationRelayHops). relayPM is passed in from
 			// LoadChunked (fetched before any chunk cursor opened).
-			s.indexObservationRelayHops(tx, nullStrVal(resolvedPathStr), obsPJ, obsIDStr, relayPM, hopsSeen, coldLoadAmbiguousHopsSkipped)
+			s.indexObservationRelayHops(tx, decodePersistedRelayPath(nullStrVal(resolvedPathStr)), obsPJ, obsIDStr, relayPM, hopsSeen, coldLoadAmbiguousHopsSkipped)
 
 			tx.mergeObservedPathHashSize(obsPJ)
 			tx.Observations = append(tx.Observations, obs)
diff --git a/cmd/server/sharegrowth_158_test.go b/cmd/server/sharegrowth_158_test.go
index 8d3e4d7ac..4924d0cba 100644
--- a/cmd/server/sharegrowth_158_test.go
+++ b/cmd/server/sharegrowth_158_test.go
@@ -372,6 +372,36 @@ func TestLateObservationIndexesLikeLoad_NullResolvedPath_158(t *testing.T) {
 	}
 }
 
+// decodePersistedRelayPath is what the live ingest paths call before taking
+// s.mu. A NULL/empty column must stay distinguishable from a persisted path
+// whose hops are all null: the first takes the path_json fallback, the
+// second is indexed as given (nothing).
+func TestDecodePersistedRelayPath_158(t *testing.T) {
+	for _, c := range []struct {
+		name, in string
+		want     persistedRelayPath
+	}{
+		{"NULL column", "", persistedRelayPath{}},
+		{"all hops null", `[null,null]`, persistedRelayPath{present: true}},
+		{"partly resolved", `["` + shareA1 + `",null,"` + shareB2 + `"]`, persistedRelayPath{pubkeys: []string{shareA1, shareB2}, present: true}},
+		{"empty list", `[]`, persistedRelayPath{present: true}},
+		{"corrupt json", `[`, persistedRelayPath{present: true}},
+		{"non-string hop", `[1,"` + shareA1 + `"]`, persistedRelayPath{present: true}},
+		{"empty string hop", `["","` + shareB2 + `"]`, persistedRelayPath{pubkeys: []string{shareB2}, present: true}},
+	} {
+		got := decodePersistedRelayPath(c.in)
+		if got.present != c.want.present || strings.Join(got.pubkeys, ",") != strings.Join(c.want.pubkeys, ",") {
+			t.Errorf("%s: decodePersistedRelayPath(%q) = %+v, want %+v", c.name, c.in, got, c.want)
+		}
+		// Oracle: the two functions the decode replaces.
+		if c.in != "" {
+			if o := extractResolvedPubkeys(unmarshalResolvedPath(c.in)); strings.Join(o, ",") != strings.Join(got.pubkeys, ",") {
+				t.Errorf("%s: %q decodes to %v, old extractResolvedPubkeys(unmarshalResolvedPath) gives %v", c.name, c.in, got.pubkeys, o)
+			}
+		}
+	}
+}
+
 // BenchmarkIngestNewFromDB_158 drives the real live ingest (SQL scan +
 // index + broadcast build) of a batch of 20 new transmissions with 11
 // observations each, as the ingestor writes them (resolved_path set).
diff --git a/cmd/server/store.go b/cmd/server/store.go
index 691450f70..49efc9e8d 100644
--- a/cmd/server/store.go
+++ b/cmd/server/store.go
@@ -1043,7 +1043,7 @@ func (s *PacketStore) Load() error {
 			}
 
 			// Decode-window: feed the relay-hop indexes, don't store on struct.
-			s.indexObservationRelayHops(tx, nullStrVal(resolvedPathStr), obsPJ, obsIDStr, relayPM, hopsSeen, &coldLoadAmbiguousHopsSkipped)
+			s.indexObservationRelayHops(tx, decodePersistedRelayPath(nullStrVal(resolvedPathStr)), obsPJ, obsIDStr, relayPM, hopsSeen, &coldLoadAmbiguousHopsSkipped)
 
 			tx.mergeObservedPathHashSize(obsPJ)
 			tx.Observations = append(tx.Observations, obs)
@@ -1878,6 +1878,45 @@ func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen
 	s.addToResolvedPubkeyIndex(tx.ID, pks)
 }
 
+// persistedRelayPath is a decoded observations.resolved_path column value.
+// present distinguishes a persisted (possibly all-null) path from a NULL
+// column, which takes the path_json fallback in indexObservationRelayHops.
+type persistedRelayPath struct {
+	pubkeys []string
+	present bool
+}
+
+// decodePersistedRelayPath decodes a resolved_path column value. It is pure
+// and takes no lock: the live ingest paths call it while scanning rows,
+// before s.mu is taken, so the JSON work does not extend the critical
+// section (#158 follow-up).
+//
+// The result equals extractResolvedPubkeys(unmarshalResolvedPath(s)). It
+// decodes straight into []string (a JSON null becomes "", which is what
+// extractResolvedPubkeys drops) and compacts in place, saving the *string
+// allocated per hop and the second slice on a path that runs once per
+// observation.
+func decodePersistedRelayPath(resolvedPath string) persistedRelayPath {
+	if resolvedPath == "" {
+		return persistedRelayPath{}
+	}
+	var hops []string
+	if json.Unmarshal([]byte(resolvedPath), &hops) != nil {
+		return persistedRelayPath{present: true} // as unmarshalResolvedPath: corrupt means no hops
+	}
+	n := 0
+	for _, h := range hops {
+		if h != "" {
+			hops[n] = h
+			n++
+		}
+	}
+	if n == 0 {
+		return persistedRelayPath{present: true}
+	}
+	return persistedRelayPath{pubkeys: hops[:n], present: true}
+}
+
 // indexObservationRelayHops feeds the relay-hop indexes from one stored
 // observation. Every path that materializes an observation (Load,
 // scanAndMergeChunk, IngestNewFromDB, IngestNewObservations) goes through
@@ -1893,10 +1932,12 @@ func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen
 // traffic share counted live transmissions for more relays than loaded
 // ones and its sum rose with uptime (#158 follow-up).
 //
-// skipped counts ambiguous cold-load hops (may be nil). Must hold s.mu.
-func (s *PacketStore) indexObservationRelayHops(tx *StoreTx, resolvedPath, pathJSON, observerID string, relayPM *prefixMap, hopsSeen map[string]bool, skipped *int) {
-	if resolvedPath != "" {
-		s.indexResolvedPathHops(tx, extractResolvedPubkeys(unmarshalResolvedPath(resolvedPath)), hopsSeen)
+// persisted is the decoded resolved_path (decodePersistedRelayPath), taken
+// outside the lock by the live paths. skipped counts ambiguous cold-load
+// hops (may be nil). Must hold s.mu.
+func (s *PacketStore) indexObservationRelayHops(tx *StoreTx, persisted persistedRelayPath, pathJSON, observerID string, relayPM *prefixMap, hopsSeen map[string]bool, skipped *int) {
+	if persisted.present {
+		s.indexResolvedPathHops(tx, persisted.pubkeys, hopsSeen)
 		return
 	}
 	if relayPM == nil || pathJSON == "" || pathJSON == "[]" {
@@ -2881,7 +2922,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		obsID                                                              *int
 		observerID, observerName, observerIATA, direction, pathJSON, obsTS string
 		obsRawHex                                                          string
-		resolvedPath                                                       string
+		resolvedPath                                                       persistedRelayPath
 		scopeName                                                          string
 		routeMask                                                          sql.NullInt64
 		snr, rssi                                                          *float64
@@ -2948,7 +2989,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 			pathJSON:     nullStrVal(pathJSON),
 			obsTS:        nullStrVal(obsTimestamp),
 			obsRawHex:    nullStrVal(obsRawHex),
-			resolvedPath: nullStrVal(resolvedPath),
+			resolvedPath: decodePersistedRelayPath(nullStrVal(resolvedPath)), // outside s.mu
 			scopeName:    nullStrVal(scopeName),
 			routeMask:    routeMask,
 			snr:          nullFloatPtr(snrVal),
@@ -3332,7 +3373,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		score        *int
 		pathJSON     string
 		rawHex       string
-		resolvedPath string
+		resolvedPath persistedRelayPath
 		timestamp    string
 		routeMask    sql.NullInt64
 	}
@@ -3374,7 +3415,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 			score:        nullIntPtr(score),
 			pathJSON:     nullStrVal(pathJSON),
 			rawHex:       nullStrVal(obsRawHex),
-			resolvedPath: nullStrVal(resolvedPath),
+			resolvedPath: decodePersistedRelayPath(nullStrVal(resolvedPath)), // outside s.mu
 			timestamp:    nullStrVal(ts),
 			routeMask:    routeMask,
 		})

From 504b5147837e24b6b30694075eda8dfd36d798ec Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 08:26:51 +0000
Subject: [PATCH 6/7] docs(store): correct the resolved_path comments in Load
 and LoadChunked (#158)

Both said resolved_path is NULL on every live deployment since #1287.
That contradicts indexObservationRelayHops and #1547: the ingestor
persists it with its conservative resolver (#1560), and it is NULL only
where that resolver could not resolve the path (a large share, #184).
Comment-only change.

Relates to #158

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017nop1P1pX4SrfRMHAernzQ
---
 cmd/server/store.go | 24 ++++++++++++++----------
 1 file changed, 14 insertions(+), 10 deletions(-)

diff --git a/cmd/server/store.go b/cmd/server/store.go
index 49efc9e8d..bcc179e4d 100644
--- a/cmd/server/store.go
+++ b/cmd/server/store.go
@@ -925,11 +925,14 @@ func (s *PacketStore) Load() error {
 			ORDER BY t.first_seen ASC, o.timestamp DESC`
 	}
 
-	// Relay-hop fallback inputs. When resolved_path is empty (always, on
-	// live, since #1287 the ingestor persists relay data as neighbor_edges
-	// instead) we re-resolve relay hops from path_json using the prefix
-	// map. PR #1643 R1 munger #1: cold load resolves ONLY when the prefix
-	// is unique (no affinity tiebreak against ≤168h-old observations,
+	// Relay-hop inputs. The ingestor persists observations.resolved_path
+	// (#1547) with its conservative resolver (#1560), so a persisted value
+	// feeds indexResolvedPathHops. It is NULL when the ingestor could not
+	// resolve the path (a large share of observations, see #184); then we
+	// re-resolve relay hops from path_json using the prefix map, into
+	// byNode only (indexObservationRelayHops). PR #1643 R1 munger #1: cold
+	// load resolves ONLY when the prefix is unique (no affinity tiebreak
+	// against ≤168h-old observations,
 	// which would silently mis-attribute hops). Fetched BEFORE opening
 	// the rows cursor below: getCachedNodesAndPM issues its own DB query,
 	// which would deadlock against the still-open cursor on a single-
@@ -1226,11 +1229,12 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 			ORDER BY t.first_seen ASC, o.timestamp DESC`
 	}
 
-	// Relay-hop fallback inputs. observations.resolved_path is NULL on
-	// every live deployment (since #1287 the ingestor persists relay data
-	// as aggregate neighbor_edges, not per-observation resolved_path), so
-	// for this background-loaded older window we re-resolve relay hops
-	// from the persisted path_json using the prefix map. PR #1643 R1
+	// Relay-hop inputs. A persisted observations.resolved_path (#1547,
+	// conservative resolver #1560) feeds indexResolvedPathHops; where it is
+	// NULL (the ingestor could not resolve the path, a large share of
+	// observations, see #184) this background-loaded older window
+	// re-resolves relay hops from the persisted path_json using the prefix
+	// map, into byNode only (indexObservationRelayHops). PR #1643 R1
 	// munger #1: cold load resolves ONLY when the prefix is unique
 	// (affinity-tier resolution against ≤168h-old observations would
 	// silently mis-attribute hops). Fetched BEFORE opening the rows

From 287e8534a9e1c5bdc51e4dd53501c7da5f47b1a2 Mon Sep 17 00:00:00 2001
From: dborup <kontakt@meshview.dk>
Date: Sat, 3 Oct 2026 13:52:13 +0000
Subject: [PATCH 7/7] test(store): catch a SELECT built from a fresh
 schema-flag read; doc the pks contract (#158)

Review round 2 nits 3 and 4.

- ingestAfterColsHook (test-only) runs right after the ingestCols snapshot
  in IngestNewFromDB and IngestNewObservations. The flag-flip test now flips
  the flag there as well as after the query, so a SELECT built from a fresh
  flag read while Scan uses the snapshot fails it (mutants M3c/M3d killed).
- indexResolvedPathHops' caller contract: pks may also be the pubkeys of
  decodePersistedRelayPath, which drops nils and empties as well.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018mVL1VB4jtY7caZQnMWJpW
---
 cmd/server/schema_snapshot_158_test.go | 125 ++++++++++++++-----------
 cmd/server/store.go                    |  16 +++-
 2 files changed, 84 insertions(+), 57 deletions(-)

diff --git a/cmd/server/schema_snapshot_158_test.go b/cmd/server/schema_snapshot_158_test.go
index a3e4ee6d3..d05efaf35 100644
--- a/cmd/server/schema_snapshot_158_test.go
+++ b/cmd/server/schema_snapshot_158_test.go
@@ -14,8 +14,10 @@ import (
 // swallowed and the row is silently skipped. One snapshot per call keeps
 // query and Scan consistent.
 //
-// The hook flips the flag after the query ran and before the rows are
-// scanned, which is the window the bug lives in.
+// The hooks flip the flag in the two windows a regression could use: after
+// the query ran and before the rows are scanned (a Scan built from a fresh
+// read), and after the snapshot and before the query is built (a SELECT
+// built from a fresh read).
 
 func TestIngestSchemaFlagFlipBetweenQueryAndScan_158(t *testing.T) {
 	type flagCase struct {
@@ -30,69 +32,82 @@ func TestIngestSchemaFlagFlipBetweenQueryAndScan_158(t *testing.T) {
 	}
 	base := time.Now().UTC().Add(-30 * time.Minute)
 
-	for _, c := range cases {
-		for _, start := range []bool{true, false} {
-			dir := "true_to_false"
-			if !start {
-				dir = "false_to_true"
-			}
+	// The flag flips either right after the ingestCols snapshot (a query
+	// built from a fresh flag read would not match the Scan) or between the
+	// query and the Scan (a Scan built from a fresh read would not match).
+	setFlip := func(store *PacketStore, at string, flip func()) {
+		if at == "after_cols" {
+			store.ingestAfterColsHook = flip
+		} else {
+			store.ingestAfterQueryHook = flip
+		}
+	}
 
-			t.Run("IngestNewFromDB/"+c.name+"/"+dir, func(t *testing.T) {
-				db := setupTestDB(t)
-				defer db.conn.Close()
-				seedShareGrowth(t, db)
-				ensureColumn158(t, db, c.table, c.name)
-				store := loadedStore158(t, db)
-				for i := 0; i < 3; i++ {
-					insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
-					insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+	for _, c := range cases {
+		for _, at := range []string{"after_cols", "after_query"} {
+			for _, start := range []bool{true, false} {
+				dir := at + "/true_to_false"
+				if !start {
+					dir = at + "/false_to_true"
 				}
-				f := c.flag(db)
-				f.v.Store(start)
-				store.ingestAfterQueryHook = func() { f.v.Store(!start) }
 
-				store.IngestNewFromDB(0, 100)
+				t.Run("IngestNewFromDB/"+c.name+"/"+dir, func(t *testing.T) {
+					db := setupTestDB(t)
+					defer db.conn.Close()
+					seedShareGrowth(t, db)
+					ensureColumn158(t, db, c.table, c.name)
+					store := loadedStore158(t, db)
+					for i := 0; i < 3; i++ {
+						insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+						insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+					}
+					f := c.flag(db)
+					f.v.Store(start)
+					setFlip(store, at, func() { f.v.Store(!start) })
+
+					store.IngestNewFromDB(0, 100)
 
-				store.mu.RLock()
-				defer store.mu.RUnlock()
-				for id := 1; id <= 3; id++ {
-					if store.byTxID[id] == nil {
-						t.Errorf("tx %d skipped: %s flag flipped %s between query and scan", id, c.name, dir)
+					store.mu.RLock()
+					defer store.mu.RUnlock()
+					for id := 1; id <= 3; id++ {
+						if store.byTxID[id] == nil {
+							t.Errorf("tx %d skipped: %s flag flipped (%s)", id, c.name, dir)
+						}
 					}
-				}
-			})
+				})
 
-			if c.name == "scope_name" {
-				continue // IngestNewObservations does not select it
-			}
-			t.Run("IngestNewObservations/"+c.name+"/"+dir, func(t *testing.T) {
-				db := setupTestDB(t)
-				defer db.conn.Close()
-				seedShareGrowth(t, db)
-				ensureColumn158(t, db, c.table, c.name)
-				for i := 0; i < 3; i++ {
-					insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
-					insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+				if c.name == "scope_name" {
+					continue // IngestNewObservations does not select it
 				}
-				store := loadedStore158(t, db)
-				since := maxObsID158(t, db)
-				for i := 0; i < 3; i++ {
-					insertShareObsPath(t, db, i, 2, `["A1","B2"]`, `["`+shareA1+`","`+shareB2+`"]`, base.Add(time.Duration(i*10+2)*time.Second))
-				}
-				f := c.flag(db)
-				f.v.Store(start)
-				store.ingestAfterQueryHook = func() { f.v.Store(!start) }
+				t.Run("IngestNewObservations/"+c.name+"/"+dir, func(t *testing.T) {
+					db := setupTestDB(t)
+					defer db.conn.Close()
+					seedShareGrowth(t, db)
+					ensureColumn158(t, db, c.table, c.name)
+					for i := 0; i < 3; i++ {
+						insertShareTx(t, db, i, base.Add(time.Duration(i)*time.Second))
+						insertShareObs(t, db, i, 1, base.Add(time.Duration(i*10+1)*time.Second))
+					}
+					store := loadedStore158(t, db)
+					since := maxObsID158(t, db)
+					for i := 0; i < 3; i++ {
+						insertShareObsPath(t, db, i, 2, `["A1","B2"]`, `["`+shareA1+`","`+shareB2+`"]`, base.Add(time.Duration(i*10+2)*time.Second))
+					}
+					f := c.flag(db)
+					f.v.Store(start)
+					setFlip(store, at, func() { f.v.Store(!start) })
 
-				store.IngestNewObservations(since, 100)
+					store.IngestNewObservations(since, 100)
 
-				store.mu.RLock()
-				defer store.mu.RUnlock()
-				for id := 1; id <= 3; id++ {
-					if n := len(store.byTxID[id].Observations); n != 2 {
-						t.Errorf("tx %d has %d observations, want 2: row skipped when %s flag flipped %s between query and scan", id, n, c.name, dir)
+					store.mu.RLock()
+					defer store.mu.RUnlock()
+					for id := 1; id <= 3; id++ {
+						if n := len(store.byTxID[id].Observations); n != 2 {
+							t.Errorf("tx %d has %d observations, want 2: row skipped when %s flag flipped (%s)", id, n, c.name, dir)
+						}
 					}
-				}
-			})
+				})
+			}
 		}
 	}
 }
diff --git a/cmd/server/store.go b/cmd/server/store.go
index bcc179e4d..f05b6c828 100644
--- a/cmd/server/store.go
+++ b/cmd/server/store.go
@@ -588,6 +588,11 @@ type PacketStore struct {
 	// its rows are scanned. Test-only: flips an optional-column schema flag
 	// between query construction and Scan (#158 follow-up).
 	ingestAfterQueryHook func()
+	// ingestAfterColsHook, if non-nil, runs in the same two functions right
+	// after the ingestCols snapshot and before the query is built. Test-only:
+	// flips a flag so a query built from a fresh flag read instead of the
+	// snapshot is caught too.
+	ingestAfterColsHook func()
 }
 
 // Precomputed distance records for fast analytics aggregation.
@@ -1863,8 +1868,9 @@ func pathFirstHop(pathJSON string) string {
 // Caller contract:
 //   - Must hold s.mu write lock (addToByNode / addToResolvedPubkeyIndex /
 //     addResolvedPubkeysToPathHopIndex all mutate store state).
-//   - pks should be the output of extractResolvedPubkeys (no nils, no
-//     empties); the helper is a no-op when pks is empty.
+//   - pks should be the output of extractResolvedPubkeys, or the pubkeys
+//     of decodePersistedRelayPath (both drop nils and 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
@@ -2867,6 +2873,9 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 	// each path itself for the live broadcast.
 	// One snapshot of the optional-column flags for query AND Scan (ingestCols).
 	cols := s.db.ingestCols()
+	if s.ingestAfterColsHook != nil {
+		s.ingestAfterColsHook()
+	}
 	var querySQL string
 	obsRHCol := ""
 	if cols.obsRawHex {
@@ -3321,6 +3330,9 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 
 	// One snapshot of the optional-column flags for query AND Scan (ingestCols).
 	cols := s.db.ingestCols()
+	if s.ingestAfterColsHook != nil {
+		s.ingestAfterColsHook()
+	}
 	var querySQL string
 	obsRHCol2 := ""
 	if cols.obsRawHex {