Add bounded topology snapshots and validate TRACE route evidence

This commit is contained in:
n30nex
2026-10-13 04:21:34 -04:00
parent bc84be577c
commit 1d3f7ce621
43 changed files with 1746 additions and 144 deletions
+2 -2
View File
@@ -4,7 +4,7 @@ name: CI
on:
push:
branches: [main, dev]
branches: [main, dev, dev1]
pull_request:
jobs:
@@ -58,7 +58,7 @@ jobs:
- name: Verify stats and endpoint queries against PostgreSQL 16
env:
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v
run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|TraceQuality|TopologyLinks|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v
- name: Verify backup command against PostgreSQL 16
run: |
+2 -1
View File
@@ -5,7 +5,7 @@ name: Build and Publish Docker Image
on:
push:
branches: [main, dev]
branches: [main, dev, dev1]
tags: ["v*"]
env:
@@ -45,6 +45,7 @@ jobs:
tags: |
type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
type=raw,value=dev,enable=${{ github.ref == 'refs/heads/dev' }}
type=raw,value=dev1,enable=${{ github.ref == 'refs/heads/dev1' }}
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=sha
+2 -1
View File
@@ -45,7 +45,7 @@ import (
var version = "dev"
// @title MeshCore Beacon API
// @version 2.0.6
// @version 2.1.0
// @description MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.
// @description REST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget.
// @description WebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open.
@@ -443,6 +443,7 @@ func main() {
// CachedReader. Only response projections receive the geographic annotation.
reader = api.WithLocalBorders(reader, localBorders)
r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, router.Options{
ScopeCatalogues: scopeImporter,
MaxConnsPerIP: resolved.MaxConnsPerIP,
MaxConnectsPerMinute: resolved.MaxConnectsPerMinute,
WSAllowedOrigins: cfg.WebSocket.AllowedOrigins,
+20 -3
View File
@@ -15,6 +15,7 @@ import (
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/MeshCore-Beacon/beacon-server/internal/ingest"
"github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"github.com/meshcore-go/meshcore-go"
@@ -406,7 +407,15 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
// compute it once rather than per observation.
var traceRawHashes [][]byte
if row.PayloadType == int16(meshcore.PayloadTypeTrace) {
if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil {
quality := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), -1)
for _, obs := range obsRows {
q := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), int(obs.PathLengthByte))
for _, reason := range q.Reasons {
quality.Add(reason)
}
}
p.TraceQuality = &quality
if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil && quality.Status == "supported" {
hashSize := int(trace.PathHashSize())
for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize {
traceRawHashes = append(traceRawHashes, trace.PathHashes[i:i+hashSize])
@@ -442,8 +451,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
obs.PropagationTimeMs = &prop
resolvedPath := []api.ResolvedHop{}
if row.PayloadType == int16(meshcore.PayloadTypeTrace) {
rawPath := hex.EncodeToString(v.PathBytes)
obs.WirePathBytes = &rawPath
obs.TraceQuality = p.TraceQuality
if len(traceRawHashes) > 0 {
resolved, err := s.ResolvePathHashes(ctx, v.Iata, traceRawHashes)
resolved, err := s.ResolveTracePathHashes(ctx, v.Iata, traceRawHashes)
if err != nil {
slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err)
} else {
@@ -491,7 +503,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
}
p.Observations = append(p.Observations, obs)
}
if row.PayloadType == 9 && len(obsRows) > 0 {
if row.PayloadType == 9 && len(obsRows) > 0 && p.TraceQuality.Status == "supported" {
iatas := make([]string, 0, len(obsRows))
seen := make(map[string]struct{})
for _, v := range obsRows {
@@ -503,6 +515,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
var parsed tracePayload
if err := json.Unmarshal(row.ParsedPayload, &parsed); err == nil {
p.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, iatas)
for _, hop := range p.ResolvedRoute {
if hop.Confidence == "ambiguous" {
p.TraceQuality.Add("ambiguous_prefix")
}
}
}
}
return p, nil
+35 -1
View File
@@ -455,6 +455,8 @@ WHERE p.packet_hash = $1;
-- Return distinct observation IATAs in first-heard order for path resolution,
-- without fetching full observations separately for every trace packet.
SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex,
p.raw_payload, p.payload_version,
EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -1429,7 +1431,11 @@ SELECT
WHERE ti.trace_tag = t.trace_tag
AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count,
COALESCE(t.trace_type, '')::text AS trace_type,
t.best_payload
t.best_payload,
ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas,
EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version,
EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct,
EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path
FROM trace_tags t
WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS (
SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[])))
@@ -1920,3 +1926,31 @@ ON CONFLICT (country) DO UPDATE SET
-- name: PruneZoneBoundaries :many
-- Drops imports for IATAs no longer configured, so their manual border returns.
DELETE FROM meshmapper_zone_boundaries WHERE NOT (iata = ANY(@keep::text[])) RETURNING iata;
-- name: ResolveTraceHashes :many
-- TRACE can terminate at companions. Preserve the full prefix (including eight
-- bytes), and return every candidate across heard regions instead of picking one.
SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key
FROM unnest(@hashes::bytea[]) AS h(hash)
JOIN node_short_ids si ON si.iata = ANY(@iatas::bpchar[]) AND
((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR
(octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR
(octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4)))
JOIN nodes n ON n.id=si.node_id
WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash
ORDER BY si.iata, h.hash, n.id;
-- name: GetTopologyLinks :many
-- Reduce route history to unique adjacent pairs before transferring it to a browser.
WITH pairs AS (
SELECT h.node_id AS a, r.node_ids[(h.ord+1)::int] AS b
FROM known_routes r CROSS JOIN LATERAL unnest(r.node_ids) WITH ORDINALITY h(node_id,ord)
WHERE r.last_seen >= @since::timestamptz AND r.last_seen < @until::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]),0)=0 OR r.iata=ANY(@iatas::bpchar[]))
AND h.ord < cardinality(r.node_ids)
)
SELECT LEAST(a,b)::uuid AS from_id, GREATEST(a,b)::uuid AS to_id
FROM pairs WHERE a IS NOT NULL AND b IS NOT NULL AND a<>b
GROUP BY LEAST(a,b), GREATEST(a,b)
ORDER BY LEAST(a,b), GREATEST(a,b)
LIMIT @link_limit::int;
+31 -1
View File
@@ -3,7 +3,7 @@
//
// Generated by this command:
//
// mockgen -source=db/sqlc/querier.go -destination=db/sqlc/mock/querier.go -package=mockdb
// mockgen -source db/sqlc/querier.go -destination db/sqlc/mock/querier.go -package mockdb
//
// Package mockdb is a generated GoMock package.
@@ -1166,6 +1166,21 @@ func (mr *MockQuerierMockRecorder) GetTopNodes(ctx, arg any) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopNodes", reflect.TypeOf((*MockQuerier)(nil).GetTopNodes), ctx, arg)
}
// GetTopologyLinks mocks base method.
func (m *MockQuerier) GetTopologyLinks(ctx context.Context, arg db.GetTopologyLinksParams) ([]db.GetTopologyLinksRow, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetTopologyLinks", ctx, arg)
ret0, _ := ret[0].([]db.GetTopologyLinksRow)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetTopologyLinks indicates an expected call of GetTopologyLinks.
func (mr *MockQuerierMockRecorder) GetTopologyLinks(ctx, arg any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopologyLinks", reflect.TypeOf((*MockQuerier)(nil).GetTopologyLinks), ctx, arg)
}
// GetTransportScopeByName mocks base method.
func (m *MockQuerier) GetTransportScopeByName(ctx context.Context, name string) (int32, error) {
m.ctrl.T.Helper()
@@ -2029,6 +2044,21 @@ func (mr *MockQuerierMockRecorder) ResolveRelayHashPairsP4(ctx, arg any) *gomock
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveRelayHashPairsP4", reflect.TypeOf((*MockQuerier)(nil).ResolveRelayHashPairsP4), ctx, arg)
}
// ResolveTraceHashes mocks base method.
func (m *MockQuerier) ResolveTraceHashes(ctx context.Context, arg db.ResolveTraceHashesParams) ([]db.ResolveTraceHashesRow, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ResolveTraceHashes", ctx, arg)
ret0, _ := ret[0].([]db.ResolveTraceHashesRow)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// ResolveTraceHashes indicates an expected call of ResolveTraceHashes.
func (mr *MockQuerierMockRecorder) ResolveTraceHashes(ctx, arg any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveTraceHashes", reflect.TypeOf((*MockQuerier)(nil).ResolveTraceHashes), ctx, arg)
}
// RollAdvertHearings mocks base method.
func (m *MockQuerier) RollAdvertHearings(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
+5
View File
@@ -162,6 +162,8 @@ type Querier interface {
// Nodes by ADVERT hearings since the given hour. Live names win over the rolled snapshot;
// node_id is NULL once the node row is gone. iata is a representative one.
GetTopNodes(ctx context.Context, arg GetTopNodesParams) ([]GetTopNodesRow, error)
// Reduce route history to unique adjacent pairs before transferring it to a browser.
GetTopologyLinks(ctx context.Context, arg GetTopologyLinksParams) ([]GetTopologyLinksRow, error)
GetTransportScopeByName(ctx context.Context, name string) (int32, error)
GetTransportScopes(ctx context.Context) ([]GetTransportScopesRow, error)
// ============================================================
@@ -335,6 +337,9 @@ type Querier interface {
ResolveRelayHashPairsP3(ctx context.Context, arg ResolveRelayHashPairsP3Params) ([]ResolveRelayHashPairsP3Row, error)
// Batch form of ResolvePathHashesP4 across IATAs; callers check uniqueness per IATA.
ResolveRelayHashPairsP4(ctx context.Context, arg ResolveRelayHashPairsP4Params) ([]ResolveRelayHashPairsP4Row, error)
// TRACE can terminate at companions. Preserve the full prefix (including eight
// bytes), and return every candidate across heard regions instead of picking one.
ResolveTraceHashes(ctx context.Context, arg ResolveTraceHashesParams) ([]ResolveTraceHashesRow, error)
RollAdvertHearings(ctx context.Context, hour pgtype.Timestamptz) error
// Each distinct ADVERT packet counts once, under the exact set of IATAs that heard it this hour.
RollAdvertSets(ctx context.Context, hour pgtype.Timestamptz) error
+146 -16
View File
@@ -1422,6 +1422,8 @@ func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (GetPa
const getPacketsByTraceTag = `-- name: GetPacketsByTraceTag :many
SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex,
p.raw_payload, p.payload_version,
EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -1442,14 +1444,17 @@ ORDER BY p.first_heard_at ASC
`
type GetPacketsByTraceTagRow struct {
PacketHashHex string `json:"packet_hash_hex"`
RouteType int16 `json:"route_type"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
ParsedPayload []byte `json:"parsed_payload"`
ScopeID *int32 `json:"scope_id"`
ScopeName *string `json:"scope_name"`
Iatas []string `json:"iatas"`
PacketHashHex string `json:"packet_hash_hex"`
RawPayload []byte `json:"raw_payload"`
PayloadVersion int16 `json:"payload_version"`
InvalidSnrPath bool `json:"invalid_snr_path"`
RouteType int16 `json:"route_type"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
ParsedPayload []byte `json:"parsed_payload"`
ScopeID *int32 `json:"scope_id"`
ScopeName *string `json:"scope_name"`
Iatas []string `json:"iatas"`
}
// Return distinct observation IATAs in first-heard order for path resolution,
@@ -1465,6 +1470,9 @@ func (q *Queries) GetPacketsByTraceTag(ctx context.Context, decode string) ([]Ge
var i GetPacketsByTraceTagRow
if err := rows.Scan(
&i.PacketHashHex,
&i.RawPayload,
&i.PayloadVersion,
&i.InvalidSnrPath,
&i.RouteType,
&i.FirstHeardAt,
&i.LastHeardAt,
@@ -2254,6 +2262,59 @@ func (q *Queries) GetTopNodes(ctx context.Context, arg GetTopNodesParams) ([]Get
return items, nil
}
const getTopologyLinks = `-- name: GetTopologyLinks :many
WITH pairs AS (
SELECT h.node_id AS a, r.node_ids[(h.ord+1)::int] AS b
FROM known_routes r CROSS JOIN LATERAL unnest(r.node_ids) WITH ORDINALITY h(node_id,ord)
WHERE r.last_seen >= $2::timestamptz AND r.last_seen < $3::timestamptz
AND (COALESCE(cardinality($4::bpchar[]),0)=0 OR r.iata=ANY($4::bpchar[]))
AND h.ord < cardinality(r.node_ids)
)
SELECT LEAST(a,b)::uuid AS from_id, GREATEST(a,b)::uuid AS to_id
FROM pairs WHERE a IS NOT NULL AND b IS NOT NULL AND a<>b
GROUP BY LEAST(a,b), GREATEST(a,b)
ORDER BY LEAST(a,b), GREATEST(a,b)
LIMIT $1::int
`
type GetTopologyLinksParams struct {
LinkLimit int32 `json:"link_limit"`
Since pgtype.Timestamptz `json:"since"`
Until pgtype.Timestamptz `json:"until"`
Iatas []string `json:"iatas"`
}
type GetTopologyLinksRow struct {
FromID uuid.UUID `json:"from_id"`
ToID uuid.UUID `json:"to_id"`
}
// Reduce route history to unique adjacent pairs before transferring it to a browser.
func (q *Queries) GetTopologyLinks(ctx context.Context, arg GetTopologyLinksParams) ([]GetTopologyLinksRow, error) {
rows, err := q.db.Query(ctx, getTopologyLinks,
arg.LinkLimit,
arg.Since,
arg.Until,
arg.Iatas,
)
if err != nil {
return nil, err
}
defer rows.Close()
items := []GetTopologyLinksRow{}
for rows.Next() {
var i GetTopologyLinksRow
if err := rows.Scan(&i.FromID, &i.ToID); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const getTransportScopeByName = `-- name: GetTransportScopeByName :one
SELECT id FROM transport_scopes WHERE name = $1
`
@@ -4194,7 +4255,11 @@ SELECT
WHERE ti.trace_tag = t.trace_tag
AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count,
COALESCE(t.trace_type, '')::text AS trace_type,
t.best_payload
t.best_payload,
ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas,
EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version,
EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct,
EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path
FROM trace_tags t
WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS (
SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[])))
@@ -4222,13 +4287,17 @@ type ListTraceTagsParams struct {
}
type ListTraceTagsRow struct {
TraceTag string `json:"trace_tag"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
PacketCount int64 `json:"packet_count"`
IataCount int64 `json:"iata_count"`
TraceType string `json:"trace_type"`
BestPayload []byte `json:"best_payload"`
TraceTag string `json:"trace_tag"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
PacketCount int64 `json:"packet_count"`
IataCount int64 `json:"iata_count"`
TraceType string `json:"trace_type"`
BestPayload []byte `json:"best_payload"`
Iatas []string `json:"iatas"`
UnsupportedVersion bool `json:"unsupported_version"`
NonDirect bool `json:"non_direct"`
InvalidSnrPath bool `json:"invalid_snr_path"`
}
// ============================================================
@@ -4261,6 +4330,10 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([
&i.IataCount,
&i.TraceType,
&i.BestPayload,
&i.Iatas,
&i.UnsupportedVersion,
&i.NonDirect,
&i.InvalidSnrPath,
); err != nil {
return nil, err
}
@@ -5103,6 +5176,63 @@ func (q *Queries) ResolveRelayHashPairsP4(ctx context.Context, arg ResolveRelayH
return items, nil
}
const resolveTraceHashes = `-- name: ResolveTraceHashes :many
SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key
FROM unnest($1::bytea[]) AS h(hash)
JOIN node_short_ids si ON si.iata = ANY($2::bpchar[]) AND
((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR
(octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR
(octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4)))
JOIN nodes n ON n.id=si.node_id
WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash
ORDER BY si.iata, h.hash, n.id
`
type ResolveTraceHashesParams struct {
Hashes [][]byte `json:"hashes"`
Iatas []string `json:"iatas"`
}
type ResolveTraceHashesRow struct {
Hash []byte `json:"hash"`
Iata string `json:"iata"`
NodeID uuid.UUID `json:"node_id"`
Name *string `json:"name"`
Latitude *float64 `json:"latitude"`
Longitude *float64 `json:"longitude"`
PublicKey []byte `json:"public_key"`
}
// TRACE can terminate at companions. Preserve the full prefix (including eight
// bytes), and return every candidate across heard regions instead of picking one.
func (q *Queries) ResolveTraceHashes(ctx context.Context, arg ResolveTraceHashesParams) ([]ResolveTraceHashesRow, error) {
rows, err := q.db.Query(ctx, resolveTraceHashes, arg.Hashes, arg.Iatas)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ResolveTraceHashesRow{}
for rows.Next() {
var i ResolveTraceHashesRow
if err := rows.Scan(
&i.Hash,
&i.Iata,
&i.NodeID,
&i.Name,
&i.Latitude,
&i.Longitude,
&i.PublicKey,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const saveChannelCatalogue = `-- name: SaveChannelCatalogue :exec
INSERT INTO meshmapper_channel_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error)
VALUES ($1, $2, $3::jsonb, $4::text,
+25
View File
@@ -0,0 +1,25 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"time"
)
func (s *Store) GetTopologyLinks(ctx context.Context, iatas []string, since, until time.Time) (*api.TopologyLinks, error) {
const limit = 100000
rows, err := s.q.GetTopologyLinks(ctx, sqlc.GetTopologyLinksParams{Iatas: iatas, Since: pgtype.Timestamptz{Time: since, Valid: true}, Until: pgtype.Timestamptz{Time: until, Valid: true}, LinkLimit: limit + 1})
if err != nil {
return nil, err
}
result := &api.TopologyLinks{Links: make([][2]uuid.UUID, 0, min(len(rows), limit)), Capped: len(rows) > limit, Since: since.UnixMilli(), Until: until.UnixMilli()}
for _, r := range rows[:min(len(rows), limit)] {
result.Links = append(result.Links, [2]uuid.UUID{r.FromID, r.ToID})
}
return result, nil
}
+54
View File
@@ -0,0 +1,54 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/google/uuid"
"testing"
"time"
)
func TestTopologyLinksPostgres(t *testing.T) {
ctx, pool := schemaPool(t)
if err := RunMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
tx, err := pool.Begin(ctx)
if err != nil {
t.Fatal(err)
}
defer tx.Rollback(context.Background())
if _, err = tx.Exec(ctx, "CREATE TEMP TABLE known_routes (LIKE public.known_routes INCLUDING ALL) ON COMMIT DROP"); err != nil {
t.Fatal(err)
}
ids := []uuid.UUID{uuid.New(), uuid.New(), uuid.New()}
now := time.Now().UTC().Truncate(time.Minute)
// Thousands of duplicate/reversed routes transfer only their unique adjacent pairs.
_, err = tx.Exec(ctx, `INSERT INTO known_routes (path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen)
SELECT decode(md5(i::text),'hex'),CASE WHEN i%2=0 THEN $1::uuid[] ELSE $2::uuid[] END,ARRAY['\xaa'::bytea,'\xbb'::bytea],
'YKF',2,$3,$3 FROM generate_series(1,1000) i`, ids[:2], []uuid.UUID{ids[1], ids[0]}, now)
if err != nil {
t.Fatal(err)
}
_, err = tx.Exec(ctx, `INSERT INTO known_routes(path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen) VALUES
(decode(md5('other'),'hex'),$1,ARRAY['\xbb'::bytea,'\xcc'::bytea],'YOW',2,$3,$3),
(decode(md5('gap'),'hex'),$2,ARRAY['\xaa'::bytea,'\x00'::bytea,'\xcc'::bytea],'YKF',3,$3,$3),
(decode(md5('old'),'hex'),$1,ARRAY['\xbb'::bytea,'\xcc'::bytea],'YKF',2,$3-interval '25 hours',$3-interval '25 hours')`, ids[1:], []*uuid.UUID{&ids[0], nil, &ids[2]}, now)
if err != nil {
t.Fatal(err)
}
store := &Store{q: sqlc.New(tx)}
result, err := store.GetTopologyLinks(ctx, []string{"YKF"}, now.Add(-24*time.Hour), now.Add(time.Minute))
if err != nil || result.Capped || len(result.Links) != 1 {
t.Fatalf("scoped: %+v %v", result, err)
}
all, err := store.GetTopologyLinks(ctx, nil, now.Add(-24*time.Hour), now.Add(time.Minute))
if err != nil || len(all.Links) != 2 {
t.Fatalf("all regions: %+v %v", all, err)
}
if result.Since != now.Add(-24*time.Hour).UnixMilli() {
t.Fatal("window changed")
}
}
+2 -1
View File
@@ -16,6 +16,7 @@ import (
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/MeshCore-Beacon/beacon-server/internal/ingest"
"github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
"github.com/jackc/pgx/v5"
)
@@ -89,7 +90,7 @@ func TestTraceTagPaginationPostgres(t *testing.T) {
paths := map[string][]string{"a": {"aa", "bb", "cc"}, "b": {"bb", "11", "22", "33"}, "c": {"cc"}, "d": {"dd"}}
snrs := map[string][]float32{"a": {1, 2, 3}, "b": {2, 3, 4, 5}, "c": {3}, "d": {4}}
summary := func(tag string, first, last int, count, iatas int64, kind string) api.TraceTagSummary {
return api.TraceTagSummary{TraceTag: strings.Repeat(tag, 8), FirstHeardAt: at(first).UnixMilli(), LastHeardAt: at(last).UnixMilli(), PacketCount: count, IATACount: iatas, TraceType: kind, PathHashes: paths[tag], SNRValues: snrs[tag]}
return api.TraceTagSummary{Quality: tracequality.Quality{Status: "suspect", Reasons: []string{"non_direct_trace"}}, TraceTag: strings.Repeat(tag, 8), FirstHeardAt: at(first).UnixMilli(), LastHeardAt: at(last).UnixMilli(), PacketCount: count, IATACount: iatas, TraceType: kind, PathHashes: paths[tag], SNRValues: snrs[tag]}
}
a, b, c, d := summary("a", 1, 40, 3, 2, "TRACE"), summary("b", 4, 30, 2, 1, "TRACE"), summary("c", 6, 20, 1, 1, "TRACE"), summary("d", 7, 15, 1, 1, "PING")
for _, tc := range []struct {
+77
View File
@@ -0,0 +1,77 @@
package db
import (
"context"
"encoding/hex"
"os"
"testing"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/jackc/pgx/v5"
)
func TestTraceQualityPostgres(t *testing.T) {
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set BEACON_TEST_POSTGRES_DSN")
}
ctx := context.Background()
conn, err := pgx.Connect(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer conn.Close(ctx)
tx, err := conn.Begin(ctx)
if err != nil {
t.Fatal(err)
}
defer tx.Rollback(ctx)
for _, name := range []string{"nodes", "node_short_ids"} {
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+name+" (LIKE public."+name+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
t.Fatal(err)
}
}
_, err = tx.Exec(ctx, `
INSERT INTO nodes (id,public_key,node_type,name) VALUES
('00000000-0000-0000-0000-000000000001',decode('aabbccdd11223344'||repeat('00',24),'hex'),1,'Companion'),
('00000000-0000-0000-0000-000000000002',decode('aabbccdd55667788'||repeat('00',24),'hex'),2,'Repeater');
INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES
('00000000-0000-0000-0000-000000000001','YKF','\xaabbccdd'),
('00000000-0000-0000-0000-000000000002','YKF','\xaabbccdd'),
('00000000-0000-0000-0000-000000000001','YHM','\xaabbccdd');`)
if err != nil {
t.Fatal(err)
}
store := &Store{q: sqlc.New(tx)}
for _, width := range []int{1, 2, 4, 8} {
full := "aabbccdd11223344"
h := full[:width*2]
b, _ := hex.DecodeString(h)
candidates, err := store.traceCandidates(ctx, []string{"YKF", "YHM"}, [][]byte{b})
if err != nil {
t.Fatal(err)
}
flags := byte(0)
switch width {
case 2:
flags = 1
case 4:
flags = 2
case 8:
flags = 3
}
route := traceRoute(&tracePayload{Flags: flags, PathHashes: []string{h}}, []string{"YKF", "YHM"}, candidates)
count := 2
confidence := "ambiguous"
if width == 8 {
count = 1
confidence = "high"
}
if len(route) != 1 || len(route[0].Nodes) != count || route[0].Confidence != confidence {
t.Fatalf("width %d: %+v", width, route)
}
if route[0].Nodes[0].PublicKey[:16] != "aabbccdd11223344" {
t.Fatal("companion excluded or eight-byte hash truncated")
}
}
}
+127 -52
View File
@@ -8,6 +8,8 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
"github.com/google/uuid"
"time"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
@@ -17,9 +19,11 @@ import (
)
type tracePayload struct {
PathHashes []string `json:"pathHashes"`
Flags byte `json:"flags"`
SNRValues []float32 `json:"snrValues"`
Raw string `json:"raw"`
Quality tracequality.Quality `json:"quality"`
PathHashes []string `json:"pathHashes"`
Flags byte `json:"flags"`
SNRValues []float32 `json:"snrValues"`
}
func (s *Store) RecordTrace(ctx context.Context, h ingest.TraceHearing) error {
@@ -71,12 +75,41 @@ func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope, traceT
return nil, err
}
items := make([]api.TraceTagSummary, 0, len(rows))
payloads := make([]tracePayload, 0, len(rows))
allIATAs := map[string]bool{}
allHashes := map[string][]byte{}
for _, r := range rows {
var best tracePayload
if len(r.BestPayload) > 0 {
_ = json.Unmarshal(r.BestPayload, &best)
}
quality := tracequality.Stored(best.Raw, best.Flags, best.PathHashes, best.SNRValues)
for _, reason := range best.Quality.Reasons {
if reason != "ambiguous_prefix" {
quality.Add(reason)
}
}
if r.UnsupportedVersion {
quality.Add("unsupported_version")
}
if r.InvalidSnrPath {
quality.Add("invalid_snr_path")
}
if r.NonDirect {
quality.Add("non_direct_trace")
}
payloads = append(payloads, best)
if quality.Status == "supported" {
for _, iata := range r.Iatas {
allIATAs[iata] = true
}
for _, h := range best.PathHashes {
b, _ := hex.DecodeString(h)
allHashes[h] = b
}
}
items = append(items, api.TraceTagSummary{
Quality: quality,
TraceTag: r.TraceTag,
FirstHeardAt: r.FirstHeardAt.Time.UnixMilli(),
LastHeardAt: r.LastHeardAt.Time.UnixMilli(),
@@ -87,6 +120,30 @@ func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope, traceT
SNRValues: best.SNRValues,
})
}
// One bounded candidate query for the entire page, including all hash widths.
regions := make([]string, 0, len(allIATAs))
for iata := range allIATAs {
regions = append(regions, iata)
}
hashes := make([][]byte, 0, len(allHashes))
for _, h := range allHashes {
hashes = append(hashes, h)
}
candidates, err := s.traceCandidates(ctx, regions, hashes)
if err != nil {
return nil, err
}
for i := range items {
if items[i].Quality.Status != "supported" {
continue
}
for _, hop := range traceRoute(&payloads[i], rows[i].Iatas, candidates) {
if hop.Confidence == "ambiguous" {
items[i].Quality.Add("ambiguous_prefix")
break
}
}
}
return items, nil
}
@@ -125,76 +182,94 @@ func (s *Store) GetTraceByTag(ctx context.Context, tag string) (*api.TraceDetail
}
packet.RawPath = rawPath
}
if len(r.Iatas) > 0 {
packet.Quality = tracequality.Assess(r.RawPayload, int(r.PayloadVersion), int(r.RouteType), len(parsed.SNRValues))
if r.InvalidSnrPath {
packet.Quality.Add("invalid_snr_path")
}
stored := tracequality.Stored(parsed.Raw, parsed.Flags, parsed.PathHashes, parsed.SNRValues)
for _, reason := range stored.Reasons {
packet.Quality.Add(reason)
}
if len(r.Iatas) > 0 && packet.Quality.Status == "supported" {
packet.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, r.Iatas)
}
for _, hop := range packet.ResolvedRoute {
if hop.Confidence == "ambiguous" {
packet.Quality.Add("ambiguous_prefix")
}
}
detail.Packets = append(detail.Packets, packet)
}
return detail, nil
}
// Candidates remain separated by hearing region until a trace is assembled.
// Never replace an ambiguous region with a convenient singleton from another.
type traceCandidateMap map[string]map[string][]api.ResolvedPathEntry
func (s *Store) traceCandidates(ctx context.Context, iatas []string, hashes [][]byte) (traceCandidateMap, error) {
out := traceCandidateMap{}
if len(iatas) == 0 || len(hashes) == 0 {
return out, nil
}
rows, err := s.q.ResolveTraceHashes(ctx, sqlc.ResolveTraceHashesParams{Iatas: iatas, Hashes: hashes})
if err != nil {
return nil, err
}
for _, r := range rows {
if out[r.Iata] == nil {
out[r.Iata] = map[string][]api.ResolvedPathEntry{}
}
h := hex.EncodeToString(r.Hash)
out[r.Iata][h] = append(out[r.Iata][h], api.ResolvedPathEntry{NodeID: r.NodeID, Name: r.Name, Latitude: r.Latitude, Longitude: r.Longitude, PublicKey: r.PublicKey})
}
return out, nil
}
func (s *Store) ResolveTracePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) {
candidates, err := s.traceCandidates(ctx, []string{iata}, hashes)
return candidates[iata], err
}
func (s *Store) resolveTraceRoute(ctx context.Context, payload *tracePayload, iatas []string) []api.ResolvedHop {
if payload == nil || len(payload.PathHashes) == 0 {
if payload == nil || tracequality.Stored(payload.Raw, payload.Flags, payload.PathHashes, payload.SNRValues).Status != "supported" {
return nil
}
hashSize := int(1 << (payload.Flags & 0x03))
hashes := make([][]byte, 0, len(payload.PathHashes))
for _, h := range payload.PathHashes {
b, err := hex.DecodeString(h)
if err == nil {
hashes = append(hashes, b)
}
b, _ := hex.DecodeString(h)
hashes = append(hashes, b)
}
confidenceRank := map[string]int{"none": 0, "ambiguous": 1, "high": 2}
type hopResult struct {
confidence string
entries []api.ResolvedPathEntry
candidates, err := s.traceCandidates(ctx, iatas, hashes)
if err != nil {
return nil
}
merged := make([]hopResult, len(hashes))
for i := range merged {
merged[i] = hopResult{confidence: "none"}
}
for _, iata := range iatas {
resolved, err := s.ResolvePathHashes(ctx, iata, hashes)
if err != nil {
continue
}
for i, hash := range hashes {
key := hex.EncodeToString(hash[:hashSize])
entries := resolved[key]
var confidence string
switch len(entries) {
case 0:
confidence = "none"
case 1:
confidence = "high"
default:
confidence = "ambiguous"
}
if confidenceRank[confidence] > confidenceRank[merged[i].confidence] {
merged[i] = hopResult{confidence: confidence, entries: entries}
return traceRoute(payload, iatas, candidates)
}
func traceRoute(payload *tracePayload, iatas []string, candidates traceCandidateMap) []api.ResolvedHop {
route := make([]api.ResolvedHop, 0, len(payload.PathHashes))
for i, h := range payload.PathHashes {
hop := api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}}
seen := map[uuid.UUID]bool{}
for _, iata := range iatas {
for _, e := range candidates[iata][h] {
if seen[e.NodeID] {
continue
}
seen[e.NodeID] = true
hop.Nodes = append(hop.Nodes, api.ResolvedNode{ID: e.NodeID, Name: e.Name, Latitude: e.Latitude, Longitude: e.Longitude, PublicKey: hex.EncodeToString(e.PublicKey)})
}
}
}
route := make([]api.ResolvedHop, 0, len(hashes))
for i, hr := range merged {
hop := api.ResolvedHop{
Confidence: hr.confidence,
Nodes: make([]api.ResolvedNode, 0, len(hr.entries)),
if len(hop.Nodes) == 1 {
hop.Confidence = "high"
} else if len(hop.Nodes) > 1 {
hop.Confidence = "ambiguous"
}
if i < len(payload.SNRValues) {
snr := payload.SNRValues[i]
hop.SNR = &snr
}
for _, e := range hr.entries {
hop.Nodes = append(hop.Nodes, api.ResolvedNode{
ID: e.NodeID,
Name: e.Name,
Latitude: e.Latitude,
Longitude: e.Longitude,
PublicKey: hex.EncodeToString(e.PublicKey),
})
}
route = append(route, hop)
}
return route
+8 -5
View File
@@ -59,7 +59,7 @@ INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES
('00000000-0000-0000-0000-000000000001','YVR','\xaa000000'),
('00000000-0000-0000-0000-000000000002','YYJ','\xaa010000');
INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,trace_tag,parsed_payload,first_heard_at,last_heard_at)
SELECT decode(lpad(to_hex(i),64,'0'),'hex'),9,0,1,'\x00','\x00','\x01020304',
SELECT decode(lpad(to_hex(i),64,'0'),'hex'),9,0,2,'\x010203040000000000aa','\x26','\x01020304',
CASE WHEN i=100 THEN '{}'::jsonb ELSE '{"flags":0,"pathHashes":["aa"],"snrValues":[12.5]}'::jsonb END,
'2026-01-01 00:00:00+00'::timestamptz+i*interval '1 second', '2026-01-01 00:00:00+00'::timestamptz+i*interval '1 second'+interval '1 second'
FROM generate_series(1,100) i;
@@ -89,7 +89,7 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_
anchor := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
for i, p := range detail.Packets {
number := i + 1
if p.PacketHash != fmt.Sprintf("%064x", number) || p.RouteType != 1 || p.FirstHeardAt != anchor.Add(time.Duration(number)*time.Second).UnixMilli() || p.LastHeardAt != p.FirstHeardAt+1000 {
if p.PacketHash != fmt.Sprintf("%064x", number) || p.RouteType != 2 || p.FirstHeardAt != anchor.Add(time.Duration(number)*time.Second).UnixMilli() || p.LastHeardAt != p.FirstHeardAt+1000 {
t.Fatalf("wrong packet fields or order at %d", number)
}
if number == 100 {
@@ -109,7 +109,10 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_
}
want := "YVR relay"
if number == 1 {
want = "YYJ relay"
if len(p.ResolvedRoute) != 1 || p.ResolvedRoute[0].Confidence != "ambiguous" || len(p.ResolvedRoute[0].Nodes) != 2 || p.Quality.Status != "ambiguous" {
t.Fatal("conflicting regions must retain both candidates")
}
continue
}
if len(p.ResolvedRoute) != 1 || p.ResolvedRoute[0].Confidence != "high" || len(p.ResolvedRoute[0].Nodes) != 1 || p.ResolvedRoute[0].Nodes[0].Name == nil || *p.ResolvedRoute[0].Nodes[0].Name != want {
t.Fatalf("wrong region precedence/resolution at %d", number)
@@ -122,8 +125,8 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_
if counter.observations != 0 {
t.Errorf("extra observation queries: %d; want 0", counter.observations)
}
if counter.queries != 100 {
t.Errorf("got %d queries; want one packet query and 99 unchanged path-resolution queries", counter.queries)
if counter.queries != 99 {
t.Errorf("got %d queries; want one packet query and 98 batched path-resolution queries", counter.queries)
}
counter.queries, counter.observations = 0, 0
missing, err := store.GetTraceByTag(ctx, "00000000")
+7 -6
View File
@@ -99,7 +99,7 @@ func TestGetTraceByTag_WithPacket(t *testing.T) {
firstHeard := pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}
lastHeard := pgtype.Timestamptz{Time: time.UnixMilli(1700000001000), Valid: true}
// aabbccdd is valid hex for the packet hash
parsedPayload := []byte(`{"pathHashes":["aabb"],"snrValues":[15.0],"flags":0}`)
parsedPayload := []byte(`{"pathHashes":["aabb"],"snrValues":[15.0],"flags":1}`)
scopeName := "default"
mock.EXPECT().
@@ -107,7 +107,8 @@ func TestGetTraceByTag_WithPacket(t *testing.T) {
Return([]sqlc.GetPacketsByTraceTagRow{
{
PacketHashHex: "aabbccdd",
RouteType: 1,
RouteType: 2,
RawPayload: []byte{0, 0, 0, 0, 0, 0, 0, 0, 1, 0xaa, 0xbb},
ScopeName: &scopeName,
FirstHeardAt: firstHeard,
LastHeardAt: lastHeard,
@@ -117,11 +118,11 @@ func TestGetTraceByTag_WithPacket(t *testing.T) {
}, nil)
mock.EXPECT().
ResolvePathHashesP2(gomock.Any(), sqlc.ResolvePathHashesP2Params{
Iata: "YVR",
Column2: [][]byte{{0xaa, 0xbb}},
ResolveTraceHashes(gomock.Any(), sqlc.ResolveTraceHashesParams{
Iatas: []string{"YVR"},
Hashes: [][]byte{{0xaa, 0xbb}},
}).
Return([]sqlc.ResolvePathHashesP2Row{}, nil)
Return([]sqlc.ResolveTraceHashesRow{}, nil)
store := &Store{q: mock}
detail, err := store.GetTraceByTag(context.Background(), "trace-001")
+179 -1
View File
@@ -2343,6 +2343,52 @@ const docTemplate = `{
}
}
},
"/routes/topology": {
"get": {
"description": "Returns up to 100000 undirected node-ID pairs, deduplicated across known routes. Minute-aligned window includes the current minute. Existing rate limits apply; no per-node lookup or route pagination is needed. Capped is true when more links exist.",
"produces": [
"application/json"
],
"tags": [
"Routes"
],
"summary": "Unique adjacent route links for a topology snapshot",
"parameters": [
{
"type": "string",
"description": "History window: 15m (default), 1h, or 24h",
"name": "window",
"in": "query"
},
{
"type": "string",
"description": "Comma-separated reception IATA codes",
"name": "iatas",
"in": "query"
}
],
"responses": {
"200": {
"description": "OK",
"schema": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks"
}
},
"400": {
"description": "Bad Request",
"schema": {
"$ref": "#/definitions/internal_api_handlers.APIError"
}
},
"503": {
"description": "Service Unavailable",
"schema": {
"$ref": "#/definitions/internal_api_handlers.APIError"
}
}
}
}
},
"/routes/{iata}/{pathKey}/observations": {
"get": {
"description": "Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope.",
@@ -2429,6 +2475,29 @@ const docTemplate = `{
}
}
},
"/scope-catalogues": {
"get": {
"description": "Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.",
"produces": [
"application/json"
],
"tags": [
"Scopes"
],
"summary": "Cached MeshMapper regional scope catalogues",
"responses": {
"200": {
"description": "OK",
"schema": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue"
}
}
}
}
}
},
"/scopes": {
"get": {
"description": "Without filters, lists every stored scope name, including imported names retained after an importer is disabled or a source is removed. With IATA or region filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; observed traffic does not add scopes.",
@@ -4996,6 +5065,9 @@ const docTemplate = `{
"description": "matched transport scope name e.g. \"#bc\"",
"type": "string"
},
"traceQuality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"transportCodes": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes"
}
@@ -5127,6 +5199,13 @@ const docTemplate = `{
},
"sourceBroker": {
"type": "string"
},
"traceQuality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"wirePathBytes": {
"description": "Exact on-air TRACE SNR bytes, before the display path substitution.",
"type": "string"
}
}
},
@@ -6045,6 +6124,29 @@ const docTemplate = `{
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks": {
"type": "object",
"properties": {
"capped": {
"type": "boolean"
},
"links": {
"type": "array",
"items": {
"type": "array",
"items": {
"type": "string"
}
}
},
"since": {
"type": "integer"
},
"until": {
"type": "integer"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail": {
"type": "object",
"properties": {
@@ -6076,6 +6178,9 @@ const docTemplate = `{
"description": "hex-encoded packet hash",
"type": "string"
},
"quality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"rawPath": {
"description": "hops as received in the packet",
"type": "array",
@@ -6130,6 +6235,9 @@ const docTemplate = `{
"type": "string"
}
},
"quality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"snrValues": {
"description": "SNR per hop from the most complete observation",
"type": "array",
@@ -6186,6 +6294,76 @@ const docTemplate = `{
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope": {
"type": "object",
"properties": {
"default": {
"type": "integer"
},
"monitored": {
"type": "boolean"
},
"name": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"wardriving": {
"type": "boolean"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue": {
"type": "object",
"properties": {
"checkedAt": {
"type": "integer"
},
"freshUntil": {
"type": "integer"
},
"generatedAt": {
"type": "integer"
},
"iata": {
"type": "string"
},
"lastError": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"scoped": {
"type": "integer"
},
"scopes": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope"
}
},
"url": {
"type": "string"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality": {
"type": "object",
"properties": {
"reasons": {
"type": "array",
"items": {
"type": "string"
}
},
"status": {
"description": "supported, suspect, or ambiguous; supported is not authenticated",
"type": "string"
}
}
},
"internal_api_handlers.APIError": {
"type": "object",
"properties": {
@@ -6265,7 +6443,7 @@ const docTemplate = `{
// SwaggerInfo holds exported Swagger Info so clients can modify it
var SwaggerInfo = &swag.Spec{
Version: "2.0.6",
Version: "2.1.0",
Host: "localhost:8080",
BasePath: "/api/v1",
Schemes: []string{"http", "https"},
+179 -1
View File
@@ -15,7 +15,7 @@
"license": {
"name": "AGPL-3-or-later"
},
"version": "2.0.6"
"version": "2.1.0"
},
"host": "localhost:8080",
"basePath": "/api/v1",
@@ -2341,6 +2341,52 @@
}
}
},
"/routes/topology": {
"get": {
"description": "Returns up to 100000 undirected node-ID pairs, deduplicated across known routes. Minute-aligned window includes the current minute. Existing rate limits apply; no per-node lookup or route pagination is needed. Capped is true when more links exist.",
"produces": [
"application/json"
],
"tags": [
"Routes"
],
"summary": "Unique adjacent route links for a topology snapshot",
"parameters": [
{
"type": "string",
"description": "History window: 15m (default), 1h, or 24h",
"name": "window",
"in": "query"
},
{
"type": "string",
"description": "Comma-separated reception IATA codes",
"name": "iatas",
"in": "query"
}
],
"responses": {
"200": {
"description": "OK",
"schema": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks"
}
},
"400": {
"description": "Bad Request",
"schema": {
"$ref": "#/definitions/internal_api_handlers.APIError"
}
},
"503": {
"description": "Service Unavailable",
"schema": {
"$ref": "#/definitions/internal_api_handlers.APIError"
}
}
}
}
},
"/routes/{iata}/{pathKey}/observations": {
"get": {
"description": "Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope.",
@@ -2427,6 +2473,29 @@
}
}
},
"/scope-catalogues": {
"get": {
"description": "Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.",
"produces": [
"application/json"
],
"tags": [
"Scopes"
],
"summary": "Cached MeshMapper regional scope catalogues",
"responses": {
"200": {
"description": "OK",
"schema": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue"
}
}
}
}
}
},
"/scopes": {
"get": {
"description": "Without filters, lists every stored scope name, including imported names retained after an importer is disabled or a source is removed. With IATA or region filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; observed traffic does not add scopes.",
@@ -4994,6 +5063,9 @@
"description": "matched transport scope name e.g. \"#bc\"",
"type": "string"
},
"traceQuality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"transportCodes": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes"
}
@@ -5125,6 +5197,13 @@
},
"sourceBroker": {
"type": "string"
},
"traceQuality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"wirePathBytes": {
"description": "Exact on-air TRACE SNR bytes, before the display path substitution.",
"type": "string"
}
}
},
@@ -6043,6 +6122,29 @@
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks": {
"type": "object",
"properties": {
"capped": {
"type": "boolean"
},
"links": {
"type": "array",
"items": {
"type": "array",
"items": {
"type": "string"
}
}
},
"since": {
"type": "integer"
},
"until": {
"type": "integer"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail": {
"type": "object",
"properties": {
@@ -6074,6 +6176,9 @@
"description": "hex-encoded packet hash",
"type": "string"
},
"quality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"rawPath": {
"description": "hops as received in the packet",
"type": "array",
@@ -6128,6 +6233,9 @@
"type": "string"
}
},
"quality": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality"
},
"snrValues": {
"description": "SNR per hop from the most complete observation",
"type": "array",
@@ -6184,6 +6292,76 @@
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope": {
"type": "object",
"properties": {
"default": {
"type": "integer"
},
"monitored": {
"type": "boolean"
},
"name": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"wardriving": {
"type": "boolean"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue": {
"type": "object",
"properties": {
"checkedAt": {
"type": "integer"
},
"freshUntil": {
"type": "integer"
},
"generatedAt": {
"type": "integer"
},
"iata": {
"type": "string"
},
"lastError": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"scoped": {
"type": "integer"
},
"scopes": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope"
}
},
"url": {
"type": "string"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality": {
"type": "object",
"properties": {
"reasons": {
"type": "array",
"items": {
"type": "string"
}
},
"status": {
"description": "supported, suspect, or ambiguous; supported is not authenticated",
"type": "string"
}
}
},
"internal_api_handlers.APIError": {
"type": "object",
"properties": {
+125 -1
View File
@@ -978,6 +978,8 @@ definitions:
scope:
description: matched transport scope name e.g. "#bc"
type: string
traceQuality:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality'
transportCodes:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes'
type: object
@@ -1077,6 +1079,11 @@ definitions:
type: number
sourceBroker:
type: string
traceQuality:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality'
wirePathBytes:
description: Exact on-air TRACE SNR bytes, before the display path substitution.
type: string
type: object
github_com_MeshCore-Beacon_beacon-server_internal_api.PacketObservationSummary:
properties:
@@ -1698,6 +1705,21 @@ definitions:
senderName:
type: string
type: object
github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks:
properties:
capped:
type: boolean
links:
items:
items:
type: string
type: array
type: array
since:
type: integer
until:
type: integer
type: object
github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail:
properties:
packets:
@@ -1720,6 +1742,8 @@ definitions:
packetHash:
description: hex-encoded packet hash
type: string
quality:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality'
rawPath:
description: hops as received in the packet
items:
@@ -1759,6 +1783,8 @@ definitions:
items:
type: string
type: array
quality:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality'
snrValues:
description: SNR per hop from the most complete observation
items:
@@ -1796,6 +1822,52 @@ definitions:
requires_restart:
type: boolean
type: object
github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope:
properties:
default:
type: integer
monitored:
type: boolean
name:
type: string
repeaters:
type: integer
wardriving:
type: boolean
type: object
github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue:
properties:
checkedAt:
type: integer
freshUntil:
type: integer
generatedAt:
type: integer
iata:
type: string
lastError:
type: string
repeaters:
type: integer
scoped:
type: integer
scopes:
items:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope'
type: array
url:
type: string
type: object
github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality:
properties:
reasons:
items:
type: string
type: array
status:
description: supported, suspect, or ambiguous; supported is not authenticated
type: string
type: object
internal_api_handlers.APIError:
properties:
code:
@@ -1825,7 +1897,7 @@ info:
name: AGPL-3-or-later
termsOfService: https://github.com/MeshCore-Beacon/beacon-server
title: MeshCore Beacon API
version: 2.0.6
version: 2.1.0
paths:
/admin/accounts:
get:
@@ -3474,6 +3546,58 @@ paths:
summary: Search known routes by source and destination hash
tags:
- Routes
/routes/topology:
get:
description: Returns up to 100000 undirected node-ID pairs, deduplicated across
known routes. Minute-aligned window includes the current minute. Existing
rate limits apply; no per-node lookup or route pagination is needed. Capped
is true when more links exist.
parameters:
- description: 'History window: 15m (default), 1h, or 24h'
in: query
name: window
type: string
- description: Comma-separated reception IATA codes
in: query
name: iatas
type: string
produces:
- application/json
responses:
"200":
description: OK
schema:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks'
"400":
description: Bad Request
schema:
$ref: '#/definitions/internal_api_handlers.APIError'
"503":
description: Service Unavailable
schema:
$ref: '#/definitions/internal_api_handlers.APIError'
summary: Unique adjacent route links for a topology snapshot
tags:
- Routes
/scope-catalogues:
get:
description: Discovered sources only. Counts describe the source region, not
individual nodes or links. No upstream request is made. An empty list means
no configured importer; checkedAt=0 means no successful check. lastError or
a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence
are separate.
produces:
- application/json
responses:
"200":
description: OK
schema:
items:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue'
type: array
summary: Cached MeshMapper regional scope catalogues
tags:
- Scopes
/scopes:
get:
description: Without filters, lists every stored scope name, including imported
+1
View File
@@ -24,6 +24,7 @@ import (
func RoutesRouter(reader api.Reader) http.Handler {
r := chi.NewRouter()
r.Get("/", listKnownRoutes(reader))
r.Get("/topology", getTopologyLinks(reader))
r.Get("/cross", searchCrossIATARoutes(reader))
r.Get("/search", searchKnownRoutes(reader))
r.Get("/{iata}/{pathKey}/observations", getRouteEvidence(reader))
+33
View File
@@ -0,0 +1,33 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package handlers
import (
"net/http"
"github.com/MeshCore-Beacon/beacon-server/internal/meshmapper"
)
// ScopeCatalogueSource supplies immutable, already-cached public metadata.
type ScopeCatalogueSource interface {
Catalogues() []meshmapper.PublicScopeCatalogue
}
// ScopeCatalogues godoc
// @Summary Cached MeshMapper regional scope catalogues
// @Description Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.
// @Tags Scopes
// @Produce json
// @Success 200 {array} meshmapper.PublicScopeCatalogue
// @Router /scope-catalogues [get]
func ScopeCatalogues(source ScopeCatalogueSource) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
items := []meshmapper.PublicScopeCatalogue{}
if source != nil {
items = source.Catalogues()
}
w.Header().Set("Cache-Control", "public, max-age=60")
respond(w, http.StatusOK, items)
}
}
@@ -403,6 +403,10 @@ func (s stubReader) GetRouteEvidence(ctx context.Context, iata, key string, q ap
return nil, nil
}
func (s stubReader) GetTopologyLinks(_ context.Context, _ []string, _ time.Time, _ time.Time) (*api.TopologyLinks, error) {
return &api.TopologyLinks{}, nil
}
func (s stubReader) ListObserverDirectory(ctx context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) {
if s.listObserverDirectory != nil {
return s.listObserverDirectory(ctx, q)
+49
View File
@@ -0,0 +1,49 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package handlers
import (
"context"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"net/http"
"time"
)
// getTopologyLinks godoc
// @Summary Unique adjacent route links for a topology snapshot
// @Description Returns up to 100000 undirected node-ID pairs, deduplicated across known routes. Minute-aligned window includes the current minute. Existing rate limits apply; no per-node lookup or route pagination is needed. Capped is true when more links exist.
// @Tags Routes
// @Produce json
// @Param window query string false "History window: 15m (default), 1h, or 24h"
// @Param iatas query string false "Comma-separated reception IATA codes"
// @Success 200 {object} api.TopologyLinks
// @Failure 400 {object} handlers.APIError
// @Failure 503 {object} handlers.APIError
// @Router /routes/topology [get]
func getTopologyLinks(reader api.Reader) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
window := r.URL.Query().Get("window")
if window == "" {
window = "15m"
}
duration, ok := map[string]time.Duration{"15m": 15 * time.Minute, "1h": time.Hour, "24h": 24 * time.Hour}[window]
if !ok {
respondError(w, http.StatusBadRequest, "window must be 15m, 1h or 24h")
return
}
iatas, err := routeIATAs(r)
if err != nil {
respondError(w, http.StatusBadRequest, err.Error())
return
}
until := time.Now().UTC().Truncate(time.Minute).Add(time.Minute)
ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second)
defer cancel()
links, err := reader.GetTopologyLinks(ctx, iatas, until.Add(-duration), until)
if err != nil {
respondError(w, http.StatusServiceUnavailable, "route snapshot unavailable; try again shortly")
return
}
respond(w, http.StatusOK, links)
}
}
+33
View File
@@ -0,0 +1,33 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package handlers
import (
"net/http/httptest"
"strings"
"testing"
)
func TestTopologyWindowValidation(t *testing.T) {
for _, window := range []string{"", "15m", "1h", "24h", "72h", "0", "garbage"} {
w := httptest.NewRecorder()
RoutesRouter(stubReader{}).ServeHTTP(w, httptest.NewRequest("GET", "/topology?window="+window, nil))
want := 200
if window == "72h" || window == "0" || window == "garbage" {
want = 400
}
if w.Code != want {
t.Fatalf("%s: %d", window, w.Code)
}
}
}
func TestTopologyRejectsInvalidRegions(t *testing.T) {
for _, query := range []string{"iata=TOOLONG", "iatas=" + strings.Join([]string{"YKF", "invalid"}, ",")} {
w := httptest.NewRecorder()
RoutesRouter(stubReader{}).ServeHTTP(w, httptest.NewRequest("GET", "/topology?"+query, nil))
if w.Code != 400 {
t.Fatalf("%s: %d", query, w.Code)
}
}
}
+18 -13
View File
@@ -3,6 +3,8 @@
package api
import "github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
import (
"encoding/hex"
"encoding/json"
@@ -53,19 +55,21 @@ type PacketPathLength struct {
// PacketObservationDetail is a full observation including radio settings and resolved path.
type PacketObservationDetail struct {
ID int64 `json:"id"`
ObserverID uuid.UUID `json:"observerId"`
ObserverName *string `json:"observerName,omitempty"`
IATA string `json:"iata"`
HeardAt int64 `json:"heardAt"` // epoch ms
PathLength PacketPathLength `json:"pathLength"`
PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes
RSSI *int16 `json:"rssi,omitempty"`
SNR *float32 `json:"snr,omitempty"`
PropagationTimeMs *int32 `json:"propagationTimeMs"` // ms since first observation; 0 for first
Radio *PacketRadio `json:"radio,omitempty"`
SourceBroker string `json:"sourceBroker"`
ResolvedPath []ResolvedHop `json:"resolvedPath"` // per-observation resolved path hashes
WirePathBytes *string `json:"wirePathBytes,omitempty"` // Exact on-air TRACE SNR bytes, before the display path substitution.
TraceQuality *tracequality.Quality `json:"traceQuality,omitempty"`
ID int64 `json:"id"`
ObserverID uuid.UUID `json:"observerId"`
ObserverName *string `json:"observerName,omitempty"`
IATA string `json:"iata"`
HeardAt int64 `json:"heardAt"` // epoch ms
PathLength PacketPathLength `json:"pathLength"`
PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes
RSSI *int16 `json:"rssi,omitempty"`
SNR *float32 `json:"snr,omitempty"`
PropagationTimeMs *int32 `json:"propagationTimeMs"` // ms since first observation; 0 for first
Radio *PacketRadio `json:"radio,omitempty"`
SourceBroker string `json:"sourceBroker"`
ResolvedPath []ResolvedHop `json:"resolvedPath"` // per-observation resolved path hashes
// ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type
// carries one. Prefer the snapshot captured at ingest; legacy observations without a
// snapshot use the current node registry. Endpoint matching itself is unchanged:
@@ -134,6 +138,7 @@ type PacketTransportCodes struct {
// Packet is the full packet representation including all observations and resolved paths.
type Packet struct {
TraceQuality *tracequality.Quality `json:"traceQuality,omitempty"`
PacketHash string `json:"packetHash"`
Header PacketHeader `json:"header"`
TransportCodes *PacketTransportCodes `json:"transportCodes,omitempty"`
+2
View File
@@ -226,6 +226,8 @@ type Reader interface {
// GetTraceByTag returns all packets for a given trace tag with resolved routes.
GetTraceByTag(ctx context.Context, tag string) (*TraceDetail, error)
GetTopologyLinks(ctx context.Context, iatas []string, since, until time.Time) (*TopologyLinks, error)
// ListKnownRoutes returns known routes filtered by IATA and optional hop count.
// A nonzero cursorID breaks ties within the cursor's millisecond.
ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]KnownRoute, error)
+1
View File
@@ -14,6 +14,7 @@ import (
// Options names the startup settings used by the router. Adding an optional
// route does not change New's signature or require unrelated callers to change.
type Options struct {
ScopeCatalogues handlers.ScopeCatalogueSource
MaxConnsPerIP int
MaxConnectsPerMinute int
WSAllowedOrigins []string
+1
View File
@@ -119,6 +119,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, opts Options)
r.Mount("/routes", handlers.RoutesRouter(reader))
r.Mount("/scopes", handlers.ScopesRouter(reader, opts.Scopes))
r.Mount("/stats", handlers.StatsRouter(reader, handlers.StatsOptions{Scopes: opts.Scopes, SeriesWindow: opts.RollupRetention}))
r.Get("/scope-catalogues", handlers.ScopeCatalogues(opts.ScopeCatalogues))
r.Mount("/traces", handlers.TracesRouter(reader))
r.Mount("/info", handlers.InfoRouter(info))
})
@@ -0,0 +1,28 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package router
import (
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/MeshCore-Beacon/beacon-server/internal/meshmapper"
)
func TestScopeCataloguesDisabledAndReadOnly(t *testing.T) {
var importer *meshmapper.Importer
r := New(nil, nil, nil, Options{ScopeCatalogues: importer})
for _, method := range []string{http.MethodGet, http.MethodPost} {
w := httptest.NewRecorder()
r.ServeHTTP(w, httptest.NewRequest(method, "/api/v1/scope-catalogues", nil))
if method == http.MethodGet && (w.Code != 200 || strings.TrimSpace(w.Body.String()) != "[]" || w.Header().Get("Cache-Control") != "public, max-age=60") {
t.Fatal(w.Code, w.Body.String())
}
if method == http.MethodPost && w.Code != 405 {
t.Fatal(w.Code)
}
}
}
+14
View File
@@ -0,0 +1,14 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package api
import "github.com/google/uuid"
// TopologyLinks is a bounded set of undirected, adjacent known-route segments.
// It does not imply radio reachability or join across missing path entries.
type TopologyLinks struct {
Links [][2]uuid.UUID `json:"links"`
Capped bool `json:"capped"`
Since int64 `json:"since"`
Until int64 `json:"until"`
}
+20 -16
View File
@@ -3,29 +3,33 @@
package api
import "github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
// TraceTagSummary is a single trace tag with aggregate metadata.
type TraceTagSummary struct {
TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag
FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
PacketCount int64 `json:"packetCount"` // number of packets with this trace tag
IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard
TraceType string `json:"traceType"` // TRACE or PING
PathHashes []string `json:"pathHashes"` // hops from the most complete observation
SNRValues []float32 `json:"snrValues"` // SNR per hop from the most complete observation
Quality tracequality.Quality `json:"quality"`
TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag
FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
PacketCount int64 `json:"packetCount"` // number of packets with this trace tag
IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard
TraceType string `json:"traceType"` // TRACE or PING
PathHashes []string `json:"pathHashes"` // hops from the most complete observation
SNRValues []float32 `json:"snrValues"` // SNR per hop from the most complete observation
}
// TracePacket is a single packet within a trace series, including its
// resolved route derived from the trace path hashes.
type TracePacket struct {
PacketHash string `json:"packetHash"` // hex-encoded packet hash
RouteType int16 `json:"routeType"` // numeric route type
RouteTypeName string `json:"routeTypeName"` // human-readable route type
Scope *string `json:"scope,omitempty"` // transport scope name, if known
FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
RawPath []RawHop `json:"rawPath"` // hops as received in the packet
ResolvedRoute []ResolvedHop `json:"resolvedRoute"` // hops resolved to known nodes
Quality tracequality.Quality `json:"quality"`
PacketHash string `json:"packetHash"` // hex-encoded packet hash
RouteType int16 `json:"routeType"` // numeric route type
RouteTypeName string `json:"routeTypeName"` // human-readable route type
Scope *string `json:"scope,omitempty"` // transport scope name, if known
FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
RawPath []RawHop `json:"rawPath"` // hops as received in the packet
ResolvedRoute []ResolvedHop `json:"resolvedRoute"` // hops resolved to known nodes
}
// RawHop is a single hop in a trace path as received in the packet.
+4
View File
@@ -451,6 +451,10 @@ func (s *stubReader) GetRouteEvidence(_ context.Context, _, _ string, _ api.Rout
return nil, nil
}
func (s *stubReader) GetTopologyLinks(_ context.Context, _ []string, _ time.Time, _ time.Time) (*api.TopologyLinks, error) {
return &api.TopologyLinks{}, nil
}
func (s *stubReader) ListObserverDirectory(context.Context, api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) {
return nil, nil
}
+9
View File
@@ -374,6 +374,15 @@ func (cr *CachedReader) ListTraceTags(ctx context.Context, iatas []string, scope
return cr.inner.ListTraceTags(ctx, iatas, scope, traceType, since, until, cursor, cursorTag, limit)
}
func (cr *CachedReader) GetTopologyLinks(ctx context.Context, iatas []string, since, until time.Time) (*api.TopologyLinks, error) {
regions := append([]string{}, iatas...)
sort.Strings(regions)
key := fmt.Sprintf("beacon:topology:%s:%d:%d", strings.Join(regions, ","), since.Unix(), until.Unix())
return getOrSet(ctx, cr.c, key, 30*time.Second, func(ctx context.Context) (*api.TopologyLinks, error) {
return cr.inner.GetTopologyLinks(ctx, iatas, since, until)
})
}
func (cr *CachedReader) ListObserverDirectory(ctx context.Context, query api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) {
return cr.inner.ListObserverDirectory(ctx, query)
}
+34
View File
@@ -5,6 +5,7 @@ package ingest
import (
"context"
"encoding/hex"
"encoding/json"
"testing"
"time"
@@ -73,3 +74,36 @@ func TestChannelMessageScopeLive(t *testing.T) {
})
}
}
func TestChannelPublicMarkerUsesDecryptionKey(t *testing.T) {
for _, public := range []bool{false, true} {
w, base := newTestWorker()
w.db = &scopeMessageDB{stubDB: base, next: &InsertedChannelMessage{ID: 1}}
key := make([]byte, 16)
if public {
key, _ = hex.DecodeString("8b3387e9c5cdea6ac9e5edbaa115cd72")
}
w.keys = &mapKeys{entries: map[byte][]keystore.Entry{0x11: {{Key: key, Name: "Public"}}}}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
go w.hub.Run()
client := w.hub.NewClient()
defer w.hub.Remove(client)
w.hub.AddScope(client, "public-marker", hub.Scope{Events: []hub.EventType{hub.EventChannelMessage, hub.EventObserverStatus}})
waitForSummarySubscriber(t, ctx, w.hub, client)
packet := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeGrpTxt, 0), Payload: encryptedGroupText(t, 0x11, key, "Fixture", "test")}
w.handlePayloadTypeSideEffects(ctx, packet, "YKF", []byte{1}, RadioSettings{}, nil, nil, nil, 0)
select {
case event := <-client.Send:
var got channelMessageEvent
if err := json.Unmarshal(event.Payload, &got); err != nil {
t.Fatal(err)
}
if got.IsPublic != public {
t.Fatal("channel name/hash used instead of actual decryption key")
}
case <-ctx.Done():
t.Fatal("no channel event")
}
}
}
+1
View File
@@ -158,6 +158,7 @@ type DB interface {
// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error)
ResolveTracePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error)
// ResolveEndpointHashes matches one-byte logical endpoints, including companions.
ResolveEndpointHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error)
+3
View File
@@ -266,6 +266,9 @@ func (s *stubDB) GetObserverScopes(_ context.Context, _ uuid.UUID) ([]string, er
func (s *stubDB) ResolvePathHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) {
return nil, nil
}
func (s *stubDB) ResolveTracePathHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) {
return nil, nil
}
func (s *stubDB) ResolveEndpointHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) {
return nil, nil
+45 -13
View File
@@ -18,6 +18,7 @@ import (
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/MeshCore-Beacon/beacon-server/internal/lora"
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
"github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
"github.com/google/uuid"
"github.com/meshcore-go/meshcore-go"
)
@@ -202,13 +203,14 @@ type parsedGroupEnvelope struct {
// is the accumulated per-hop SNR sequence carried in the packet header's
// path field (a protocol-level repurposing distinct from the payload).
type parsedTrace struct {
Raw string `json:"raw"`
Type string `json:"type"`
TraceTag string `json:"traceTag"`
AuthCode uint32 `json:"authCode"`
Flags byte `json:"flags"`
PathHashes []string `json:"pathHashes"`
SNRValues []float32 `json:"snrValues"`
Quality tracequality.Quality `json:"quality"`
Raw string `json:"raw"`
Type string `json:"type"`
TraceTag string `json:"traceTag"`
AuthCode uint32 `json:"authCode"`
Flags byte `json:"flags"`
PathHashes []string `json:"pathHashes"`
SNRValues []float32 `json:"snrValues"`
}
// parsedAck is the parsed form of an ACK payload (§2.12). Checksum is the
@@ -557,7 +559,10 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
hashes = append(hashes, hex.EncodeToString(h))
rawHashes = append(rawHashes, h)
}
traceRawHashes = rawHashes
quality := tracequality.Assess(packet.Payload, int(packet.PayloadVer()), int(packet.RouteType()), int(packet.PathLength))
if quality.Status == "supported" {
traceRawHashes = rawHashes
}
// SNR values are in packet.Path, one signed int8 per consumed hop
snrValues := make([]float32, 0, len(packet.Path))
for _, b := range packet.Path {
@@ -568,6 +573,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
traceType = "PING"
}
pt := parsedTrace{
Quality: quality,
Raw: hex.EncodeToString(packet.Payload),
Type: traceType,
TraceTag: hex.EncodeToString(uint32ToBytes(trace.Tag)),
@@ -585,8 +591,8 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
// originator isn't in PathHashes) and is skipped. Either side
// resolving ambiguously (>1 candidate) or not at all (0
// candidates) skips that specific pair.
if len(rawHashes) >= 2 {
resolved, rErr := w.db.ResolvePathHashes(ctx, iata, rawHashes)
if quality.Status == "supported" && len(rawHashes) >= 2 {
resolved, rErr := w.db.ResolveTracePathHashes(ctx, iata, rawHashes)
if rErr == nil {
for i := 1; i < len(rawHashes); i++ {
if i >= len(snrValues) {
@@ -822,12 +828,20 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
if packet.PayloadType() == meshcore.PayloadTypeTrace {
hashes = traceRawHashes
}
resolved, err := w.db.ResolvePathHashes(ctx, iata, hashes)
var resolved map[string][]api.ResolvedPathEntry
if packet.PayloadType() == meshcore.PayloadTypeTrace {
resolved, err = w.db.ResolveTracePathHashes(ctx, iata, hashes)
} else {
resolved, err = w.db.ResolvePathHashes(ctx, iata, hashes)
}
if err != nil {
w.log.Error("path resolution failed", "error", err)
}
var resolvedIDs []uuid.UUID
for _, entries := range resolved {
if len(entries) != 1 {
continue
}
for _, e := range entries {
resolvedIDs = append(resolvedIDs, e.NodeID)
}
@@ -846,13 +860,31 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
nodeIDs = append(nodeIDs, entries[0].NodeID)
hashPrefixes = append(hashPrefixes, hash)
}
if allHigh && len(nodeIDs) > 1 {
if allHigh && len(nodeIDs) > 1 && packet.PayloadType() != meshcore.PayloadTypeTrace {
if err := w.db.UpsertKnownRoute(ctx, nodeIDs, hashPrefixes, iata, int32(len(nodeIDs))); err != nil {
w.log.Error("failed to upsert known route", "error", err)
}
}
}
w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs)
hashSize := packet.PathHashSize()
if packet.PayloadType() == meshcore.PayloadTypeTrace {
hashSize = 1
if len(traceRawHashes) > 0 {
hashSize = uint8(len(traceRawHashes[0]))
}
// Only consumed, uniquely resolved hops demonstrate trace capability.
resolvedIDs = nil
for i, hash := range traceRawHashes {
if i >= len(packet.Path) {
break
}
entries := resolved[hex.EncodeToString(hash)]
if len(entries) == 1 {
resolvedIDs = append(resolvedIDs, entries[0].NodeID)
}
}
}
w.runCapabilityDetection(ctx, packet.PayloadType(), hashSize, resolvedIDs)
if inserted || repeat {
if inserted {
+2
View File
@@ -48,6 +48,7 @@ type InsertedChannelMessage struct {
// channelMessageEvent is the JSON payload for a channelMessage WS event.
type channelMessageEvent struct {
IsPublic bool `json:"isPublic"` // True only when decryption used the documented default Public key, never inferred from a one-byte channel hash.
ID int64 `json:"id"`
ChannelID int `json:"channelId"`
ChannelHash string `json:"channelHash"` // hex-encoded single byte
@@ -223,6 +224,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
if result.Message != nil {
evt := channelMessageEvent{
IsPublic: hex.EncodeToString(result.Entry.Key) == "8b3387e9c5cdea6ac9e5edbaa115cd72",
ID: result.Message.ID,
Scope: result.Message.Scope,
ScopeStatus: result.Message.ScopeStatus,
+75
View File
@@ -0,0 +1,75 @@
package ingest
import (
"context"
"encoding/hex"
"encoding/json"
"testing"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/google/uuid"
"github.com/meshcore-go/meshcore-go"
)
type traceSafetyDB struct {
stubDB
saved UpsertPacketParams
resolutions, neighbors, routes int
}
func (s *traceSafetyDB) UpsertPacket(_ context.Context, p UpsertPacketParams) (bool, error) {
s.saved = p
return true, nil
}
func (s *traceSafetyDB) ResolveTracePathHashes(_ context.Context, _ string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) {
if len(hashes) > 0 {
s.resolutions++
}
out := map[string][]api.ResolvedPathEntry{}
for i, h := range hashes {
id := uuid.UUID{}
id[15] = byte(i + 1)
out[hex.EncodeToString(h)] = []api.ResolvedPathEntry{{NodeID: id}}
}
return out, nil
}
func (s *traceSafetyDB) UpsertNodeNeighbor(_ context.Context, _, _ uuid.UUID, _ string, _ *float32, _ *string) error {
s.neighbors++
return nil
}
func (s *traceSafetyDB) UpsertKnownRoute(_ context.Context, _ []uuid.UUID, _ [][]byte, _ string, _ int32) error {
s.routes++
return nil
}
func TestTraceQualityDoesNotPromoteMalformedOrUnvisitedHops(t *testing.T) {
for _, bad := range []bool{false, true} {
w, _ := newTestWorker()
db := &traceSafetyDB{}
w.db = db
payload, _ := (&meshcore.Trace{Tag: 7, Flags: 1, PathHashes: []byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}}).ToBytes()
version := byte(0)
if bad {
version = 1
}
pkt := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeDirect, meshcore.PayloadTypeTrace, version), PathLength: 2, Path: []byte{0, 252}, Payload: payload}
w.handlePacket(context.Background(), "YKF", "0102", packetEnvelope(t, pkt))
if len(db.saved.RawPayload) == 0 || len(db.traceHearings) != 1 {
t.Fatal("raw evidence lost")
}
var parsed parsedTrace
if err := json.Unmarshal(db.saved.ParsedPayload, &parsed); err != nil {
t.Fatal(err)
}
if bad {
if parsed.Quality.Status != "suspect" || db.neighbors != 0 || db.resolutions != 0 || len(db.setCapabilityCalls) != 0 {
t.Fatalf("malformed trace promoted: %+v", db)
}
} else if parsed.Quality.Status != "supported" || db.neighbors != 1 || len(db.setCapabilityCalls) != 2 {
t.Fatalf("valid consumed links/capabilities missing: %+v", db)
}
if db.routes != 0 {
t.Fatal("planned route promoted as observed")
}
}
}
+75
View File
@@ -0,0 +1,75 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package meshmapper
import (
"encoding/json"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
)
// CatalogueScope is regional source metadata, not per-node forwarding evidence.
type CatalogueScope struct {
Name string `json:"name"`
Repeaters int `json:"repeaters"`
Default int `json:"default"`
Monitored bool `json:"monitored"`
Wardriving bool `json:"wardriving"`
}
type PublicScopeCatalogue struct {
IATA string `json:"iata"`
URL string `json:"url"`
GeneratedAt int64 `json:"generatedAt"`
CheckedAt int64 `json:"checkedAt"`
FreshUntil int64 `json:"freshUntil"`
LastError string `json:"lastError,omitempty"`
Repeaters int `json:"repeaters"`
Scoped int `json:"scoped"`
Scopes []CatalogueScope `json:"scopes"`
}
// Catalogues returns an immutable snapshot of discovered sources. No request
// triggers an upstream fetch or reads ingestion tables. Callers must not mutate it.
func (i *Importer) Catalogues() []PublicScopeCatalogue {
if i != nil {
if snapshot := i.catalogues.Load(); snapshot != nil {
return snapshot.([]PublicScopeCatalogue)
}
}
return []PublicScopeCatalogue{}
}
func (i *Importer) publishCatalogues() {
result := make([]PublicScopeCatalogue, 0, len(i.sources))
for _, s := range i.sources {
c := PublicScopeCatalogue{IATA: s.iata, URL: s.url, LastError: s.cache.LastError, Scopes: []CatalogueScope{}}
if !s.cache.CheckedAt.IsZero() {
c.CheckedAt = s.cache.CheckedAt.UnixMilli()
c.FreshUntil = s.cache.CheckedAt.Add(i.interval + 5*time.Minute).UnixMilli()
}
if len(s.cache.Payload) > 0 {
// Reuse the importer validation before exposing saved metadata. Invalid
// saved catalogues are omitted just as they are from the scope matcher.
if _, generated, err := decode(s.cache.Payload, s.iata); err == nil {
var payload struct {
Repeaters int
Scoped int
Scopes []CatalogueScope
}
if json.Unmarshal(s.cache.Payload, &payload) == nil {
c.GeneratedAt = generated.UnixMilli()
c.Repeaters, c.Scoped = payload.Repeaters, payload.Scoped
for _, scope := range payload.Scopes {
scope.Name = scopestore.FromName(scope.Name).Name
c.Scopes = append(c.Scopes, scope)
}
}
}
}
result = append(result, c)
}
i.catalogues.Store(result)
}
+104
View File
@@ -0,0 +1,104 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package meshmapper
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/config"
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
)
func TestPublicCataloguesFreshnessAndIsolation(t *testing.T) {
status := http.StatusOK
calls := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
calls++
w.WriteHeader(status)
if status == http.StatusOK {
_, _ = w.Write(catalogue("yow", "YOW", "monitored-zero"))
}
}))
defer server.Close()
cfg := config.MeshMapperScopesConfig{Enabled: true}
store := newMemoryStore("YOW")
imp, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory(), ""), scopestore.New(), nil)
if err != nil {
t.Fatal(err)
}
if got := imp.Catalogues(); len(got) != 1 || got[0].CheckedAt != 0 || len(got[0].Scopes) != 0 {
t.Fatal(got)
}
imp.sources[0].url = server.URL
imp.client = server.Client()
now := time.Now()
if err := imp.refresh(context.Background(), &imp.sources[0], now); err != nil {
t.Fatal(err)
}
imp.publishCatalogues()
old := imp.Catalogues()
if old[0].Scopes[0].Name != "#yow" || old[0].Scopes[1].Name != "#YOW" || len(old[0].Scopes) != 3 || old[0].Repeaters != 5 || old[0].Scoped != 3 || old[0].FreshUntil <= old[0].CheckedAt {
t.Fatal(old)
}
status = http.StatusServiceUnavailable
if err := imp.refresh(context.Background(), &imp.sources[0], now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
imp.publishCatalogues()
current := imp.Catalogues()[0]
if current.LastError != "HTTP 503" || current.CheckedAt != old[0].CheckedAt || current.GeneratedAt != old[0].GeneratedAt || len(current.Scopes) != 3 || old[0].LastError != "" {
t.Fatal(current, old)
}
// Readers only access published immutable data; the race suite exercises concurrent refreshes.
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
for range 100 {
_, _ = json.Marshal(imp.Catalogues())
}
}()
for range 100 {
imp.publishCatalogues()
}
wg.Wait()
if calls != 2 {
t.Fatal("reading public metadata performed upstream requests", calls)
}
disabled, err := New(context.Background(), config.MeshMapperScopesConfig{}, store, NewDirectory(newZoneListMemory(), ""), scopestore.New(), nil)
if err != nil || len(disabled.Catalogues()) != 0 {
t.Fatal("disabled importer exposed old membership")
}
store.rows[sourceKey{"YOW", server.URL}] = Cache{Payload: []byte(`{"scopes":"invalid"}`)}
restored, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory(), ""), scopestore.New(), nil)
if err != nil || len(restored.Catalogues()[0].Scopes) != 0 || restored.Catalogues()[0].LastError == "" {
t.Fatal("invalid saved metadata leaked")
}
}
func TestPublicCataloguesIncludeDiscoveredAndUnlistedRegions(t *testing.T) {
ctx := context.Background()
store := newMemoryStore("YOW")
dir := NewDirectory(newZoneListMemory(), "")
dir.lists["CA"] = &zoneList{zones: map[string]zoneEntry{}, fetchedAt: time.Now()}
imp, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopestore.New(), nil)
if err != nil {
t.Fatal(err)
}
before := imp.Catalogues()
store.iatas = append(store.iatas, "YKF")
if err := imp.Refresh(ctx); err != nil {
t.Fatal(err)
}
after := imp.Catalogues()
if len(before) != 1 || before[0].LastError != "" || len(after) != 2 || after[1].IATA != "YKF" || after[1].LastError != "not listed" {
t.Fatal(before, after)
}
}
+14 -10
View File
@@ -12,6 +12,7 @@ import (
"net/http"
"slices"
"strings"
"sync/atomic"
"time"
"unicode"
"unicode/utf8"
@@ -58,16 +59,17 @@ type source struct {
// Importer is owned by one background task; ScopeStore synchronizes its consumers.
type Importer struct {
store Store
scopes *scopestore.ScopeStore
manual []scopestore.Entry
dir *Directory
scopesURL func(site string) (string, bool)
seen map[string]bool
sources []source
interval time.Duration
client *http.Client
onChange func(context.Context)
catalogues atomic.Value // immutable []PublicScopeCatalogue, off the ingest path
store Store
scopes *scopestore.ScopeStore
manual []scopestore.Entry
dir *Directory
scopesURL func(site string) (string, bool)
seen map[string]bool
sources []source
interval time.Duration
client *http.Client
onChange func(context.Context)
}
// SetCacheInvalidator is wired once at startup, before the background task starts.
@@ -97,6 +99,7 @@ func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, di
}
i.track(iatas, latest)
i.publish()
i.publishCatalogues()
return i, nil
}
@@ -131,6 +134,7 @@ func (i *Importer) track(iatas []string, saved map[string]Catalogue) {
// Refresh checks only one due source. The scheduler serializes calls every 15s.
func (i *Importer) Refresh(ctx context.Context) (err error) {
defer i.publishCatalogues()
parent := ctx
defer func() {
if parent.Err() != nil {
+96
View File
@@ -0,0 +1,96 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
// Package tracequality checks TRACE structure, not authenticity or RF plausibility.
package tracequality
import "encoding/hex"
type Quality struct {
Status string `json:"status"` // supported, suspect, or ambiguous; supported is not authenticated
Reasons []string `json:"reasons"`
}
func (q *Quality) Add(reason string) {
for _, r := range q.Reasons {
if r == reason {
return
}
}
q.Reasons = append(q.Reasons, reason)
if reason == "ambiguous_prefix" && q.Status != "suspect" {
q.Status = "ambiguous"
} else if reason != "ambiguous_prefix" {
q.Status = "suspect"
}
}
// Assess uses TRACE's power-of-two payload hash widths, NOT the ordinary
// packet path's 1/2/3-byte widths. The path field contains consumed-hop SNRs.
// -1 means a header/path field is unavailable in a retained summary.
// Mesh.cpp onRecvPacket/sendDirect: a trace uses direct routing, flags bits
// 0..1, and fewer than 64 SNR bytes. Partial probes and return loops are valid.
func Assess(raw []byte, version, route, pathLength int) Quality {
q := Quality{Status: "supported", Reasons: []string{}}
if version > 0 {
q.Add("unsupported_version")
}
if route >= 0 && route != 2 && route != 3 {
q.Add("non_direct_trace")
}
if len(raw) < 9 {
q.Add("incomplete_payload")
return q
}
if len(raw) > 184 {
q.Add("payload_too_large")
}
if raw[8]&0xfc != 0 {
q.Add("unsupported_flags")
}
size := 1 << (raw[8] & 3)
if (len(raw)-9)%size != 0 {
q.Add("misaligned_hashes")
}
count := (len(raw) - 9) / size
if count > 63 {
q.Add("too_many_hops")
}
if pathLength > 63 || pathLength > count {
q.Add("invalid_snr_path")
}
return q
}
// Stored also checks old parsed rows, without rewriting or deleting the evidence.
func Stored(raw string, flags byte, hashes []string, snrs []float32) Quality {
b, err := hex.DecodeString(raw)
if err != nil {
return Quality{Status: "suspect", Reasons: []string{"incomplete_payload"}}
}
if raw == "" { // Older exports may contain only decoded fields.
b = make([]byte, 9)
b[8] = flags
for _, h := range hashes {
v, e := hex.DecodeString(h)
if e != nil {
return Quality{Status: "suspect", Reasons: []string{"misaligned_hashes"}}
}
b = append(b, v...)
}
}
q := Assess(b, -1, -1, len(snrs))
size := 1 << (flags & 3)
if len(b) >= 9 && b[8] != flags {
q.Add("misaligned_hashes")
}
for _, h := range hashes {
if len(h) != 2*size {
q.Add("misaligned_hashes")
}
}
if len(b) >= 9 && len(hashes) != (len(b)-9)/size {
q.Add("misaligned_hashes")
}
return q
}
+54
View File
@@ -0,0 +1,54 @@
package tracequality
import (
"bytes"
"slices"
"testing"
)
func TestTraceFormat(t *testing.T) {
if q := Assess(make([]byte, 9), 0, 2, 0); q.Status != "supported" {
t.Fatalf("zero-hop trace header must remain inspectable: %+v", q)
}
for flags := byte(0); flags < 4; flags++ {
raw := make([]byte, 9)
raw[8] = flags
raw = append(raw, bytes.Repeat([]byte{0xaa}, 3*(1<<flags))...)
for _, consumed := range []int{0, 1, 3} {
if q := Assess(raw, 0, 2, consumed); q.Status != "supported" {
t.Fatalf("valid width %d, partial/loop trace: %+v", 1<<flags, q)
}
}
}
cases := []struct {
name string
version, route, path, flags, count int
reason string
}{
{"version", 1, 2, 0, 0, 3, "unsupported_version"},
{"flood", 0, 1, 0, 0, 3, "non_direct_trace"},
{"flags", 0, 2, 0, 133, 74, "unsupported_flags"},
{"101 hops", 0, 2, 55, 0, 101, "too_many_hops"},
{"145 hops", 0, 2, 10, 0, 145, "too_many_hops"},
{"ordinary multibyte metadata is not trace SNR count", 0, 2, 65, 1, 3, "invalid_snr_path"},
{"extra SNR", 0, 2, 4, 0, 3, "invalid_snr_path"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
raw := make([]byte, 9)
raw[8] = byte(tc.flags)
raw = append(raw, bytes.Repeat([]byte{0xaa}, tc.count*(1<<(tc.flags&3)))...)
q := Assess(raw, tc.version, tc.route, tc.path)
if q.Status != "suspect" || !slices.Contains(q.Reasons, tc.reason) {
t.Fatalf("%+v", q)
}
})
}
if q := Assess([]byte{0, 0, 0, 0, 0, 0, 0, 0, 2, 0, 1, 2}, 0, 2, 0); !slices.Contains(q.Reasons, "misaligned_hashes") {
t.Fatal(q)
}
// Legitimate missing reports, negative SNR, and a returning loop are not corruption.
if q := Stored("", 0, []string{"aa", "bb", "aa"}, []float32{-12.25, 0}); q.Status != "supported" {
t.Fatal(q)
}
}