Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/deploy.yml
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,7 @@ jobs:
node test-a11y-1716-rf-range-btn-active.js
node test-issue-1705-subpath-contrast.js
node test-analytics-table-ids-unique.js
node test-analytics-relay-airtime-dumbbell.js
node test-live-vcr-mode-contrast.js
node test-issue-1770-mobile-row-clamp.js
node test-a11y-axe-routes-coverage.js
Expand Down
148 changes: 148 additions & 0 deletions cmd/ingestor/advert_route_first_ingested_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
package main

import (
"strings"
"testing"
)

// The same ADVERT can reach the ingestor on both route classes: MeshCore's
// BaseChatMesh::shareContactZeroHop re-sends a stored advert unchanged as a
// TRANSPORT_DIRECT zero-hop packet. ComputeContentHash ignores the route bits,
// transport codes and path, so both variants share one transmissions row, and
// InsertTransmission never rewrites route_type or raw_hex for an existing hash.
// The stored route is therefore that of the first observation *inserted*, not
// the earliest received. Relay Airtime Share's ADVERT route classes are built
// on this contract; these tests pin it so any change to it is deliberate.

var firstIngestedAdvertPayload = "" +
strings.Repeat("86", 32) + // public key
"00e1f565" + // timestamp
strings.Repeat("00", 64) + // signature (not validated here)
"81" + "50523836" // app data: flags + name "PR86"

var (
// ROUTE_TYPE_FLOOD (header 0x11 = ADVERT<<2 | 1), two 1-byte relay hops.
firstIngestedFloodRaw = "11" + "02" + "a1b2" + firstIngestedAdvertPayload
// ROUTE_TYPE_TRANSPORT_DIRECT (0x13), transport codes {0,0}, path_len 0:
// exactly what shareContactZeroHop sends.
firstIngestedZeroHopRaw = "13" + "00000000" + "00" + firstIngestedAdvertPayload
)

type advertObservation struct {
raw, observer, rxTime string
}

func advertPacketData(t *testing.T, o advertObservation) *PacketData {
t.Helper()
decoded, err := DecodePacket(o.raw, nil, false)
if err != nil {
t.Fatalf("decode %s: %v", o.raw[:10], err)
}
pd := BuildPacketData(&MQTTPacketMessage{Raw: o.raw}, decoded, o.observer, "", nil)
pd.Timestamp = o.rxTime
return pd
}

type storedAdvert struct {
rows int
routeType int
rawHex string
firstSeen string
observations int
}

func ingestAdverts(t *testing.T, obs ...advertObservation) (storedAdvert, string) {
t.Helper()
s := newTestStore(t)
var hash string
for i, o := range obs {
pd := advertPacketData(t, o)
if pd.PayloadType != PayloadADVERT {
t.Fatalf("observation %d decoded as payload %d, want ADVERT", i, pd.PayloadType)
}
if i == 0 {
hash = pd.Hash
} else if pd.Hash != hash {
t.Fatalf("observation %d hash %s differs from %s; route variants must share a content hash", i, pd.Hash, hash)
}
if _, err := s.InsertTransmission(pd); err != nil {
t.Fatalf("insert observation %d: %v", i, err)
}
}
var got storedAdvert
if err := s.db.QueryRow(`SELECT COUNT(*), MAX(route_type), MAX(raw_hex), MAX(first_seen) FROM transmissions WHERE hash = ?`, hash).
Scan(&got.rows, &got.routeType, &got.rawHex, &got.firstSeen); err != nil {
t.Fatalf("query transmissions: %v", err)
}
if err := s.db.QueryRow(`SELECT COUNT(*) FROM observations o JOIN transmissions t ON t.id = o.transmission_id WHERE t.hash = ?`, hash).
Scan(&got.observations); err != nil {
t.Fatalf("query observations: %v", err)
}
return got, hash
}

func TestAdvertRouteVariantsShareContentHash(t *testing.T) {
flood := advertPacketData(t, advertObservation{firstIngestedFloodRaw, "obs-a", "2026-09-23T10:00:00Z"})
zeroHop := advertPacketData(t, advertObservation{firstIngestedZeroHopRaw, "obs-b", "2026-09-23T10:05:00Z"})
if flood.RouteType != RouteFlood || zeroHop.RouteType != RouteTransportDirect {
t.Fatalf("route types = %d/%d, want %d/%d", flood.RouteType, zeroHop.RouteType, RouteFlood, RouteTransportDirect)
}
if flood.Hash != zeroHop.Hash {
t.Fatalf("hashes differ: flood %s, zero-hop %s", flood.Hash, zeroHop.Hash)
}
}

func TestInsertTransmission_AdvertRouteIsFirstIngested(t *testing.T) {
cases := []struct {
name string
obs []advertObservation
wantRoute int
wantRaw string
wantFirstSeen string
}{
{
name: "flood then zero-hop",
obs: []advertObservation{
{firstIngestedFloodRaw, "obs-a", "2026-09-23T10:00:00Z"},
{firstIngestedZeroHopRaw, "obs-b", "2026-09-23T10:05:00Z"},
},
wantRoute: RouteFlood, wantRaw: firstIngestedFloodRaw, wantFirstSeen: "2026-09-23T10:00:00Z",
},
{
name: "zero-hop then flood",
obs: []advertObservation{
{firstIngestedZeroHopRaw, "obs-b", "2026-09-23T10:00:00Z"},
{firstIngestedFloodRaw, "obs-a", "2026-09-23T10:05:00Z"},
},
wantRoute: RouteTransportDirect, wantRaw: firstIngestedZeroHopRaw, wantFirstSeen: "2026-09-23T10:00:00Z",
},
{
// A delayed zero-hop observation inserted first keeps its route even
// though the flood observation was received earlier: first_seen moves
// back, route_type and raw_hex do not.
name: "zero-hop inserted first but received later",
obs: []advertObservation{
{firstIngestedZeroHopRaw, "obs-b", "2026-09-23T10:05:00Z"},
{firstIngestedFloodRaw, "obs-a", "2026-09-23T10:00:00Z"},
},
wantRoute: RouteTransportDirect, wantRaw: firstIngestedZeroHopRaw, wantFirstSeen: "2026-09-23T10:00:00Z",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got, _ := ingestAdverts(t, tc.obs...)
if got.rows != 1 || got.observations != 2 {
t.Fatalf("rows=%d observations=%d, want one transmission with two observations", got.rows, got.observations)
}
if got.routeType != tc.wantRoute {
t.Errorf("route_type = %d, want %d (first inserted observation)", got.routeType, tc.wantRoute)
}
if got.rawHex != tc.wantRaw {
t.Errorf("raw_hex header %s, want %s: raw_hex must stay consistent with route_type", got.rawHex[:2], tc.wantRaw[:2])
}
if got.firstSeen != tc.wantFirstSeen {
t.Errorf("first_seen = %s, want %s (earliest rx time)", got.firstSeen, tc.wantFirstSeen)
}
})
}
}
6 changes: 6 additions & 0 deletions cmd/server/openapi.go
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,12 @@ func routeDescriptions() map[string]routeMeta {
"GET /api/analytics/subpaths-bulk": {Summary: "Bulk subpath analysis", Tag: "analytics"},
"GET /api/analytics/subpath-detail": {Summary: "Subpath detail", Tag: "analytics"},
"GET /api/analytics/neighbor-graph": {Summary: "Neighbor graph", Description: "Full neighbor affinity graph for visualization.", Tag: "analytics"},
"GET /api/analytics/relay-airtime-share": {Summary: "Relay airtime share per payload type", Description: "Overview tab's Relay Airtime Share. Each distinct packet (deduplicated by content hash) scores LoRa Time-on-Air of its raw bytes × distinct repeaters that relayed it, under the assumed preset returned in preset (config analytics.loraPreset); originator TX is excluded. rows[] = {payload_type, type, route_class, count, count_pct, score, airtime_pct}, sorted by airtime_pct desc. type is the numeric payload type; payload_type is a display label and may change. ADVERT is split into up to three rows that all have type 4: route_class \"flood\" (route_type 0/1, label \"ADVERT (flood)\"), \"zero_hop\" (route_type 2/3, label \"ADVERT (zero-hop)\") and \"legacy\" (NULL or out-of-range route_type, label \"ADVERT\"). route_class is null on every other row, so key rows by (type, route_class), never by type alone. The route is the one stored on the transmission: the first observation the ingestor inserted for that content hash, even if later observations of the same hash arrive on the other route class (e.g. a contact re-shared as a zero-hop advert). count_pct and airtime_pct are shares of total_count and total_score (nanoseconds). Payload Type Mix (/api/analytics/rf payloadTypes) is not split. Cached in the RF analytics cache (60s default).", Tag: "analytics",
QueryParams: []paramMeta{
{Name: "window", Description: "Relative window: 1h, 24h/1d, 3d, 7d/1w or 30d (default: all loaded packets)", Type: "string"},
{Name: "from", Description: "RFC3339 start; if from or to is given, window is ignored (an unparseable value leaves that bound open)", Type: "string"},
{Name: "to", Description: "RFC3339 end; if from or to is given, window is ignored (an unparseable value leaves that bound open)", Type: "string"},
}},
"GET /api/analytics/wardriving": {Summary: "Wardriving channel analytics", Description: "Activity/entry-point/coverage/signal/session analytics for the #wardriving channel (or another channel via ?channel=): message volume over time, top senders, path[0] entry-point hash-prefix tallies (resolve names via /api/resolve-hops), per-observer coverage (observer's known IATA-derived coordinates, not the sender's — MeshMapper's wardriving messages normally carry an anonymous session token, not live GPS), average SNR/RSSI over the same time buckets as the activity series, each sender's messages grouped into distinct sessions/runs (split on a 15-minute gap, each with an AirtimeMs field — LoRa Time-on-Air × distinct relaying repeaters, same formula as the Overview tab's Relay Airtime Share, omitted in DB-only mode), and any senders who explicitly shared their own position (some clients append plaintext \"<lat>,<lon>\" after the token — a deliberate choice by that sender, confirmed empirically, not something CoreScope infers). Cached 30s per window+channel.", Tag: "analytics",
QueryParams: []paramMeta{
{Name: "window", Description: "Time window: 1h, 24h (default), or 7d", Type: "string"},
Expand Down
1 change: 0 additions & 1 deletion cmd/server/openapi_known_gaps.json
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
"routes": [
"/api/admin/prune-geo-filter",
"/api/admin/prune-geo-filter/status",
"/api/analytics/relay-airtime-share",
"/api/analytics/roles",
"/api/config/areas",
"/api/config/areas/polygons",
Expand Down
126 changes: 109 additions & 17 deletions cmd/server/relay_airtime_share.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,8 @@ import (
// preset 869.6 MHz / BW 62.5 kHz / SF 8 / CR 4/5, with the
// SF-dependent preamble pulled from internal/lora.PreambleForSF.
//
// Aggregated by payload_type. Originator TX is deliberately excluded — a
// Aggregated by payload_type, except ADVERT packets which are split into flood
// and zero-hop route classes. Originator TX is deliberately excluded — a
// never-relayed direct message scores 0, which is the correct framing for a
// "relay amplification" metric. In-memory only; no SQL, no new index.

Expand Down Expand Up @@ -197,12 +198,100 @@ func (s *PacketStore) AirtimeAndRelayCountForTransmission(txID int64) (total tim
return total, relays, true
}

// computeRelayAirtimeShare aggregates relay-airtime-share per payload_type.
type relayAirtimeAdvertRoute uint8

const (
relayAirtimeAdvertUnknown relayAirtimeAdvertRoute = iota
relayAirtimeAdvertFlood
relayAirtimeAdvertZeroHop
)

type relayAirtimeBucketKey struct {
payloadType int
advertRoute relayAirtimeAdvertRoute
}

// relayAirtimeKey keeps non-ADVERT aggregation unchanged while separating the
// two ADVERT behaviours defined by MeshCore's route bits. Unknown/legacy route
// values deliberately keep the historical unsuffixed ADVERT bucket rather than
// being silently assigned to either behaviour. tx.PayloadType must be non-nil.
func relayAirtimeKey(tx *StoreTx) relayAirtimeBucketKey {
key := relayAirtimeBucketKey{payloadType: *tx.PayloadType}
if key.payloadType != PayloadADVERT || tx.RouteType == nil {
return key
}
switch *tx.RouteType {
case RouteTransportFlood, RouteFlood:
key.advertRoute = relayAirtimeAdvertFlood
case RouteDirect, RouteTransportDirect:
key.advertRoute = relayAirtimeAdvertZeroHop
}
return key
}

func relayAirtimeBucketName(key relayAirtimeBucketKey) string {
name := payloadTypeNames[key.payloadType]
if name == "" {
name = "UNK"
}
if key.payloadType != PayloadADVERT {
return name
}
switch key.advertRoute {
case relayAirtimeAdvertFlood:
return name + " (flood)"
case relayAirtimeAdvertZeroHop:
return name + " (zero-hop)"
default:
return name
}
}

// Machine-readable ADVERT route classes, returned as route_class. They are
// part of the API contract and deliberately independent of the display
// labels built by relayAirtimeBucketName.
const (
relayAirtimeRouteClassFlood = "flood"
relayAirtimeRouteClassZeroHop = "zero_hop"
relayAirtimeRouteClassLegacy = "legacy"
)

// relayAirtimeRouteClass returns the route_class value for a row: one of the
// constants above for ADVERT rows and nil (JSON null) for every other payload
// type, so (type, route_class) identifies each row.
func relayAirtimeRouteClass(key relayAirtimeBucketKey) *string {
if key.payloadType != PayloadADVERT {
return nil
}
var class string
switch key.advertRoute {
case relayAirtimeAdvertFlood:
class = relayAirtimeRouteClassFlood
case relayAirtimeAdvertZeroHop:
class = relayAirtimeRouteClassZeroHop
default:
class = relayAirtimeRouteClassLegacy
}
return &class
}

// computeRelayAirtimeShare aggregates relay-airtime-share per payload_type,
// separating ADVERT rows by their flood and zero-hop route classes.
//
// The route class is the route_type stored on the transmission, which is the
// route of the first observation the ingestor inserted for that content hash.
// The content hash ignores route bits, so a contact re-shared as a zero-hop
// advert has the same hash as the original flood advert; later observations
// never change the stored route (see cmd/ingestor InsertTransmission).
//
// Returns:
//
// {
// "rows": [{payload_type, type, count, count_pct, score, airtime_pct}, ...] sorted by airtime_pct desc,
// "rows": [{payload_type, type, route_class, count, count_pct, score, airtime_pct}, ...]
// sorted by airtime_pct desc, where type is the numeric payload type,
// payload_type the display label and route_class "flood" / "zero_hop" /
// "legacy" on ADVERT rows and null otherwise; up to three ADVERT rows
// share type 4, and (type, route_class) identifies each row,
// "total_count": int,
// "total_score": int64 (nanoseconds of LoRa Time-on-Air × repeater-count, summed across packets),
// "window": window label,
Expand All @@ -212,14 +301,13 @@ func (s *PacketStore) computeRelayAirtimeShare(window TimeWindow) map[string]int
s.mu.RLock()
defer s.mu.RUnlock()

ptNames := payloadTypeNames
preset := s.resolveLoRaPreset()

type bucket struct {
count int
score int64 // sum of ToA(payload) × relays, in nanoseconds
}
buckets := make(map[int]*bucket)
buckets := make(map[relayAirtimeBucketKey]*bucket)
seenHash := make(map[string]bool, len(s.packets))
totalCount := 0
var totalScore int64
Expand All @@ -240,11 +328,11 @@ func (s *PacketStore) computeRelayAirtimeShare(window TimeWindow) map[string]int
}
seenHash[tx.Hash] = true
}
pt := *tx.PayloadType
b := buckets[pt]
key := relayAirtimeKey(tx)
b := buckets[key]
if b == nil {
b = &bucket{}
buckets[pt] = b
buckets[key] = b
}
b.count++
totalCount++
Expand All @@ -261,11 +349,8 @@ func (s *PacketStore) computeRelayAirtimeShare(window TimeWindow) map[string]int
}

rows := make([]map[string]interface{}, 0, len(buckets))
for pt, b := range buckets {
name := ptNames[pt]
if name == "" {
name = "UNK"
}
for key, b := range buckets {
name := relayAirtimeBucketName(key)
var countPct, airtimePct float64
if totalCount > 0 {
countPct = float64(b.count) / float64(totalCount) * 100.0
Expand All @@ -275,16 +360,17 @@ func (s *PacketStore) computeRelayAirtimeShare(window TimeWindow) map[string]int
}
rows = append(rows, map[string]interface{}{
"payload_type": name,
"type": pt,
"type": key.payloadType,
"route_class": relayAirtimeRouteClass(key),
"count": b.count,
"count_pct": countPct,
"score": b.score,
"airtime_pct": airtimePct,
})
}

// Sort descending by airtime_pct; tiebreak count desc, then name asc
// for deterministic ordering.
// Sort descending by airtime_pct; tiebreak count desc, then name asc,
// then numeric type asc for deterministic ordering.
sort.SliceStable(rows, func(i, j int) bool {
ai, _ := rows[i]["airtime_pct"].(float64)
aj, _ := rows[j]["airtime_pct"].(float64)
Expand All @@ -298,7 +384,13 @@ func (s *PacketStore) computeRelayAirtimeShare(window TimeWindow) map[string]int
}
ni, _ := rows[i]["payload_type"].(string)
nj, _ := rows[j]["payload_type"].(string)
return ni < nj
if ni != nj {
return ni < nj
}
// Unnamed payload types all share the "UNK" label.
ti, _ := rows[i]["type"].(int)
tj, _ := rows[j]["type"].(int)
return ti < tj
})

label := ""
Expand Down
Loading
Loading