fix(store): index live observations from the persisted resolved_path, as Load does (#158)

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, decodePersistedRelayPath(nullStrVal(resolvedPathStr)), obsPJ, obsIDStr, relayPM, hopsSeen, coldLoadAmbiguousHopsSkipped)
 
 			tx.mergeObservedPathHashSize(obsPJ)
 			tx.Observations = append(tx.Observations, obs)

cmd/server/db.go

@@ -156,6 +156,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)

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)

cmd/server/schema_snapshot_158_test.go

@@ -0,0 +1,142 @@
+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 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 {
+		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)
+
+	// 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
+		}
+	}
+
+	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"
+				}
+
+				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)", 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)
+					setFlip(store, at, 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)", 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")
+	}
+}