From bbba9a7800adb4a5b5b2371d730ac43c898d69fa Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 10:47:59 +0200 Subject: [PATCH 01/17] feat(ingestor): resolve the last hop of a flood path from the observer (#188) A non-advert observation has no anchor for its first hop, so with 1-byte hashes (about 7 nodes per prefix on staging) every hop stayed nil and the row was stored as NULL. The observer is a better anchor: a flood forwarder appends its hash before retransmitting (MeshCore Mesh.cpp routeRecvPacket) and the observer logs the packet before its own routing step (Dispatcher.cpp logRx), so the last hash is the node it heard, a direct neighbour. resolveObservationPath keeps the forward chain from fromPubkey and, for flood route types only, adds a backward chain: the last hop is the unique neighbour of the observer, each earlier hop the unique neighbour of the hop after it. The merge keeps every forward hop and fills nil hops from the backward chain; if the two disagree on a hop, or would place one node twice, the backward result is dropped for the row. No geo, GPS or count tie-break. DIRECT and TRACE paths are not anchored. path_resolver.go is gofmt-ed as a whole (one comment table moved). Relates to #188, #184 Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/db.go | 8 +- cmd/ingestor/path_resolver.go | 118 ++++++- .../path_resolver_observer_188_test.go | 321 ++++++++++++++++++ 3 files changed, 439 insertions(+), 8 deletions(-) create mode 100644 cmd/ingestor/path_resolver_observer_188_test.go 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/path_resolver.go b/cmd/ingestor/path_resolver.go index 7ca94479f..92552ac79 100644 --- a/cmd/ingestor/path_resolver.go +++ b/cmd/ingestor/path_resolver.go @@ -134,12 +134,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 @@ -223,3 +224,108 @@ func (s *Store) RefreshNeighborGraph() error { s.neighborGraph.store(g) return nil } + +// Observer anchor (#188) +// +// Firmware basis (MeshCore src/): a flood forwarder appends its own hash +// before it retransmits (Mesh.cpp routeRecvPacket: copyHashTo(&path[n*sz])), +// and an observer logs a packet on reception, before its own routing step +// (Dispatcher.cpp: logRx(pkt, ...) runs before processRecvPacket). 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 removeSelfFromPath), nor for TRACE +// (the path carries SNR bytes), 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 +// (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. +func resolveObservationPath(hops []string, fromPubkey, observer string, routeType int, graph *NeighborGraph, idx prefixIndex) []*string { + fwd := resolvePathWithContext(hops, fromPubkey, graph, idx) + if graph == nil || idx == nil || observer == "" || !isFloodRoute(routeType) || !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. +func resolvePathBackward(hops []string, fromPubkey, observer string, graph *NeighborGraph, idx prefixIndex) []*string { + out := make([]*string, len(hops)) + anchor := strings.ToLower(observer) + seen := make(map[string]struct{}, len(hops)+1) + if fp := strings.ToLower(fromPubkey); fp != "" { + seen[fp] = struct{}{} + } + for i := len(hops) - 1; i >= 0; i-- { + r := resolveHopWithContext(hops[i], anchor, graph, idx, seen) + 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_observer_188_test.go b/cmd/ingestor/path_resolver_observer_188_test.go new file mode 100644 index 000000000..349258002 --- /dev/null +++ b/cmd/ingestor/path_resolver_observer_188_test.go @@ -0,0 +1,321 @@ +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), and a TRACE path carries SNR bytes. +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) +} From dc08e63bdd6b792c3854eccfc7cb174ef7776a85 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 10:52:39 +0200 Subject: [PATCH 02/17] feat(ingestor): build the hop prefix index from relay roles only (#188) The ingestor's prefix index held every node, so a prefix shared by one repeater and a companion counted as ambiguous, and a prefix unique to a companion resolved to it (64 companion pubkeys are stored as relays on staging). buildPrefixIndex now keeps only nodes for which isRelayRole is true: the server's canAppearInPath (repeater, room_server, room), with the same test cases. Companions and sensors ship with forwarding off in the firmware (companion_radio NodePrefs.h, simple_sensor SensorMesh.cpp); a companion that opted in to repeating stays unresolved, as on the server. The index also feeds the neighbour-edge builder, so edges to companions are no longer derived from a hop prefix. Existing fixtures that seeded hop nodes without a role now give them role 'repeater'; expectations are unchanged. Relates to #188, #184 Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/neighbor_builder.go | 32 ++++- cmd/ingestor/neighbor_builder_delta_test.go | 3 +- cmd/ingestor/neighbor_builder_test.go | 9 +- cmd/ingestor/prefix_index_relay_188_test.go | 127 ++++++++++++++++++++ cmd/ingestor/resolved_path_test.go | 6 +- 5 files changed, 165 insertions(+), 12 deletions(-) create mode 100644 cmd/ingestor/prefix_index_relay_188_test.go diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index 63cda0567..a571dbb20 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -336,11 +336,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 +348,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 +367,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_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/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_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 { From f1d408c3c8f38cb17473d00b2da3fdf1b211dc4c Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 10:57:16 +0200 Subject: [PATCH 03/17] feat(ingestor): backfill NULL resolved_path rows after the index is primed (#188) Rows stored before the observer anchor and the relay-only index, and rows drained from the start-up buffer before the prefix index is primed, keep resolved_path = NULL. RunResolvedPathBackfill re-resolves them once per ingestor start, with resolveObservationPath, for observation ids between the persisted watermark and MAX(id) at the start of the pass. Each batch reads at most batchSize rows by primary key without writerMu, resolves them outside any lock, and commits one WriterTx that sets resolved_path only where it is still NULL and moves the watermark (resolved_path_backfill_state, one row). A restart repeats at most the uncommitted batch. Batches are paced by a pause. Defaults: 500 rows, 250 ms, configurable as resolvedPathBackfill.{batchSize,pauseMs,disabled}. main.go starts the pass right after StartNeighborEdgesBuilder has primed the index and graph. The server indexes a row when it first polls it or at Load, so backfilled values reach its indexes on its next restart; the server is unchanged. Relates to #188, #184 Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/config.go | 4 + cmd/ingestor/main.go | 7 + cmd/ingestor/resolved_path_backfill.go | 282 +++++++++++++++ .../resolved_path_backfill_188_test.go | 335 ++++++++++++++++++ 4 files changed, 628 insertions(+) create mode 100644 cmd/ingestor/resolved_path_backfill.go create mode 100644 cmd/ingestor/resolved_path_backfill_188_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 493905dd6..6b8b4b8ca 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -103,6 +103,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/main.go b/cmd/ingestor/main.go index db2e816f3..454765777 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -488,6 +488,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/resolved_path_backfill.go b/cmd/ingestor/resolved_path_backfill.go new file mode 100644 index 000000000..79f724cd1 --- /dev/null +++ b/cmd/ingestor/resolved_path_backfill.go @@ -0,0 +1,282 @@ +package main + +import ( + "context" + "database/sql" + "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). +// +// 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. +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 +} + +// StartResolvedPathBackfill runs one backfill pass in the background. Call it +// after the prefix index and neighbour graph are primed +// (StartNeighborEdgesBuilder). 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) + } + }() + if _, err := s.RunResolvedPathBackfill(ctx, batchSize, pause); err != nil && ctx.Err() == nil { + log.Printf("[%s] error: %v", resolvedPathBackfillComponent, err) + } + }() + 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 + } + 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 + 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). + graph, idx := s.neighborGraph.load(), s.prefixIdx.load() + 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..19ac7366e --- /dev/null +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -0,0 +1,335 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "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 +} + +func primeIndexAndGraph188(t *testing.T, store *Store) { + t.Helper() + if err := store.RefreshPrefixIndex(); err != nil { + t.Fatal(err) + } + if err := store.RefreshNeighborGraph(); 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") + } +} From 1cfb42d91eedf9f2c2527989e2f0e27d9f672668 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 11:02:10 +0200 Subject: [PATCH 04/17] fix(ingestor): resolved_path backfill refuses to run before the index is primed (#188) Started before StartNeighborEdgesBuilder had primed the prefix index and graph, a pass would resolve nothing yet move its watermark past every row, so those rows would never be retried. RunResolvedPathBackfill now returns an error and leaves the watermark unchanged until both are loaded. Relates to #188 Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/resolved_path_backfill.go | 5 +++++ .../resolved_path_backfill_188_test.go | 22 +++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/cmd/ingestor/resolved_path_backfill.go b/cmd/ingestor/resolved_path_backfill.go index 79f724cd1..87ab707ca 100644 --- a/cmd/ingestor/resolved_path_backfill.go +++ b/cmd/ingestor/resolved_path_backfill.go @@ -132,6 +132,11 @@ func (s *Store) RunResolvedPathBackfill(ctx context.Context, batchSize int, paus if batchSize <= 0 { batchSize = defaultResolvedPathBackfillBatchSize } + // Without the index and graph every row would stay NULL while the + // watermark moved past it for good. + if s.prefixIdx.load() == nil || s.neighborGraph.load() == nil { + return res, fmt.Errorf("prefix index or neighbour graph not primed") + } if err := ensureResolvedPathBackfillState(s.db); err != nil { return res, fmt.Errorf("ensure state table: %w", err) } diff --git a/cmd/ingestor/resolved_path_backfill_188_test.go b/cmd/ingestor/resolved_path_backfill_188_test.go index 19ac7366e..8a918f90b 100644 --- a/cmd/ingestor/resolved_path_backfill_188_test.go +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -333,3 +333,25 @@ func TestResolvedPathBackfillSettings_Defaults_188(t *testing.T) { 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)) + } +} From f9bae200b21d624eb390a8a3d261d0d611eb0821 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:32:00 +0000 Subject: [PATCH 05/17] test(ingestor): reproduce the backfill running on a pre-build or empty graph (#188) PR #190 review, finding 1. StartNeighborEdgesBuilder loads the graph snapshot before its warm-up build, and the backfill only checks that the graph is non-nil. With empty neighbor_edges the pass resolves only unique prefixes and moves its watermark past every row for good. Three tests, red by assertion on 1cfb42d9: - the pass right after StartNeighborEdgesBuilder uses the pre-build graph; - a pass and a single batch on an empty graph commit a watermark; - the background pass started before the first build runs at once. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- .../resolved_path_backfill_188_test.go | 150 ++++++++++++++++++ 1 file changed, 150 insertions(+) diff --git a/cmd/ingestor/resolved_path_backfill_188_test.go b/cmd/ingestor/resolved_path_backfill_188_test.go index 8a918f90b..a9857d68f 100644 --- a/cmd/ingestor/resolved_path_backfill_188_test.go +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -355,3 +355,153 @@ func TestResolvedPathBackfill_RefusesBeforePriming_188(t *testing.T) { 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) + ids := seedNullRows188(t, store, 10) + seedEdgeSourceRow188(t, store) + // The state before the warm-up build: index and an (empty) 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) +} From cbaa9ca69679307ead8e7960524ac95a3e78ca69 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:44:37 +0000 Subject: [PATCH 06/17] fix(ingestor): backfill waits for a post-build, non-empty neighbour graph (#188) PR #190 review, finding 1. The backfill could run on the graph snapshot that StartNeighborEdgesBuilder loads before its warm-up build, or on an empty graph. It then resolved only unique prefixes and moved its watermark past every other row for good. - StartNeighborEdgesBuilder publishes the graph again after a successful warm-up build, and after each successful tick, as a post-build snapshot (neighborGraphHolder.storeBuilt). - The pass and every batch run only with a non-empty prefix index and a post-build graph that has at least one edge; otherwise they write nothing, not even the watermark (errResolvedPathBackfillNotReady). - StartResolvedPathBackfill waits for the next post-build graph and retries; the retry resumes at the persisted watermark. - Test setup primes the way the builder does: build, then publish. - TestResolvedPathBackfillReady_188 pins each readiness condition. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/neighbor_builder.go | 19 +++++- cmd/ingestor/path_resolver.go | 51 ++++++++++++++ cmd/ingestor/resolved_path_backfill.go | 67 ++++++++++++++++--- .../resolved_path_backfill_188_test.go | 41 +++++++++++- 4 files changed, 163 insertions(+), 15 deletions(-) diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index a571dbb20..5609a565b 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -79,10 +79,12 @@ 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) } + var wuErr error for { n, err := s.buildAndPersistNeighborEdges() if err != nil { log.Printf("[neighbor-build] initial build error: %v", err) + wuErr = err break } wuTotal += n @@ -91,6 +93,14 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { } } 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. If the warm-up failed, the first successful tick does it. + if wuErr == nil { + if err := s.refreshBuiltNeighborGraph(); err != nil { + log.Printf("[neighbor-build] post-build neighbor-graph refresh error: %v", err) + } + } var stopOnce sync.Once go func() { @@ -109,8 +119,13 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { n, err := s.buildAndPersistNeighborEdges() // 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 it is a post-build snapshot (#188). + refresh := s.RefreshNeighborGraph + if err == nil { + refresh = s.refreshBuiltNeighborGraph + } + if grErr := refresh(); grErr != nil { log.Printf("[neighbor-build] neighbor-graph refresh error: %v", grErr) } dur := time.Since(start) diff --git a/cmd/ingestor/path_resolver.go b/cmd/ingestor/path_resolver.go index 92552ac79..835018798 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 successful +// neighbor_edges build 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 successful neighbor_edges build, +// 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). @@ -225,6 +264,18 @@ func (s *Store) RefreshNeighborGraph() error { return nil } +// refreshBuiltNeighborGraph is RefreshNeighborGraph for a caller that has +// just completed a successful buildAndPersistNeighborEdges: 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/): a flood forwarder appends its own hash diff --git a/cmd/ingestor/resolved_path_backfill.go b/cmd/ingestor/resolved_path_backfill.go index 87ab707ca..a6a5a9d23 100644 --- a/cmd/ingestor/resolved_path_backfill.go +++ b/cmd/ingestor/resolved_path_backfill.go @@ -3,6 +3,7 @@ package main import ( "context" "database/sql" + "errors" "fmt" "log" "strings" @@ -15,6 +16,14 @@ import ( // 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 +// successful neighbor_edges build, 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: @@ -101,10 +110,22 @@ func (s *Store) resolvedPathBackfillWatermark() (int64, error) { return w, err } -// StartResolvedPathBackfill runs one backfill pass in the background. Call it -// after the prefix index and neighbour graph are primed -// (StartNeighborEdgesBuilder). The returned stop function cancels the pass and -// waits for it; the watermark of the last committed batch is kept. +// errResolvedPathBackfillNotReady: no prefix index, or no neighbour graph +// from a successful edge build, or that graph has no edge. +var errResolvedPathBackfillNotReady = errors.New("prefix index or neighbour graph not ready (no successful neighbour-edge build 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 a successful edge build 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{}) @@ -115,8 +136,26 @@ func (s *Store) StartResolvedPathBackfill(batchSize int, pause time.Duration) fu log.Printf("[%s] panic recovered: %v", resolvedPathBackfillComponent, r) } }() - if _, err := s.RunResolvedPathBackfill(ctx, batchSize, pause); err != nil && ctx.Err() == nil { - log.Printf("[%s] error: %v", resolvedPathBackfillComponent, err) + 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() { @@ -132,10 +171,10 @@ func (s *Store) RunResolvedPathBackfill(ctx context.Context, batchSize int, paus if batchSize <= 0 { batchSize = defaultResolvedPathBackfillBatchSize } - // Without the index and graph every row would stay NULL while the - // watermark moved past it for good. - if s.prefixIdx.load() == nil || s.neighborGraph.load() == nil { - return res, fmt.Errorf("prefix index or neighbour graph not primed") + // 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) @@ -200,6 +239,13 @@ type resolvedPathBackfillRow struct { // 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 @@ -232,7 +278,6 @@ func (s *Store) resolvedPathBackfillBatch(ctx context.Context, after, ceiling in // Resolve outside any lock, with the same inputs InsertTransmission uses // (fromPubkey only for ADVERTs, as buildPacketData sets it). - graph, idx := s.neighborGraph.load(), s.prefixIdx.load() type update struct { id int64 rp string diff --git a/cmd/ingestor/resolved_path_backfill_188_test.go b/cmd/ingestor/resolved_path_backfill_188_test.go index a9857d68f..dd69effe7 100644 --- a/cmd/ingestor/resolved_path_backfill_188_test.go +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -38,12 +38,18 @@ func backfillFixture188(t *testing.T, path string, seed bool) *Store { 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.RefreshNeighborGraph(); err != nil { + if _, err := store.buildAndPersistNeighborEdges(); err != nil { + t.Fatal(err) + } + if err := store.refreshBuiltNeighborGraph(); err != nil { t.Fatal(err) } } @@ -477,9 +483,14 @@ 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 an (empty) graph loaded. + // The state before the warm-up build: index and graph loaded. if err := store.RefreshPrefixIndex(); err != nil { t.Fatal(err) } @@ -505,3 +516,29 @@ func TestResolvedPathBackfill_StartWaitsForFirstEdgeBuild_188(t *testing.T) { } 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) + } + } +} From d1db161c2f0d19e797c3442746c8487866bc82fc Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:45:34 +0000 Subject: [PATCH 07/17] docs(config): document resolvedPathBackfill and its zero-means-default rule (#188) PR #190 review, finding 2. The three keys were missing from config.example.json. They now sit in a resolvedPathBackfill block with a _comment, like the other ingestor blocks, and in the configuration guide. pauseMs: 0 cannot switch the pause off, because 0 means the default. That is documented rather than changed: a pass with no pause at all would hold the single write connection back to back. The smallest pause is 1 ms. Tests: - TestConfigExample_ResolvedPathBackfill_188 parses config.example.json and checks the block and its defaults (red before the block was added); - TestResolvedPathBackfillSettings_ZeroMeansDefault_188 pins 0 and negative values to the defaults and 1 to 1 ms. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/resolved_path_backfill.go | 5 +- .../resolved_path_backfill_188_test.go | 54 +++++++++++++++++++ config.example.json | 6 +++ docs/user-guide/configuration.md | 10 ++++ 4 files changed, 74 insertions(+), 1 deletion(-) diff --git a/cmd/ingestor/resolved_path_backfill.go b/cmd/ingestor/resolved_path_backfill.go index a6a5a9d23..1a60bedb4 100644 --- a/cmd/ingestor/resolved_path_backfill.go +++ b/cmd/ingestor/resolved_path_backfill.go @@ -52,7 +52,10 @@ const ( defaultResolvedPathBackfillPause = 250 * time.Millisecond ) -// ResolvedPathBackfillConfig is the "resolvedPathBackfill" config block. +// 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 diff --git a/cmd/ingestor/resolved_path_backfill_188_test.go b/cmd/ingestor/resolved_path_backfill_188_test.go index dd69effe7..1654785b9 100644 --- a/cmd/ingestor/resolved_path_backfill_188_test.go +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -3,7 +3,9 @@ package main import ( "context" "database/sql" + "encoding/json" "fmt" + "os" "path/filepath" "strings" "testing" @@ -542,3 +544,55 @@ func TestResolvedPathBackfillReady_188(t *testing.T) { } } } + +// 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) + } + } +} diff --git a/config.example.json b/config.example.json index a5446416c..c70083394 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 for the first successful neighbour-edge build with 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 89b61eaed..2e0a0a2d9 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 for the first neighbour-edge build. 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 | From 0c623dbafc36bea4721f83eab7f79f8b87b46cfe Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:46:50 +0000 Subject: [PATCH 08/17] test(ingestor): kill the resolver and backfill mutants that survived (#188) PR #190 review, finding 5. Three mutants survived the whole ingestor suite. Each test passes on the code and fails on its mutant: - m9, the backward walk keeps its anchor across a nil hop: TestObserverAnchor_BackwardWalkBreaksOnNilHop_188 (the reviewer's [a1 b2 c3] probe, hop 1 ambiguous). - m10, the backward walk without its exclusion set. It was reported as practically equivalent, but it is not: when a prefix repeats, the exclusion leaves one candidate where there would be two. TestObserverAnchor_BackwardWalkExcludesResolvedHops_188. - m12, the backfill passes from_pubkey for non-ADVERT rows: TestResolvedPathBackfill_FromPubkeyOnlyForAdverts_188. Today only ADVERTs carry from_pubkey, so on real data this mutant has no effect. The test keeps the backfill consistent with InsertTransmission if that changes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- .../resolved_path_mutants_188_test.go | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) create mode 100644 cmd/ingestor/resolved_path_mutants_188_test.go 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) + } +} From f5add6787febbaed873c9a9fbc7782a6846f0604 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:47:26 +0000 Subject: [PATCH 09/17] test(ingestor): a tick publishes the post-build graph after a failed warm-up (#188) Follow-up to the finding-1 fix. If the warm-up build fails, the backfill waits for the first successful tick. The mutant "a tick never publishes a post-build graph" survived the targeted suite; this test kills it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- .../resolved_path_backfill_188_test.go | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/cmd/ingestor/resolved_path_backfill_188_test.go b/cmd/ingestor/resolved_path_backfill_188_test.go index 1654785b9..cd8417c72 100644 --- a/cmd/ingestor/resolved_path_backfill_188_test.go +++ b/cmd/ingestor/resolved_path_backfill_188_test.go @@ -596,3 +596,35 @@ func TestConfigExample_ResolvedPathBackfill_188(t *testing.T) { } } } + +// 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") + } +} From e610dbb845fcc24f33645dac8e3e9a257d23ae20 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:47:26 +0000 Subject: [PATCH 10/17] test(ingestor): reproduce the observer resolved as its own last hop (#188) PR #190 review, finding 4. In a flood path an observer reports, the last hop is the node it heard, never itself. With the observer's prefix unique, both chains name the observer: path ["0b"] resolves to obs188. With path ["c3","0b"], the backward walk then anchors hop 0 on it. TestObserverAnchor_ObserverIsNeverItsOwnLastHop_188 is red by assertion. TestObserverAnchor_ObserverMayBeAnEarlierHop_188 is green and pins the scope: an observer logs a packet before de-duplication, so it can report an echo of a flood it forwarded, with its own hash earlier in the path. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- .../path_resolver_observer_188_test.go | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/cmd/ingestor/path_resolver_observer_188_test.go b/cmd/ingestor/path_resolver_observer_188_test.go index 349258002..05d0013f4 100644 --- a/cmd/ingestor/path_resolver_observer_188_test.go +++ b/cmd/ingestor/path_resolver_observer_188_test.go @@ -319,3 +319,41 @@ func TestInsertTransmission_ObserverAnchorResolvesLastHop_188(t *testing.T) { } 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) +} From 503ea758aebe166f2d2e1dab706136864f6244b7 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:49:25 +0000 Subject: [PATCH 11/17] fix(ingestor): the observer is never its own last hop in a flood path (#188) PR #190 review, finding 4. A radio does not receive its own transmission, so the last hop of a flood path an observer reports is never the observer. Both chains now leave that hop nil instead: - the forward chain (resolvePathForward, the walk behind resolvePathWithContext) does not take the observer as a candidate for the last hop of a flood path. With a unique prefix, the forward chain alone named the observer, so a fix in the backward walk alone would not change the stored row; - the backward walk excludes the observer from the last hop only. It may still be an earlier hop: it logs a packet before de-duplication (Dispatcher.cpp logRx before processRecvPacket), so it can report the echo of a flood it forwarded. DIRECT routes and resolvePathWithContext's other callers are unchanged. Perf (hot path, InsertTransmission): BenchmarkResolveObservationPath_188 is new, because the PR's benchmark was not in the repo. It uses 700 relays, 60 observers and 1-5 hop 1-byte flood paths; one op is one observation. Run interleaved, n=12 each, on the same machine: - before: median 3,875 ns/op, 119 B/op, 5 allocs/op; - after: median 3,890 ns/op, 119 B/op, 5 allocs/op. That is within noise. The change adds one map insert per flood observation. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/path_resolver.go | 47 ++++++++++--- cmd/ingestor/path_resolver_bench_188_test.go | 72 ++++++++++++++++++++ 2 files changed, 110 insertions(+), 9 deletions(-) create mode 100644 cmd/ingestor/path_resolver_bench_188_test.go diff --git a/cmd/ingestor/path_resolver.go b/cmd/ingestor/path_resolver.go index 835018798..47acdd863 100644 --- a/cmd/ingestor/path_resolver.go +++ b/cmd/ingestor/path_resolver.go @@ -226,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 } @@ -234,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 { @@ -307,13 +317,22 @@ func isFloodRoute(routeType int) bool { // resolveObservationPath resolves one observation's hops for // observations.resolved_path. It runs the forward chain from fromPubkey -// (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. +// (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 { - fwd := resolvePathWithContext(hops, fromPubkey, graph, idx) - if graph == nil || idx == nil || observer == "" || !isFloodRoute(routeType) || !hasNil(fwd) { + 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)) @@ -323,15 +342,25 @@ func resolveObservationPath(hops []string, fromPubkey, observer string, routeTyp // 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 := strings.ToLower(observer) - seen := make(map[string]struct{}, len(hops)+1) + 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{}{} 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) + } +} From 2e2886af5dd741cc3aa5a5ec5a0edceeacd8b0e9 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 13:50:18 +0000 Subject: [PATCH 12/17] docs(ingestor): correct the TRACE note and firmware references (#188) PR #190 review, finding 8. A TRACE's header path carries SNR bytes, but the ingestor's decoder replaces the hops with the planned route from the payload. TRACE is not anchored because it is always DIRECT: sendFlood refuses it. The comment now says so. Firmware references are checked against MeshCore a366955, the version cloned here; the PR cited 0679dbe: - routeRecvPacket is at Mesh.cpp:344-356, with copyHashTo at :349; - removeSelfFromPath is at :334; - the front-of-path match is at :89; - the SNR append is at :60-61; - logRx is at Dispatcher.cpp:238, before processRecvPacket at :246/:256. Comments only. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/path_resolver.go | 22 +++++++++++-------- .../path_resolver_observer_188_test.go | 3 ++- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/cmd/ingestor/path_resolver.go b/cmd/ingestor/path_resolver.go index 47acdd863..981a5e8b3 100644 --- a/cmd/ingestor/path_resolver.go +++ b/cmd/ingestor/path_resolver.go @@ -288,15 +288,19 @@ func (s *Store) refreshBuiltNeighborGraph() error { // Observer anchor (#188) // -// Firmware basis (MeshCore src/): a flood forwarder appends its own hash -// before it retransmits (Mesh.cpp routeRecvPacket: copyHashTo(&path[n*sz])), -// and an observer logs a packet on reception, before its own routing step -// (Dispatcher.cpp: logRx(pkt, ...) runs before processRecvPacket). 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 removeSelfFromPath), nor for TRACE -// (the path carries SNR bytes), so only flood route types are anchored. +// 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 diff --git a/cmd/ingestor/path_resolver_observer_188_test.go b/cmd/ingestor/path_resolver_observer_188_test.go index 05d0013f4..e9967a325 100644 --- a/cmd/ingestor/path_resolver_observer_188_test.go +++ b/cmd/ingestor/path_resolver_observer_188_test.go @@ -119,7 +119,8 @@ func TestObserverAnchor_TwoNeighborCandidatesStayNil_188(t *testing.T) { // 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), and a TRACE path carries SNR bytes. +// 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}) From e369ae69146a08751b5a9cc7ef00a660a8ae51d3 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 15:12:49 +0000 Subject: [PATCH 13/17] test(ingestor): reproduce the warm-up stopping after a full batch with few edges (#188) PR #190, second review, P2-1. The warm-up loop in StartNeighborEdgesBuilder stops once buildAndPersistNeighborEdges returns fewer than neighborBuilderMaxBatch edges, but each call reads up to that many observations. A full batch that yields few edges ends the warm-up with observations left unscanned. The backfill then runs on that partial graph and moves its watermark past rows it could not resolve. Two tests, red by assertion: - one full batch (50,000 rows) with a single edge, and the edge the rows need in the second batch: the warm-up stops early; - a full batch with no edge newer than the watermark, so the build cannot move past it: the backfill runs anyway, because the graph holds an older edge. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- .../neighbor_builder_review_190_test.go | 124 ++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 cmd/ingestor/neighbor_builder_review_190_test.go 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..c62b439f0 --- /dev/null +++ b/cmd/ingestor/neighbor_builder_review_190_test.go @@ -0,0 +1,124 @@ +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"]`}) + + stop := store.StartNeighborEdgesBuilder(time.Hour) + defer stop() + 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") + } +} From da0494c481f62f754b6f4a7e43c36210ae0871ca Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 15:15:27 +0000 Subject: [PATCH 14/17] fix(ingestor): end the edge-build warm-up on rows scanned, not edges (#188) PR #190, second review, P2-1. buildAndPersistNeighborEdges reads at most neighborBuilderMaxBatch observations per call, but the warm-up loop stopped once a call returned fewer edges than that. A full batch with few edges ended the warm-up with observations unscanned, and the backfill then ran on that partial graph. - buildNeighborEdges reports both the edge upserts and the observation rows read; buildAndPersistNeighborEdges keeps its old result for existing callers. - The warm-up continues while a call reads a full batch. It stops when a call reads fewer rows: it has caught up. - A full batch with no edge cannot move the watermark (MAX(last_seen)), and the next call would read the same rows. The warm-up then stops, logs it, and leaves the graph unpublished. - Only a build that caught up publishes a post-build graph, in the warm-up and in a tick. The backfill therefore never runs on a graph the builder has not finished, and waits instead. - Comments, config.example.json and the guide now say "caught up". TestNeighborEdgesBuilder_WarmUpScansPastFullBatchWithFewEdges_190 and TestResolvedPathBackfill_WaitsWhileEdgeBuildCannotCatchUp_190 (red in e369ae69) pass. The second one now also lets ticks run. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/neighbor_builder.go | 74 ++++++++++++++----- .../neighbor_builder_review_190_test.go | 11 ++- cmd/ingestor/path_resolver.go | 8 +- cmd/ingestor/resolved_path_backfill.go | 9 ++- config.example.json | 2 +- docs/user-guide/configuration.md | 2 +- 6 files changed, 76 insertions(+), 30 deletions(-) diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index 5609a565b..7db91b01a 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -79,24 +79,35 @@ 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) } - var wuErr error + // + // 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) - wuErr = 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. If the warm-up failed, the first successful tick does it. - if wuErr == nil { + // 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) } @@ -116,13 +127,15 @@ 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. After a - // successful build it is a post-build snapshot (#188). + // successful build that caught up, it is a post-build + // snapshot (#188). refresh := s.RefreshNeighborGraph - if err == nil { + if err == nil && b.caughtUp() { refresh = s.refreshBuiltNeighborGraph } if grErr := refresh(); grErr != nil { @@ -152,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. +// 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 @@ -176,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 @@ -188,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 != "" { @@ -211,12 +245,13 @@ 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 decodedJSON, fromPubkey, pathJSON, observerID string var epochTs int64 @@ -289,7 +324,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 @@ -319,9 +354,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 diff --git a/cmd/ingestor/neighbor_builder_review_190_test.go b/cmd/ingestor/neighbor_builder_review_190_test.go index c62b439f0..049fefc9f 100644 --- a/cmd/ingestor/neighbor_builder_review_190_test.go +++ b/cmd/ingestor/neighbor_builder_review_190_test.go @@ -110,8 +110,17 @@ func TestResolvedPathBackfill_WaitsWhileEdgeBuildCannotCatchUp_190(t *testing.T) ids := seedObservations190(t, store, 0, repeatPath190(`["c3"]`, neighborBuilderMaxBatch)) seedObservations190(t, store, int64(neighborBuilderMaxBatch), []string{`["c355"]`}) - stop := store.StartNeighborEdgesBuilder(time.Hour) + // 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") } diff --git a/cmd/ingestor/path_resolver.go b/cmd/ingestor/path_resolver.go index 981a5e8b3..3e78196ee 100644 --- a/cmd/ingestor/path_resolver.go +++ b/cmd/ingestor/path_resolver.go @@ -97,8 +97,8 @@ func (g *NeighborGraph) empty() bool { // path. atomic.Value lets the 60s rebuild publish without a read-side // lock. // -// It also records whether a snapshot loaded after a successful -// neighbor_edges build has been published (storeBuilt). Before that, the +// 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 { @@ -120,7 +120,7 @@ func (h *neighborGraphHolder) store(g *NeighborGraph) { h.v.Store(g) } -// storeBuilt publishes g, loaded after a successful neighbor_edges build, +// 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) @@ -275,7 +275,7 @@ func (s *Store) RefreshNeighborGraph() error { } // refreshBuiltNeighborGraph is RefreshNeighborGraph for a caller that has -// just completed a successful buildAndPersistNeighborEdges: it marks the +// 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) diff --git a/cmd/ingestor/resolved_path_backfill.go b/cmd/ingestor/resolved_path_backfill.go index 1a60bedb4..7463a9715 100644 --- a/cmd/ingestor/resolved_path_backfill.go +++ b/cmd/ingestor/resolved_path_backfill.go @@ -17,7 +17,8 @@ import ( // ingest is drained before the index is primed (main.go). // // The pass needs a prefix index and a neighbour graph loaded after a -// successful neighbor_edges build, with at least one edge +// 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 @@ -114,8 +115,8 @@ func (s *Store) resolvedPathBackfillWatermark() (int64, error) { } // errResolvedPathBackfillNotReady: no prefix index, or no neighbour graph -// from a successful edge build, or that graph has no edge. -var errResolvedPathBackfillNotReady = errors.New("prefix index or neighbour graph not ready (no successful neighbour-edge build yet, or no edges)") +// 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. @@ -125,7 +126,7 @@ func (s *Store) resolvedPathBackfillReady(idx prefixIndex, graph *NeighborGraph) } // StartResolvedPathBackfill runs one backfill pass in the background. Until -// a neighbour graph from a successful edge build with at least one edge is +// 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. diff --git a/config.example.json b/config.example.json index c55f0c289..8cfd4e5f9 100644 --- a/config.example.json +++ b/config.example.json @@ -30,7 +30,7 @@ "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 for the first successful neighbour-edge build with 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." + "_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, diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index cfdddd15d..65d4e780e 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -116,7 +116,7 @@ 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 for the first neighbour-edge build. The server sees the new values after its next restart. +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 | |-------|---------|-------------| From addccb8831a77ac1661a03204470b3222ed0e028 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 15:16:19 +0000 Subject: [PATCH 15/17] test(ingestor): reproduce observer edges built from DIRECT paths (#188) PR #190, second review, P2-2. The neighbour builder adds an observer <-> last-hop edge for every observation, whatever its route type. For a DIRECT packet the path is the remaining planned route: forwarders strip themselves from the front (Mesh.cpp:89, removeSelfFromPath :334-342), and only flood forwarders append their hash (routeRecvPacket, :346-350). The last hop of a DIRECT path is therefore the route's far end, not the node the observer heard. TestNeighborEdgesBuilder_ObserverEdgeOnlyForFloodRoutes_190 mixes flood, transport flood, DIRECT and TRANSPORT_DIRECT observations. It is red by assertion: there are edges from DIRECT paths, and the flood resolver leaves an ambiguous last hop nil because of the false edge. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- .../neighbor_builder_review_190_test.go | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/cmd/ingestor/neighbor_builder_review_190_test.go b/cmd/ingestor/neighbor_builder_review_190_test.go index 049fefc9f..e56603b1e 100644 --- a/cmd/ingestor/neighbor_builder_review_190_test.go +++ b/cmd/ingestor/neighbor_builder_review_190_test.go @@ -131,3 +131,69 @@ func TestResolvedPathBackfill_WaitsWhileEdgeBuildCannotCatchUp_190(t *testing.T) 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, `["c355"]`}, // transport flood: same edge + {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) + } + if !g.IsAdjacent(obs188, c3a) { + t.Fatal("no observer<->last-hop edge from the flood observations") + } + 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) +} From 194512d2fe5ffbb2784a0a9a37760fdd77079d95 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 15:18:17 +0000 Subject: [PATCH 16/17] fix(ingestor): build observer<->last-hop edges from flood routes only (#188) PR #190, second review, P2-2. On a DIRECT route the path is the remaining planned route: - each forwarder matches itself at the front and strips itself (MeshCore a366955, Mesh.cpp:89, removeSelfFromPath :334-342); - only flood forwarders append their hash (routeRecvPacket :346-350). So the last hop of a DIRECT path is the far end of the route, not the node the observer heard. The edge it produced was false, and the flood resolver relies on those edges to pick a unique candidate. The builder now reads transmissions.route_type and adds the observer edge only for TRANSPORT_FLOOD (0) and FLOOD (1). An unknown route type (NULL) is skipped as well. Other edges are unchanged: the originator<->first-hop edge of ADVERTs, and the interior hop-to-hop edges. Consecutive hops of a DIRECT route are neighbours. This changes neighbor_edges for every consumer: the Neighbors panel, bridge nodes and the estimates built on them. The server reads the table and has no other production edge source; its extractEdgesFromObs has the same rule but no caller outside tests. DIRECT edges already persisted stay until the neighbour-edge prune drops them (5 days by default). TestNeighborEdgesBuilder_ObserverEdgeOnlyForFloodRoutes_190 (red in the previous commit) passes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/neighbor_builder.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index 7db91b01a..c72e69da2 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -186,7 +186,7 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { // buildNeighborEdges scans transmissions + observations, // extracts edge candidates (originator↔first-hop on ADVERTs; -// observer↔last-hop on all packet types) and upserts them into +// 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. // @@ -233,6 +233,7 @@ func (s *Store) buildNeighborEdges() (neighborEdgesBuild, 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, ''), @@ -253,9 +254,10 @@ func (s *Store) buildNeighborEdges() (neighborEdgesBuild, error) { 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) @@ -278,7 +280,14 @@ func (s *Store) buildNeighborEdges() (neighborEdgesBuild, 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)) From 0f4c167a91dedf3e8a2339c93aabe1afc7621f37 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 3 Oct 2026 15:18:32 +0000 Subject: [PATCH 17/17] test(ingestor): pin the TRANSPORT_FLOOD observer edge separately (#188) In the P2-2 test the TRANSPORT_FLOOD row used the same path as the FLOOD row, so a builder that dropped route type 0 still passed. The row now names its own relay (a1a), and the test expects both observer edges. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QbNdgonR8kq7SPEU7LcmXD --- cmd/ingestor/neighbor_builder_review_190_test.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/cmd/ingestor/neighbor_builder_review_190_test.go b/cmd/ingestor/neighbor_builder_review_190_test.go index e56603b1e..554ac0e77 100644 --- a/cmd/ingestor/neighbor_builder_review_190_test.go +++ b/cmd/ingestor/neighbor_builder_review_190_test.go @@ -157,7 +157,7 @@ func TestNeighborEdgesBuilder_ObserverEdgeOnlyForFloodRoutes_190(t *testing.T) { path string }{ {routeFlood188, `["c355"]`}, // flood: observer heard c3a - {routeTransportFlood188, `["c355"]`}, // transport flood: same edge + {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 } { @@ -178,8 +178,10 @@ func TestNeighborEdgesBuilder_ObserverEdgeOnlyForFloodRoutes_190(t *testing.T) { if err != nil { t.Fatal(err) } - if !g.IsAdjacent(obs188, c3a) { - t.Fatal("no observer<->last-hop edge from the flood observations") + 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) {