mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-11-11 12:18:47 +00:00
Part of #1869. ## Problem `observers.radio` keeps only the **last** `MHz,kHz,SF,CR` an observer reported. Once an observer is retuned (say from SF7 to the EU narrow SF8 setting to listen to a neighbouring mesh), every packet it ever heard reads as heard on the new setting, and there is no way to tell an SF7 packet from an SF8 one. ## What this does As #1869 proposes, the packet data model is untouched: the setting of an observation is resolved from its observer and time. **Ingestor** (all schema in `dbschema.Apply`) - `observer_radios`: one row per stretch of time an observer, and each radio of a multi-radio node, stayed on one setting. Every live status extends the current row and a change starts a new one; retained replays never extend a row. On the first run it is seeded with every observer's current radio, which is what the UI already showed for the past. - Settings are stored canonical through `dbschema.CanonicalRadio`, so `869.4320068,62.5,7,5` and `869.432,62.50,7,5` are one setting. - `observation_radios` (`observation_id → rx_radio_id`): multi-radio openHop nodes tag each packet with `rx_radio_id` and list `radios: [{id, radio}]` in their status, so their observations resolve per radio. - The table is sparse: single-radio nodes never send the field, so `observations` and every query that scans it stay untouched. - The row id comes from the observation upsert via `RETURNING id`. - Rows are pruned with their transmissions, and a purged observer takes its history along. **Server** (read-only) - No `AssertReady` change: an absent table reads as empty and is probed again later, the same rule as `advert_route_evidence`. - The history is held in memory. The live poller checks it on every tick (two primary-key lookups), so a retune or a new `rx_radio_id` is known before the observation is broadcast. A full reload runs every 10 min. - `radio` is added to observations, packets, grouped rows (taken from the header observation) and the live feed. - `GET /api/packets?radio=MHz,kHz,SF,CR` filters by setting. - Combined with `observer=`, it means *that* observer heard it on that setting. - A radio query is served from memory instead of falling back to SQL, which would return the window unfiltered. - `GET /api/radios` (named struct) and `radio_history` on `GET /api/observers/{id}`. **Frontend** - Packet filter fields `radio`, `freq`, `bw`, `sf`, `cr`, e.g. `sf == 8` or `freq > 869.5 && bw == 62.5`. - A Radio row in the packet detail and a Radio column in its observations table (only when the server reports one). - A radio history table on the observer page once there is more than one stretch. - `RfParams.radioText` formats the setting. It escapes all five characters and returns non-numeric input as it came, escaped. ## Performance `txMatchesRadio` runs once per observation inside `filterPackets`. - Each observation costs one map lookup by observer id. - Observers with a single stretch (nearly all) answer from that lookup directly. - Retuned observers add a binary search over their few stretches. - Multi-radio observers add one lookup by observation id. - The snapshot is loaded once per filter call. `BenchmarkTxMatchesRadio`, 30K packets / 150K observations (20 observers, 2 of them retuned): ``` BenchmarkTxMatchesRadio-24 602 2056924 ns/op 0 B/op 0 allocs/op ``` Memory is one entry per stretch plus one per multi-radio observation; nothing is added per observation otherwise. ## Tests - **dbschema:** seed runs once, canonical spellings, the tables exist without `observers.radio`. - **ingestor:** - stretches extend and split; canonical spellings do not split - two radios of one node hearing one transmission over two paths each keep their own radio - a retained status does not write history - prune and purge - `rx_radio_id` reaches `PacketData` - **server:** - resolution before, between and after stretches, and with an unknown radio id - `observer` + `radio`, and grouped rows carry `radio` - incremental refresh, `/api/radios`, `radio_history`, missing tables - the benchmark above - **frontend:** the new filter fields in `tests/unit/test-packet-filter.js`, and XSS cases through the real `RfParams.radioText` and the history renderer. - **E2E:** `tests/e2e/test-radio-setting-1869-e2e.js` checks the grouped rows' radio, `sf ==` on the packets page, the detail Radio row and the observer history. It is added to `scripts/non-unit-tests.json` and to shard 2. `go test -race` in `cmd/server`, `cmd/ingestor` and `internal/dbschema` passes, along with `sh test-all.sh`, `eslint@8` (0 errors) and the XSS diff gate. The new E2E test and `test-e2e-playwright.js` pass against the fixture. ## Browser validation This runs on a live instance with six observers, one of them retuned to 869.618/62.5/SF8/CR8: - The packet detail shows `Radio: 869.432 MHz · SF7 · BW 62.5 kHz · CR 4/5`. - `sf == 7` shows 106 of 106 grouped rows and `sf == 8` shows 0 of 107; the SF8 observer has not heard anything yet. - `/api/radios` lists both settings. - No page errors. ## Notes - **The customizer (rule 8)** gets nothing: the only constants are server-side refresh intervals. - **Attribution timing:** packets heard between a retune and the observer's next status message (minutes) are attributed to the previous setting, because a stretch starts when the status is handled. This is documented at `RecordObserverRadios`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
267 lines
9.7 KiB
Go
267 lines
9.7 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type packetQueryDoc struct {
|
|
Name string `json:"name"`
|
|
In string `json:"in"`
|
|
Description string `json:"description"`
|
|
Required bool `json:"required"`
|
|
Schema struct {
|
|
Type string `json:"type"`
|
|
} `json:"schema"`
|
|
}
|
|
|
|
func packetQueryDocs(t *testing.T, router http.Handler) map[string]packetQueryDoc {
|
|
t.Helper()
|
|
w := httptest.NewRecorder()
|
|
router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/api/spec", nil))
|
|
if w.Code != http.StatusOK {
|
|
t.Fatalf("GET /api/spec: status %d", w.Code)
|
|
}
|
|
var spec struct {
|
|
Paths map[string]struct {
|
|
Get struct {
|
|
Parameters []packetQueryDoc `json:"parameters"`
|
|
} `json:"get"`
|
|
} `json:"paths"`
|
|
}
|
|
if err := json.Unmarshal(w.Body.Bytes(), &spec); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
params := make(map[string]packetQueryDoc)
|
|
for _, param := range spec.Paths["/api/packets"].Get.Parameters {
|
|
if _, duplicate := params[param.Name]; duplicate {
|
|
t.Errorf("duplicate packet query parameter %q", param.Name)
|
|
}
|
|
params[param.Name] = param
|
|
}
|
|
return params
|
|
}
|
|
|
|
func TestOpenAPIPacketQueryParameters(t *testing.T) {
|
|
_, router := setupTestServer(t)
|
|
params := packetQueryDocs(t, router)
|
|
want := map[string]string{
|
|
"limit": "integer", "offset": "integer", "order": "string",
|
|
"type": "integer", "route": "integer", "observer": "string",
|
|
"hash": "string", "since": "string", "until": "string",
|
|
"region": "string", "node": "string", "nodes": "string",
|
|
"channel": "string", "area": "string", "expand": "string",
|
|
"radio": "string", "groupByHash": "boolean",
|
|
}
|
|
for name, typ := range want {
|
|
param, ok := params[name]
|
|
if !ok {
|
|
t.Errorf("GET /api/packets is missing supported query parameter %q", name)
|
|
continue
|
|
}
|
|
if param.In != "query" || param.Required || param.Schema.Type != typ || param.Description == "" {
|
|
t.Errorf("%s: want optional, described query parameter of type %s; got %+v", name, typ, param)
|
|
}
|
|
}
|
|
for name := range params {
|
|
if _, supported := want[name]; !supported {
|
|
t.Errorf("GET /api/packets documents unsupported query parameter %q", name)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Couple the served documentation to observable route behavior, including the
|
|
// mode switches that bypass otherwise-supported packet filters (#1670).
|
|
func TestOpenAPIPacketQueryBehavior(t *testing.T) {
|
|
srv, router := setupTestServer(t)
|
|
params := packetQueryDocs(t, router)
|
|
get := func(t *testing.T, query url.Values) PacketResult {
|
|
t.Helper()
|
|
w := httptest.NewRecorder()
|
|
router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/api/packets?"+query.Encode(), nil))
|
|
if w.Code != http.StatusOK {
|
|
t.Fatalf("GET /api/packets?%s: status %d: %s", query.Encode(), w.Code, w.Body.String())
|
|
}
|
|
var result PacketResult
|
|
if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return result
|
|
}
|
|
described := func(t *testing.T, name string, phrases ...string) {
|
|
t.Helper()
|
|
param, ok := params[name]
|
|
if !ok {
|
|
t.Errorf("working query parameter %q is missing from served spec", name)
|
|
}
|
|
for _, phrase := range phrases {
|
|
if !strings.Contains(param.Description, phrase) {
|
|
t.Errorf("%s description must include %q; got %q", name, phrase, param.Description)
|
|
}
|
|
}
|
|
}
|
|
assertHashes := func(t *testing.T, result PacketResult, want ...string) {
|
|
t.Helper()
|
|
got := make([]string, len(result.Packets))
|
|
for i, packet := range result.Packets {
|
|
got[i], _ = packet["hash"].(string)
|
|
}
|
|
if !slices.Equal(got, want) {
|
|
t.Errorf("packet hashes = %v, want %v", got, want)
|
|
}
|
|
}
|
|
const advert = "abc123def4567890"
|
|
const message = "1234567890abcdef"
|
|
const olderAdvert = "def456abc1230099"
|
|
cutoff := time.Now().UTC().Add(-12 * time.Hour).Format(time.RFC3339)
|
|
|
|
// Use the existing seeded nodes and transmissions, with one configured area.
|
|
srv.store.config = &Config{Areas: map[string]AreaEntry{
|
|
"TEST": {Polygon: [][2]float64{{37.45, -122.05}, {37.55, -122.05}, {37.55, -121.95}, {37.45, -121.95}}},
|
|
"OTHER": {Polygon: [][2]float64{{37.55, -122.15}, {37.65, -122.15}, {37.65, -122.05}, {37.55, -122.05}}},
|
|
}}
|
|
for _, tc := range []struct {
|
|
name, value string
|
|
want []string
|
|
}{
|
|
{"hash", strings.ToUpper(advert), []string{advert}},
|
|
{"hash", advert[:8], nil}, // Exact hash, not a prefix search.
|
|
{"type", "5", []string{message}},
|
|
{"route", "2", nil},
|
|
{"observer", "missing, obs2", []string{advert}},
|
|
{"region", "SFO", []string{advert}},
|
|
{"node", "TestRepeater", []string{olderAdvert, advert}},
|
|
{"nodes", "missing, TestRepeater", []string{olderAdvert, advert}},
|
|
{"channel", "#test", []string{message}},
|
|
{"area", "TEST", []string{olderAdvert, advert}},
|
|
{"area", "OTHER", nil},
|
|
{"since", cutoff, []string{advert}},
|
|
{"until", cutoff, []string{olderAdvert, message}},
|
|
{"order", "asc", []string{advert, message, olderAdvert}},
|
|
} {
|
|
t.Run(tc.name+"="+tc.value, func(t *testing.T) {
|
|
result := get(t, url.Values{tc.name: {tc.value}})
|
|
assertHashes(t, result, tc.want...)
|
|
described(t, tc.name)
|
|
})
|
|
}
|
|
|
|
t.Run("defaults and pagination", func(t *testing.T) {
|
|
result := get(t, nil)
|
|
assertHashes(t, result, olderAdvert, message, advert)
|
|
described(t, "limit", fmt.Sprintf("Defaults to %d", result.Limit))
|
|
described(t, "offset", fmt.Sprintf("Defaults to %d", result.Offset))
|
|
described(t, "order", "Defaults to desc", "ingest ID")
|
|
described(t, "groupByHash", "Defaults to false")
|
|
if _, grouped := result.Packets[0]["count"]; grouped {
|
|
t.Error("default packet list unexpectedly uses grouped summaries")
|
|
}
|
|
page := get(t, url.Values{"limit": {"1"}, "offset": {"1"}})
|
|
assertHashes(t, page, message)
|
|
if page.Limit != 1 || page.Offset != 1 || page.Total != result.Total {
|
|
t.Errorf("unexpected pagination metadata: %+v", page)
|
|
}
|
|
for _, value := range []string{"0", "-1", "invalid"} {
|
|
if got := get(t, url.Values{"limit": {value}}).Limit; got != result.Limit {
|
|
t.Errorf("limit=%s: got %d, want default %d", value, got, result.Limit)
|
|
}
|
|
}
|
|
described(t, "limit", "non-positive", "invalid", "listLimits.packetsMax", strconv.Itoa(srv.cfg.ListLimits.PacketsMax))
|
|
originalMax := srv.cfg.ListLimits.PacketsMax
|
|
t.Cleanup(func() { srv.cfg.ListLimits.PacketsMax = originalMax })
|
|
srv.cfg.ListLimits.PacketsMax = 2
|
|
capped := get(t, url.Values{"limit": {"99999"}})
|
|
if capped.Limit != 2 || len(capped.Packets) != 2 {
|
|
t.Errorf("configured cap not applied: %+v", capped)
|
|
}
|
|
})
|
|
|
|
t.Run("grouping and expansion", func(t *testing.T) {
|
|
for _, expand := range []string{"", "observations", "other"} {
|
|
result := get(t, url.Values{"hash": {advert}, "expand": {expand}})
|
|
assertHashes(t, result, advert)
|
|
_, expanded := result.Packets[0]["observations"]
|
|
if expanded != (expand == "observations") {
|
|
t.Errorf("expand=%q: observations present = %v", expand, expanded)
|
|
}
|
|
}
|
|
described(t, "expand", "observations", "in-memory", "ungrouped")
|
|
grouped := get(t, url.Values{"hash": {advert}, "groupByHash": {"true"}, "order": {"asc"}, "expand": {"observations"}})
|
|
assertHashes(t, grouped, advert)
|
|
if grouped.Packets[0]["count"] != float64(2) || grouped.Packets[0]["observations"] != nil {
|
|
t.Errorf("unexpected grouped summary: %+v", grouped.Packets[0])
|
|
}
|
|
described(t, "groupByHash", "latest observation", "order", "expand")
|
|
for _, order := range []string{"asc", "desc"} {
|
|
grouped := get(t, url.Values{"groupByHash": {"true"}, "order": {order}})
|
|
if grouped.Packets[0]["hash"] != advert {
|
|
t.Errorf("grouped order=%s: latest observation must be first", order)
|
|
}
|
|
}
|
|
})
|
|
|
|
t.Run("time bounds depend on query mode", func(t *testing.T) {
|
|
boundary := get(t, url.Values{"hash": {advert}}).Packets[0]["first_seen"].(string)
|
|
store := srv.store
|
|
t.Cleanup(func() { srv.store = store })
|
|
for _, tc := range []struct {
|
|
mode, param string
|
|
want []string
|
|
}{
|
|
{"memory", "since", nil},
|
|
{"memory", "until", []string{olderAdvert, message}},
|
|
{"SQL", "since", []string{advert}},
|
|
{"SQL", "until", []string{olderAdvert, message, advert}},
|
|
{"nodes", "since", []string{advert}},
|
|
{"nodes", "until", []string{olderAdvert, advert}},
|
|
} {
|
|
t.Run(tc.mode+"/"+tc.param, func(t *testing.T) {
|
|
srv.store = store
|
|
query := url.Values{tc.param: {boundary}}
|
|
if tc.mode == "SQL" {
|
|
srv.store = nil
|
|
} else if tc.mode == "nodes" {
|
|
query.Set("nodes", "TestRepeater")
|
|
}
|
|
assertHashes(t, get(t, query), tc.want...)
|
|
})
|
|
}
|
|
described(t, "since", "first_seen > since", "observations >= since", "first_seen >= since")
|
|
described(t, "until", "first_seen < until", "observations <= until", "first_seen <= until")
|
|
})
|
|
|
|
t.Run("multi-node mode", func(t *testing.T) {
|
|
result := get(t, url.Values{"nodes": {"TestRepeater"}, "hash": {"missing"}, "type": {"5"}, "groupByHash": {"true"}})
|
|
assertHashes(t, result, olderAdvert, advert)
|
|
described(t, "nodes", fmt.Sprintf("%d non-empty", maxMultiNodePubkeys), "only limit, offset, order, since and until")
|
|
tooMany := url.Values{"nodes": {strings.Repeat("missing,", maxMultiNodePubkeys) + "missing"}}
|
|
w := httptest.NewRecorder()
|
|
router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/api/packets?"+tooMany.Encode(), nil))
|
|
if w.Code != http.StatusBadRequest {
|
|
t.Errorf("over-cap nodes list: status %d, want 400", w.Code)
|
|
}
|
|
})
|
|
|
|
t.Run("SQL fallback limitations", func(t *testing.T) {
|
|
assertHashes(t, get(t, url.Values{"hash": {advert}, "area": {"OTHER"}}))
|
|
store := srv.store
|
|
t.Cleanup(func() { srv.store = store })
|
|
srv.store = nil
|
|
result := get(t, url.Values{"hash": {strings.ToUpper(advert)}, "area": {"OTHER"}, "expand": {"observations"}})
|
|
assertHashes(t, result, advert)
|
|
if result.Packets[0]["observations"] != nil {
|
|
t.Error("SQL result unexpectedly expanded observations")
|
|
}
|
|
described(t, "area", "SQL", "ignored")
|
|
described(t, "expand", "SQL", "ignored")
|
|
})
|
|
}
|