From dd7126ffd284629b76a18c7c7938c27cc2cafc71 Mon Sep 17 00:00:00 2001 From: efiten Date: Thu, 15 Oct 2026 15:23:58 +0200 Subject: [PATCH] fix(observers): an observer's own transmissions are not receptions (#667) Observations with direction "tx" are packets the observer's own radio sent. A repeater-observer appends its own hash before forwarding a flood, so the tx copy ends on the observer itself, and its own advert goes out with an empty path. Counted as receptions, the node detail "Heard By" card listed the observer as having heard itself, and the neighbor graph counted each forwarded advert twice (once per rx and tx copy). Skip tx observations in buildDirectHeardIndex, the in-memory neighbor graph build and the ingestor's neighbor_edges builder. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/neighbor_builder.go | 1 + cmd/ingestor/neighbor_builder_test.go | 52 +++++++++++++++++++++++++++ cmd/server/direct_heard.go | 2 +- cmd/server/direct_heard_test.go | 26 ++++++++++++++ cmd/server/neighbor_graph.go | 3 ++ cmd/server/neighbor_graph_test.go | 34 ++++++++++++++++++ cmd/server/store.go | 5 +++ 7 files changed, 122 insertions(+), 1 deletion(-) diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index 7a49ecc9a..c66d4fc27 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -198,6 +198,7 @@ func (s *Store) buildAndPersistNeighborEdges(trust *packetpath.TrustConfig) (int JOIN transmissions t ON t.id = o.transmission_id LEFT JOIN observers obs ON obs.rowid = o.observer_idx WHERE o.timestamp > ? + AND o.direction IS NOT 'tx' -- the observer's own transmission, not a reception (#667) ORDER BY o.timestamp LIMIT ?`, watermarkEpoch, neighborBuilderMaxBatch) if err != nil { diff --git a/cmd/ingestor/neighbor_builder_test.go b/cmd/ingestor/neighbor_builder_test.go index 5cd777df9..a78c09369 100644 --- a/cmd/ingestor/neighbor_builder_test.go +++ b/cmd/ingestor/neighbor_builder_test.go @@ -332,3 +332,55 @@ func TestNeighborEdgesBuilderPathTrustAllowsTwoByte(t *testing.T) { t.Fatalf("2-byte hop must still produce the edge, got %d", got) } } + +// TestNeighborEdgesBuilderSkipsOwnTransmissions pins #667: an observation +// with direction 'tx' is the observer's own radio transmitting, not a +// reception, so it must not produce an edge. +func TestNeighborEdgesBuilderSkipsOwnTransmissions(t *testing.T) { + store, err := OpenStore(filepath.Join(t.TempDir(), "tx.db")) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + defer store.Close() + + if _, err := store.db.Exec( + `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + "aaaaaaaaaa", "from-node", + "bbbbbbbbbb", "first-hop", + ); err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec(`INSERT INTO observers (id, name) VALUES (?, ?)`, "obs-1", "observer-1"); err != nil { + t.Fatal(err) + } + var obsRowid int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, "obs-1").Scan(&obsRowid); err != nil { + t.Fatal(err) + } + res, err := store.db.Exec( + `INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, from_pubkey) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + "", "h-tx", "2026-01-01T00:00:00Z", 0, payloadADVERT, 0, "{}", "aaaaaaaaaa", + ) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + if _, err := store.db.Exec( + `INSERT INTO observations (transmission_id, observer_idx, direction, path_json, timestamp) VALUES (?, ?, 'tx', ?, ?)`, + txID, obsRowid, `["bb"]`, int64(1735689600), + ); err != nil { + t.Fatal(err) + } + + if _, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()); err != nil { + t.Fatalf("buildAndPersistNeighborEdges: %v", err) + } + var got int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM neighbor_edges`).Scan(&got); err != nil { + t.Fatal(err) + } + if got != 0 { + t.Fatalf("a tx observation produced %d neighbor_edges rows, want 0", got) + } +} diff --git a/cmd/server/direct_heard.go b/cmd/server/direct_heard.go index 0fa3e01c3..b99408670 100644 --- a/cmd/server/direct_heard.go +++ b/cmd/server/direct_heard.go @@ -157,7 +157,7 @@ func buildDirectHeardIndex(packets []*StoreTx, pm *prefixMap) directHeardIndex { idx := make(directHeardIndex, 256) for _, tx := range packets { for _, obs := range tx.Observations { - if obs.ObserverID == "" { + if obs.ObserverID == "" || obs.isOwnTransmission() { continue } pk := directHeardNode(tx, obs, pm) diff --git a/cmd/server/direct_heard_test.go b/cmd/server/direct_heard_test.go index b5101ec38..d93702513 100644 --- a/cmd/server/direct_heard_test.go +++ b/cmd/server/direct_heard_test.go @@ -308,3 +308,29 @@ func TestBuildDirectObserverRows_CanRelayTriState(t *testing.T) { t.Fatalf("repeater must report can_relay true, got %v", got["repeater"]) } } + +// #667: a "tx" observation is the observer's own radio transmitting, not a +// reception. When a repeater-observer forwards a flood it appends its own +// hash first (firmware Mesh.cpp), so the tx copy ends on the observer itself, +// and its own advert goes out with an empty path. Counted as receptions, both +// made the observer the top entry of its own "Heard By" card. +func TestBuildDirectHeardIndex_SkipsOwnTransmissions(t *testing.T) { + forwarded := dhTx(RouteFlood, PayloadTXT_MSG, "") + forwarded.Observations = []*StoreObs{ + {ObserverID: dhUnique, ObserverName: "self", Direction: "tx", PathJSON: `["A433","BB11"]`}, + {ObserverID: dhUnique, ObserverName: "self", Direction: "rx", PathJSON: `["A433"]`}, + } + ownAdvert := dhTx(RouteFlood, PayloadADVERT, `{"type":"ADVERT","pubKey":"`+dhUnique+`"}`) + ownAdvert.Observations = []*StoreObs{ + {ObserverID: dhUnique, ObserverName: "self", Direction: "tx", PathJSON: `[]`}, + } + + idx := buildDirectHeardIndex([]*StoreTx{forwarded, ownAdvert}, buildPrefixMap(directHeardNodes)) + + if got := idx[dhUnique]; len(got) != 0 { + t.Fatalf("own transmissions credited as receptions: %#v", got) + } + if a := idx[dhCollideA][dhUnique]; a == nil || a.Count != 1 { + t.Fatalf("rx copy of the forwarded packet = %#v, want Count 1", a) + } +} diff --git a/cmd/server/neighbor_graph.go b/cmd/server/neighbor_graph.go index 19ecc63bf..155e3db98 100644 --- a/cmd/server/neighbor_graph.go +++ b/cmd/server/neighbor_graph.go @@ -318,6 +318,9 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG } for _, obs := range tx.Observations { + if obs.isOwnTransmission() { + continue + } path := parsePathJSON(obs.PathJSON) observerPK := cachedToLower(lowerCache, obs.ObserverID) diff --git a/cmd/server/neighbor_graph_test.go b/cmd/server/neighbor_graph_test.go index 5f332fc25..8020be57d 100644 --- a/cmd/server/neighbor_graph_test.go +++ b/cmd/server/neighbor_graph_test.go @@ -1019,3 +1019,37 @@ func TestBuildNeighborGraph_CountsByMode(t *testing.T) { t.Errorf("CountsByMode[3] = %d, want 1", got) } } + +// #667: the tx copy of a packet the observer forwarded carries the same +// originator↔path[0] evidence as its rx copy, so counting it doubles the edge. +// A tx row with an empty path is not a zero-hop hearing either. +func TestBuildNeighborGraph_SkipsOwnTransmissions(t *testing.T) { + nodes := []nodeInfo{ + {Role: "repeater", PublicKey: "aaaa1111", Name: "NodeX"}, + {Role: "repeater", PublicKey: "r1aabbcc", Name: "R1"}, + {Role: "repeater", PublicKey: "obs00001", Name: "Observer"}, + } + rx := ngMakeObs("obs00001", `["r1aa"]`, nowStr, nil) + rx.Direction = "rx" + tx := ngMakeObs("obs00001", `["r1aa","obs0"]`, nowStr, nil) + tx.Direction = "tx" + emptyTx := ngMakeObs("obs00001", `[]`, nowStr, nil) + emptyTx.Direction = "tx" + + store := ngTestStore(nodes, []*StoreTx{ + ngMakeTx(1, 4, ngFromNodeJSON("aaaa1111"), []*StoreObs{rx, tx}), + ngMakeTx(2, 4, ngFromNodeJSON("aaaa1111"), []*StoreObs{emptyTx}), + }) + g := BuildFromStore(store) + + for _, e := range g.AllEdges() { + if (e.NodeA == "aaaa1111" && e.NodeB == "r1aabbcc") || (e.NodeA == "r1aabbcc" && e.NodeB == "aaaa1111") { + if e.Count != 1 { + t.Errorf("X↔R1 count = %d, want 1 (the tx copy must not count)", e.Count) + } + } + if e.NodeA == "aaaa1111" && e.NodeB == "obs00001" || e.NodeA == "obs00001" && e.NodeB == "aaaa1111" { + t.Errorf("empty-path tx row produced a zero-hop edge X↔Observer") + } + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 3c8ffbc18..ce5cfb4dc 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -75,6 +75,11 @@ type StoreTx struct { accountedBytes int64 } +// isOwnTransmission reports whether the observation is the observer's own +// radio transmitting (direction "tx" from the MQTT bridge) rather than a +// reception. Aggregates of who heard whom must skip it (#667). +func (o *StoreObs) isOwnTransmission() bool { return o.Direction == "tx" } + // StoreObs is a lean in-memory observation (no duplication of transmission fields). type StoreObs struct { ID int