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 9e407dfc6..9429a638f 100644 --- a/cmd/server/neighbor_graph.go +++ b/cmd/server/neighbor_graph.go @@ -328,6 +328,9 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG } for _, obs := range p.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 959289caa..194ca96c7 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 a21758220..80cb8156e 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