diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index 314481508..3d8343db7 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, decodePersistedRelayPath(nullStrVal(resolvedPathStr)), obsPJ, obsIDStr, relayPM, hopsSeen, coldLoadAmbiguousHopsSkipped) tx.mergeObservedPathHashSize(obsPJ) tx.Observations = append(tx.Observations, obs) diff --git a/cmd/server/db.go b/cmd/server/db.go index 8401081dc..72602cb83 100644 --- a/cmd/server/db.go +++ b/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) 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/schema_snapshot_158_test.go b/cmd/server/schema_snapshot_158_test.go new file mode 100644 index 000000000..d05efaf35 --- /dev/null +++ b/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, ¬null, &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/sharegrowth_158_test.go b/cmd/server/sharegrowth_158_test.go new file mode 100644 index 000000000..4924d0cba --- /dev/null +++ b/cmd/server/sharegrowth_158_test.go @@ -0,0 +1,436 @@ +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) + } +} + +// 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) + } +} + +// 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). +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) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 5fcd95244..f05b6c828 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -583,6 +583,16 @@ 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() + // 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. @@ -920,11 +930,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- @@ -1037,29 +1050,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, decodePersistedRelayPath(nullStrVal(resolvedPathStr)), obsPJ, obsIDStr, relayPM, hopsSeen, &coldLoadAmbiguousHopsSkipped) tx.mergeObservedPathHashSize(obsPJ) tx.Observations = append(tx.Observations, obs) @@ -1242,11 +1234,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 @@ -1875,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 @@ -1894,6 +1888,76 @@ 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 +// 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). +// +// 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 == "[]" { + 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,29 +2867,39 @@ 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. + // 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 s.db.hasObsRawHex() { + if cols.obsRawHex { obsRHCol = ", o.raw_hex" } + rpCol := "" + 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() { 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 +2909,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 @@ -2849,6 +2923,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 { @@ -2858,6 +2935,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac obsID *int observerID, observerName, observerIATA, direction, pathJSON, obsTS string obsRawHex string + resolvedPath persistedRelayPath scopeName string routeMask sql.NullInt64 snr, rssi *float64 @@ -2877,6 +2955,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 @@ -2884,13 +2963,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.hasScopeName() { + if cols.resolvedPath { + scanArgs2 = append(scanArgs2, &resolvedPath) + } + 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 { @@ -2920,6 +3002,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac pathJSON: nullStrVal(pathJSON), obsTS: nullStrVal(obsTimestamp), obsRawHex: nullStrVal(obsRawHex), + resolvedPath: decodePersistedRelayPath(nullStrVal(resolvedPath)), // outside s.mu scopeName: nullStrVal(scopeName), routeMask: routeMask, snr: nullFloatPtr(snrVal), @@ -3032,15 +3115,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 { @@ -3246,21 +3328,31 @@ 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() + if s.ingestAfterColsHook != nil { + s.ingestAfterColsHook() + } 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 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" } 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 +3360,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 > ? @@ -3282,6 +3374,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 @@ -3294,6 +3389,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] score *int pathJSON string rawHex string + resolvedPath persistedRelayPath timestamp string routeMask sql.NullInt64 } @@ -3305,14 +3401,18 @@ 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, &snr, &rssi, &score, &pathJSON, &ts} - if s.db.hasObsRawHex() { + if cols.obsRawHex { scanArgs3 = append(scanArgs3, &obsRawHex) } - if s.db.hasRouteMask() { + if cols.resolvedPath { + scanArgs3 = append(scanArgs3, &resolvedPath) + } + if cols.routeMask { scanArgs3 = append(scanArgs3, &routeMask) } if err := rows.Scan(scanArgs3...); err != nil { @@ -3331,6 +3431,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] score: nullIntPtr(score), pathJSON: nullStrVal(pathJSON), rawHex: nullStrVal(obsRawHex), + resolvedPath: decodePersistedRelayPath(nullStrVal(resolvedPath)), // outside s.mu timestamp: nullStrVal(ts), routeMask: routeMask, }) @@ -3395,16 +3496,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 {