diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 13c134ee7..a54fdf778 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -107,6 +107,10 @@ type Config struct { // (internal/channelregistry). Approved channels are always loaded into // the channel keys, whether or not new submissions are enabled. ChannelProposals *channelregistry.Config `json:"channelProposals,omitempty"` + + // ResolvedPathBackfill tunes the start-up re-resolution of NULL + // observations.resolved_path rows (#188, resolved_path_backfill.go). + ResolvedPathBackfill *ResolvedPathBackfillConfig `json:"resolvedPathBackfill,omitempty"` } // NeighborEdgesDaysOrDefault returns the configured pruning window or 5. diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index 042af9153..03ee56a23 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -1215,10 +1215,14 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) { // the ingestor now. Per #1560: use the context-aware resolver so // 1-byte prefix collisions are disambiguated via NeighborGraph // adjacency (anchored on from_pubkey for ADVERTs, previous hop - // otherwise). Empty resolved JSON → NULL via nilIfEmpty. - resolved := resolvePathWithContext( + // otherwise). Per #188: flood paths are also walked backwards from the + // observer, whose neighbour the last hop is. Empty resolved JSON → NULL + // via nilIfEmpty. + resolved := resolveObservationPath( parsePathArray(data.PathJSON), strings.ToLower(data.FromPubkey), + data.ObserverID, + data.RouteType, s.neighborGraph.load(), s.prefixIdx.load(), ) diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 9323d0915..9f98b23b0 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -440,6 +440,13 @@ func main() { defer stopNeighborBuilder() log.Printf("[neighbor-build] enabled (interval=%s)", NeighborEdgesBuilderInterval) + // #188: re-resolve NULL resolved_path rows now that the prefix index + // and neighbour graph are primed (StartNeighborEdgesBuilder above). + if enabled, batch, pause := cfg.ResolvedPathBackfillSettings(); enabled { + stopResolvedPathBackfill := store.StartResolvedPathBackfill(batch, pause) + defer stopResolvedPathBackfill() + } + // #1212: per-source stall watchdog. Detects "silently dead" sources // where the client reports connected but no messages have flowed. Logs // a WARN line every minute for any source silent for >5m. Scan every diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index 63cda0567..c72e69da2 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -79,18 +79,39 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { if err := s.RefreshNeighborGraph(); err != nil { log.Printf("[neighbor-build] initial neighbor-graph refresh error: %v", err) } + // + // Each call reads at most neighborBuilderMaxBatch observations, so the + // loop ends when a call read fewer than that (caught up), not when it + // produced few edges (PR #190 review). Every row read is newer than the + // watermark, so a call that persists an edge moves the watermark; a full + // batch with no edge cannot, and the next call would read the same rows. + caughtUp := false for { - n, err := s.buildAndPersistNeighborEdges() + b, err := s.buildNeighborEdges() if err != nil { log.Printf("[neighbor-build] initial build error: %v", err) break } - wuTotal += n - if n < neighborBuilderMaxBatch { + wuTotal += b.edges + if b.caughtUp() { + caughtUp = true + break + } + if b.edges == 0 { + log.Printf("[neighbor-build] initial build cannot move past its watermark: %d observations yielded no edge; neighbor_edges stays incomplete and the resolved_path backfill waits", b.scanned) break } } log.Printf("[neighbor-build] initial build: %d edges upserted in %s", wuTotal, time.Since(wuStart)) + // Publish the graph with the edges the warm-up just persisted (#188): + // the snapshot primed above predates them, and on a fresh or restored DB + // it is empty. Only a build that caught up publishes a post-build graph; + // otherwise the first tick that catches up does it. + if caughtUp { + if err := s.refreshBuiltNeighborGraph(); err != nil { + log.Printf("[neighbor-build] post-build neighbor-graph refresh error: %v", err) + } + } var stopOnce sync.Once go func() { @@ -106,11 +127,18 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { if err := s.RefreshPrefixIndex(); err != nil { log.Printf("[neighbor-build] prefix-index refresh error: %v", err) } - n, err := s.buildAndPersistNeighborEdges() + b, err := s.buildNeighborEdges() + n := b.edges // Refresh the neighbor-graph snapshot after the edges // build (#1560) so the context-aware resolver picks up - // newly persisted adjacencies on the next ingest. - if grErr := s.RefreshNeighborGraph(); grErr != nil { + // newly persisted adjacencies on the next ingest. After a + // successful build that caught up, it is a post-build + // snapshot (#188). + refresh := s.RefreshNeighborGraph + if err == nil && b.caughtUp() { + refresh = s.refreshBuiltNeighborGraph + } + if grErr := refresh(); grErr != nil { log.Printf("[neighbor-build] neighbor-graph refresh error: %v", grErr) } dur := time.Since(start) @@ -137,10 +165,30 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { } } -// buildAndPersistNeighborEdges scans transmissions + observations, +// neighborEdgesBuild reports one buildNeighborEdges call. +type neighborEdgesBuild struct { + edges int // edge rows upserted + scanned int // observation rows read, at most neighborBuilderMaxBatch +} + +// caughtUp reports whether the scan read every observation newer than the +// watermark it started from: it read fewer rows than the cap. +func (b neighborEdgesBuild) caughtUp() bool { + return b.scanned < neighborBuilderMaxBatch +} + +// buildAndPersistNeighborEdges is buildNeighborEdges reporting only the +// number of edge upserts. +func (s *Store) buildAndPersistNeighborEdges() (int, error) { + b, err := s.buildNeighborEdges() + return b.edges, err +} + +// buildNeighborEdges scans transmissions + observations, // extracts edge candidates (originator↔first-hop on ADVERTs; -// observer↔last-hop on all packet types) and upserts them into -// neighbor_edges. Returns count of attempted upserts. +// observer↔last-hop on flood routes) and upserts them into +// neighbor_edges. It reports the edge upserts and the observation rows +// read; fewer rows than neighborBuilderMaxBatch means it caught up. // // Watermark / delta semantics (#1339): the builder derives a watermark // from MAX(neighbor_edges.last_seen). On an empty edges table (fresh @@ -161,10 +209,11 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { // SELECT of (lowered) pubkey prefixes from nodes. Prefixes with // multiple candidates are skipped (matches the conservative // resolution rule in cmd/server/extractEdgesFromObs). -func (s *Store) buildAndPersistNeighborEdges() (int, error) { +func (s *Store) buildNeighborEdges() (neighborEdgesBuild, error) { + var b neighborEdgesBuild prefixIdx, err := buildPrefixIndex(s.db) if err != nil { - return 0, fmt.Errorf("build prefix index: %w", err) + return b, fmt.Errorf("build prefix index: %w", err) } // Derive the watermark from the existing edges table. RFC3339 @@ -173,7 +222,7 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { // query and the parse return zero → full warm-up scan. var watermarkRFC sql.NullString if err := s.db.QueryRow(`SELECT MAX(last_seen) FROM neighbor_edges`).Scan(&watermarkRFC); err != nil { - return 0, fmt.Errorf("read watermark: %w", err) + return b, fmt.Errorf("read watermark: %w", err) } var watermarkEpoch int64 if watermarkRFC.Valid && watermarkRFC.String != "" { @@ -184,6 +233,7 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { rows, err := s.db.Query(`SELECT t.payload_type, + COALESCE(t.route_type, -1), t.decoded_json, COALESCE(t.from_pubkey, ''), COALESCE(o.path_json, ''), @@ -196,16 +246,18 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { ORDER BY o.timestamp LIMIT ?`, watermarkEpoch, neighborBuilderMaxBatch) if err != nil { - return 0, fmt.Errorf("scan observations: %w", err) + return b, fmt.Errorf("scan observations: %w", err) } defer rows.Close() var edges []edgeRow for rows.Next() { + b.scanned++ var payloadType sql.NullInt64 + var routeType int var decodedJSON, fromPubkey, pathJSON, observerID string var epochTs int64 - if err := rows.Scan(&payloadType, &decodedJSON, &fromPubkey, &pathJSON, &observerID, &epochTs); err != nil { + if err := rows.Scan(&payloadType, &routeType, &decodedJSON, &fromPubkey, &pathJSON, &observerID, &epochTs); err != nil { continue } fromNode := strings.ToLower(fromPubkey) @@ -228,7 +280,14 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { edges = append(edges, canonEdge(fromNode, resolved, ts)) } } - if observerPK != "" { + // The last hop is the node the observer heard only on a flood + // route: flood forwarders append their hash (MeshCore Mesh.cpp:346-350 + // routeRecvPacket, at a366955). A DIRECT path is the remaining + // planned route, from which each forwarder strips itself at the + // front (Mesh.cpp:89, removeSelfFromPath :334-342), so its last hop + // is the route's far end (PR #190 review). Unknown route types are + // skipped too. + if observerPK != "" && isFloodRoute(routeType) { last := path[len(path)-1] if resolved, ok := resolvePrefix(prefixIdx, last); ok && resolved != observerPK { edges = append(edges, canonEdge(observerPK, resolved, ts)) @@ -274,7 +333,7 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { } if len(edges) == 0 { - return 0, nil + return b, nil } // Wrap the whole edge-persist tx under writer-perf instrumentation @@ -304,9 +363,10 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { return nil }) if err != nil { - return 0, err + return b, err } - return inserted, nil + b.edges = inserted + return b, nil } // canonEdge orders the pair so node_a <= node_b (matches the existing @@ -336,11 +396,11 @@ func parsePathArray(s string) []string { // considered ambiguous and skipped during resolution. type prefixIndex map[string][]string -// buildPrefixIndex reads nodes.public_key and builds the prefix → pubkey -// map. We index every 1-byte (2 hex char) prefix length the firmware -// uses (1, 2, 3, 4, 6, 8). Memory cost is O(nodes × len(prefixLens)). +// buildPrefixIndex reads the relay nodes (isRelayRole) and builds the +// prefix → pubkey map. We index every 1-byte (2 hex char) prefix length the +// firmware uses (1, 2, 3, 4, 6, 8). Memory cost is O(nodes × len(prefixLens)). func buildPrefixIndex(db *sql.DB) (prefixIndex, error) { - rows, err := db.Query(`SELECT public_key FROM nodes`) + rows, err := db.Query(`SELECT public_key, COALESCE(role, '') FROM nodes`) if err != nil { return nil, err } @@ -348,8 +408,11 @@ func buildPrefixIndex(db *sql.DB) (prefixIndex, error) { idx := make(prefixIndex, 1024) var prefixLens = []int{1 * 2, 2 * 2, 3 * 2, 4 * 2, 6 * 2, 8 * 2} for rows.Next() { - var pk string - if err := rows.Scan(&pk); err != nil { + var pk, role string + if err := rows.Scan(&pk, &role); err != nil { + continue + } + if !isRelayRole(role) { continue } pkLower := strings.ToLower(pk) @@ -364,6 +427,23 @@ func buildPrefixIndex(db *sql.DB) (prefixIndex, error) { return idx, nil } +// isRelayRole reports whether a node of this role can appear as a hop in a +// path (#188). It is the server's canAppearInPath (cmd/server/store.go), +// duplicated because the two binaries share no package for it; the test +// cases mirror the server's TestCanAppearInPath. +// +// Firmware: repeaters and room servers forward unless disable_fwd is set +// (examples/simple_repeater and simple_room_server MyMesh::allowPacketForward). +// Companions and sensors ship with forwarding off (companion_radio +// NodePrefs.h repeat.disable_fwd = 1; simple_sensor SensorMesh.cpp +// disable_fwd = true) and can only opt in, so a hop through an opted-in +// companion stays unresolved, or resolves to the one relay that shares its +// prefix. That is the server's trade-off too. +func isRelayRole(role string) bool { + r := strings.ToLower(role) + return strings.Contains(r, "repeater") || strings.Contains(r, "room_server") || r == "room" +} + // resolvePrefix returns the single resolved pubkey if exactly one // candidate matches, otherwise (zero || multiple), it returns ok=false // (matches the conservative server-side resolver in diff --git a/cmd/ingestor/neighbor_builder_delta_test.go b/cmd/ingestor/neighbor_builder_delta_test.go index 307fb9784..14685860d 100644 --- a/cmd/ingestor/neighbor_builder_delta_test.go +++ b/cmd/ingestor/neighbor_builder_delta_test.go @@ -35,8 +35,9 @@ func TestNeighborEdgesBuilderDeltaScan(t *testing.T) { } defer store.Close() + // Hop nodes need a relay role: the prefix index holds relays only (#188). if _, err := store.db.Exec( - `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + `INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater'), (?, ?, 'repeater')`, "aaaaaaaaaa", "from-node", "bbbbbbbbbb", "first-hop", ); err != nil { diff --git a/cmd/ingestor/neighbor_builder_review_190_test.go b/cmd/ingestor/neighbor_builder_review_190_test.go new file mode 100644 index 000000000..554ac0e77 --- /dev/null +++ b/cmd/ingestor/neighbor_builder_review_190_test.go @@ -0,0 +1,201 @@ +package main + +import ( + "context" + "fmt" + "path/filepath" + "strings" + "testing" + "time" +) + +// PR #190, second review (P2-1): the warm-up build must not stop after a +// full batch of observations just because that batch produced few edges, and +// the backfill must not start until the build has caught up. + +const ts190 = int64(1780272000) // 2026-06-01T00:00:00Z + +// seedObservations190 writes one transmission + observation per path, at +// timestamps ts190+offset+i, from the obs188 observer. Flood route, GRP_TXT. +func seedObservations190(t *testing.T, store *Store, offset int64, paths []string) []int64 { + t.Helper() + var obsIdx int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, strings.ToUpper(obs188)).Scan(&obsIdx); err != nil { + t.Fatal(err) + } + tx, err := store.db.Begin() + if err != nil { + t.Fatal(err) + } + insTx, err := tx.Prepare(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES ('00', ?, '2026-06-01T00:00:00Z', 1, 5, '{}')`) + if err != nil { + t.Fatal(err) + } + insObs, err := tx.Prepare(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, ?, ?)`) + if err != nil { + t.Fatal(err) + } + ids := make([]int64, 0, len(paths)) + for i, p := range paths { + res, err := insTx.Exec(fmt.Sprintf("h190-%d-%d", offset, i)) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + res, err = insObs.Exec(txID, obsIdx, p, ts190+offset+int64(i)) + if err != nil { + t.Fatal(err) + } + id, _ := res.LastInsertId() + ids = append(ids, id) + } + insTx.Close() + insObs.Close() + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + return ids +} + +func repeatPath190(p string, n int) []string { + out := make([]string, n) + for i := range out { + out[i] = p + } + return out +} + +// A full first batch (neighborBuilderMaxBatch observations) that yields one +// edge must not end the warm-up: the edge observer <-> c3a comes from the +// next observation, in the second batch. +func TestNeighborEdgesBuilder_WarmUpScansPastFullBatchWithFewEdges_190(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + clearEdges188(t, store) + if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, 'b2a', 'repeater')`, b2a); err != nil { + t.Fatal(err) + } + // Batch 1: 49,999 ambiguous ["c3"] rows (no edge) and, last, ["b233"] + // (edge observer <-> b2a). Batch 2: ["c355"] (edge observer <-> c3a). + first := append(repeatPath190(`["c3"]`, neighborBuilderMaxBatch-1), `["b233"]`) + ids := seedObservations190(t, store, 0, first) + seedObservations190(t, store, int64(neighborBuilderMaxBatch), []string{`["c355"]`}) + + stop := store.StartNeighborEdgesBuilder(time.Hour) + defer stop() + if !store.neighborGraph.load().IsAdjacent(obs188, c3a) { + t.Fatal("the warm-up stopped after a full batch with one edge: no observer<->c3a edge from the second batch") + } + res, err := store.RunResolvedPathBackfill(context.Background(), defaultResolvedPathBackfillBatchSize, 0) + if err != nil { + t.Fatal(err) + } + if res.Resolved < len(ids)-1 { + t.Fatalf("pass = %+v, want the %d ambiguous rows resolved", res, len(ids)-1) + } + wantAllResolvedTo188(t, store, ids[:3], c3a) +} + +// A full batch that yields no edge newer than the watermark leaves the +// builder unable to move past it. The graph is then incomplete, so the +// backfill must not run on it, even though the graph is not empty. +func TestResolvedPathBackfill_WaitsWhileEdgeBuildCannotCatchUp_190(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + clearEdges188(t, store) + // An older edge: the graph is not empty, and it sets the watermark. + if _, err := store.db.Exec(`INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, '2026-05-01T00:00:00Z')`, a1a, b2a); err != nil { + t.Fatal(err) + } + ids := seedObservations190(t, store, 0, repeatPath190(`["c3"]`, neighborBuilderMaxBatch)) + seedObservations190(t, store, int64(neighborBuilderMaxBatch), []string{`["c355"]`}) + + // Ticks every 100 ms: a tick that has not caught up must not publish a + // post-build graph either. + stop := store.StartNeighborEdgesBuilder(100 * time.Millisecond) + defer stop() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if built, _ := store.neighborGraph.buildState(); built { + t.Fatal("a post-build graph was published although no build caught up") + } + time.Sleep(20 * time.Millisecond) + } + if _, err := store.RunResolvedPathBackfill(context.Background(), defaultResolvedPathBackfillBatchSize, 0); err == nil { + t.Fatal("the pass ran although the edge build has not caught up with the observations") + } + if w := backfillWatermarkOrZero188(t, store); w != 0 { + t.Fatalf("watermark moved to %d on an incomplete graph", w) + } + if resolvedPathOf188(t, store, ids[0]).Valid { + t.Fatal("a row was resolved on an incomplete graph") + } +} + +// PR #190, second review (P2-2): the observer <-> last-hop edge holds only +// for flood routes. A DIRECT path is the remaining planned route: each +// forwarder matches itself at the front and strips itself +// (firmware Mesh.cpp:89, removeSelfFromPath :334-342), and only flood +// forwarders append their hash (routeRecvPacket :346-350). So the last hop +// of a DIRECT path is the destination end of the route, not the node the +// observer heard. +func TestNeighborEdgesBuilder_ObserverEdgeOnlyForFloodRoutes_190(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + clearEdges188(t, store) + for _, pk := range []string{a1a, b2a} { + if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater')`, pk, pk[:4]); err != nil { + t.Fatal(err) + } + } + var obsIdx int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, strings.ToUpper(obs188)).Scan(&obsIdx); err != nil { + t.Fatal(err) + } + for i, o := range []struct { + route int + path string + }{ + {routeFlood188, `["c355"]`}, // flood: observer heard c3a + {routeTransportFlood188, `["a111"]`}, // transport flood: observer heard a1a + {routeDirect188, `["c366"]`}, // DIRECT: c3b is the route's far end + {3, `["a111","b233"]`}, // TRANSPORT_DIRECT: b2a is the far end + } { + res, err := store.db.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES ('00', ?, '2026-06-01T00:00:00Z', ?, 5, '{}')`, + fmt.Sprintf("h190-route-%d", i), o.route) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + if _, err := store.db.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, ?, ?)`, txID, obsIdx, o.path, ts190+int64(i)); err != nil { + t.Fatal(err) + } + } + if _, err := store.buildAndPersistNeighborEdges(); err != nil { + t.Fatal(err) + } + g, err := loadNeighborGraph(store.db) + if err != nil { + t.Fatal(err) + } + for _, pk := range []string{c3a, a1a} { + if !g.IsAdjacent(obs188, pk) { + t.Errorf("no observer<->%s edge from a flood observation", pk[:4]) + } + } + for _, pk := range []string{c3b, b2a} { + if g.IsAdjacent(obs188, pk) { + t.Errorf("observer<->%s edge built from a DIRECT path", pk[:4]) + } + } + // A DIRECT route's consecutive hops are still neighbours. + if !g.IsAdjacent(a1a, b2a) { + t.Fatal("the interior edge of the DIRECT route is gone") + } + + // The effect on the flood resolver: with only the flood edge, the + // ambiguous last hop "c3" of a flood path resolves to c3a. A false + // DIRECT edge to c3b would leave it nil. + got := resolveObservationPath([]string{"c3"}, "", obs188, routeFlood188, g, idx188(c3a, c3b, a1a, b2a)) + wantPath188(t, got, c3a) +} diff --git a/cmd/ingestor/neighbor_builder_test.go b/cmd/ingestor/neighbor_builder_test.go index 7c2d14505..6b568c31b 100644 --- a/cmd/ingestor/neighbor_builder_test.go +++ b/cmd/ingestor/neighbor_builder_test.go @@ -26,8 +26,9 @@ func TestNeighborEdgesBuilderUpsertsFromObservations(t *testing.T) { defer store.Close() // Seed two nodes whose pubkey prefixes will be used as hops. + // Hop nodes need a relay role: the prefix index holds relays only (#188). if _, err := store.db.Exec( - `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + `INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater'), (?, ?, 'repeater')`, "aaaaaaaaaa", "from-node", "bbbbbbbbbb", "first-hop", ); err != nil { @@ -101,8 +102,9 @@ func TestNeighborEdgesBuilderInteriorHopEdges(t *testing.T) { defer store.Close() // Three repeaters along the path, each with a unique 2-hex prefix. + // Hop nodes need a relay role: the prefix index holds relays only (#188). if _, err := store.db.Exec( - `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?), (?, ?)`, + `INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater'), (?, ?, 'repeater'), (?, ?, 'repeater')`, "bbbbbbbbbb", "hop-b", "cccccccccc", "hop-c", "dddddddddd", "hop-d", @@ -182,8 +184,9 @@ func TestNeighborEdgesBuilderInteriorHopEdges_ExcludesOneByte(t *testing.T) { // Same three repeaters as the sibling test, but referenced by their // 1-byte (2-hex-char) prefixes below instead of 2-byte. + // Hop nodes need a relay role: the prefix index holds relays only (#188). if _, err := store.db.Exec( - `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?), (?, ?)`, + `INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater'), (?, ?, 'repeater'), (?, ?, 'repeater')`, "bbbbbbbbbb", "hop-b", "cccccccccc", "hop-c", "dddddddddd", "hop-d", diff --git a/cmd/ingestor/path_resolver.go b/cmd/ingestor/path_resolver.go index 7ca94479f..3e78196ee 100644 --- a/cmd/ingestor/path_resolver.go +++ b/cmd/ingestor/path_resolver.go @@ -3,6 +3,7 @@ package main import ( "database/sql" "strings" + "sync" "sync/atomic" ) @@ -87,11 +88,25 @@ func (g *NeighborGraph) IsAdjacent(a, b string) bool { return present } +// empty reports whether the graph has no edge. +func (g *NeighborGraph) empty() bool { + return g == nil || len(g.adj) == 0 +} + // neighborGraphHolder caches the graph for the InsertTransmission hot // path. atomic.Value lets the 60s rebuild publish without a read-side // lock. +// +// It also records whether a snapshot loaded after a neighbor_edges build +// that succeeded and caught up has been published (storeBuilt). Before that, the +// snapshot may predate the edges the warm-up build derives; the +// resolved_path backfill waits for it (#188, PR #190 review). type neighborGraphHolder struct { v atomic.Value // holds *NeighborGraph + + mu sync.Mutex + built bool // a post-build snapshot has been published + next chan struct{} // closed when the next post-build snapshot is published } func (h *neighborGraphHolder) load() *NeighborGraph { @@ -105,6 +120,30 @@ func (h *neighborGraphHolder) store(g *NeighborGraph) { h.v.Store(g) } +// storeBuilt publishes g, loaded after a neighbor_edges build that caught up, +// and wakes everyone waiting on buildState's channel. +func (h *neighborGraphHolder) storeBuilt(g *NeighborGraph) { + h.store(g) + h.mu.Lock() + h.built = true + if h.next != nil { + close(h.next) + h.next = nil + } + h.mu.Unlock() +} + +// buildState reports whether a post-build snapshot has been published, and +// returns a channel that is closed when the next one is. +func (h *neighborGraphHolder) buildState() (built bool, next <-chan struct{}) { + h.mu.Lock() + defer h.mu.Unlock() + if h.next == nil { + h.next = make(chan struct{}) + } + return h.built, h.next +} + // loadNeighborGraph reads neighbor_edges and returns an in-memory // adjacency snapshot. Safe to call against a fresh DB (returns an // empty graph). @@ -134,12 +173,13 @@ func loadNeighborGraph(db *sql.DB) (*NeighborGraph, error) { // does not revisit a node). // // Behavior matrix: -// len(candidates) | anchor | graph | result -// 0 | — | — | nil -// 1 | — | — | candidates[0] -// >1 | "" or no graph|— | nil -// >1 | non-empty | set | unique adjacent candidate -// (or nil if 0 or >1 survive) +// +// len(candidates) | anchor | graph | result +// 0 | — | — | nil +// 1 | — | — | candidates[0] +// >1 | "" or no graph|— | nil +// >1 | non-empty | set | unique adjacent candidate +// (or nil if 0 or >1 survive) func resolveHopWithContext(hop string, anchor string, graph *NeighborGraph, idx prefixIndex, exclude map[string]struct{}) *string { if idx == nil { return nil @@ -186,6 +226,13 @@ func resolveHopWithContext(hop string, anchor string, graph *NeighborGraph, idx // node. Returns a `[]*string` shape compatible with // marshalResolvedPath (and the all-nil clobber-guard from PR #1548). func resolvePathWithContext(hops []string, fromPubkey string, graph *NeighborGraph, idx prefixIndex) []*string { + return resolvePathForward(hops, fromPubkey, "", graph, idx) +} + +// resolvePathForward is the forward walk of resolvePathWithContext. notLast +// (lower-case), when set, is not a candidate for the last hop (#188: the +// observer of a flood path, see resolveObservationPath). +func resolvePathForward(hops []string, fromPubkey, notLast string, graph *NeighborGraph, idx prefixIndex) []*string { if len(hops) == 0 { return nil } @@ -194,11 +241,14 @@ func resolvePathWithContext(hops []string, fromPubkey string, graph *NeighborGra return out } prevAnchor := strings.ToLower(fromPubkey) - seen := make(map[string]struct{}, len(hops)+1) + seen := make(map[string]struct{}, len(hops)+2) if prevAnchor != "" { seen[prevAnchor] = struct{}{} } for i, hop := range hops { + if notLast != "" && i == len(hops)-1 { + seen[notLast] = struct{}{} + } r := resolveHopWithContext(hop, prevAnchor, graph, idx, seen) out[i] = r if r != nil { @@ -223,3 +273,143 @@ func (s *Store) RefreshNeighborGraph() error { s.neighborGraph.store(g) return nil } + +// refreshBuiltNeighborGraph is RefreshNeighborGraph for a caller that has +// just completed a buildNeighborEdges that caught up: it marks the +// snapshot as post-build (neighborGraphHolder.storeBuilt). +func (s *Store) refreshBuiltNeighborGraph() error { + g, err := loadNeighborGraph(s.db) + if err != nil { + return err + } + s.neighborGraph.storeBuilt(g) + return nil +} + +// Observer anchor (#188) +// +// Firmware basis (MeshCore src/, line numbers at a366955): a flood forwarder +// appends its own hash before it retransmits (Mesh.cpp:344-356 +// routeRecvPacket, copyHashTo at :349), and an observer logs a packet on +// reception, before its own routing step (Dispatcher.cpp:238 logRx runs +// before processRecvPacket at :246/:256). So the last hash of a FLOOD path +// an observer reports is the node it heard the packet from: a direct +// neighbour of the observer. This does not hold for DIRECT routes: the path +// is the remaining planned route, and each forwarder strips itself from the +// front (Mesh.cpp:89, removeSelfFromPath at :334). TRACE is DIRECT too +// (sendFlood refuses it, Mesh.cpp:638). Its header path carries SNR bytes +// (Mesh.cpp:60-61), and the decoder replaces the hops with the planned route +// from the payload (decoder.go, TRACE branch). So only flood route types are +// anchored. +// +// Observer identity: the observer is the segment of the MQTT +// topic meshcore///packets, which is the observer node's +// pubkey. The neighbour builder writes observer<->last-hop edges keyed by +// that id lower-cased (neighbor_builder.go), and NeighborGraph lower-cases +// on lookup, so an id that is not a node (e.g. "companion") has no edges and +// anchors nothing. + +// Route types (MeshCore src/Packet.h ROUTE_TYPE_*). +const ( + routeTypeTransportFlood = 0 + routeTypeFlood = 1 +) + +func isFloodRoute(routeType int) bool { + return routeType == routeTypeTransportFlood || routeType == routeTypeFlood +} + +// resolveObservationPath resolves one observation's hops for +// observations.resolved_path. It runs the forward chain from fromPubkey +// (resolvePathForward, the walk behind resolvePathWithContext); for a flood +// packet whose forward chain left a hop nil, it also runs a backward chain +// from the observer and merges the two (mergeForwardBackward). Both chains +// use resolveHopWithContext, so a hop resolves only to a unique candidate: +// no geo, GPS or count tie-break. +// +// In a flood path, the observer is never the last hop: a radio does not +// receive its own transmission. Both chains leave that hop nil rather than +// name the observer, even when its prefix is unique (PR #190 review). +func resolveObservationPath(hops []string, fromPubkey, observer string, routeType int, graph *NeighborGraph, idx prefixIndex) []*string { + if !isFloodRoute(routeType) { + return resolvePathWithContext(hops, fromPubkey, graph, idx) + } + observer = strings.ToLower(observer) + fwd := resolvePathForward(hops, fromPubkey, observer, graph, idx) + if graph == nil || idx == nil || observer == "" || !hasNil(fwd) { + return fwd + } + return mergeForwardBackward(fwd, resolvePathBackward(hops, fromPubkey, observer, graph, idx)) +} + +// resolvePathBackward walks the hops from last to first. The last hop is +// anchored on the observer, each earlier hop on the hop after it; a hop that +// does not resolve breaks the chain, as in the forward walk. fromPubkey and +// already resolved hops are excluded from later candidate pools. +// +// The observer (lower-case) is excluded from the last hop only. It stays a +// candidate for earlier hops: it logs a packet before de-duplication +// (Dispatcher.cpp logRx runs before processRecvPacket), so it can report the +// echo of a flood it forwarded itself. +func resolvePathBackward(hops []string, fromPubkey, observer string, graph *NeighborGraph, idx prefixIndex) []*string { + out := make([]*string, len(hops)) + anchor := observer + seen := make(map[string]struct{}, len(hops)+2) + if fp := strings.ToLower(fromPubkey); fp != "" { + seen[fp] = struct{}{} + } + _, observerSeen := seen[observer] + seen[observer] = struct{}{} + for i := len(hops) - 1; i >= 0; i-- { + r := resolveHopWithContext(hops[i], anchor, graph, idx, seen) + if i == len(hops)-1 && !observerSeen { + delete(seen, observer) + } + out[i] = r + if r != nil { + seen[*r] = struct{}{} + anchor = *r + } else { + anchor = "" + } + } + return out +} + +// mergeForwardBackward combines the two chains. A hop keeps the forward +// result when there is one and takes the backward result otherwise. The two +// must never contradict each other in one row: if they resolve a hop to +// different nodes, or the merge would put one node at two positions, the +// backward evidence is dropped for the whole row and fwd is returned as is. +func mergeForwardBackward(fwd, bwd []*string) []*string { + out := make([]*string, len(fwd)) + seen := make(map[string]struct{}, len(fwd)) + for i, f := range fwd { + b := bwd[i] + if f != nil && b != nil && *f != *b { + return fwd + } + r := f + if r == nil { + r = b + } + if r == nil { + continue + } + if _, dup := seen[*r]; dup { + return fwd + } + seen[*r] = struct{}{} + out[i] = r + } + return out +} + +func hasNil(rp []*string) bool { + for _, p := range rp { + if p == nil { + return true + } + } + return false +} diff --git a/cmd/ingestor/path_resolver_bench_188_test.go b/cmd/ingestor/path_resolver_bench_188_test.go new file mode 100644 index 000000000..40a207c49 --- /dev/null +++ b/cmd/ingestor/path_resolver_bench_188_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "fmt" + "math/rand" + "testing" +) + +// BenchmarkResolveObservationPath_188 measures resolveObservationPath, the +// per-observation call in InsertTransmission, on a synthetic mesh: 700 +// relays, 60 observers, 1-byte hops, flood paths of 1-5 hops built as walks +// over the neighbour graph that end at a neighbour of the observer. One op is +// one observation. +func BenchmarkResolveObservationPath_188(b *testing.B) { + rng := rand.New(rand.NewSource(188)) + key := func() string { + buf := make([]byte, 32) + rng.Read(buf) + return fmt.Sprintf("%x", buf) + } + relays := make([]string, 700) + for i := range relays { + relays[i] = key() + } + observers := make([]string, 60) + for i := range observers { + observers[i] = key() + } + idx := idx188(relays...) + g := NewNeighborGraph() + nbrs := map[string][]string{} + link := func(a, c string) { + g.AddEdge(a, c) + nbrs[a] = append(nbrs[a], c) + nbrs[c] = append(nbrs[c], a) + } + for _, r := range relays { + for k := 0; k < 3; k++ { + link(r, relays[rng.Intn(len(relays))]) + } + } + for _, o := range observers { + for k := 0; k < 8; k++ { + link(o, relays[rng.Intn(len(relays))]) + } + } + + type obs struct { + hops []string + observer string + } + sample := make([]obs, 4096) + for i := range sample { + o := observers[rng.Intn(len(observers))] + cur := nbrs[o][rng.Intn(len(nbrs[o]))] + n := 1 + rng.Intn(5) + hops := make([]string, n) + for j := n - 1; j >= 0; j-- { + hops[j] = cur[:2] + next := nbrs[cur] + cur = next[rng.Intn(len(next))] + } + sample[i] = obs{hops, o} + } + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s := sample[i%len(sample)] + resolveObservationPath(s.hops, "", s.observer, routeFlood188, g, idx) + } +} diff --git a/cmd/ingestor/path_resolver_observer_188_test.go b/cmd/ingestor/path_resolver_observer_188_test.go new file mode 100644 index 000000000..e9967a325 --- /dev/null +++ b/cmd/ingestor/path_resolver_observer_188_test.go @@ -0,0 +1,360 @@ +package main + +import ( + "database/sql" + "fmt" + "math/rand" + "path/filepath" + "strings" + "testing" +) + +// Issue #188: the observer anchors the last hop of a FLOOD path, and a +// backward walk resolves earlier hops from it. See resolveObservationPath. + +// Pubkeys for 1-byte collisions: each prefix has an "a" and a "b" node. +const ( + obs188 = "0b5e000000000000000000000000000000000000000000000000000000000001" + a1a = "a1110000000000000000000000000000000000000000000000000000000000aa" + a1b = "a1220000000000000000000000000000000000000000000000000000000000bb" + b2a = "b2330000000000000000000000000000000000000000000000000000000000aa" + b2b = "b2440000000000000000000000000000000000000000000000000000000000bb" + c3a = "c3550000000000000000000000000000000000000000000000000000000000aa" + c3b = "c3660000000000000000000000000000000000000000000000000000000000bb" + from188 = "f0f0000000000000000000000000000000000000000000000000000000000001" +) + +const ( + routeTransportFlood188 = 0 + routeFlood188 = 1 + routeDirect188 = 2 +) + +func idx188(keys ...string) prefixIndex { + idx := prefixIndex{} + for _, pk := range keys { + for _, n := range []int{2, 4, 6, 8, 12, 16} { + idx[pk[:n]] = append(idx[pk[:n]], pk) + } + } + return idx +} + +func graph188(edges ...[2]string) *NeighborGraph { + g := NewNeighborGraph() + for _, e := range edges { + g.AddEdge(e[0], e[1]) + } + return g +} + +func path188(rp []*string) string { + parts := make([]string, len(rp)) + for i, p := range rp { + if p == nil { + parts[i] = "nil" + } else { + parts[i] = (*p)[:4] + } + } + return "[" + strings.Join(parts, " ") + "]" +} + +func wantPath188(t *testing.T, got []*string, want ...string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("len = %d, want %d (%s)", len(got), len(want), path188(got)) + } + for i := range want { + g := "" + if got[i] != nil { + g = *got[i] + } + if g != want[i] { + t.Fatalf("hop %d = %q, want %q (path %s)", i, g, want[i], path188(got)) + } + } +} + +var allNodes188 = []string{obs188, a1a, a1b, b2a, b2b, c3a, c3b, from188} + +func TestObserverAnchor_LastHopResolvesViaObserver_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}) + got := resolveObservationPath([]string{"b2", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, "", c3a) + + // The observer id arrives upper-case from the MQTT topic. + got = resolveObservationPath([]string{"C3"}, "", strings.ToUpper(obs188), routeTransportFlood188, g, idx) + wantPath188(t, got, c3a) +} + +func TestObserverAnchor_BackwardWalkResolvesEarlierHops_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}, [2]string{c3a, b2a}, [2]string{b2a, a1a}, + // decoys that are adjacent to the wrong anchors + [2]string{c3b, b2b}, [2]string{b2b, a1b}) + got := resolveObservationPath([]string{"a1", "b2", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, a1a, b2a, c3a) + + // The walk stops at the first ambiguous hop: hop 0 has no unique + // neighbour of the resolved hop 1, so it stays nil. + g2 := graph188([2]string{obs188, c3a}, [2]string{c3a, b2a}, [2]string{b2a, a1a}, [2]string{b2a, a1b}) + got = resolveObservationPath([]string{"a1", "b2", "c3"}, "", obs188, routeFlood188, g2, idx) + wantPath188(t, got, "", b2a, c3a) +} + +func TestObserverAnchor_TwoNeighborCandidatesStayNil_188(t *testing.T) { + idx := idx188(allNodes188...) + // Both c3 candidates are neighbours of the observer. + g := graph188([2]string{obs188, c3a}, [2]string{obs188, c3b}, [2]string{c3a, b2a}) + got := resolveObservationPath([]string{"b2", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, "", "") + + // Last hop resolves, but both b2 candidates neighbour it. + g = graph188([2]string{obs188, c3a}, [2]string{c3a, b2a}, [2]string{c3a, b2b}) + got = resolveObservationPath([]string{"b2", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, "", c3a) +} + +// The observer anchor only holds for flood routing: a DIRECT packet's path is +// the remaining planned route (each forwarder strips itself from the front, +// firmware Mesh.cpp removeSelfFromPath). TRACE is always DIRECT; its hops are +// the planned route the decoder takes from the payload. +func TestObserverAnchor_DirectRouteIsNotAnchored_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}) + for _, rt := range []int{routeDirect188, 3} { + got := resolveObservationPath([]string{"c3"}, "", obs188, rt, g, idx) + wantPath188(t, got, "") + } +} + +func TestObserverAnchor_UnknownObserverHasNoEffect_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}) + for _, o := range []string{"", "companion", "not-a-node"} { + got := resolveObservationPath([]string{"c3"}, "", o, routeFlood188, g, idx) + wantPath188(t, got, "") + } + // No graph: nothing to anchor on. + got := resolveObservationPath([]string{"c3"}, "", obs188, routeFlood188, nil, idx) + wantPath188(t, got, "") +} + +// An ADVERT resolved by the forward chain keeps exactly that result; the +// backward walk only fills hops the forward chain left nil. +func TestObserverAnchor_AdvertForwardResultUnchanged_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{from188, a1a}, [2]string{a1a, b2a}, [2]string{b2a, c3a}, [2]string{obs188, c3a}) + hops := []string{"a1", "b2", "c3"} + fwd := resolvePathWithContext(hops, from188, g, idx) + wantPath188(t, fwd, a1a, b2a, c3a) + got := resolveObservationPath(hops, from188, obs188, routeFlood188, g, idx) + wantPath188(t, got, a1a, b2a, c3a) + + // Forward resolves hop 0 only; the backward walk fills hops 1 and 2. + g = graph188([2]string{from188, a1a}, [2]string{obs188, c3a}, [2]string{c3a, b2a}) + got = resolveObservationPath(hops, from188, obs188, routeFlood188, g, idx) + wantPath188(t, got, a1a, b2a, c3a) +} + +// Forward and backward never contradict each other in one row: when they +// resolve the same hop differently, or would put one node at two positions, +// the backward result is dropped for the whole row. +func TestObserverAnchor_ForwardBackwardMeetOrNil_188(t *testing.T) { + idx := idx188(allNodes188...) + hops := []string{"a1", "b2", "c3"} + + t.Run("agree", func(t *testing.T) { + // Forward: a1a -> b2a, then c3 is ambiguous (both neighbour b2a). + // Backward: observer -> c3a -> b2a -> a1a, agreeing on hops 0-1. + g := graph188([2]string{from188, a1a}, [2]string{a1a, b2a}, [2]string{b2a, c3a}, [2]string{b2a, c3b}, + [2]string{obs188, c3a}) + fwd := resolvePathWithContext(hops, from188, g, idx) + wantPath188(t, fwd, a1a, b2a, "") + got := resolveObservationPath(hops, from188, obs188, routeFlood188, g, idx) + wantPath188(t, got, a1a, b2a, c3a) + }) + t.Run("disagree on a hop keeps the forward result", func(t *testing.T) { + // Forward: a1a -> b2a, then c3 has no neighbour of b2a. + // Backward: observer -> c3b -> b2b. Hop 1 conflicts (b2a vs b2b). + g := graph188([2]string{from188, a1a}, [2]string{a1a, b2a}, + [2]string{obs188, c3b}, [2]string{c3b, b2b}) + fwd := resolvePathWithContext(hops, from188, g, idx) + wantPath188(t, fwd, a1a, b2a, "") + got := resolveObservationPath(hops, from188, obs188, routeFlood188, g, idx) + wantPath188(t, got, a1a, b2a, "") + }) + t.Run("same node at two positions keeps the forward result", func(t *testing.T) { + // A path that names the same prefix twice: forward puts c3a at + // hop 0, backward puts c3a at hop 1. + g := graph188([2]string{from188, c3a}, [2]string{obs188, c3a}) + got := resolveObservationPath([]string{"c3", "c3"}, from188, obs188, routeFlood188, g, idx) + wantPath188(t, got, c3a, "") + }) +} + +// Invariant over random graphs and paths: every hop the forward chain +// resolved is kept, every resolved hop matches its prefix, and no node +// appears twice in a row. +func TestObserverAnchor_NeverContradictsForward_Random_188(t *testing.T) { + rng := rand.New(rand.NewSource(188)) + var keys []string + for p := 0; p < 6; p++ { + for v := 0; v < 3; v++ { + keys = append(keys, fmt.Sprintf("%02x%02x", 0xa0+p, v)+strings.Repeat("0", 60)) + } + } + keys = append(keys, obs188, from188) + idx := idx188(keys...) + resolvedSome := 0 + for iter := 0; iter < 2000; iter++ { + g := NewNeighborGraph() + for e := 0; e < 25; e++ { + g.AddEdge(keys[rng.Intn(len(keys))], keys[rng.Intn(len(keys))]) + } + n := 1 + rng.Intn(5) + hops := make([]string, n) + for i := range hops { + hops[i] = fmt.Sprintf("%02x", 0xa0+rng.Intn(6)) + } + from := "" + if rng.Intn(2) == 0 { + from = from188 + } + fwd := resolvePathWithContext(hops, from, g, idx) + got := resolveObservationPath(hops, from, obs188, routeFlood188, g, idx) + seen := map[string]bool{} + for i := range hops { + if fwd[i] != nil && (got[i] == nil || *got[i] != *fwd[i]) { + t.Fatalf("iter %d: hop %d changed from forward %s to %s", iter, i, path188(fwd), path188(got)) + } + if got[i] == nil { + continue + } + if !strings.HasPrefix(*got[i], hops[i]) { + t.Fatalf("iter %d: hop %d %q does not match prefix %q", iter, i, *got[i], hops[i]) + } + if seen[*got[i]] { + t.Fatalf("iter %d: %s appears twice in %s", iter, (*got[i])[:4], path188(got)) + } + seen[*got[i]] = true + if fwd[i] == nil { + resolvedSome++ + } + } + } + if resolvedSome == 0 { + t.Fatal("the backward walk never resolved a hop; the invariant was not exercised") + } +} + +func TestObserverAnchor_HashSizes_188(t *testing.T) { + for _, size := range []int{1, 2, 3} { + t.Run(fmt.Sprintf("%d-byte", size), func(t *testing.T) { + // Two relays that share the first `size` bytes. + shared := strings.Repeat("d4", size) + x := shared + "11" + strings.Repeat("0", 62-2*size) + y := shared + "22" + strings.Repeat("0", 62-2*size) + idx := idx188(x, y, obs188) + g := graph188([2]string{obs188, y}) + got := resolveObservationPath([]string{shared}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, y) + if fwd := resolvePathWithContext([]string{shared}, "", g, idx); fwd[0] != nil { + t.Fatalf("setup: the forward chain already resolves %s", shared) + } + }) + } +} + +// End to end through InsertTransmission: a non-advert flood packet with a +// 1-byte path is stored with its last hop resolved from the observer's +// neighbour edge. On master the row is NULL. +func TestInsertTransmission_ObserverAnchorResolvesLastHop_188(t *testing.T) { + store, err := OpenStore(filepath.Join(t.TempDir(), "ingest.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + store.backfillWg.Wait() + + for _, n := range []struct{ pk, role string }{{c3a, "repeater"}, {c3b, "repeater"}, {b2a, "repeater"}, {b2b, "room"}} { + if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, ?, ?)`, n.pk, n.pk[:4], n.role); err != nil { + t.Fatal(err) + } + } + observerID := strings.ToUpper(obs188) // as it arrives in the MQTT topic + if err := store.UpsertObserver(observerID, "observer-188", "", nil); err != nil { + t.Fatal(err) + } + // The neighbour builder writes observer<->last-hop edges lower-cased. + if _, err := store.db.Exec(`INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, '2026-06-01T00:00:00Z')`, obs188, c3a); err != nil { + t.Fatal(err) + } + if err := store.RefreshPrefixIndex(); err != nil { + t.Fatal(err) + } + if err := store.RefreshNeighborGraph(); err != nil { + t.Fatal(err) + } + pkt := &PacketData{ + RawHex: "c0ffee", + Timestamp: "2026-06-01T00:00:00Z", + ObserverID: observerID, + Hash: "h-188-anchor", + RouteType: routeFlood188, + PayloadType: 5, // GRP_TXT: no fromPubkey + PathJSON: `["b2","c3"]`, + DecodedJSON: "{}", + } + if _, err := store.InsertTransmission(pkt); err != nil { + t.Fatal(err) + } + var rp sql.NullString + if err := store.db.QueryRow(`SELECT resolved_path FROM observations WHERE transmission_id = (SELECT id FROM transmissions WHERE hash = ?)`, pkt.Hash).Scan(&rp); err != nil { + t.Fatal(err) + } + if !rp.Valid { + t.Fatal("resolved_path is NULL: the observer did not anchor the last hop") + } + wantPath188(t, unmarshalResolvedPathLocal(rp.String), "", c3a) +} + +// PR #190 review, finding 4: an observer cannot hear its own transmission, +// so in a flood path it reports it is never the last hop. Both chains must +// leave that hop nil rather than name the observer. obs188 is the only node +// with the 1-byte prefix "0b". +func TestObserverAnchor_ObserverIsNeverItsOwnLastHop_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}) + + // The unique prefix names the observer: the forward chain alone would + // resolve it. + got := resolveObservationPath([]string{"0b"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, "") + got = resolveObservationPath([]string{"0B"}, "", strings.ToUpper(obs188), routeTransportFlood188, g, idx) + wantPath188(t, got, "") + + // The backward walk must not anchor hop 0 on the observer named as the + // last hop either. + got = resolveObservationPath([]string{"c3", "0b"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, "", "") + + // DIRECT routes are unchanged: there the observer is the next hop, at + // the front of the remaining route (firmware Mesh.cpp:89). + got = resolveObservationPath([]string{"0b"}, "", obs188, routeDirect188, g, idx) + wantPath188(t, got, obs188) +} + +// The exclusion is for the last hop only. An observer logs every packet it +// receives before de-duplication (firmware Dispatcher.cpp logRx before +// processRecvPacket), so it can report the echo of a flood it forwarded +// itself, with its own hash earlier in the path. +func TestObserverAnchor_ObserverMayBeAnEarlierHop_188(t *testing.T) { + obsTwin := "0b77000000000000000000000000000000000000000000000000000000000077" // shares "0b" with obs188 + idx := idx188(append([]string{obsTwin}, allNodes188...)...) + g := graph188([2]string{obs188, c3a}) + got := resolveObservationPath([]string{"0b", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, obs188, c3a) +} diff --git a/cmd/ingestor/prefix_index_relay_188_test.go b/cmd/ingestor/prefix_index_relay_188_test.go new file mode 100644 index 000000000..fe7f77abe --- /dev/null +++ b/cmd/ingestor/prefix_index_relay_188_test.go @@ -0,0 +1,127 @@ +package main + +import ( + "database/sql" + "path/filepath" + "testing" +) + +// Issue #188 point 2: the prefix index holds relay roles only, with the same +// definition as the server's canAppearInPath (cmd/server/store.go). The +// cases mirror cmd/server/prefix_map_role_test.go TestCanAppearInPath. +func TestIsRelayRole_MatchesServerCanAppearInPath_188(t *testing.T) { + cases := []struct { + role string + want bool + }{ + {"repeater", true}, + {"Repeater", true}, + {"REPEATER", true}, + {"room_server", true}, + {"Room_Server", true}, + {"room", true}, + {"companion", false}, + {"sensor", false}, + {"", false}, + {"unknown", false}, + } + for _, tc := range cases { + if got := isRelayRole(tc.role); got != tc.want { + t.Errorf("isRelayRole(%q) = %v, want %v", tc.role, got, tc.want) + } + } +} + +func openRelayTestStore188(t *testing.T) *Store { + t.Helper() + store, err := OpenStore(filepath.Join(t.TempDir(), "ingest.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { store.Close() }) + store.backfillWg.Wait() + return store +} + +func seedNodes188(t *testing.T, store *Store, nodes ...[2]string) { + t.Helper() + for _, n := range nodes { + var role interface{} = n[1] + if n[1] == "" { + role = nil + } + if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, ?, ?)`, n[0], n[0][:4], role); err != nil { + t.Fatal(err) + } + } +} + +func TestBuildPrefixIndex_RelayRolesOnly_188(t *testing.T) { + store := openRelayTestStore188(t) + const ( + rep = "aa110000000000000000000000000000000000000000000000000000000000aa" + comp = "aa220000000000000000000000000000000000000000000000000000000000bb" + room = "bb330000000000000000000000000000000000000000000000000000000000aa" + sens = "cc440000000000000000000000000000000000000000000000000000000000aa" + none = "dd550000000000000000000000000000000000000000000000000000000000aa" + ) + seedNodes188(t, store, [2]string{rep, "repeater"}, [2]string{comp, "companion"}, [2]string{room, "room"}, + [2]string{sens, "sensor"}, [2]string{none, ""}) + idx, err := buildPrefixIndex(store.db) + if err != nil { + t.Fatal(err) + } + if got := idx["aa"]; len(got) != 1 || got[0] != rep { + t.Errorf(`idx["aa"] = %v, want only the repeater (the companion sharing the prefix cannot relay)`, got) + } + if got := idx["bb"]; len(got) != 1 || got[0] != room { + t.Errorf(`idx["bb"] = %v, want the room`, got) + } + for _, p := range []string{"aa22", "cc", "dd"} { + if got := idx[p]; len(got) != 0 { + t.Errorf("idx[%q] = %v, want empty (companion, sensor and role-less nodes are not relays)", p, got) + } + } +} + +// A companion is never stored as a relay hop, even when its prefix is +// unique in the node table; a prefix it shares with one repeater now +// resolves to that repeater. +func TestInsertTransmission_CompanionNeverStoredAsRelay_188(t *testing.T) { + store := openRelayTestStore188(t) + const ( + rep = "aa110000000000000000000000000000000000000000000000000000000000aa" + comp = "aa220000000000000000000000000000000000000000000000000000000000bb" + lone = "ee660000000000000000000000000000000000000000000000000000000000cc" + ) + seedNodes188(t, store, [2]string{rep, "repeater"}, [2]string{comp, "companion"}, [2]string{lone, "companion"}) + if err := store.UpsertObserver("obs-188", "observer", "", nil); err != nil { + t.Fatal(err) + } + if err := store.RefreshPrefixIndex(); err != nil { + t.Fatal(err) + } + insert := func(hash, path string) sql.NullString { + t.Helper() + pkt := &PacketData{RawHex: "c0ffee", Timestamp: "2026-06-01T00:00:00Z", ObserverID: "obs-188", Hash: hash, + RouteType: 1, PayloadType: 5, PathJSON: path, DecodedJSON: "{}"} + if _, err := store.InsertTransmission(pkt); err != nil { + t.Fatal(err) + } + var rp sql.NullString + if err := store.db.QueryRow(`SELECT resolved_path FROM observations WHERE transmission_id = (SELECT id FROM transmissions WHERE hash = ?)`, hash).Scan(&rp); err != nil { + t.Fatal(err) + } + return rp + } + if rp := insert("h-188-lone", `["ee"]`); rp.Valid { + t.Errorf("a hop whose only match is a companion was stored as %s, want NULL", rp.String) + } + rp := insert("h-188-shared", `["aa"]`) + if !rp.Valid { + t.Fatal(`hop "aa" (one repeater + one companion) stayed NULL, want the repeater`) + } + if got := unmarshalResolvedPathLocal(rp.String); len(got) != 1 || got[0] == nil || *got[0] != rep { + t.Errorf("resolved_path = %s, want [%s]", rp.String, rep) + } +} diff --git a/cmd/ingestor/resolved_path_backfill.go b/cmd/ingestor/resolved_path_backfill.go new file mode 100644 index 000000000..7463a9715 --- /dev/null +++ b/cmd/ingestor/resolved_path_backfill.go @@ -0,0 +1,336 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "strings" + "time" +) + +// Issue #188 point 4: re-resolve observations stored with +// resolved_path = NULL once the prefix index and neighbour graph are primed. +// Those rows come from the resolver before #188 (no observer anchor, +// companions in the index) and from the start-up window in which buffered +// ingest is drained before the index is primed (main.go). +// +// The pass needs a prefix index and a neighbour graph loaded after a +// neighbor_edges build that succeeded and caught up with the observations +// (neighborEdgesBuild.caughtUp), with at least one edge +// (resolvedPathBackfillReady). On anything less it would resolve only +// unique prefixes and still move the watermark past every row, so it does +// not start, and a batch whose snapshot is not usable writes nothing. The +// background pass waits for the next post-build graph and then resumes from +// the persisted watermark. +// +// One pass per ingestor start covers observation ids in (watermark, ceiling], +// where ceiling is MAX(observations.id) when the pass starts and watermark is +// persisted in resolved_path_backfill_state. Each batch: +// - reads at most batchSize rows by id on the shared connection, without +// writerMu (a bounded primary-key range scan); +// - resolves them in Go with resolveObservationPath, holding no lock; +// - writes in one WriterTx: an UPDATE ... WHERE resolved_path IS NULL per +// row that now resolves, and the new watermark. +// +// Updating only NULL rows and moving the watermark in the same transaction +// makes the pass idempotent and resumable: a crash or restart repeats at most +// the batch that did not commit. Rows that still do not resolve stay NULL and +// are not retried in a later pass (the watermark has passed them). +// +// Server visibility: the server indexes an observation's resolved_path when +// it first polls the row (live ingest, #182) or loads it at start-up. A value +// backfilled after that reaches the server's indexes only on its next restart +// (Load). This change does not touch the server. + +const resolvedPathBackfillComponent = "resolved_path_backfill" + +// Defaults; ResolvedPathBackfillConfig overrides them. Candidates for the +// customizer later (AGENTS.md rule 8). +const ( + defaultResolvedPathBackfillBatchSize = 500 + defaultResolvedPathBackfillPause = 250 * time.Millisecond +) + +// ResolvedPathBackfillConfig is the "resolvedPathBackfill" config block +// (documented in config.example.json). For BatchSize and PauseMs, 0 or a +// negative value means the default, so the pause cannot be turned off; the +// smallest pause is 1 ms. +type ResolvedPathBackfillConfig struct { + Disabled bool `json:"disabled,omitempty"` + BatchSize int `json:"batchSize,omitempty"` // rows per batch; default 500 + PauseMs int `json:"pauseMs,omitempty"` // pause between batches; default 250 +} + +// ResolvedPathBackfillSettings returns the effective settings. +func (c *Config) ResolvedPathBackfillSettings() (enabled bool, batchSize int, pause time.Duration) { + enabled, batchSize, pause = true, defaultResolvedPathBackfillBatchSize, defaultResolvedPathBackfillPause + if b := c.ResolvedPathBackfill; b != nil { + enabled = !b.Disabled + if b.BatchSize > 0 { + batchSize = b.BatchSize + } + if b.PauseMs > 0 { + pause = time.Duration(b.PauseMs) * time.Millisecond + } + } + return enabled, batchSize, pause +} + +// resolvedPathBatch reports one batch. +type resolvedPathBatch struct { + Scanned int // observation rows read + Resolved int // rows whose resolved_path was set + Last int64 // highest id read: the new watermark + Hold time.Duration // duration of the write transaction (writerMu + SQLite write lock) +} + +// ResolvedPathBackfillResult reports one pass. +type ResolvedPathBackfillResult struct { + From, Ceiling int64 + Batches int + Scanned int + Resolved int + MaxHold time.Duration +} + +func ensureResolvedPathBackfillState(db *sql.DB) error { + // PREFLIGHT: async=true reason="CREATE TABLE IF NOT EXISTS for a one-row state table; constant cost at any scale" + _, err := db.Exec(`CREATE TABLE IF NOT EXISTS resolved_path_backfill_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + watermark INTEGER NOT NULL, + updated_at TEXT NOT NULL + )`) + return err +} + +func (s *Store) resolvedPathBackfillWatermark() (int64, error) { + var w int64 + err := s.db.QueryRow(`SELECT watermark FROM resolved_path_backfill_state WHERE id = 1`).Scan(&w) + if err == sql.ErrNoRows { + return 0, nil + } + return w, err +} + +// errResolvedPathBackfillNotReady: no prefix index, or no neighbour graph +// from an edge build that caught up, or that graph has no edge. +var errResolvedPathBackfillNotReady = errors.New("prefix index or neighbour graph not ready (no neighbour-edge build has caught up yet, or no edges)") + +// resolvedPathBackfillReady reports whether the pass can run on this index +// and graph. +func (s *Store) resolvedPathBackfillReady(idx prefixIndex, graph *NeighborGraph) bool { + built, _ := s.neighborGraph.buildState() + return built && len(idx) > 0 && !graph.empty() +} + +// StartResolvedPathBackfill runs one backfill pass in the background. Until +// a neighbour graph from an edge build that caught up, with at least one edge, is +// published (StartNeighborEdgesBuilder), it waits and retries on each new +// post-build graph. The returned stop function cancels the pass and waits for +// it; the watermark of the last committed batch is kept. +func (s *Store) StartResolvedPathBackfill(batchSize int, pause time.Duration) func() { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + defer func() { + if r := recover(); r != nil { + log.Printf("[%s] panic recovered: %v", resolvedPathBackfillComponent, r) + } + }() + for logged := false; ; { + // Take the wake-up channel before the readiness check, so a + // graph published in between is not missed. + _, next := s.neighborGraph.buildState() + _, err := s.RunResolvedPathBackfill(ctx, batchSize, pause) + if !errors.Is(err, errResolvedPathBackfillNotReady) { + if err != nil && ctx.Err() == nil { + log.Printf("[%s] error: %v", resolvedPathBackfillComponent, err) + } + return + } + if !logged { + log.Printf("[%s] waiting: %v", resolvedPathBackfillComponent, err) + logged = true + } + select { + case <-ctx.Done(): + return + case <-next: + } + } + }() + return func() { + cancel() + <-done + } +} + +// RunResolvedPathBackfill runs one pass from the persisted watermark up to the +// current MAX(observations.id). +func (s *Store) RunResolvedPathBackfill(ctx context.Context, batchSize int, pause time.Duration) (ResolvedPathBackfillResult, error) { + var res ResolvedPathBackfillResult + if batchSize <= 0 { + batchSize = defaultResolvedPathBackfillBatchSize + } + // Without a usable index and graph most rows would stay NULL while the + // watermark moved past them for good. + if !s.resolvedPathBackfillReady(s.prefixIdx.load(), s.neighborGraph.load()) { + return res, errResolvedPathBackfillNotReady + } + if err := ensureResolvedPathBackfillState(s.db); err != nil { + return res, fmt.Errorf("ensure state table: %w", err) + } + from, err := s.resolvedPathBackfillWatermark() + if err != nil { + return res, fmt.Errorf("read watermark: %w", err) + } + if err := s.db.QueryRow(`SELECT COALESCE(MAX(id), 0) FROM observations`).Scan(&res.Ceiling); err != nil { + return res, fmt.Errorf("read ceiling: %w", err) + } + res.From = from + if from >= res.Ceiling { + log.Printf("[%s] nothing to do (watermark %d, max id %d)", resolvedPathBackfillComponent, from, res.Ceiling) + return res, nil + } + start := time.Now() + log.Printf("[%s] starting: ids (%d, %d], batch %d, pause %s", resolvedPathBackfillComponent, from, res.Ceiling, batchSize, pause) + for after := from; after < res.Ceiling; { + if err := ctx.Err(); err != nil { + return res, err + } + b, err := s.resolvedPathBackfillBatch(ctx, after, res.Ceiling, batchSize) + if err != nil { + return res, err + } + if b.Scanned == 0 { + break + } + after = b.Last + res.Batches++ + res.Scanned += b.Scanned + res.Resolved += b.Resolved + if b.Hold > res.MaxHold { + res.MaxHold = b.Hold + } + if res.Batches%100 == 0 { + log.Printf("[%s] progress: watermark %d of %d, %d scanned, %d resolved", resolvedPathBackfillComponent, after, res.Ceiling, res.Scanned, res.Resolved) + } + select { + case <-ctx.Done(): + return res, ctx.Err() + case <-time.After(pause): + } + } + log.Printf("[%s] done: %d rows scanned, %d resolved, %d batches, max write hold %s, in %s", + resolvedPathBackfillComponent, res.Scanned, res.Resolved, res.Batches, res.MaxHold.Round(time.Microsecond), time.Since(start).Round(time.Millisecond)) + return res, nil +} + +type resolvedPathBackfillRow struct { + id int64 + pathJSON string + isNull bool + routeType int + payloadType int + fromPubkey string + observerID string +} + +// resolvedPathBackfillBatch processes ids in (after, ceiling], at most limit +// rows, and persists the new watermark. +func (s *Store) resolvedPathBackfillBatch(ctx context.Context, after, ceiling int64, limit int) (resolvedPathBatch, error) { + var b resolvedPathBatch + // The pass checked readiness when it started; check the snapshot this + // batch uses as well, and write nothing (not even the watermark) if it + // is not usable. + graph, idx := s.neighborGraph.load(), s.prefixIdx.load() + if !s.resolvedPathBackfillReady(idx, graph) { + return b, errResolvedPathBackfillNotReady + } + rows, err := s.db.QueryContext(ctx, `SELECT o.id, COALESCE(o.path_json, ''), o.resolved_path IS NULL, + COALESCE(t.route_type, -1), COALESCE(t.payload_type, -1), COALESCE(t.from_pubkey, ''), COALESCE(obs.id, '') + FROM observations o + JOIN transmissions t ON t.id = o.transmission_id + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.id > ? AND o.id <= ? + ORDER BY o.id + LIMIT ?`, after, ceiling, limit) + if err != nil { + return b, fmt.Errorf("select batch: %w", err) + } + batch := make([]resolvedPathBackfillRow, 0, limit) + for rows.Next() { + var r resolvedPathBackfillRow + if err := rows.Scan(&r.id, &r.pathJSON, &r.isNull, &r.routeType, &r.payloadType, &r.fromPubkey, &r.observerID); err != nil { + rows.Close() + return b, fmt.Errorf("scan batch: %w", err) + } + batch = append(batch, r) + } + rows.Close() + if err := rows.Err(); err != nil { + return b, fmt.Errorf("read batch: %w", err) + } + if len(batch) == 0 { + return b, nil + } + b.Scanned = len(batch) + b.Last = batch[len(batch)-1].id + + // Resolve outside any lock, with the same inputs InsertTransmission uses + // (fromPubkey only for ADVERTs, as buildPacketData sets it). + type update struct { + id int64 + rp string + } + var updates []update + for _, r := range batch { + if !r.isNull { + continue + } + hops := parsePathArray(r.pathJSON) + if len(hops) == 0 { + continue + } + from := "" + if r.payloadType == int(payloadADVERT) { + from = strings.ToLower(r.fromPubkey) + } + if rp := marshalResolvedPath(resolveObservationPath(hops, from, r.observerID, r.routeType, graph, idx)); rp != "" { + updates = append(updates, update{r.id, rp}) + } + } + + holdStart := time.Now() + err = s.WriterTx(resolvedPathBackfillComponent, func(tx *sql.Tx) error { + if len(updates) > 0 { + stmt, err := tx.Prepare(`UPDATE observations SET resolved_path = ? WHERE id = ? AND resolved_path IS NULL`) + if err != nil { + return err + } + defer stmt.Close() + for _, u := range updates { + res, err := stmt.Exec(u.rp, u.id) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n > 0 { + b.Resolved++ + } + } + } + _, err := tx.Exec(`INSERT INTO resolved_path_backfill_state (id, watermark, updated_at) VALUES (1, ?, ?) + ON CONFLICT(id) DO UPDATE SET watermark = excluded.watermark, updated_at = excluded.updated_at`, + b.Last, time.Now().UTC().Format(time.RFC3339)) + return err + }) + b.Hold = time.Since(holdStart) + if err != nil { + b.Resolved = 0 + return b, fmt.Errorf("write batch: %w", err) + } + return b, nil +} diff --git a/cmd/ingestor/resolved_path_backfill_188_test.go b/cmd/ingestor/resolved_path_backfill_188_test.go new file mode 100644 index 000000000..cd8417c72 --- /dev/null +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -0,0 +1,630 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// Issue #188 point 4: backfill of NULL observations.resolved_path. + +// backfillFixture188 opens a store at path with two relays sharing the +// 1-byte prefix "c3" and an observer whose neighbour edge picks c3a, so a +// non-advert flood observation with path ["c3"] resolves to c3a. +func backfillFixture188(t *testing.T, path string, seed bool) *Store { + t.Helper() + store, err := OpenStore(path) + if err != nil { + t.Fatal(err) + } + store.backfillWg.Wait() + if seed { + for _, pk := range []string{c3a, c3b} { + if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater')`, pk, pk[:4]); err != nil { + t.Fatal(err) + } + } + if err := store.UpsertObserver(strings.ToUpper(obs188), "observer-188", "", nil); err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec(`INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, '2026-06-01T00:00:00Z')`, obs188, c3a); err != nil { + t.Fatal(err) + } + } + return store +} + +// primeIndexAndGraph188 does what StartNeighborEdgesBuilder's warm-up does +// before the backfill may run: prime the index, build neighbor_edges, and +// publish the post-build graph. +func primeIndexAndGraph188(t *testing.T, store *Store) { + t.Helper() + if err := store.RefreshPrefixIndex(); err != nil { + t.Fatal(err) + } + if _, err := store.buildAndPersistNeighborEdges(); err != nil { + t.Fatal(err) + } + if err := store.refreshBuiltNeighborGraph(); err != nil { + t.Fatal(err) + } +} + +// seedNullRows188 writes n transmissions with one observation each, path +// ["c3"], resolved_path NULL: rows stored before #188. Returns their ids. +func seedNullRows188(t testing.TB, store *Store, n int) []int64 { + t.Helper() + var obsIdx int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, strings.ToUpper(obs188)).Scan(&obsIdx); err != nil { + t.Fatal(err) + } + var base int + store.db.QueryRow(`SELECT COUNT(*) FROM transmissions`).Scan(&base) + tx, err := store.db.Begin() + if err != nil { + t.Fatal(err) + } + ids := make([]int64, 0, n) + for i := 0; i < n; i++ { + res, err := tx.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES ('00', ?, '2026-06-01T00:00:00Z', 1, 5, '{}')`, + fmt.Sprintf("h188-%d", base+i)) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + res, err = tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, '["c3"]', 1780272000)`, txID, obsIdx) + if err != nil { + t.Fatal(err) + } + id, _ := res.LastInsertId() + ids = append(ids, id) + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + return ids +} + +func resolvedPathOf188(t *testing.T, store *Store, id int64) sql.NullString { + t.Helper() + var rp sql.NullString + if err := store.db.QueryRow(`SELECT resolved_path FROM observations WHERE id = ?`, id).Scan(&rp); err != nil { + t.Fatal(err) + } + return rp +} + +func persistedWatermark188(t *testing.T, store *Store) int64 { + t.Helper() + w, err := store.resolvedPathBackfillWatermark() + if err != nil { + t.Fatal(err) + } + return w +} + +// Rows ingested before the prefix index is primed (the start-up window in +// main.go) are stored NULL; the backfill that starts after priming resolves +// them. +func TestResolvedPathBackfill_ResolvesRowsIngestedBeforePriming_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + if store.prefixIdx.load() != nil { + t.Fatal("setup: the prefix index is already primed") + } + pkt := &PacketData{RawHex: "c0ffee", Timestamp: "2026-06-01T00:00:00Z", ObserverID: strings.ToUpper(obs188), + Hash: "h-188-window", RouteType: 1, PayloadType: 5, PathJSON: `["c3"]`, DecodedJSON: "{}"} + if _, err := store.InsertTransmission(pkt); err != nil { + t.Fatal(err) + } + var id int64 + if err := store.db.QueryRow(`SELECT id FROM observations ORDER BY id DESC LIMIT 1`).Scan(&id); err != nil { + t.Fatal(err) + } + if rp := resolvedPathOf188(t, store, id); rp.Valid { + t.Fatalf("setup: row ingested before priming was resolved: %s", rp.String) + } + + primeIndexAndGraph188(t, store) + if _, err := store.RunResolvedPathBackfill(context.Background(), 100, 0); err != nil { + t.Fatal(err) + } + rp := resolvedPathOf188(t, store, id) + if !rp.Valid { + t.Fatal("the backfill left the start-up window row NULL") + } + wantPath188(t, unmarshalResolvedPathLocal(rp.String), c3a) +} + +func TestResolvedPathBackfill_BatchBoundAndWatermark_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + ids := seedNullRows188(t, store, 25) + primeIndexAndGraph188(t, store) + if err := ensureResolvedPathBackfillState(store.db); err != nil { + t.Fatal(err) + } + ceiling := ids[19] // rows above the ceiling arrived after the pass started + + b, err := store.resolvedPathBackfillBatch(context.Background(), 0, ceiling, 10) + if err != nil { + t.Fatal(err) + } + if b.Scanned != 10 || b.Resolved != 10 || b.Last != ids[9] { + t.Fatalf("first batch = %+v, want 10 scanned, 10 resolved, last id %d", b, ids[9]) + } + if w := persistedWatermark188(t, store); w != ids[9] { + t.Fatalf("persisted watermark = %d, want %d", w, ids[9]) + } + for i, id := range ids { + if got, want := resolvedPathOf188(t, store, id).Valid, i < 10; got != want { + t.Fatalf("row %d resolved = %v after one batch of 10, want %v", i, got, want) + } + } + + b, err = store.resolvedPathBackfillBatch(context.Background(), b.Last, ceiling, 10) + if err != nil { + t.Fatal(err) + } + if b.Scanned != 10 || b.Last != ids[19] { + t.Fatalf("second batch = %+v, want 10 scanned up to id %d", b, ids[19]) + } + b, err = store.resolvedPathBackfillBatch(context.Background(), b.Last, ceiling, 10) + if err != nil || b.Scanned != 0 { + t.Fatalf("batch past the ceiling = %+v, %v; want nothing", b, err) + } + for _, id := range ids[20:] { + if resolvedPathOf188(t, store, id).Valid { + t.Fatalf("row %d above the ceiling was touched", id) + } + } +} + +// Re-running never rewrites a row: only NULL rows are updated, and a row +// that already has a resolved_path is left as it is. +func TestResolvedPathBackfill_Idempotent_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + ids := seedNullRows188(t, store, 12) + const kept = `["kept-by-ingest"]` + if _, err := store.db.Exec(`UPDATE observations SET resolved_path = ? WHERE id = ?`, kept, ids[3]); err != nil { + t.Fatal(err) + } + primeIndexAndGraph188(t, store) + + first, err := store.RunResolvedPathBackfill(context.Background(), 5, 0) + if err != nil { + t.Fatal(err) + } + if first.Scanned != 12 || first.Resolved != 11 || first.Batches != 3 { + t.Fatalf("first pass = %+v, want 12 scanned, 11 resolved, 3 batches", first) + } + before := map[int64]string{} + for _, id := range ids { + before[id] = resolvedPathOf188(t, store, id).String + } + if before[ids[3]] != kept { + t.Fatalf("a row that already had a resolved_path was rewritten to %s", before[ids[3]]) + } + + // Forget the watermark so the same rows are scanned again. + if _, err := store.db.Exec(`DELETE FROM resolved_path_backfill_state`); err != nil { + t.Fatal(err) + } + second, err := store.RunResolvedPathBackfill(context.Background(), 5, 0) + if err != nil { + t.Fatal(err) + } + if second.Scanned != 12 || second.Resolved != 0 { + t.Fatalf("second pass = %+v, want 12 scanned, 0 resolved", second) + } + for _, id := range ids { + if got := resolvedPathOf188(t, store, id).String; got != before[id] { + t.Fatalf("row %d changed on re-run: %s -> %s", id, before[id], got) + } + } + // A pass with nothing new does no work. + third, err := store.RunResolvedPathBackfill(context.Background(), 5, 0) + if err != nil || third.Scanned != 0 { + t.Fatalf("third pass = %+v, %v; want nothing to do", third, err) + } +} + +// After a restart the pass resumes at the persisted watermark: rows below it +// are not scanned again, rows above it are. +func TestResolvedPathBackfill_ResumesAfterRestart_188(t *testing.T) { + path := filepath.Join(t.TempDir(), "ingest.db") + store := backfillFixture188(t, path, true) + ids := seedNullRows188(t, store, 30) + primeIndexAndGraph188(t, store) + if err := ensureResolvedPathBackfillState(store.db); err != nil { + t.Fatal(err) + } + // The process stops after its first committed batch. + if _, err := store.resolvedPathBackfillBatch(context.Background(), 0, ids[29], 10); err != nil { + t.Fatal(err) + } + // Mark a row below the watermark: a rescan would resolve it again. + if _, err := store.db.Exec(`UPDATE observations SET resolved_path = NULL WHERE id = ?`, ids[0]); err != nil { + t.Fatal(err) + } + store.Close() + + store = backfillFixture188(t, path, false) + defer store.Close() + primeIndexAndGraph188(t, store) + res, err := store.RunResolvedPathBackfill(context.Background(), 10, 0) + if err != nil { + t.Fatal(err) + } + if res.From != ids[9] || res.Scanned != 20 || res.Resolved != 20 { + t.Fatalf("resumed pass = %+v, want from %d, 20 scanned, 20 resolved", res, ids[9]) + } + if resolvedPathOf188(t, store, ids[0]).Valid { + t.Fatal("a row below the persisted watermark was scanned again") + } + for _, id := range ids[10:] { + if !resolvedPathOf188(t, store, id).Valid { + t.Fatalf("row %d above the watermark is still NULL", id) + } + } + if w := persistedWatermark188(t, store); w != ids[29] { + t.Fatalf("final watermark = %d, want %d", w, ids[29]) + } +} + +// Stopping the background pass keeps the last committed watermark. +func TestResolvedPathBackfill_StopKeepsWatermark_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + ids := seedNullRows188(t, store, 50) + primeIndexAndGraph188(t, store) + stop := store.StartResolvedPathBackfill(10, time.Hour) // first batch, then a long pause + deadline := time.Now().Add(5 * time.Second) + for { + var w sql.NullInt64 + store.db.QueryRow(`SELECT watermark FROM resolved_path_backfill_state WHERE id = 1`).Scan(&w) + if w.Valid { + break + } + if time.Now().After(deadline) { + t.Fatal("the background pass committed no batch") + } + time.Sleep(10 * time.Millisecond) + } + stop() + if w := persistedWatermark188(t, store); w != ids[9] { + t.Fatalf("watermark after stop = %d, want %d (one batch)", w, ids[9]) + } +} + +// resolvedPathBackfillHoldBudget is the stated per-batch budget for the write +// transaction at the default batch size (500 rows). It bounds how long a live +// InsertTransmission can wait behind one batch. +const resolvedPathBackfillHoldBudget = 250 * time.Millisecond + +func TestResolvedPathBackfill_WriteHoldUnderBudget_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + seedNullRows188(t, store, 5000) + primeIndexAndGraph188(t, store) + res, err := store.RunResolvedPathBackfill(context.Background(), defaultResolvedPathBackfillBatchSize, 0) + if err != nil { + t.Fatal(err) + } + if res.Resolved != 5000 || res.Batches != 10 { + t.Fatalf("pass = %+v, want 5000 resolved in 10 batches", res) + } + t.Logf("max write hold per %d-row batch: %s (budget %s)", defaultResolvedPathBackfillBatchSize, res.MaxHold, resolvedPathBackfillHoldBudget) + if res.MaxHold > resolvedPathBackfillHoldBudget { + t.Fatalf("max write hold %s exceeds the budget %s", res.MaxHold, resolvedPathBackfillHoldBudget) + } +} + +func TestResolvedPathBackfillSettings_Defaults_188(t *testing.T) { + var c Config + if on, b, p := c.ResolvedPathBackfillSettings(); !on || b != 500 || p != 250*time.Millisecond { + t.Fatalf("defaults = %v, %d, %s", on, b, p) + } + c.ResolvedPathBackfill = &ResolvedPathBackfillConfig{BatchSize: 50, PauseMs: 1000} + if on, b, p := c.ResolvedPathBackfillSettings(); !on || b != 50 || p != time.Second { + t.Fatalf("overrides = %v, %d, %s", on, b, p) + } + c.ResolvedPathBackfill = &ResolvedPathBackfillConfig{Disabled: true} + if on, _, _ := c.ResolvedPathBackfillSettings(); on { + t.Fatal("disabled config still enabled") + } +} + +// A pass that started before the prefix index and graph were primed would +// move the watermark past rows it could not resolve, and they would never be +// retried. It must refuse to run and leave the watermark alone. +func TestResolvedPathBackfill_RefusesBeforePriming_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + ids := seedNullRows188(t, store, 5) + if _, err := store.RunResolvedPathBackfill(context.Background(), 10, 0); err == nil { + t.Fatal("the pass ran with no prefix index") + } + var n int + store.db.QueryRow(`SELECT COUNT(*) FROM resolved_path_backfill_state`).Scan(&n) + if n != 0 { + t.Fatal("the unprimed pass persisted a watermark") + } + primeIndexAndGraph188(t, store) + res, err := store.RunResolvedPathBackfill(context.Background(), 10, 0) + if err != nil || res.Resolved != len(ids) { + t.Fatalf("pass after priming = %+v, %v; want %d resolved", res, err, len(ids)) + } +} + +// PR #190 review, finding 1: the pass must use a neighbour graph loaded after +// a real neighbor_edges build, and must not move its watermark over rows it +// failed to resolve because the graph was empty. + +// clearEdges188 empties neighbor_edges: a fresh or restored DB, or one whose +// edges were pruned. +func clearEdges188(t *testing.T, store *Store) { + t.Helper() + if _, err := store.db.Exec(`DELETE FROM neighbor_edges`); err != nil { + t.Fatal(err) + } +} + +// seedEdgeSourceRow188 adds an observation with the 2-byte path ["c355"]. +// It names c3a uniquely, so a neighbor_edges build derives the edge +// observer <-> c3a from it, and that edge resolves the 1-byte ["c3"] rows. +func seedEdgeSourceRow188(t *testing.T, store *Store) { + t.Helper() + var obsIdx int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, strings.ToUpper(obs188)).Scan(&obsIdx); err != nil { + t.Fatal(err) + } + res, err := store.db.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES ('00', 'h188-edge-source', '2026-06-01T00:00:00Z', 1, 5, '{}')`) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + if _, err := store.db.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, '["c355"]', 1780272000)`, txID, obsIdx); err != nil { + t.Fatal(err) + } +} + +// backfillWatermarkOrZero188 is the persisted watermark, or 0 when the pass +// never created its state table. +func backfillWatermarkOrZero188(t *testing.T, store *Store) int64 { + t.Helper() + var n int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'resolved_path_backfill_state'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n == 0 { + return 0 + } + return persistedWatermark188(t, store) +} + +func wantAllResolvedTo188(t *testing.T, store *Store, ids []int64, want string) { + t.Helper() + for _, id := range ids { + rp := resolvedPathOf188(t, store, id) + if !rp.Valid { + t.Fatalf("row %d is still NULL", id) + } + wantPath188(t, unmarshalResolvedPathLocal(rp.String), want) + } +} + +// main.go starts the pass right after StartNeighborEdgesBuilder returns. The +// graph the pass uses must be the one loaded after the builder's warm-up +// build, not the snapshot loaded before it: on a DB with empty +// neighbor_edges that snapshot is empty. +func TestResolvedPathBackfill_UsesGraphFromFirstEdgeBuild_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + clearEdges188(t, store) + ids := seedNullRows188(t, store, 10) + seedEdgeSourceRow188(t, store) + + stop := store.StartNeighborEdgesBuilder(time.Hour) + defer stop() + if _, err := store.RunResolvedPathBackfill(context.Background(), 100, 0); err != nil { + t.Fatal(err) + } + wantAllResolvedTo188(t, store, ids, c3a) +} + +// A neighbour edge build that finds no edges leaves an empty graph. A pass on +// it resolves only unique prefixes, so it must refuse to run and keep its +// watermark; once a build finds edges, the same rows resolve. +func TestResolvedPathBackfill_EmptyGraphKeepsWatermark_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + clearEdges188(t, store) + ids := seedNullRows188(t, store, 5) // "c3" is ambiguous: the build derives no edge from it + + stop := store.StartNeighborEdgesBuilder(time.Hour) + if _, err := store.RunResolvedPathBackfill(context.Background(), 10, 0); err == nil { + t.Fatal("the pass ran on an empty neighbour graph") + } + // A batch on its own must not commit a watermark either. + if err := ensureResolvedPathBackfillState(store.db); err != nil { + t.Fatal(err) + } + if _, err := store.resolvedPathBackfillBatch(context.Background(), 0, ids[len(ids)-1], 10); err == nil { + t.Fatal("a batch ran on an empty neighbour graph") + } + if w := backfillWatermarkOrZero188(t, store); w != 0 { + t.Fatalf("watermark moved to %d on an empty graph", w) + } + stop() + + // Edges appear (here inserted directly; normally a later build derives them). + if _, err := store.db.Exec(`INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, '2026-06-01T00:00:00Z')`, obs188, c3a); err != nil { + t.Fatal(err) + } + stop = store.StartNeighborEdgesBuilder(time.Hour) + defer stop() + res, err := store.RunResolvedPathBackfill(context.Background(), 10, 0) + if err != nil || res.Resolved != len(ids) { + t.Fatalf("pass after the edges appeared = %+v, %v; want %d resolved", res, err, len(ids)) + } + wantAllResolvedTo188(t, store, ids, c3a) +} + +// The background pass started before the first edge build (the order is a +// caller's choice, and #141 rewrites main.go) waits for that build instead of +// running on the snapshot it finds, then runs. +func TestResolvedPathBackfill_StartWaitsForFirstEdgeBuild_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + clearEdges188(t, store) + // An edge persisted earlier, so the pre-build graph is not empty: only + // the missing build holds the pass back. + if _, err := store.db.Exec(`INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, '2026-05-01T00:00:00Z')`, a1a, b2a); err != nil { + t.Fatal(err) + } + ids := seedNullRows188(t, store, 10) + seedEdgeSourceRow188(t, store) + // The state before the warm-up build: index and graph loaded. + if err := store.RefreshPrefixIndex(); err != nil { + t.Fatal(err) + } + if err := store.RefreshNeighborGraph(); err != nil { + t.Fatal(err) + } + + stopBackfill := store.StartResolvedPathBackfill(100, time.Millisecond) + defer stopBackfill() + time.Sleep(200 * time.Millisecond) + if w := backfillWatermarkOrZero188(t, store); w != 0 { + t.Fatalf("the pass committed watermark %d before the first edge build", w) + } + + stopBuilder := store.StartNeighborEdgesBuilder(time.Hour) + defer stopBuilder() + deadline := time.Now().Add(5 * time.Second) + for !resolvedPathOf188(t, store, ids[len(ids)-1]).Valid { + if time.Now().After(deadline) { + t.Fatal("the pass did not run after the first edge build") + } + time.Sleep(10 * time.Millisecond) + } + wantAllResolvedTo188(t, store, ids, c3a) +} + +func TestResolvedPathBackfillReady_188(t *testing.T) { + idx := idx188(c3a, c3b) + g := graph188([2]string{obs188, c3a}) + store := &Store{} + if store.resolvedPathBackfillReady(idx, g) { + t.Fatal("ready before any post-build graph was published") + } + store.neighborGraph.storeBuilt(g) + for _, c := range []struct { + name string + idx prefixIndex + graph *NeighborGraph + want bool + }{ + {"index and graph", idx, g, true}, + {"nil index", nil, g, false}, + {"empty index", prefixIndex{}, g, false}, + {"nil graph", idx, nil, false}, + {"empty graph", idx, NewNeighborGraph(), false}, + } { + if got := store.resolvedPathBackfillReady(c.idx, c.graph); got != c.want { + t.Errorf("%s: ready = %v, want %v", c.name, got, c.want) + } + } +} + +// PR #190 review, finding 2: 0 (or a negative value) means "default" for +// both numbers, so pauseMs cannot turn the pause off; the smallest pause is +// 1 ms. config.example.json documents exactly this. +func TestResolvedPathBackfillSettings_ZeroMeansDefault_188(t *testing.T) { + for _, c := range []struct { + cfg ResolvedPathBackfillConfig + wantBatch int + wantPause time.Duration + }{ + {ResolvedPathBackfillConfig{BatchSize: 0, PauseMs: 0}, 500, 250 * time.Millisecond}, + {ResolvedPathBackfillConfig{BatchSize: -1, PauseMs: -1}, 500, 250 * time.Millisecond}, + {ResolvedPathBackfillConfig{BatchSize: 1, PauseMs: 1}, 1, time.Millisecond}, + } { + cfg := Config{ResolvedPathBackfill: &c.cfg} + if on, b, p := cfg.ResolvedPathBackfillSettings(); !on || b != c.wantBatch || p != c.wantPause { + t.Errorf("%+v: settings = %v, %d, %s; want true, %d, %s", c.cfg, on, b, p, c.wantBatch, c.wantPause) + } + } +} + +// config.example.json documents the resolvedPathBackfill block with its +// defaults. +func TestConfigExample_ResolvedPathBackfill_188(t *testing.T) { + data, err := os.ReadFile(filepath.Join("..", "..", "config.example.json")) + if err != nil { + t.Fatal(err) + } + var cfg Config + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatal(err) + } + if cfg.ResolvedPathBackfill == nil { + t.Fatal("config.example.json has no resolvedPathBackfill block") + } + if on, b, p := cfg.ResolvedPathBackfillSettings(); !on || b != defaultResolvedPathBackfillBatchSize || p != defaultResolvedPathBackfillPause { + t.Fatalf("example settings = %v, %d, %s; want the defaults", on, b, p) + } + var raw map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + t.Fatal(err) + } + var block map[string]json.RawMessage + if err := json.Unmarshal(raw["resolvedPathBackfill"], &block); err != nil { + t.Fatal(err) + } + for _, k := range []string{"disabled", "batchSize", "pauseMs", "_comment"} { + if _, ok := block[k]; !ok { + t.Errorf("resolvedPathBackfill in config.example.json lacks %q", k) + } + } +} + +// If the warm-up build fails, no post-build graph is published, and the +// first successful tick publishes it instead. Here the warm-up fails because +// neighbor_edges is missing; the table comes back before the next tick. +func TestNeighborEdgesBuilder_TickPublishesBuiltGraphAfterFailedWarmUp_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + if _, err := store.db.Exec(`ALTER TABLE neighbor_edges RENAME TO neighbor_edges_away`); err != nil { + t.Fatal(err) + } + stop := store.StartNeighborEdgesBuilder(50 * time.Millisecond) + defer stop() + if built, _ := store.neighborGraph.buildState(); built { + t.Fatal("a failed warm-up build published a post-build graph") + } + if _, err := store.db.Exec(`ALTER TABLE neighbor_edges_away RENAME TO neighbor_edges`); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + for { + if built, _ := store.neighborGraph.buildState(); built { + break + } + if time.Now().After(deadline) { + t.Fatal("no tick published a post-build graph") + } + time.Sleep(10 * time.Millisecond) + } + if !store.neighborGraph.load().IsAdjacent(obs188, c3a) { + t.Fatal("the published graph lacks the persisted edge") + } +} diff --git a/cmd/ingestor/resolved_path_mutants_188_test.go b/cmd/ingestor/resolved_path_mutants_188_test.go new file mode 100644 index 000000000..8f14944bf --- /dev/null +++ b/cmd/ingestor/resolved_path_mutants_188_test.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "fmt" + "path/filepath" + "strings" + "testing" +) + +// PR #190 review, finding 5: tests that kill the mutants the review found +// surviving the whole ingestor suite. + +// PR #190 review, finding 5 (mutant m9): the backward walk breaks on a hop it +// cannot resolve. Hop 1 is ambiguous (both b2 nodes neighbour c3a); hop 0 +// must not be anchored on c3a, two hops away, even though exactly one a1 +// node neighbours it. +func TestObserverAnchor_BackwardWalkBreaksOnNilHop_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}, [2]string{c3a, b2a}, [2]string{c3a, b2b}, [2]string{c3a, a1a}) + got := resolveObservationPath([]string{"a1", "b2", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, "", "", c3a) +} + +// PR #190 review, finding 5 (mutant m10, reported as practically +// equivalent): the backward walk excludes the nodes it has already resolved. +// Path c3b -> b2a -> c3a -> observer. Hop 0 has the prefix "c3" again; both +// c3 nodes neighbour b2a, but c3a is already hop 2, so c3b is the one +// candidate left. Without the exclusion hop 0 stays nil. +func TestObserverAnchor_BackwardWalkExcludesResolvedHops_188(t *testing.T) { + idx := idx188(allNodes188...) + g := graph188([2]string{obs188, c3a}, [2]string{c3a, b2a}, [2]string{b2a, c3b}) + got := resolveObservationPath([]string{"c3", "b2", "c3"}, "", obs188, routeFlood188, g, idx) + wantPath188(t, got, c3b, b2a, c3a) +} + +// Mutant m12: the backfill passes transmissions.from_pubkey as the forward +// anchor only for ADVERTs, as InsertTransmission does (buildPacketData sets +// FromPubkey only for ADVERTs). Two rows with the same path ["a1"] and the +// same from_pubkey, whose neighbour edge would pick a1a: the ADVERT resolves, +// the GRP_TXT row stays NULL. The observer has no a1 neighbour, so it +// anchors nothing. +func TestResolvedPathBackfill_FromPubkeyOnlyForAdverts_188(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + for _, pk := range []string{a1a, a1b} { + if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater')`, pk, pk[:4]); err != nil { + t.Fatal(err) + } + } + if _, err := store.db.Exec(`INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, '2026-06-01T00:00:00Z')`, a1a, from188); err != nil { + t.Fatal(err) + } + var obsIdx int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, strings.ToUpper(obs188)).Scan(&obsIdx); err != nil { + t.Fatal(err) + } + ids := map[int]int64{} + for _, payloadType := range []int{int(payloadADVERT), 5} { + res, err := store.db.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) VALUES ('00', ?, '2026-06-01T00:00:00Z', 1, ?, '{}', ?)`, + fmt.Sprintf("h188-from-%d", payloadType), payloadType, from188) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + res, err = store.db.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, '["a1"]', 1780272000)`, txID, obsIdx) + if err != nil { + t.Fatal(err) + } + ids[payloadType], _ = res.LastInsertId() + } + primeIndexAndGraph188(t, store) + if _, err := store.RunResolvedPathBackfill(context.Background(), 10, 0); err != nil { + t.Fatal(err) + } + advert := resolvedPathOf188(t, store, ids[int(payloadADVERT)]) + if !advert.Valid { + t.Fatal("setup: the ADVERT row was not resolved from its from_pubkey") + } + wantPath188(t, unmarshalResolvedPathLocal(advert.String), a1a) + if rp := resolvedPathOf188(t, store, ids[5]); rp.Valid { + t.Fatalf("the GRP_TXT row was resolved from from_pubkey: %s", rp.String) + } +} diff --git a/cmd/ingestor/resolved_path_test.go b/cmd/ingestor/resolved_path_test.go index 308868506..b8c4be092 100644 --- a/cmd/ingestor/resolved_path_test.go +++ b/cmd/ingestor/resolved_path_test.go @@ -102,8 +102,9 @@ func TestInsertTransmissionWritesResolvedPath(t *testing.T) { defer store.Close() // Seed nodes with unique 1-byte prefixes. + // Hop nodes need a relay role: the prefix index holds relays only (#188). if _, err := store.db.Exec( - `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + `INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater'), (?, ?, 'repeater')`, "aaaaaaaaaa", "from-node", "bbbbbbbbbb", "first-hop", ); err != nil { @@ -380,8 +381,9 @@ func TestInsertTransmissionDoesNotClobberResolvedPathOnAllNil(t *testing.T) { } defer store.Close() + // Hop nodes need a relay role: the prefix index holds relays only (#188). if _, err := store.db.Exec( - `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + `INSERT INTO nodes (public_key, name, role) VALUES (?, ?, 'repeater'), (?, ?, 'repeater')`, "aaaaaaaaaa", "from-node", "bbbbbbbbbb", "first-hop", ); err != nil { diff --git a/config.example.json b/config.example.json index b339be9fa..8cfd4e5f9 100644 --- a/config.example.json +++ b/config.example.json @@ -26,6 +26,12 @@ "_comment": "vacuumOnStartup: run one-time full VACUUM to enable incremental auto-vacuum on existing DBs. Executed by the INGESTOR at startup, BEFORE the MQTT subscriber starts (#1283), so there is no contention with concurrent writes. Blocks ingestor startup for minutes on large DBs; requires 2x DB file size in free disk space. incrementalVacuumPages: free pages returned to OS after each retention reaper cycle (default 1024). See #919. load.chunkSize: see nested _comment (#1009).", "_comment_slowWriterMs": "#1340 — SQLite writer-lock log threshold (default 500). Any wrapped writer call (tagged neighbor_builder, mqtt_handler, prune_packets, prune_observers, prune_metrics, vacuum) whose hold_ms exceeds this emits a single [db-slow-writer] log line. Configured per-process via the CORESCOPE_DB_SLOW_WRITER_MS environment variable on the INGESTOR (e.g. CORESCOPE_DB_SLOW_WRITER_MS=200 for tighter alerting). Per-component wait_ms / hold_ms / contention_total histograms are surfaced via /api/perf/write-sources under .writer_perf regardless of this threshold." }, + "resolvedPathBackfill": { + "disabled": false, + "batchSize": 500, + "pauseMs": 250, + "_comment": "Optional (ingestor, #188). Once per ingestor start, re-resolves observations stored with resolved_path = NULL, from a persisted watermark up to the newest observation at the start of the pass. It waits until a neighbour-edge build has caught up with the stored observations and found at least one edge, then works in batches: each batch reads batchSize rows, resolves them outside any lock, and writes the results and the new watermark in one short write transaction, then sleeps pauseMs. Rows that still do not resolve stay NULL and are not retried. disabled: true skips the pass. batchSize: rows per batch, default 500. pauseMs: milliseconds between batches, default 250, which caps the pass at about 2,000 rows/s. For both numbers, 0 or omitted means the default, so the pause cannot be turned off; the smallest pause is 1. The server sees backfilled values only after its next restart." + }, "listLimits": { "packetsMax": 10000, "nodesMax": 2000, diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 06666691e..65d4e780e 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -114,6 +114,16 @@ How long (in hours) before a node is marked degraded or silent: See [Database](database.md) for details on SQLite auto-vacuum, WAL, and manual maintenance. See [#919](https://github.com/Kpa-clawbot/CoreScope/issues/919) for background. +### Resolved-path backfill (ingestor) + +Once per ingestor start, observations stored with `resolved_path = NULL` are resolved again in small batches. The pass waits until the neighbour-edge build has caught up with the stored observations. The server sees the new values after its next restart. + +| Field | Default | Description | +|-------|---------|-------------| +| `resolvedPathBackfill.disabled` | `false` | Skip the pass | +| `resolvedPathBackfill.batchSize` | `500` | Rows per batch; `0` means the default | +| `resolvedPathBackfill.pauseMs` | `250` | Milliseconds between batches; `0` means the default, so the pause cannot be turned off (minimum `1`) | + ## Channel decryption | Field | Description |