diff --git a/cmd/server/analytics_recompute_after_load_test.go b/cmd/server/analytics_recompute_after_load_test.go index 0c4eac04..b2a54cb5 100644 --- a/cmd/server/analytics_recompute_after_load_test.go +++ b/cmd/server/analytics_recompute_after_load_test.go @@ -221,8 +221,8 @@ func TestAnalyticsRecomputers_PostLoadOrder(t *testing.T) { for i, rc := range list { pos[rc.name] = i } - if len(list) != 10 || len(pos) != 10 { - t.Fatalf("want 10 distinct recomputers, got %d (%d distinct)", len(list), len(pos)) + if len(list) != 11 || len(pos) != 11 { + t.Fatalf("want 11 distinct recomputers, got %d (%d distinct)", len(list), len(pos)) } for _, name := range []string{"rf", "topology", "channels"} { if pos[name] > 2 { diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go index 3cf30c20..db6b5dbf 100644 --- a/cmd/server/analytics_recomputer.go +++ b/cmd/server/analytics_recomputer.go @@ -259,6 +259,7 @@ func (s *PacketStore) analyticsRecomputersLocked() []*analyticsRecomputer { s.recompObserversClockSkew, s.recompNodesClockSkew, s.recompRoles, s.recompRetransmissions, + s.recompDirectHeard, } } @@ -338,6 +339,18 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o return s.computeRetransmissionPressure("", TimeWindow{}, retransmissionDefaultBucket) }, ) + // Feeds the node-health "Heard By" card. Not an analytics endpoint, + // but it has the same shape: one full pass over the store that no + // request can afford, served from an atomic snapshot. See + // direct_heard.go. + s.recompDirectHeard = newAnalyticsRecomputer( + "direct-heard", defaultInterval, + func() interface{} { + idx := s.computeDirectHeard() + s.publishDirectHeard(idx) + return idx + }, + ) all := s.analyticsRecomputersLocked() s.analyticsRecomputerMu.Unlock() diff --git a/cmd/server/direct_heard.go b/cmd/server/direct_heard.go new file mode 100644 index 00000000..07e95b12 --- /dev/null +++ b/cmd/server/direct_heard.go @@ -0,0 +1,273 @@ +// Package main: direct-RF attribution for the node-health "Heard By" card. +// +// An observer "heard" a node when it received that node's own transmission +// off the air. That is a narrower relation than "saw a packet this node was +// involved in", which is what the node-health card reported before this +// file existed, and it is the only one for which the SNR and RSSI printed +// next to an observer belong to the node the row names. +// +// The rule follows the firmware: +// +// - Flood routes build the path up as they travel: a forwarding repeater +// appends its own hash before retransmitting (Mesh.cpp:349). The last +// hop is therefore the node whose transmission the observer received. +// - Direct routes carry the REMAINING route, not the travelled one. A +// forwarder matches itself against the head of the path and calls +// removeSelfFromPath (Mesh.cpp:89,103) before retransmitting, so the +// node the observer heard is not in the path at all. Direct routes +// therefore never attribute. +// - An empty flood path means the observer received the originator's own +// transmission. Only ADVERTs carry the originator's pubkey in the +// clear, so other payload types with an empty path attribute to nobody. +// - A hop prefix that matches more than one node attributes to nobody. +// Path hop sizes are chosen by the originator (Packet.h:83, +// Mesh.cpp:649) and default to one byte, so a hop often matches many +// candidates. resolveWithContext guesses in that case; this file does +// not. The gate matches resolvePathForObsColdLoad: under-attribute +// rather than credit the wrong node. +package main + +import ( + "sort" + "strings" +) + +// directHeardAgg accumulates one observer's direct receptions of one node. +type directHeardAgg struct { + ObserverName string + Count int + SNRSum float64 + SNRCount int + RSSISum float64 + RSSICount int +} + +// directHeardIndex maps node pubkey (lowercase) to observer id to aggregate. +type directHeardIndex map[string]map[string]*directHeardAgg + +// HealthObserverRow is one row of the node-health "Heard By" table. +// +// Field names and null semantics are unchanged from the map-based rows this +// replaced, so the frontend needs no migration: avgSnr and avgRssi are null +// when the observer contributed no sample, and can_relay is tri-state (null +// = no repeat field ever reported, see PR #1624). +type HealthObserverRow struct { + ObserverID string `json:"observer_id"` + ObserverName string `json:"observer_name"` + AvgSNR *float64 `json:"avgSnr"` + AvgRSSI *float64 `json:"avgRssi"` + PacketCount int `json:"packetCount"` + CanRelay *bool `json:"can_relay"` +} + +// lastPathHop returns the last quoted token of a path_json array, or "" when +// there is none. It scans backwards rather than unmarshalling: the compute +// pass runs over every observation in the store (2.9M on the reference +// deployment) and only ever needs the final element. +func lastPathHop(pathJSON string) string { + end := -1 + for i := len(pathJSON) - 1; i >= 0; i-- { + if pathJSON[i] != '"' { + continue + } + if end < 0 { + end = i + continue + } + return pathJSON[i+1 : end] + } + return "" +} + +// advertOriginPubkey returns the pubkey an ADVERT announces, or "" when the +// transmission is not an ADVERT or carries no decodable pubkey. Mirrors the +// field probing in trackAdvertPubkey. +func advertOriginPubkey(tx *StoreTx) string { + if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT || tx.DecodedJSON == "" { + return "" + } + d := tx.ParsedDecoded() + if d == nil { + return "" + } + if v, ok := d["pubKey"].(string); ok && v != "" { + return strings.ToLower(v) + } + if v, ok := d["public_key"].(string); ok && v != "" { + return strings.ToLower(v) + } + return "" +} + +// directHeardNode returns the lowercase pubkey of the node whose +// transmission this observation received off the air, or "" when that cannot +// be established. See the package comment for the rule and its firmware +// grounding. +func directHeardNode(tx *StoreTx, obs *StoreObs, pm *prefixMap) string { + if tx == nil || obs == nil || tx.RouteType == nil { + return "" + } + switch *tx.RouteType { + case RouteFlood, RouteTransportFlood: + default: + return "" + } + hop := lastPathHop(obs.PathJSON) + if hop == "" { + return advertOriginPubkey(tx) + } + if pm == nil { + return "" + } + candidates := pm.relayCandidates(hop) + if len(candidates) != 1 { + return "" + } + return strings.ToLower(candidates[0].PublicKey) +} + +// buildDirectHeardIndex folds every observation of every transmission into +// the node-to-observer aggregate. Pure over its arguments so it can be +// tested and benchmarked without a store. +func buildDirectHeardIndex(packets []*StoreTx, pm *prefixMap) directHeardIndex { + idx := make(directHeardIndex, 256) + for _, tx := range packets { + for _, obs := range tx.Observations { + if obs.ObserverID == "" { + continue + } + pk := directHeardNode(tx, obs, pm) + if pk == "" { + continue + } + byObs := idx[pk] + if byObs == nil { + byObs = make(map[string]*directHeardAgg, 4) + idx[pk] = byObs + } + agg := byObs[obs.ObserverID] + if agg == nil { + agg = &directHeardAgg{ObserverName: obs.ObserverName} + byObs[obs.ObserverID] = agg + } + agg.Count++ + if obs.SNR != nil { + agg.SNRSum += *obs.SNR + agg.SNRCount++ + } + if obs.RSSI != nil { + agg.RSSISum += *obs.RSSI + agg.RSSICount++ + } + } + } + return idx +} + +// computeDirectHeard rebuilds the whole index from the current store. Run by +// a background recomputer rather than per request: the reference deployment +// holds 232,928 transmissions and 2,887,861 observations, and one node's +// byNode slice alone can hold 1.45M observations. +// +// Rebuilding wholesale also means eviction needs no bookkeeping — a pass +// simply does not see transmissions that are gone. +func (s *PacketStore) computeDirectHeard() directHeardIndex { + s.mu.RLock() + defer s.mu.RUnlock() + _, pm := s.getCachedNodesAndPM() + return buildDirectHeardIndex(s.packets, pm) +} + +// publishDirectHeard installs a snapshot for readers. Called by the +// recomputer after each pass; a nil-safe no-op for callers that have nothing +// to publish. +func (s *PacketStore) publishDirectHeard(idx directHeardIndex) { + if idx == nil { + idx = directHeardIndex{} + } + s.directHeardSnap.Store(idx) +} + +// loadDirectHeard returns the latest snapshot, or nil before the first +// compute has published one. A nil index simply yields empty direct rows: +// the card degrades to "nobody hears this node" until the first pass lands, +// never to a wrong attribution. +func (s *PacketStore) loadDirectHeard() directHeardIndex { + idx, _ := s.directHeardSnap.Load().(directHeardIndex) + return idx +} + +// canRelaySets fetches the two inputs behind the can_relay tri-state badge +// (#1290, PR #1624): the observers that reported repeat:off, and the +// observers we have any repeat field for at all. Both are lowercase to match +// pm.nonRelay and GetNonRelayObserverPubkeys; two case conventions on the +// same upstream string would be a latent regression. A read failure degrades +// to "no badge" rather than a wrong badge. +func (s *PacketStore) canRelaySets() (nonRelay, seen map[string]struct{}) { + nonRelay = map[string]struct{}{} + seen = map[string]struct{}{} + if s.db == nil || s.db.conn == nil { + return nonRelay, seen + } + if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil { + for _, pk := range pks { + nonRelay[strings.ToLower(pk)] = struct{}{} + } + } + if pks, err := s.db.GetCanRelaySeenObserverPubkeys(); err == nil { + for _, pk := range pks { + seen[strings.ToLower(pk)] = struct{}{} + } + } + return nonRelay, seen +} + +// relayOnlyObserverCount counts observers that saw traffic involving the node +// without hearing it on air. seenObservers is the set the health builders +// already collect from each transmission's representative observation. +func relayOnlyObserverCount(seenObservers map[string]struct{}, direct map[string]*directHeardAgg) int { + n := 0 + for id := range seenObservers { + if _, isDirect := direct[id]; !isDirect { + n++ + } + } + return n +} + +// buildDirectObserverRows renders one node's aggregate as sorted API rows. +// nonRelay and seen carry the can_relay tri-state; both may be nil. +func buildDirectObserverRows(byObs map[string]*directHeardAgg, nonRelay, seen map[string]struct{}) []HealthObserverRow { + rows := make([]HealthObserverRow, 0, len(byObs)) + for id, agg := range byObs { + row := HealthObserverRow{ + ObserverID: id, + ObserverName: agg.ObserverName, + PacketCount: agg.Count, + } + if agg.SNRCount > 0 { + v := agg.SNRSum / float64(agg.SNRCount) + row.AvgSNR = &v + } + if agg.RSSICount > 0 { + v := agg.RSSISum / float64(agg.RSSICount) + row.AvgRSSI = &v + } + idLower := strings.ToLower(id) + if _, ok := seen[idLower]; ok { + _, isListener := nonRelay[idLower] + canRelay := !isListener + row.CanRelay = &canRelay + } + rows = append(rows, row) + } + // Packet count descending, observer id ascending as a deterministic + // tiebreak so repeated requests return a stable order. + sort.Slice(rows, func(i, j int) bool { + if rows[i].PacketCount != rows[j].PacketCount { + return rows[i].PacketCount > rows[j].PacketCount + } + return rows[i].ObserverID < rows[j].ObserverID + }) + return rows +} diff --git a/cmd/server/direct_heard_test.go b/cmd/server/direct_heard_test.go new file mode 100644 index 00000000..dc56bb00 --- /dev/null +++ b/cmd/server/direct_heard_test.go @@ -0,0 +1,278 @@ +package main + +import ( + "testing" +) + +// Coverage for the direct-RF attribution rule behind the node detail +// "Heard By" card. +// +// The card used to credit every observer that saw traffic *involving* a +// node — originated by it, addressed to it, or relayed through it — and +// printed an SNR/RSSI next to each. Those signal numbers belong to +// whichever node last transmitted the copy the observer received, not to +// the node the row names. +// +// The rule these tests pin, derived from the firmware: +// +// - Only flood routes carry a travelled path. For ROUTE_TYPE_DIRECT the +// forwarder removes itself from the front before retransmitting +// (firmware Mesh.cpp:103 removeSelfFromPath), so path_json is the +// REMAINING route and says nothing about who was heard. +// - On a flood, the last hop is the node the observer heard on air +// (firmware Mesh.cpp:349 — a repeater appends its own hash before +// retransmitting). +// - An empty flood path means the originator was heard directly. Only +// ADVERTs carry the originator's pubkey in the clear. +// - An ambiguous hop prefix credits nobody. Same gate as +// resolvePathForObsColdLoad: under-attribute rather than guess. +// +// Fixture pubkeys are lowercase hex placeholders only (AGENTS.md PII rule). + +// directHeardNodes are three repeaters, two of which collide on the 1-byte +// prefix "a4" — the shape that produced the original report, where a 433 MHz +// repeater was credited with 868 MHz traffic it could not have relayed. +var directHeardNodes = []nodeInfo{ + {PublicKey: "a433ec0000000000000000000000000000000000000000000000000000000001", Role: "repeater", Name: "collideA"}, + {PublicKey: "a4ef4b0000000000000000000000000000000000000000000000000000000002", Role: "repeater", Name: "collideB"}, + {PublicKey: "bb11220000000000000000000000000000000000000000000000000000000003", Role: "repeater", Name: "unique"}, +} + +const ( + dhCollideA = "a433ec0000000000000000000000000000000000000000000000000000000001" + dhUnique = "bb11220000000000000000000000000000000000000000000000000000000003" +) + +func dhRoute(rt int) *int { return &rt } + +func dhTx(routeType int, payloadType int, decoded string) *StoreTx { + return &StoreTx{ + RouteType: dhRoute(routeType), + PayloadType: dhRoute(payloadType), + DecodedJSON: decoded, + } +} + +func TestLastPathHop(t *testing.T) { + cases := []struct { + name, in, want string + }{ + {"empty array", "[]", ""}, + {"empty string", "", ""}, + {"single hop", `["A4"]`, "A4"}, + {"two hops", `["A4","F1"]`, "F1"}, + {"two-byte hops", `["A433","1403"]`, "1403"}, + {"long path", `["66","E8","EA","DE","7C","CA"]`, "CA"}, + {"unterminated", `["A4`, ""}, + {"not json", "garbage", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := lastPathHop(tc.in); got != tc.want { + t.Fatalf("lastPathHop(%q) = %q, want %q", tc.in, got, tc.want) + } + }) + } +} + +func TestDirectHeardNode(t *testing.T) { + pm := buildPrefixMap(directHeardNodes) + advert := `{"type":"ADVERT","pubKey":"` + dhUnique + `"}` + + cases := []struct { + name string + tx *StoreTx + pathJSON string + want string + }{ + { + name: "flood, unique two-byte last hop, credits that node", + tx: dhTx(RouteFlood, PayloadTXT_MSG, ""), + pathJSON: `["A433","BB11"]`, + want: dhUnique, + }, + { + name: "transport flood counts as flood", + tx: dhTx(RouteTransportFlood, PayloadTXT_MSG, ""), + pathJSON: `["BB11"]`, + want: dhUnique, + }, + { + name: "flood, node is an earlier hop, not the last one, credits nobody here", + tx: dhTx(RouteFlood, PayloadTXT_MSG, ""), + pathJSON: `["BB11","A433"]`, + want: dhCollideA, + }, + { + name: "ambiguous one-byte last hop credits nobody", + tx: dhTx(RouteFlood, PayloadTXT_MSG, ""), + pathJSON: `["66","E8","A4"]`, + want: "", + }, + { + name: "unknown last hop credits nobody", + tx: dhTx(RouteFlood, PayloadTXT_MSG, ""), + pathJSON: `["9999"]`, + want: "", + }, + { + name: "direct route never credits, even when the node is the last entry", + tx: dhTx(RouteDirect, PayloadTXT_MSG, ""), + pathJSON: `["A433","BB11"]`, + want: "", + }, + { + name: "transport direct never credits", + tx: dhTx(RouteTransportDirect, PayloadTXT_MSG, ""), + pathJSON: `["BB11"]`, + want: "", + }, + { + name: "flood advert with empty path credits the originator", + tx: dhTx(RouteFlood, PayloadADVERT, advert), + pathJSON: `[]`, + want: dhUnique, + }, + { + name: "flood non-advert with empty path credits nobody: originator unknown", + tx: dhTx(RouteFlood, PayloadTXT_MSG, ""), + pathJSON: `[]`, + want: "", + }, + { + name: "missing route type credits nobody", + tx: &StoreTx{PayloadType: dhRoute(PayloadTXT_MSG)}, + pathJSON: `["BB11"]`, + want: "", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + obs := &StoreObs{ObserverID: "obs1", PathJSON: tc.pathJSON} + if got := directHeardNode(tc.tx, obs, pm); got != tc.want { + t.Fatalf("directHeardNode = %q, want %q", got, tc.want) + } + }) + } +} + +// TestDirectHeardNode_AmbiguousPrefixRegression pins the reported case: a +// long 868 MHz flood path whose one-byte "A4" hop matches two repeaters must +// not credit either of them, however plausible the geo/affinity tiers of +// resolveWithContext would find one. +func TestDirectHeardNode_AmbiguousPrefixRegression(t *testing.T) { + pm := buildPrefixMap(directHeardNodes) + tx := dhTx(RouteFlood, PayloadTXT_MSG, "") + obs := &StoreObs{ + ObserverID: "obs868", + PathJSON: `["A4","F1","AE","6A","77","5C","29","ED","4D","3F","6A","E7","6A","C2","68"]`, + } + if got := directHeardNode(tx, obs, pm); got != "" { + t.Fatalf("ambiguous A4 hop credited %q; the prefix has %d candidates and must credit nobody", + got, len(pm.relayCandidates("a4"))) + } +} + +// TestDirectHeardNode_ListenerNeverCredited: an observer that reported +// repeat:off cannot have retransmitted the packet, so it must not survive as +// a last-hop candidate (#1290 parity — relayCandidates already filters it). +func TestDirectHeardNode_ListenerNeverCredited(t *testing.T) { + pm := buildPrefixMap(directHeardNodes) + pm.markNonRelay([]string{dhUnique}) + tx := dhTx(RouteFlood, PayloadTXT_MSG, "") + obs := &StoreObs{ObserverID: "obs1", PathJSON: `["BB11"]`} + if got := directHeardNode(tx, obs, pm); got != "" { + t.Fatalf("listener-only node credited as last hop: %q", got) + } +} + +func TestComputeDirectHeardAggregates(t *testing.T) { + snr := func(v float64) *float64 { return &v } + tx := dhTx(RouteFlood, PayloadTXT_MSG, "") + tx.Observations = []*StoreObs{ + {ObserverID: "obsA", ObserverName: "A", PathJSON: `["BB11"]`, SNR: snr(10), RSSI: snr(-50)}, + {ObserverID: "obsA", ObserverName: "A", PathJSON: `["BB11"]`, SNR: snr(20), RSSI: snr(-70)}, + {ObserverID: "obsB", ObserverName: "B", PathJSON: `["BB11","A433"]`, SNR: snr(5)}, + // Ambiguous last hop: contributes to nobody. + {ObserverID: "obsC", ObserverName: "C", PathJSON: `["A4"]`, SNR: snr(1)}, + } + + idx := buildDirectHeardIndex([]*StoreTx{tx}, buildPrefixMap(directHeardNodes)) + + byObs := idx[dhUnique] + if len(byObs) != 1 { + t.Fatalf("unique node has %d direct observers, want 1: %#v", len(byObs), byObs) + } + a := byObs["obsA"] + if a == nil || a.Count != 2 { + t.Fatalf("obsA aggregate = %#v, want Count 2", a) + } + if a.SNRCount != 2 || a.SNRSum != 30 { + t.Fatalf("obsA SNR = %v over %d, want 30 over 2", a.SNRSum, a.SNRCount) + } + if a.RSSICount != 2 || a.RSSISum != -120 { + t.Fatalf("obsA RSSI = %v over %d, want -120 over 2", a.RSSISum, a.RSSICount) + } + + if got := len(idx[dhCollideA]); got != 1 { + t.Fatalf("collideA has %d direct observers, want 1 (obsB heard it as last hop)", got) + } + for pk, m := range idx { + if _, ok := m["obsC"]; ok { + t.Fatalf("observer with an ambiguous last hop was credited to %s", pk) + } + } +} + +func TestBuildDirectObserverRowsSortedAndAveraged(t *testing.T) { + byObs := map[string]*directHeardAgg{ + "low": {ObserverName: "low", Count: 1, SNRSum: 3, SNRCount: 1}, + "high": {ObserverName: "high", Count: 9, SNRSum: 18, SNRCount: 2, RSSISum: -100, RSSICount: 2}, + "none": {ObserverName: "none", Count: 4}, + } + rows := buildDirectObserverRows(byObs, nil, nil) + if len(rows) != 3 { + t.Fatalf("got %d rows, want 3", len(rows)) + } + if rows[0].ObserverID != "high" || rows[1].ObserverID != "none" || rows[2].ObserverID != "low" { + t.Fatalf("rows not sorted by packet count desc: %#v", rows) + } + if rows[0].AvgSNR == nil || *rows[0].AvgSNR != 9 { + t.Fatalf("avgSnr = %v, want 9", rows[0].AvgSNR) + } + if rows[0].AvgRSSI == nil || *rows[0].AvgRSSI != -50 { + t.Fatalf("avgRssi = %v, want -50", rows[0].AvgRSSI) + } + if rows[1].AvgSNR != nil || rows[1].AvgRSSI != nil { + t.Fatalf("observer with no signal samples must report null, got %#v", rows[1]) + } +} + +// TestBuildDirectObserverRows_CanRelayTriState mirrors the nodes.js badge: +// nil means "no repeat field ever seen", false means listener, true means +// confirmed repeater (PR #1624). +func TestBuildDirectObserverRows_CanRelayTriState(t *testing.T) { + byObs := map[string]*directHeardAgg{ + "unknown": {Count: 3}, + "listener": {Count: 2}, + "repeater": {Count: 1}, + } + seen := map[string]struct{}{"listener": {}, "repeater": {}} + nonRelay := map[string]struct{}{"listener": {}} + rows := buildDirectObserverRows(byObs, nonRelay, seen) + + got := map[string]*bool{} + for _, r := range rows { + got[r.ObserverID] = r.CanRelay + } + if got["unknown"] != nil { + t.Fatalf("unknown observer must report nil can_relay, got %v", *got["unknown"]) + } + if got["listener"] == nil || *got["listener"] { + t.Fatalf("listener must report can_relay false, got %v", got["listener"]) + } + if got["repeater"] == nil || !*got["repeater"] { + t.Fatalf("repeater must report can_relay true, got %v", got["repeater"]) + } +} diff --git a/cmd/server/node_health_can_relay_case_1290_test.go b/cmd/server/node_health_can_relay_case_1290_test.go index d2785ba5..59d0aa6f 100644 --- a/cmd/server/node_health_can_relay_case_1290_test.go +++ b/cmd/server/node_health_can_relay_case_1290_test.go @@ -43,20 +43,37 @@ func TestNodeHealth_CanRelayCaseInsensitive_Issue1290(t *testing.T) { // In-memory packet with the MIXED-case observer id so the badge resolver // must lower-case both sides to match against the lower-cased pubkey set. + // The packet is a flood ADVERT heard with an empty path, which is what + // makes the observer a DIRECT receiver of nodePubkey — only direct rows + // carry the badge (see direct_heard.go). snr := 7.0 - srv.store.mu.Lock() - if srv.store.byNode == nil { - srv.store.byNode = make(map[string][]*StoreTx) - } - srv.store.byNode[nodePubkey] = append(srv.store.byNode[nodePubkey], &StoreTx{ + routeFlood := RouteFlood + payloadAdvert := PayloadADVERT + tx := &StoreTx{ Hash: "1290casebadge00", FirstSeen: now, + RouteType: &routeFlood, + PayloadType: &payloadAdvert, + DecodedJSON: `{"type":"ADVERT","pubKey":"` + nodePubkey + `"}`, SNR: &snr, ObservationCount: 1, ObserverID: obsIDMixed, ObserverName: "ListenerOnly", - }) + Observations: []*StoreObs{{ + ObserverID: obsIDMixed, + ObserverName: "ListenerOnly", + PathJSON: "[]", + SNR: &snr, + }}, + } + srv.store.mu.Lock() + if srv.store.byNode == nil { + srv.store.byNode = make(map[string][]*StoreTx) + } + srv.store.byNode[nodePubkey] = append(srv.store.byNode[nodePubkey], tx) + srv.store.packets = append(srv.store.packets, tx) srv.store.mu.Unlock() + srv.store.publishDirectHeard(srv.store.computeDirectHeard()) req := httptest.NewRequest(http.MethodGet, "/api/nodes/"+nodePubkey+"/health", nil) w := httptest.NewRecorder() diff --git a/cmd/server/node_health_direct_rf_test.go b/cmd/server/node_health_direct_rf_test.go new file mode 100644 index 00000000..5084df7a --- /dev/null +++ b/cmd/server/node_health_direct_rf_test.go @@ -0,0 +1,134 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// End-to-end guard for the reported bug: a 433 MHz repeater listed 13 +// observers on its detail page, twelve of which run on 868 MHz only and +// cannot physically have heard it. They were credited because the node's +// one-byte pubkey prefix collides with ten other repeaters and +// resolveWithContext picks a winner instead of abstaining. +// +// /api/nodes/{pk}/health must now report only observers that received the +// node's own transmission off the air, and count the rest separately. +func TestNodeHealth_RelayedObserverIsNotHeardBy(t *testing.T) { + srv, router := setupTestServer(t) + const nodePubkey = "aabbccdd11223344" // seeded by seedTestData + now := time.Now().UTC().Format(time.RFC3339) + + routeFlood := RouteFlood + payload := PayloadTXT_MSG + snr := 3.0 + rssi := -110.0 + + // A long flood path that merely passes through the node. The observer + // at the far end saw the packet; it never heard this node. + relayed := &StoreTx{ + Hash: "directrf-relayed", + FirstSeen: now, + RouteType: &routeFlood, + PayloadType: &payload, + SNR: &snr, + RSSI: &rssi, + ObservationCount: 1, + ObserverID: "farawayobserver", + ObserverName: "FarAway", + Observations: []*StoreObs{{ + ObserverID: "farawayobserver", + ObserverName: "FarAway", + PathJSON: `["AABB","1234","5678"]`, + SNR: &snr, + RSSI: &rssi, + }}, + } + + srv.store.mu.Lock() + if srv.store.byNode == nil { + srv.store.byNode = make(map[string][]*StoreTx) + } + srv.store.byNode[nodePubkey] = append(srv.store.byNode[nodePubkey], relayed) + srv.store.packets = append(srv.store.packets, relayed) + srv.store.mu.Unlock() + srv.store.publishDirectHeard(srv.store.computeDirectHeard()) + + req := httptest.NewRequest(http.MethodGet, "/api/nodes/"+nodePubkey+"/health", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d (body: %s)", w.Code, w.Body.String()) + } + + var body map[string]interface{} + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("json: %v", err) + } + + obs, _ := body["observers"].([]interface{}) + for _, raw := range obs { + row, ok := raw.(map[string]interface{}) + if ok && row["observer_id"] == "farawayobserver" { + t.Fatalf("an observer that only saw relayed traffic was listed as having heard the node: %v", row) + } + } + + relayCount, ok := body["relayObserverCount"].(float64) + if !ok { + t.Fatalf("relayObserverCount missing or not a number: %T %v", + body["relayObserverCount"], body["relayObserverCount"]) + } + if relayCount < 1 { + t.Fatalf("relayObserverCount = %v, want at least 1 (the relayed observer must still be counted)", relayCount) + } +} + +// The walk runs over every observation in the store on each recompute pass. +// The reference deployment holds 2.9M of them; this pins that a full pass +// stays well inside one recompute interval. +func BenchmarkBuildDirectHeardIndex(b *testing.B) { + const ( + txCount = 60000 + perTx = 50 // 3M observations total + hopCount = 8 + ) + nodes := make([]nodeInfo, 0, 64) + for i := 0; i < 64; i++ { + nodes = append(nodes, nodeInfo{ + Role: "repeater", + PublicKey: string([]byte{hexDigit(i / 16), hexDigit(i % 16)}) + "00112233445566778899aabbccddeeff00112233445566778899aabbccddee", + }) + } + pm := buildPrefixMap(nodes) + + routeFlood := RouteFlood + payload := PayloadTXT_MSG + path := `["AABB","1234","5678","9ABC","DEF0","0011","2233","4455"]` + packets := make([]*StoreTx, 0, txCount) + for i := 0; i < txCount; i++ { + obsList := make([]*StoreObs, 0, perTx) + for j := 0; j < perTx; j++ { + obsList = append(obsList, &StoreObs{ObserverID: "obs", PathJSON: path}) + } + packets = append(packets, &StoreTx{ + RouteType: &routeFlood, + PayloadType: &payload, + Observations: obsList, + }) + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + _ = buildDirectHeardIndex(packets, pm) + } + _ = hopCount +} + +func hexDigit(v int) byte { + if v < 10 { + return byte('0' + v) + } + return byte('a' + v - 10) +} diff --git a/cmd/server/store.go b/cmd/server/store.go index cee8c609..df94d8c6 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -224,8 +224,13 @@ type PacketStore struct { recompObserversClockSkew *analyticsRecomputer recompNodesClockSkew *analyticsRecomputer recompRetransmissions *analyticsRecomputer - cacheHits int64 - cacheMisses int64 + recompDirectHeard *analyticsRecomputer + // directHeardSnap holds the latest directHeardIndex published by + // recompDirectHeard. Separate from the recomputer's own cache so + // readers never touch analyticsRecomputerMu. See direct_heard.go. + directHeardSnap atomic.Value + cacheHits int64 + cacheMisses int64 // Rate-limited invalidation (fixes #533: caches cleared faster than hit) lastInvalidated time.Time pendingInv *cacheInvalidation // accumulated dirty flags during cooldown @@ -9376,6 +9381,10 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string areaNodes = s.resolveAreaNodes(area) } + // Loaded before s.mu so the lock order stays s.mu → analyticsRecomputerMu. + directHeard := s.loadDirectHeard() + nonRelaySet, seenSet := s.canRelaySets() + s.mu.RLock() defer s.mu.RUnlock() @@ -9455,11 +9464,10 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string var snrSum float64 var snrCount int var lastHeard string - observerStats := map[string]*struct { - name string - snrSum, rssiSum float64 - snrCount, rssiCount, count int - }{} + // See GetNodeHealth: this set is "saw traffic involving the node", + // which is not "heard the node". Only the direct-RF rows below carry + // signal numbers. + relayObservers := map[string]struct{}{} totalObservations := 0 for _, pkt := range packets { @@ -9477,46 +9485,14 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string if lastHeard == "" || pkt.FirstSeen > lastHeard { lastHeard = pkt.FirstSeen } - obsID := pkt.ObserverID - if obsID != "" { - obs := observerStats[obsID] - if obs == nil { - obs = &struct { - name string - snrSum, rssiSum float64 - snrCount, rssiCount, count int - }{name: pkt.ObserverName} - observerStats[obsID] = obs - } - obs.count++ - if pkt.SNR != nil { - obs.snrSum += *pkt.SNR - obs.snrCount++ - } - if pkt.RSSI != nil { - obs.rssiSum += *pkt.RSSI - obs.rssiCount++ - } + if pkt.ObserverID != "" { + relayObservers[pkt.ObserverID] = struct{}{} } } - observerRows := make([]map[string]interface{}, 0) - for id, o := range observerStats { - var avgSnr, avgRssi interface{} - if o.snrCount > 0 { - avgSnr = o.snrSum / float64(o.snrCount) - } - if o.rssiCount > 0 { - avgRssi = o.rssiSum / float64(o.rssiCount) - } - observerRows = append(observerRows, map[string]interface{}{ - "observer_id": id, "observer_name": o.name, - "avgSnr": avgSnr, "avgRssi": avgRssi, "packetCount": o.count, - }) - } - sort.Slice(observerRows, func(i, j int) bool { - return observerRows[i]["packetCount"].(int) > observerRows[j]["packetCount"].(int) - }) + directByObs := directHeard[strings.ToLower(n.pk)] + observerRows := buildDirectObserverRows(directByObs, nonRelaySet, seenSet) + relayObserverCount := relayOnlyObserverCount(relayObservers, directByObs) var avgSnr interface{} if snrCount > 0 { @@ -9541,7 +9517,8 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string "avgSnr": avgSnr, "lastHeard": lhVal, }, - "observers": observerRows, + "observers": observerRows, + "relayObserverCount": relayObserverCount, }) } @@ -9574,6 +9551,10 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro } } + // Loaded before taking s.mu so the lock order stays s.mu → + // analyticsRecomputerMu everywhere (computeDirectHeard takes s.mu). + directHeard := s.loadDirectHeard() + s.mu.RLock() defer s.mu.RUnlock() @@ -9587,11 +9568,12 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro var lastHeard string totalObservations := 0 - observerStats := map[string]*struct { - name string - snrSum, rssiSum float64 - snrCount, rssiCount, count int - }{} + // Observers that saw traffic involving this node — as originator, as a + // destination, or as a resolved relay hop. Seeing a packet is not + // hearing the node: the SNR/RSSI on such a transmission belongs to + // whichever node last transmitted the copy this observer received. Only + // the direct-RF set below may carry signal numbers. See direct_heard.go. + relayObservers := map[string]struct{}{} for _, pkt := range packets { totalObservations += pkt.ObservationCount @@ -9611,85 +9593,15 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro totalHops += len(hops) hopCount++ } - // Observer stats - obsID := pkt.ObserverID - if obsID != "" { - obs := observerStats[obsID] - if obs == nil { - obs = &struct { - name string - snrSum, rssiSum float64 - snrCount, rssiCount, count int - }{name: pkt.ObserverName} - observerStats[obsID] = obs - } - obs.count++ - if pkt.SNR != nil { - obs.snrSum += *pkt.SNR - obs.snrCount++ - } - if pkt.RSSI != nil { - obs.rssiSum += *pkt.RSSI - obs.rssiCount++ - } + if pkt.ObserverID != "" { + relayObservers[pkt.ObserverID] = struct{}{} } } - observerRows := make([]map[string]interface{}, 0) - // Issue #1290: surface listener/repeater hint on node detail by - // looking up can_relay for each observer that heard this node. - // One-shot fetch of the non-relay set keeps this O(observers) on - // rare events; nil on error degrades to "neither badge" client-side. - // Issue #1290: keep this set lowercase to match the convention used - // by the resolver (cmd/server/store.go pm.nonRelay) and by - // GetNonRelayObserverPubkeys (which already returns LOWER(id)). - // Two case conventions on the same upstream string would be a - // latent regression waiting for any refactor that touches the - // observer-id normalization layer. - nonRelaySet := map[string]struct{}{} - // PR #1624 MAJOR-2: tri-state badge needs to distinguish "confirmed - // repeater" (seen=1, can_relay=1) from "unknown" (seen=0). Build - // the set of observers we have NO repeat-field record for so the - // badge is nil/omitted for them — matches nodes.js:679 tri-state. - seenSet := map[string]struct{}{} - if s.db != nil && s.db.conn != nil { - if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil { - for _, pk := range pks { - nonRelaySet[strings.ToLower(pk)] = struct{}{} - } - } - if pks, err := s.db.GetCanRelaySeenObserverPubkeys(); err == nil { - for _, pk := range pks { - seenSet[strings.ToLower(pk)] = struct{}{} - } - } - } - for id, o := range observerStats { - var avgSnr, avgRssi interface{} - if o.snrCount > 0 { - avgSnr = o.snrSum / float64(o.snrCount) - } - if o.rssiCount > 0 { - avgRssi = o.rssiSum / float64(o.rssiCount) - } - idLower := strings.ToLower(id) - var canRelay interface{} // nil = unknown (no repeat field ever) - if _, seen := seenSet[idLower]; seen { - if _, isListener := nonRelaySet[idLower]; isListener { - canRelay = false - } else { - canRelay = true - } - } - observerRows = append(observerRows, map[string]interface{}{ - "observer_id": id, "observer_name": o.name, - "avgSnr": avgSnr, "avgRssi": avgRssi, "packetCount": o.count, - "can_relay": canRelay, - }) - } - sort.Slice(observerRows, func(i, j int) bool { - return observerRows[i]["packetCount"].(int) > observerRows[j]["packetCount"].(int) - }) + nonRelaySet, seenSet := s.canRelaySets() + directByObs := directHeard[strings.ToLower(pubkey)] + observerRows := buildDirectObserverRows(directByObs, nonRelaySet, seenSet) + relayObserverCount := relayOnlyObserverCount(relayObservers, directByObs) var avgSnr interface{} if snrCount > 0 { @@ -9717,8 +9629,14 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro } return map[string]interface{}{ - "node": node, + "node": node, + // Direct-RF only: observers that received this node's own + // transmission off the air. "observers": observerRows, + // Observers that saw traffic through this node without hearing it. + // The stats below count that relayed traffic too, so the card needs + // the number to stay consistent with them. + "relayObserverCount": relayObserverCount, "stats": map[string]interface{}{ "totalTransmissions": len(packets), "totalObservations": totalObservations, diff --git a/docs/api-spec.md b/docs/api-spec.md index 21e5bd09..9ae92975 100644 --- a/docs/api-spec.md +++ b/docs/api-spec.md @@ -408,15 +408,18 @@ Returns a JSON array (not wrapped in an object): "avgSnr": number | null, "lastHeard": string (ISO) | null }, + // Direct-RF only, same shape and rule as GET /api/nodes/:pubkey/health. "observers": [ { "observer_id": string, "observer_name": string | null, "avgSnr": number | null, "avgRssi": number | null, - "packetCount": number + "packetCount": number, + "can_relay": boolean | null } - ] + ], + "relayObserverCount": number } ] ``` @@ -514,6 +517,12 @@ Detailed health information for a single node. "first_seen": string (ISO), "advert_count": number }, + // Observers that received this node's OWN transmission off the air: + // a flood packet whose last path hop resolves unambiguously to this + // node, or a flood ADVERT it originated that arrived with an empty + // path. Only here do avgSnr/avgRssi describe this node's signal. + // Direct routes never qualify: their path is the remaining route, + // not the travelled one. "observers": [ { "observer_id": string, @@ -521,9 +530,13 @@ Detailed health information for a single node. "packetCount": number, "avgSnr": number | null, "avgRssi": number | null, - "iata": string | null + "can_relay": boolean | null // null = observer never reported a repeat field } ], + // Observers that saw traffic through this node without hearing it. + // The stats below count that relayed traffic, so this keeps the two + // consistent. Most nodes have no observer in radio range at all. + "relayObserverCount": number, "stats": { "totalTransmissions": number, "totalObservations": number, diff --git a/public/nodes.js b/public/nodes.js index 5699bc95..d1322b31 100644 --- a/public/nodes.js +++ b/public/nodes.js @@ -650,6 +650,7 @@ const h = healthData || {}; const stats = h.stats || {}; const observers = h.observers || []; + const relayObserverCount = Number(h.relayObserverCount) || 0; const recent = h.recentPackets || []; const lastHeard = stats.lastHeard; @@ -777,10 +778,11 @@ `; })()} - ${observers.length ? `
+ ${observers.length || relayObserverCount ? `
${(() => { const regions = [...new Set(observers.map(o => o.iata).filter(Boolean))]; return regions.length ? `
Regions: ${regions.map(r => '' + escapeHtml(r) + '').join(' ')}
` : ''; })()} -

Heard By (${observers.length} observer${observers.length > 1 ? 's' : ''})

- +

Heard By — direct (${observers.length} observer${observers.length === 1 ? '' : 's'})

+ ${observers.length ? '' : '
No observer is within radio range of this node.
'} + ${observers.length ? `
@@ -797,7 +799,8 @@ `).join('')} -
Observer Region${o.avgRssi != null ? Number(o.avgRssi).toFixed(0) + ' dBm' : '—'}
+ ` : ''} + ${relayObserverCount ? `
Seen via relay by ${relayObserverCount} observer${relayObserverCount === 1 ? '' : 's'}. Those observers heard a repeater that forwarded this node's traffic, not this node.
` : ''}
` : ''}
@@ -1676,6 +1679,7 @@ const h = data.healthData || {}; const stats = h.stats || {}; const observers = h.observers || []; + const relayObserverCount = Number(h.relayObserverCount) || 0; const recent = h.recentPackets || []; const hasLoc = n.lat != null && n.lon != null; const nodeUrl = location.origin + '/#/nodes/' + encodeURIComponent(n.public_key); @@ -1749,9 +1753,10 @@ `; })()}
- ${observers.length ? `
+ ${observers.length || relayObserverCount ? `
${(() => { const regions = [...new Set(observers.map(o => o.iata).filter(Boolean))]; return regions.length ? `
Regions: ${regions.join(', ')}
` : ''; })()} -

Heard By (${observers.length} observer${observers.length > 1 ? 's' : ''})

+

Heard By — direct (${observers.length} observer${observers.length === 1 ? '' : 's'})

+ ${observers.length ? '' : '
No observer is within radio range of this node.
'}
${observers.map(o => { const stats = [`${o.packetCount} pkts`]; @@ -1763,6 +1768,7 @@
`; }).join('')}
+ ${relayObserverCount ? `
Seen via relay by ${relayObserverCount} observer${relayObserverCount === 1 ? '' : 's'}.
` : ''}
` : ''}
diff --git a/test-all.sh b/test-all.sh index f9622ad3..b8515b21 100755 --- a/test-all.sh +++ b/test-all.sh @@ -44,6 +44,7 @@ node tests/unit/test-confidence-indicator.js node tests/unit/test-coverage-gate.js node tests/unit/test-customizer-v2.js node tests/unit/test-drag-manager.js +node tests/unit/test-direct-rf-heard-by.js node tests/unit/test-embed-mode-1369.js node tests/unit/test-fetch-all-nodes-pagination.js node tests/unit/test-fluid-scaffolding.js diff --git a/tests/unit/test-direct-rf-heard-by.js b/tests/unit/test-direct-rf-heard-by.js new file mode 100644 index 00000000..28a219b4 --- /dev/null +++ b/tests/unit/test-direct-rf-heard-by.js @@ -0,0 +1,110 @@ +/* The node detail "Heard By" card reports direct radio reception only. + * + * Before this split the card credited every observer that saw traffic the + * node was involved in — including traffic merely relayed through it — and + * printed an SNR/RSSI next to each. Those numbers belong to whichever node + * last transmitted the copy the observer received. A 433 MHz repeater was + * listed as heard by twelve 868 MHz observers this way. + * + * The card template itself is exercised here, not a copy of it: the block is + * sliced out of public/nodes.js and evaluated as the template literal it is. + */ +'use strict'; +const REPO_ROOT = require('path').resolve(__dirname, '..', '..'); +const fs = require('fs'); +const assert = require('assert'); + +let passed = 0, failed = 0; +function test(name, fn) { + try { fn(); passed++; console.log(' ✅ ' + name); } + catch (e) { failed++; console.log(' ❌ ' + name + ': ' + e.message); } +} + +console.log('\n=== node detail: Heard By is direct-RF only ==='); + +const src = fs.readFileSync(REPO_ROOT + '/public/nodes.js', 'utf8'); + +// --- slice the full-detail card out of the renderer --------------------------- +const START = '${observers.length || relayObserverCount ? `
'; +const END = '
'; +const startIdx = src.indexOf(START); +assert.ok(startIdx >= 0, 'could not find the Heard By card in public/nodes.js'); +const endIdx = src.indexOf(END, startIdx); +assert.ok(endIdx > startIdx, 'could not find the end of the Heard By card'); +const block = src.slice(startIdx, endIdx).replace(/\s+$/, ''); + +// The block is one `${cond ? `...` : ''}` substitution, so wrapping it in +// backticks turns it back into the markup the page renders. +const renderCard = new Function( + 'observers', 'relayObserverCount', 'escapeHtml', + 'return `' + block + '`;' +); +const esc = s => String(s).replace(/&/g, '&').replace(//g, '>'); + +const directRow = { + observer_id: 'obs-direct', observer_name: 'NearbyObserver', + packetCount: 42, avgSnr: 11.25, avgRssi: -63.4, can_relay: true, +}; + +test('heading says "direct" and counts only the direct observers', () => { + const html = renderCard([directRow], 7, esc); + assert.ok(/Heard By — direct \(1 observer\)/.test(html), + 'expected a singular direct heading, got: ' + html.slice(0, 300)); + assert.ok(!/Heard By \(/.test(html), 'the old undifferentiated heading is still rendered'); +}); + +test('relayed observers are reported as a count, without signal numbers', () => { + const html = renderCard([directRow], 12, esc); + assert.ok(/Seen via relay by 12 observers/.test(html), + 'expected the relay count line, got: ' + html); + // The relay line must not carry an SNR/RSSI: no signal was measured for + // this node on those receptions. + const relayLine = html.slice(html.indexOf('Seen via relay')); + assert.ok(!/dBm?/.test(relayLine), 'relay line must not print signal values'); +}); + +test('a node nobody hears directly still renders the card, with an empty state', () => { + const html = renderCard([], 35, esc); + assert.ok(/Heard By — direct \(0 observers\)/.test(html), 'expected a zero direct heading'); + assert.ok(/No observer is within radio range of this node\./.test(html), + 'expected the empty state line'); + assert.ok(/Seen via relay by 35 observers/.test(html), 'expected the relay count'); + assert.ok(!/observer-sort-table/.test(html), + 'an empty direct list must not render a table header with no rows'); +}); + +test('a node with neither direct nor relayed observers renders nothing', () => { + assert.strictEqual(renderCard([], 0, esc).trim(), ''); +}); + +test('singular and plural agree for one relayed observer', () => { + const html = renderCard([], 1, esc); + assert.ok(/Seen via relay by 1 observer\./.test(html), 'expected singular, got: ' + html); +}); + +test('direct rows keep their signal columns', () => { + const html = renderCard([directRow], 0, esc); + assert.ok(/11\.3 dB/.test(html), 'expected the rounded avg SNR'); + assert.ok(/-63 dBm/.test(html), 'expected the rounded avg RSSI'); + assert.ok(/NearbyObserver/.test(html), 'expected the observer name'); +}); + +test('the listener/repeater badge still renders on direct rows', () => { + const listener = Object.assign({}, directRow, { can_relay: false }); + assert.ok(/badge-listener/.test(renderCard([listener], 0, esc)), 'expected the listener badge'); + assert.ok(/badge-repeater/.test(renderCard([directRow], 0, esc)), 'expected the repeater badge'); + const unknown = Object.assign({}, directRow, { can_relay: null }); + const html = renderCard([unknown], 0, esc); + assert.ok(!/badge-listener|badge-repeater/.test(html), + 'an observer with no repeat field must get no badge'); +}); + +// --- the side pane must not drift from the full page ------------------------- +test('the side pane reads relayObserverCount too', () => { + const occurrences = (src.match(/const relayObserverCount = Number\(h\.relayObserverCount\) \|\| 0;/g) || []).length; + assert.strictEqual(occurrences, 2, + 'both the full detail page and the side pane must read relayObserverCount'); +}); + +console.log(`\n${passed} passed, ${failed} failed`); +if (failed > 0) process.exit(1);