mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-23 04:11:26 +00:00
feat(meshmapper): import region boundaries from the Zones API
Imported boundaries override borderFile on the border map and for foreign marking, falling back to the manual border when MeshMapper has none.
This commit is contained in:
1 parent
689bc2326a
commit
0e242574fb
26 files changed
+1397
-49
No files matched your search
+37
-1
@@ -6,6 +6,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
@@ -22,6 +23,7 @@ import (
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api/handlers"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api/router"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/background"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/borders"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/cache"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/hub"
|
||||
@@ -111,11 +113,16 @@ func main() {
|
||||
}
|
||||
|
||||
resolved := config.Resolve(cfg)
|
||||
localBorders, err := config.LoadLocalBorders(cfg)
|
||||
borderFiles, err := config.LoadLocalBorders(cfg)
|
||||
if err != nil {
|
||||
slog.Error("invalid local border configuration", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var localBorders *borders.Live
|
||||
if cfg.Nodes.MarkForeign {
|
||||
local, _ := config.BuildLocalBorders(borderFiles, nil) // checked by LoadLocalBorders
|
||||
localBorders = borders.NewLive(local)
|
||||
}
|
||||
|
||||
slog.Info(fmt.Sprintf("config: loaded — %s", resolved), "component", "startup")
|
||||
|
||||
@@ -201,6 +208,32 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
// Always restored, so disabling the import prunes it and file borders return.
|
||||
zones := meshmapper.NewZones(cfg.MeshMapper.Zones, cfg.RegionIATAs(), store)
|
||||
if cr, ok := reader.(*cache.CachedReader); ok {
|
||||
zones.OnChange(cr.InvalidateIATABorder)
|
||||
}
|
||||
if localBorders != nil {
|
||||
zones.OnUpdate(func(imported map[string]json.RawMessage) {
|
||||
local, err := config.BuildLocalBorders(borderFiles, imported)
|
||||
if err != nil {
|
||||
slog.Error("MeshMapper boundaries rejected for foreign marking", "component", "meshmapper.zones", "error", err)
|
||||
return
|
||||
}
|
||||
localBorders.Store(local)
|
||||
})
|
||||
}
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
err = zones.Restore(restoreCtx)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper boundaries", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if localBorders != nil && !localBorders.Ready() {
|
||||
slog.Warn("nodes.mark_foreign has no boundaries yet; waiting for MeshMapper", "component", "startup")
|
||||
}
|
||||
|
||||
// ── Build channel keystore ──────────────────────────────────────────────
|
||||
entries := make(map[string][]keystore.Entry)
|
||||
|
||||
@@ -318,6 +351,9 @@ func main() {
|
||||
if scopeImporter != nil {
|
||||
tasks = append(tasks, background.Task{Name: "meshmapper.scopes", Interval: meshmapper.PollInterval, Run: scopeImporter.Refresh})
|
||||
}
|
||||
if cfg.MeshMapper.Zones.Enabled {
|
||||
tasks = append(tasks, background.Task{Name: "meshmapper.zones", Interval: meshmapper.PollInterval, Run: zones.Refresh})
|
||||
}
|
||||
profiles := configureProfiling(ctx, pool)
|
||||
defer profiles.Stop()
|
||||
for i := range tasks {
|
||||
|
||||
+14
-2
@@ -47,6 +47,7 @@ iatas:
|
||||
lat: 49.1967
|
||||
lng: -123.1815
|
||||
# borderFile: borders/yvr.geojson # optional GeoJSON Feature (Polygon/MultiPolygon)
|
||||
# # (meshmapper.zones overrides it when enabled)
|
||||
# # for the region border map; path is relative to
|
||||
# # this file's directory; validated at boot.
|
||||
YYJ:
|
||||
@@ -159,8 +160,9 @@ websocket:
|
||||
# Node staleness, deletion, and clock-drift thresholds.
|
||||
#nodes:
|
||||
# mark_foreign: false # opt in to possiblyForeign on repeater nodes
|
||||
# # uses the union of configured iatas.*.borderFile polygons;
|
||||
# # requires at least one valid border; no packet filtering
|
||||
# # uses the union of configured iatas.*.borderFile polygons,
|
||||
# # with meshmapper.zones boundaries replacing them; requires
|
||||
# # a valid border file or meshmapper.zones; no packet filtering
|
||||
# stale_threshold: 24h # default: 24h
|
||||
# delete_after: 720h # default: 30 days
|
||||
# clock_drift_threshold: 5m # default: 5m
|
||||
@@ -228,6 +230,16 @@ cache:
|
||||
# refresh_interval: 1h # 5m-24h; one source checked per 15s tick
|
||||
# sources:
|
||||
# YOW: https://yow.meshmapper.net/get_scopes.php
|
||||
#
|
||||
# Optional MeshMapper region boundaries (Zones API). Every IATA in a configured
|
||||
# region is looked up in its country's zone list, then its get_geojson.php outline
|
||||
# is imported. An imported boundary overrides that IATA's borderFile on the border
|
||||
# map and for nodes.mark_foreign. A missing or failed boundary keeps the last good
|
||||
# import, or the borderFile when there is none. Disabling, or removing an IATA from
|
||||
# every region, drops its import on restart. Logged under component=meshmapper.zones.
|
||||
# zones:
|
||||
# enabled: false
|
||||
# refresh_interval: 24h # 1h-168h; one request per 15s tick, ETag-conditional
|
||||
|
||||
# Background task intervals.
|
||||
# Shorter intervals are useful during initial deployment to confirm data is
|
||||
|
||||
@@ -47,3 +47,40 @@ func (s *Store) SaveScopeCatalogue(ctx context.Context, iata, url string, cache
|
||||
}
|
||||
return s.q.SaveScopeCatalogue(ctx, params)
|
||||
}
|
||||
|
||||
func (s *Store) PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error) {
|
||||
if keep == nil {
|
||||
keep = []string{}
|
||||
}
|
||||
return s.q.PruneZoneBoundaries(ctx, keep)
|
||||
}
|
||||
|
||||
func (s *Store) ListZoneBoundaries(ctx context.Context) ([]meshmapper.Boundary, error) {
|
||||
rows, err := s.q.ListZoneBoundaries(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]meshmapper.Boundary, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
b := meshmapper.Boundary{IATA: row.Iata, URL: row.Url, Feature: row.Feature, CheckedAt: row.CheckedAt.Time,
|
||||
AttemptedAt: row.AttemptedAt.Time, NextAttempt: row.NextAttempt.Time, LastError: row.LastError}
|
||||
if row.Etag != nil {
|
||||
b.ETag = *row.Etag
|
||||
}
|
||||
out = append(out, b)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *Store) SaveZoneBoundary(ctx context.Context, b meshmapper.Boundary) error {
|
||||
params := sqlc.SaveZoneBoundaryParams{
|
||||
Iata: b.IATA, Url: b.URL, Feature: b.Feature, LastError: b.LastError,
|
||||
CheckedAt: pgtype.Timestamptz{Time: b.CheckedAt, Valid: !b.CheckedAt.IsZero()},
|
||||
AttemptedAt: pgtype.Timestamptz{Time: b.AttemptedAt, Valid: true},
|
||||
NextAttempt: pgtype.Timestamptz{Time: b.NextAttempt, Valid: true},
|
||||
}
|
||||
if !b.CheckedAt.IsZero() {
|
||||
params.Etag = &b.ETag
|
||||
}
|
||||
return s.q.SaveZoneBoundary(ctx, params)
|
||||
}
|
||||
@@ -6,6 +6,7 @@ package db
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -137,3 +138,88 @@ func TestMeshMapperCataloguePostgres(t *testing.T) {
|
||||
t.Fatal("partially committed catalogue", count, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMeshMapperZoneBoundariesPostgres(t *testing.T) {
|
||||
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("set BEACON_TEST_POSTGRES_DSN")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := RunMigrations(ctx, pool); err != nil {
|
||||
pool.Close()
|
||||
t.Fatal(err)
|
||||
}
|
||||
pool.Close()
|
||||
conn, err := pgx.Connect(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close(context.Background())
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
for _, table := range []string{"iata_codes", "meshmapper_zone_boundaries"} {
|
||||
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
store := &Store{q: sqlc.New(tx)}
|
||||
manual := []byte(`{"type":"Feature","geometry":{"type":"Polygon","coordinates":[[[0,0],[1,0],[1,1],[0,1],[0,0]]]},"properties":{"source":"file"}}`)
|
||||
imported := []byte(`{"type":"Feature","geometry":{"type":"Polygon","coordinates":[[[0,0],[2,0],[2,2],[0,2],[0,0]]]},"properties":{"code":"YOW"}}`)
|
||||
if err := store.UpsertIATABorder(ctx, "YOW", manual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.UpsertIATA(ctx, "YYZ"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
border := func(iata string) string {
|
||||
t.Helper()
|
||||
got, err := store.GetIATABorder(ctx, iata)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(got)
|
||||
}
|
||||
if !strings.Contains(border("YOW"), `"file"`) || border("YYZ") != "" {
|
||||
t.Fatal("manual border not served without an import")
|
||||
}
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
good := meshmapper.Boundary{IATA: "YOW", URL: "https://yow.meshmapper.net/", Feature: imported, ETag: `"v1"`, CheckedAt: now, AttemptedAt: now, NextAttempt: now.Add(24 * time.Hour)}
|
||||
if err := store.SaveZoneBoundary(ctx, good); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(border("YOW"), `"YOW"`) {
|
||||
t.Fatal("imported boundary did not override the manual border")
|
||||
}
|
||||
if err := store.UpsertIATABorder(ctx, "YOW", manual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
failed := meshmapper.Boundary{IATA: "YOW", URL: good.URL, AttemptedAt: now.Add(time.Hour), NextAttempt: now.Add(2 * time.Hour), LastError: "HTTP 503"}
|
||||
if err := store.SaveZoneBoundary(ctx, failed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows, err := store.ListZoneBoundaries(ctx)
|
||||
if err != nil || len(rows) != 1 || !strings.Contains(string(rows[0].Feature), `"YOW"`) || rows[0].ETag != `"v1"` ||
|
||||
!rows[0].CheckedAt.Equal(now) || rows[0].LastError != "HTTP 503" || !rows[0].NextAttempt.Equal(failed.NextAttempt) {
|
||||
t.Fatal("failure or reseed lost the last good boundary", rows, err)
|
||||
}
|
||||
if !strings.Contains(border("YOW"), `"YOW"`) {
|
||||
t.Fatal("reseeding the manual border undid the override")
|
||||
}
|
||||
if pruned, err := store.PruneZoneBoundaries(ctx, []string{"YOW"}); err != nil || len(pruned) != 0 {
|
||||
t.Fatal("configured import pruned", pruned, err)
|
||||
}
|
||||
if pruned, err := store.PruneZoneBoundaries(ctx, nil); err != nil || len(pruned) != 1 || pruned[0] != "YOW" {
|
||||
t.Fatal(pruned, err)
|
||||
}
|
||||
if !strings.Contains(border("YOW"), `"file"`) {
|
||||
t.Fatal("manual border did not return after pruning")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
-- Copyright 2026 Beacon Contributors
|
||||
-- SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
-- Imported MeshMapper region boundaries. Kept apart from iata_codes.border so
|
||||
-- startup seeding of manual borderFile borders never undoes the override.
|
||||
CREATE TABLE IF NOT EXISTS meshmapper_zone_boundaries (
|
||||
iata TEXT PRIMARY KEY,
|
||||
url TEXT NOT NULL,
|
||||
feature JSONB,
|
||||
etag TEXT,
|
||||
checked_at TIMESTAMPTZ,
|
||||
attempted_at TIMESTAMPTZ NOT NULL,
|
||||
next_attempt TIMESTAMPTZ NOT NULL,
|
||||
last_error TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
+28
-4
@@ -25,10 +25,13 @@ ON CONFLICT (iata) DO UPDATE SET
|
||||
approx_lng = EXCLUDED.approx_lng;
|
||||
|
||||
-- name: GetIATABorder :one
|
||||
-- border is NULL when the IATA exists but has no border configured; a
|
||||
-- missing row (unknown IATA) is sql.ErrNoRows, same not-found distinction
|
||||
-- GetIATA already makes.
|
||||
SELECT border FROM iata_codes WHERE iata = $1;
|
||||
-- An imported MeshMapper boundary overrides the configured one. border is NULL
|
||||
-- when neither exists; a missing row (unknown IATA) is sql.ErrNoRows, same
|
||||
-- not-found distinction GetIATA already makes.
|
||||
SELECT COALESCE(z.feature, i.border)::jsonb AS border
|
||||
FROM iata_codes i
|
||||
LEFT JOIN meshmapper_zone_boundaries z ON z.iata = i.iata
|
||||
WHERE i.iata = $1;
|
||||
|
||||
-- name: UpsertIATABorder :exec
|
||||
-- Written by the config-file-driven seeder (internal/config/seed.go), not a
|
||||
@@ -1627,3 +1630,24 @@ ON CONFLICT (iata, url) DO UPDATE SET
|
||||
attempted_at = EXCLUDED.attempted_at,
|
||||
next_attempt = EXCLUDED.next_attempt,
|
||||
last_error = EXCLUDED.last_error;
|
||||
|
||||
-- name: ListZoneBoundaries :many
|
||||
SELECT * FROM meshmapper_zone_boundaries ORDER BY iata;
|
||||
|
||||
-- name: SaveZoneBoundary :exec
|
||||
-- NULL feature/etag/checked_at retain the last good boundary after an error or 304.
|
||||
INSERT INTO meshmapper_zone_boundaries (iata, url, feature, etag, checked_at, attempted_at, next_attempt, last_error)
|
||||
VALUES (@iata, @url, sqlc.narg(feature)::jsonb, sqlc.narg(etag)::text,
|
||||
sqlc.narg(checked_at)::timestamptz, @attempted_at, @next_attempt, @last_error)
|
||||
ON CONFLICT (iata) DO UPDATE SET
|
||||
url = EXCLUDED.url,
|
||||
feature = COALESCE(EXCLUDED.feature, meshmapper_zone_boundaries.feature),
|
||||
etag = COALESCE(EXCLUDED.etag, meshmapper_zone_boundaries.etag),
|
||||
checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_zone_boundaries.checked_at),
|
||||
attempted_at = EXCLUDED.attempted_at,
|
||||
next_attempt = EXCLUDED.next_attempt,
|
||||
last_error = EXCLUDED.last_error;
|
||||
|
||||
-- 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;
|
||||
+47
-2
@@ -11,6 +11,7 @@ package mockdb
|
||||
|
||||
import (
|
||||
context "context"
|
||||
json "encoding/json"
|
||||
reflect "reflect"
|
||||
|
||||
db "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
|
||||
@@ -265,10 +266,10 @@ func (mr *MockQuerierMockRecorder) GetIATA(ctx, iata any) *gomock.Call {
|
||||
}
|
||||
|
||||
// GetIATABorder mocks base method.
|
||||
func (m *MockQuerier) GetIATABorder(ctx context.Context, iata string) ([]byte, error) {
|
||||
func (m *MockQuerier) GetIATABorder(ctx context.Context, iata string) (json.RawMessage, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "GetIATABorder", ctx, iata)
|
||||
ret0, _ := ret[0].([]byte)
|
||||
ret0, _ := ret[0].(json.RawMessage)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
@@ -1313,6 +1314,36 @@ func (mr *MockQuerierMockRecorder) ListUndecryptedGroupTextPackets(ctx any) *gom
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListUndecryptedGroupTextPackets", reflect.TypeOf((*MockQuerier)(nil).ListUndecryptedGroupTextPackets), ctx)
|
||||
}
|
||||
|
||||
// ListZoneBoundaries mocks base method.
|
||||
func (m *MockQuerier) ListZoneBoundaries(ctx context.Context) ([]db.MeshmapperZoneBoundary, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "ListZoneBoundaries", ctx)
|
||||
ret0, _ := ret[0].([]db.MeshmapperZoneBoundary)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// ListZoneBoundaries indicates an expected call of ListZoneBoundaries.
|
||||
func (mr *MockQuerierMockRecorder) ListZoneBoundaries(ctx any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListZoneBoundaries", reflect.TypeOf((*MockQuerier)(nil).ListZoneBoundaries), ctx)
|
||||
}
|
||||
|
||||
// PruneZoneBoundaries mocks base method.
|
||||
func (m *MockQuerier) PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "PruneZoneBoundaries", ctx, keep)
|
||||
ret0, _ := ret[0].([]string)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// PruneZoneBoundaries indicates an expected call of PruneZoneBoundaries.
|
||||
func (mr *MockQuerierMockRecorder) PruneZoneBoundaries(ctx, keep any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PruneZoneBoundaries", reflect.TypeOf((*MockQuerier)(nil).PruneZoneBoundaries), ctx, keep)
|
||||
}
|
||||
|
||||
// ReconfirmNeighbors mocks base method.
|
||||
func (m *MockQuerier) ReconfirmNeighbors(ctx context.Context) error {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1586,6 +1617,20 @@ func (mr *MockQuerierMockRecorder) SaveScopeCatalogue(ctx, arg any) *gomock.Call
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveScopeCatalogue", reflect.TypeOf((*MockQuerier)(nil).SaveScopeCatalogue), ctx, arg)
|
||||
}
|
||||
|
||||
// SaveZoneBoundary mocks base method.
|
||||
func (m *MockQuerier) SaveZoneBoundary(ctx context.Context, arg db.SaveZoneBoundaryParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "SaveZoneBoundary", ctx, arg)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// SaveZoneBoundary indicates an expected call of SaveZoneBoundary.
|
||||
func (mr *MockQuerierMockRecorder) SaveZoneBoundary(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveZoneBoundary", reflect.TypeOf((*MockQuerier)(nil).SaveZoneBoundary), ctx, arg)
|
||||
}
|
||||
|
||||
// SearchKnownRoutes mocks base method.
|
||||
func (m *MockQuerier) SearchKnownRoutes(ctx context.Context, arg db.SearchKnownRoutesParams) ([]db.SearchKnownRoutesRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@@ -245,6 +245,17 @@ type MeshmapperScopeCatalogue struct {
|
||||
LastError string `json:"last_error"`
|
||||
}
|
||||
|
||||
type MeshmapperZoneBoundary struct {
|
||||
Iata string `json:"iata"`
|
||||
Url string `json:"url"`
|
||||
Feature []byte `json:"feature"`
|
||||
Etag *string `json:"etag"`
|
||||
CheckedAt pgtype.Timestamptz `json:"checked_at"`
|
||||
AttemptedAt pgtype.Timestamptz `json:"attempted_at"`
|
||||
NextAttempt pgtype.Timestamptz `json:"next_attempt"`
|
||||
LastError string `json:"last_error"`
|
||||
}
|
||||
|
||||
type MvHourlyIataStat struct {
|
||||
Iata string `json:"iata"`
|
||||
Hour pgtype.Timestamptz `json:"hour"`
|
||||
|
||||
+10
-4
@@ -6,6 +6,7 @@ package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
@@ -46,10 +47,10 @@ type Querier interface {
|
||||
GetCrossIATANeighbors(ctx context.Context, arg GetCrossIATANeighborsParams) ([]GetCrossIATANeighborsRow, error)
|
||||
GetHourlyStats(ctx context.Context, arg GetHourlyStatsParams) ([]MvHourlyIataStat, error)
|
||||
GetIATA(ctx context.Context, iata string) (IataCode, error)
|
||||
// border is NULL when the IATA exists but has no border configured; a
|
||||
// missing row (unknown IATA) is sql.ErrNoRows, same not-found distinction
|
||||
// GetIATA already makes.
|
||||
GetIATABorder(ctx context.Context, iata string) ([]byte, error)
|
||||
// An imported MeshMapper boundary overrides the configured one. border is NULL
|
||||
// when neither exists; a missing row (unknown IATA) is sql.ErrNoRows, same
|
||||
// not-found distinction GetIATA already makes.
|
||||
GetIATABorder(ctx context.Context, iata string) (json.RawMessage, error)
|
||||
GetKnownRoutesByNode(ctx context.Context, arg GetKnownRoutesByNodeParams) ([]GetKnownRoutesByNodeRow, error)
|
||||
GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow, error)
|
||||
GetNodeByPubkey(ctx context.Context, publicKey []byte) (uuid.UUID, error)
|
||||
@@ -220,6 +221,9 @@ type Querier interface {
|
||||
// to the config after they'd already been ingested -- see
|
||||
// internal/ingest.BackfillChannelMessages.
|
||||
ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error)
|
||||
ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBoundary, error)
|
||||
// Drops imports for IATAs no longer configured, so their manual border returns.
|
||||
PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error)
|
||||
// Delete node_neighbors where the neighbor has departed from node_short_ids
|
||||
// for that IATA, or where its prefix_4 is now ambiguous.
|
||||
ReconfirmNeighbors(ctx context.Context) error
|
||||
@@ -260,6 +264,8 @@ type Querier interface {
|
||||
// Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data
|
||||
// after an error or 304. Imported names never replace existing manual metadata.
|
||||
SaveScopeCatalogue(ctx context.Context, arg SaveScopeCatalogueParams) error
|
||||
// NULL feature/etag/checked_at retain the last good boundary after an error or 304.
|
||||
SaveZoneBoundary(ctx context.Context, arg SaveZoneBoundaryParams) error
|
||||
// Returns known routes containing a subsequence from source to destination hash prefix.
|
||||
// Verifies source appears before destination in the route.
|
||||
SearchKnownRoutes(ctx context.Context, arg SearchKnownRoutesParams) ([]SearchKnownRoutesRow, error)
|
||||
|
||||
+108
-6
@@ -7,6 +7,7 @@ package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
@@ -390,15 +391,18 @@ func (q *Queries) GetIATA(ctx context.Context, iata string) (IataCode, error) {
|
||||
}
|
||||
|
||||
const getIATABorder = `-- name: GetIATABorder :one
|
||||
SELECT border FROM iata_codes WHERE iata = $1
|
||||
SELECT COALESCE(z.feature, i.border)::jsonb AS border
|
||||
FROM iata_codes i
|
||||
LEFT JOIN meshmapper_zone_boundaries z ON z.iata = i.iata
|
||||
WHERE i.iata = $1
|
||||
`
|
||||
|
||||
// border is NULL when the IATA exists but has no border configured; a
|
||||
// missing row (unknown IATA) is sql.ErrNoRows, same not-found distinction
|
||||
// GetIATA already makes.
|
||||
func (q *Queries) GetIATABorder(ctx context.Context, iata string) ([]byte, error) {
|
||||
// An imported MeshMapper boundary overrides the configured one. border is NULL
|
||||
// when neither exists; a missing row (unknown IATA) is sql.ErrNoRows, same
|
||||
// not-found distinction GetIATA already makes.
|
||||
func (q *Queries) GetIATABorder(ctx context.Context, iata string) (json.RawMessage, error) {
|
||||
row := q.db.QueryRow(ctx, getIATABorder, iata)
|
||||
var border []byte
|
||||
var border json.RawMessage
|
||||
err := row.Scan(&border)
|
||||
return border, err
|
||||
}
|
||||
@@ -3975,6 +3979,64 @@ func (q *Queries) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUn
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listZoneBoundaries = `-- name: ListZoneBoundaries :many
|
||||
SELECT iata, url, feature, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_zone_boundaries ORDER BY iata
|
||||
`
|
||||
|
||||
func (q *Queries) ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBoundary, error) {
|
||||
rows, err := q.db.Query(ctx, listZoneBoundaries)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []MeshmapperZoneBoundary{}
|
||||
for rows.Next() {
|
||||
var i MeshmapperZoneBoundary
|
||||
if err := rows.Scan(
|
||||
&i.Iata,
|
||||
&i.Url,
|
||||
&i.Feature,
|
||||
&i.Etag,
|
||||
&i.CheckedAt,
|
||||
&i.AttemptedAt,
|
||||
&i.NextAttempt,
|
||||
&i.LastError,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const pruneZoneBoundaries = `-- name: PruneZoneBoundaries :many
|
||||
DELETE FROM meshmapper_zone_boundaries WHERE NOT (iata = ANY($1::text[])) RETURNING iata
|
||||
`
|
||||
|
||||
// Drops imports for IATAs no longer configured, so their manual border returns.
|
||||
func (q *Queries) PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error) {
|
||||
rows, err := q.db.Query(ctx, pruneZoneBoundaries, keep)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []string{}
|
||||
for rows.Next() {
|
||||
var iata string
|
||||
if err := rows.Scan(&iata); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, iata)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const reconfirmNeighbors = `-- name: ReconfirmNeighbors :exec
|
||||
DELETE FROM node_neighbors nn
|
||||
WHERE NOT EXISTS (
|
||||
@@ -4522,6 +4584,46 @@ func (q *Queries) SaveScopeCatalogue(ctx context.Context, arg SaveScopeCatalogue
|
||||
return err
|
||||
}
|
||||
|
||||
const saveZoneBoundary = `-- name: SaveZoneBoundary :exec
|
||||
INSERT INTO meshmapper_zone_boundaries (iata, url, feature, etag, checked_at, attempted_at, next_attempt, last_error)
|
||||
VALUES ($1, $2, $3::jsonb, $4::text,
|
||||
$5::timestamptz, $6, $7, $8)
|
||||
ON CONFLICT (iata) DO UPDATE SET
|
||||
url = EXCLUDED.url,
|
||||
feature = COALESCE(EXCLUDED.feature, meshmapper_zone_boundaries.feature),
|
||||
etag = COALESCE(EXCLUDED.etag, meshmapper_zone_boundaries.etag),
|
||||
checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_zone_boundaries.checked_at),
|
||||
attempted_at = EXCLUDED.attempted_at,
|
||||
next_attempt = EXCLUDED.next_attempt,
|
||||
last_error = EXCLUDED.last_error
|
||||
`
|
||||
|
||||
type SaveZoneBoundaryParams struct {
|
||||
Iata string `json:"iata"`
|
||||
Url string `json:"url"`
|
||||
Feature []byte `json:"feature"`
|
||||
Etag *string `json:"etag"`
|
||||
CheckedAt pgtype.Timestamptz `json:"checked_at"`
|
||||
AttemptedAt pgtype.Timestamptz `json:"attempted_at"`
|
||||
NextAttempt pgtype.Timestamptz `json:"next_attempt"`
|
||||
LastError string `json:"last_error"`
|
||||
}
|
||||
|
||||
// NULL feature/etag/checked_at retain the last good boundary after an error or 304.
|
||||
func (q *Queries) SaveZoneBoundary(ctx context.Context, arg SaveZoneBoundaryParams) error {
|
||||
_, err := q.db.Exec(ctx, saveZoneBoundary,
|
||||
arg.Iata,
|
||||
arg.Url,
|
||||
arg.Feature,
|
||||
arg.Etag,
|
||||
arg.CheckedAt,
|
||||
arg.AttemptedAt,
|
||||
arg.NextAttempt,
|
||||
arg.LastError,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const searchKnownRoutes = `-- name: SearchKnownRoutes :many
|
||||
SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count
|
||||
FROM known_routes
|
||||
|
||||
@@ -12,9 +12,9 @@ import (
|
||||
)
|
||||
|
||||
// WithLocalBorders annotates node responses after cache reads. Historical rows
|
||||
// use the current startup configuration without extra SQL or cached-policy drift.
|
||||
// use the current boundaries without extra SQL or cached-policy drift.
|
||||
// Cached values are copied; the decorator never changes its inner reader's data.
|
||||
func WithLocalBorders(reader Reader, local *borders.Local) Reader {
|
||||
func WithLocalBorders(reader Reader, local *borders.Live) Reader {
|
||||
if local == nil {
|
||||
return reader
|
||||
}
|
||||
@@ -23,7 +23,7 @@ func WithLocalBorders(reader Reader, local *borders.Local) Reader {
|
||||
|
||||
type localBorderReader struct {
|
||||
Reader
|
||||
local *borders.Local
|
||||
local *borders.Live
|
||||
}
|
||||
|
||||
func (r *localBorderReader) ListNodes(ctx context.Context, nodeType int16, iatas []string, paths, traces *bool, pubkey []byte, prefix, name, scope string, cursor int64, limit int32, neighbors bool) (Page[NodeSummary], error) {
|
||||
|
||||
@@ -52,8 +52,8 @@ func TestLocalBordersAfterCache(t *testing.T) {
|
||||
client := cache.NewClient(mr.Addr(), "", 0)
|
||||
t.Cleanup(func() { client.Close() })
|
||||
cached := cache.NewCachedReader(base, client, cache.CacheTTLs{Nodes: time.Hour})
|
||||
first := api.WithLocalBorders(cached, local)
|
||||
second := api.WithLocalBorders(cached, other)
|
||||
first := api.WithLocalBorders(cached, borders.NewLive(local))
|
||||
second := api.WithLocalBorders(cached, borders.NewLive(other))
|
||||
if api.WithLocalBorders(cached, nil) != cached {
|
||||
t.Fatal("disabled decorator changed reader")
|
||||
}
|
||||
@@ -110,7 +110,7 @@ func TestLocalBordersPreserveReaderErrors(t *testing.T) {
|
||||
}
|
||||
want := errors.New("reader unavailable")
|
||||
base := &borderReader{err: want}
|
||||
reader := api.WithLocalBorders(base, local)
|
||||
reader := api.WithLocalBorders(base, borders.NewLive(local))
|
||||
if _, err := reader.GetNode(context.Background(), uuid.New()); !errors.Is(err, want) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -114,3 +114,28 @@ func BenchmarkPossiblyForeign(b *testing.B) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLiveSwap(t *testing.T) {
|
||||
var disabled *Live
|
||||
if disabled.PossiblyForeign(2, new(15.0), new(15.0)) != nil {
|
||||
t.Fatal("nil live classified")
|
||||
}
|
||||
live := NewLive(nil)
|
||||
if disabled.Ready() || live.Ready() {
|
||||
t.Fatal("ready without boundaries")
|
||||
}
|
||||
if live.PossiblyForeign(2, new(15.0), new(15.0)) != nil {
|
||||
t.Fatal("empty live classified")
|
||||
}
|
||||
local, err := New([]orb.Polygon{{{{10, 10}, {20, 10}, {20, 20}, {10, 20}, {10, 10}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
live.Store(local)
|
||||
if !live.Ready() {
|
||||
t.Fatal("not ready after store")
|
||||
}
|
||||
if got := live.PossiblyForeign(2, new(15.0), new(15.0)); got == nil || *got {
|
||||
t.Fatal("stored classifier not used", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package borders
|
||||
|
||||
import "sync/atomic"
|
||||
|
||||
// Live lets boundary refreshes replace the classifier under concurrent readers.
|
||||
// A nil Live disables classification; an empty one reports unknown until stored.
|
||||
type Live struct{ current atomic.Pointer[Local] }
|
||||
|
||||
func NewLive(local *Local) *Live {
|
||||
live := &Live{}
|
||||
live.current.Store(local)
|
||||
return live
|
||||
}
|
||||
|
||||
func (l *Live) Store(local *Local) { l.current.Store(local) }
|
||||
|
||||
// Ready reports whether any boundary is loaded.
|
||||
func (l *Live) Ready() bool { return l != nil && l.current.Load() != nil }
|
||||
|
||||
func (l *Live) PossiblyForeign(role int16, lat, lng *float64) *bool {
|
||||
if l == nil {
|
||||
return nil
|
||||
}
|
||||
return l.current.Load().PossiblyForeign(role, lat, lng)
|
||||
}
|
||||
Vendored
+11
@@ -396,6 +396,17 @@ func TestInvalidateScopeNamesClearsDerivedKeys(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestInvalidateIATABorder(t *testing.T) {
|
||||
c, mr := newTestClient(t)
|
||||
cr := &CachedReader{inner: &stubReader{}, c: c, ttl: CacheTTLs{}}
|
||||
mr.Set(keyIATABorderPrefix+"YOW", "x")
|
||||
mr.Set(keyIATABorderPrefix+"YYZ", "keep")
|
||||
cr.InvalidateIATABorder(context.Background(), "YOW")
|
||||
if got := mr.Keys(); len(got) != 1 || got[0] != keyIATABorderPrefix+"YYZ" {
|
||||
t.Fatalf("keys left: %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCachedReader_InvalidateNode(t *testing.T) {
|
||||
c, mr := newTestClient(t)
|
||||
nodeID := uuid.MustParse("00000000-0000-0000-0000-000000000001")
|
||||
|
||||
Vendored
+5
@@ -87,6 +87,11 @@ func (cr *CachedReader) InvalidateScopeNames(ctx context.Context) {
|
||||
cr.c.delPrefix(ctx, keyScopeByNamePrefix)
|
||||
}
|
||||
|
||||
// InvalidateIATABorder makes an imported or removed MeshMapper boundary visible.
|
||||
func (cr *CachedReader) InvalidateIATABorder(ctx context.Context, iata string) {
|
||||
cr.c.del(ctx, keyIATABorderPrefix+iata)
|
||||
}
|
||||
|
||||
// InvalidateNode removes the cached entries for a node by UUID.
|
||||
// Should be called from the ingest path after a node upsert.
|
||||
func (cr *CachedReader) InvalidateNode(ctx context.Context, nodeID uuid.UUID) {
|
||||
|
||||
@@ -271,8 +271,9 @@ type RoutesConfig struct {
|
||||
|
||||
// NodesConfig controls node-derived signal thresholds.
|
||||
type NodesConfig struct {
|
||||
// MarkForeign annotates repeaters outside the union of configured IATA border
|
||||
// files. Disabled by default; enabling it requires at least one usable border.
|
||||
// MarkForeign annotates repeaters outside the union of IATA borders, with
|
||||
// imported MeshMapper boundaries replacing border files. Disabled by default;
|
||||
// enabling it requires a border file or meshmapper.zones.
|
||||
MarkForeign bool `yaml:"mark_foreign"`
|
||||
// ClockDriftThreshold is the |device clock - server clock| magnitude, measured from a
|
||||
// repeater/room server's ADVERT timestamp, above which the node API reports
|
||||
@@ -412,6 +413,9 @@ func Load(path string) (*Config, error) {
|
||||
if err := cfg.validateMeshMapper(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if cfg.Nodes.MarkForeign && !cfg.MeshMapper.Zones.Enabled && !cfg.hasBorderFile() {
|
||||
return nil, fmt.Errorf("nodes.mark_foreign requires an iatas.*.borderFile or meshmapper.zones")
|
||||
}
|
||||
for iata, details := range cfg.IATAs {
|
||||
if details.BorderFile != "" && !filepath.IsAbs(details.BorderFile) {
|
||||
details.BorderFile = filepath.Join(configDir, details.BorderFile)
|
||||
|
||||
@@ -4,31 +4,38 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"maps"
|
||||
"os"
|
||||
"sort"
|
||||
"slices"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/borders"
|
||||
"github.com/paulmach/orb"
|
||||
"github.com/paulmach/orb/geojson"
|
||||
)
|
||||
|
||||
// LoadLocalBorders builds one startup-only classifier from configured border
|
||||
// files. IATA airport coordinates and auto-discovered IATAs are not boundaries.
|
||||
func LoadLocalBorders(cfg *Config) (*borders.Local, error) {
|
||||
func (c *Config) hasBorderFile() bool {
|
||||
for _, details := range c.IATAs {
|
||||
if details.BorderFile != "" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// LoadLocalBorders reads the validated border files used for foreign marking,
|
||||
// keyed by IATA. IATA airport coordinates are not boundaries. Nil when disabled.
|
||||
func LoadLocalBorders(cfg *Config) (map[string]json.RawMessage, error) {
|
||||
if !cfg.Nodes.MarkForeign {
|
||||
return nil, nil
|
||||
}
|
||||
var codes []string
|
||||
files := map[string]json.RawMessage{}
|
||||
for code, details := range cfg.IATAs {
|
||||
if details.BorderFile != "" {
|
||||
codes = append(codes, code)
|
||||
if details.BorderFile == "" {
|
||||
continue
|
||||
}
|
||||
}
|
||||
sort.Strings(codes)
|
||||
var polygons []orb.Polygon
|
||||
for _, code := range codes {
|
||||
raw, err := os.ReadFile(cfg.IATAs[code].BorderFile)
|
||||
raw, err := os.ReadFile(details.BorderFile)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nodes.mark_foreign: reading border for %s: %w", code, err)
|
||||
}
|
||||
@@ -36,7 +43,29 @@ func LoadLocalBorders(cfg *Config) (*borders.Local, error) {
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nodes.mark_foreign: invalid border for %s: %w", code, err)
|
||||
}
|
||||
feature, err := geojson.UnmarshalFeature(validated)
|
||||
files[code] = validated
|
||||
}
|
||||
// Fail at startup rather than on the first rebuild.
|
||||
if _, err := BuildLocalBorders(files, nil); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return files, nil
|
||||
}
|
||||
|
||||
// BuildLocalBorders unions validated borders; an imported boundary replaces
|
||||
// the same IATA's file. Nil when there are none.
|
||||
func BuildLocalBorders(files, imported map[string]json.RawMessage) (*borders.Local, error) {
|
||||
merged := maps.Clone(files)
|
||||
if merged == nil {
|
||||
merged = map[string]json.RawMessage{}
|
||||
}
|
||||
maps.Copy(merged, imported)
|
||||
if len(merged) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
var polygons []orb.Polygon
|
||||
for _, code := range slices.Sorted(maps.Keys(merged)) {
|
||||
feature, err := geojson.UnmarshalFeature(merged[code])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nodes.mark_foreign: decoding border for %s: %w", code, err)
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -22,16 +23,27 @@ func TestLoadLocalBorders(t *testing.T) {
|
||||
if got, err := LoadLocalBorders(&Config{}); err != nil || got != nil {
|
||||
t.Fatalf("disabled: %v %v", got, err)
|
||||
}
|
||||
if _, err := LoadLocalBorders(&Config{Nodes: NodesConfig{MarkForeign: true}}); err == nil {
|
||||
t.Fatal("enabled with no border")
|
||||
}
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
for _, tc := range []struct {
|
||||
text string
|
||||
bad bool
|
||||
}{
|
||||
{"nodes:\n mark_foreign: true\n", true},
|
||||
{"nodes:\n mark_foreign: true\nregions:\n - slug: a\n iatas: [AAA]\nmeshmapper:\n zones:\n enabled: true\n", false},
|
||||
} {
|
||||
if err := os.WriteFile(path, []byte(tc.text), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := Load(path); (err != nil) != tc.bad {
|
||||
t.Fatalf("%q: %v", tc.text, err)
|
||||
}
|
||||
}
|
||||
border := filepath.Join(dir, "region.json")
|
||||
raw := `{"type":"Feature","geometry":{"type":"MultiPolygon","coordinates":[[[[10,10],[20,10],[20,20],[10,20],[10,10]]],[[[30,10],[40,10],[40,20],[30,20],[30,10]]]]}}`
|
||||
if err := os.WriteFile(border, []byte(raw), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
if err := os.WriteFile(path, []byte("nodes:\n mark_foreign: true\niatas:\n AAA:\n borderFile: region.json\n BBB:\n name: Airport without an operating border\n"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -39,13 +51,31 @@ func TestLoadLocalBorders(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
local, err := LoadLocalBorders(cfg)
|
||||
files, err := LoadLocalBorders(cfg)
|
||||
if err != nil || len(files) != 1 {
|
||||
t.Fatal(files, err)
|
||||
}
|
||||
local, err := BuildLocalBorders(files, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if *local.PossiblyForeign(2, new(15.0), new(15.0)) || *local.PossiblyForeign(2, new(15.0), new(35.0)) || !*local.PossiblyForeign(2, new(25.0), new(25.0)) {
|
||||
t.Fatal("polygon union not used")
|
||||
}
|
||||
imported := map[string]json.RawMessage{"AAA": json.RawMessage(`{"type":"Feature","geometry":{"type":"Polygon","coordinates":[[[50,50],[60,50],[60,60],[50,60],[50,50]]]}}`)}
|
||||
replaced, err := BuildLocalBorders(files, imported)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !*replaced.PossiblyForeign(2, new(15.0), new(15.0)) || *replaced.PossiblyForeign(2, new(55.0), new(55.0)) {
|
||||
t.Fatal("imported boundary did not replace the file border")
|
||||
}
|
||||
if len(files) != 1 || string(files["AAA"]) == string(imported["AAA"]) {
|
||||
t.Fatal("build changed its inputs")
|
||||
}
|
||||
if empty, err := BuildLocalBorders(nil, nil); err != nil || empty != nil {
|
||||
t.Fatal(empty, err)
|
||||
}
|
||||
for _, bad := range []string{`not json`, `{"type":"Feature","geometry":null}`, `{"type":"Feature","geometry":{"type":"Polygon","coordinates":[[[179,10],[-179,10],[-179,20],[179,20],[179,10]]]}}`} {
|
||||
if err := os.WriteFile(border, []byte(bad), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"slices"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -15,6 +16,30 @@ const MaxScopeSources = 16
|
||||
|
||||
type MeshMapperConfig struct {
|
||||
Scopes MeshMapperScopesConfig `yaml:"scopes"`
|
||||
Zones MeshMapperZonesConfig `yaml:"zones"`
|
||||
}
|
||||
|
||||
// MeshMapperZonesConfig imports each region IATA's published boundary, overriding borderFile.
|
||||
type MeshMapperZonesConfig struct {
|
||||
Enabled bool `yaml:"enabled"`
|
||||
RefreshInterval duration `yaml:"refresh_interval"`
|
||||
}
|
||||
|
||||
func (c MeshMapperZonesConfig) Interval() time.Duration {
|
||||
if c.RefreshInterval.Duration == 0 {
|
||||
return 24 * time.Hour
|
||||
}
|
||||
return c.RefreshInterval.Duration
|
||||
}
|
||||
|
||||
// RegionIATAs returns the sorted, unique IATAs of every configured region.
|
||||
func (c *Config) RegionIATAs() []string {
|
||||
var iatas []string
|
||||
for _, region := range c.Regions {
|
||||
iatas = append(iatas, region.IATAs...)
|
||||
}
|
||||
slices.Sort(iatas)
|
||||
return slices.Compact(iatas)
|
||||
}
|
||||
|
||||
// MeshMapperScopesConfig augments, but never replaces, the manual scopes list.
|
||||
@@ -34,6 +59,15 @@ func (c MeshMapperScopesConfig) Interval() time.Duration {
|
||||
var meshMapperHost = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*\.meshmapper\.net$`)
|
||||
|
||||
func (c *Config) validateMeshMapper() error {
|
||||
if z := c.MeshMapper.Zones; z.Enabled {
|
||||
// The Zones API asks clients not to poll more than once an hour.
|
||||
if z.Interval() < time.Hour || z.Interval() > 7*24*time.Hour {
|
||||
return fmt.Errorf("meshmapper.zones.refresh_interval must be between 1h and 168h")
|
||||
}
|
||||
if len(c.RegionIATAs()) == 0 {
|
||||
return fmt.Errorf("meshmapper.zones needs at least one configured region IATA")
|
||||
}
|
||||
}
|
||||
s := c.MeshMapper.Scopes
|
||||
if !s.Enabled {
|
||||
return nil
|
||||
|
||||
@@ -47,3 +47,37 @@ func TestMeshMapperConfig(t *testing.T) {
|
||||
t.Fatal("empty sources accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMeshMapperZonesConfig(t *testing.T) {
|
||||
valid := "regions:\n - slug: ottawa\n iatas: [YOW, YOW]\n - slug: toronto\n iatas: [YYZ]\nmeshmapper:\n zones:\n enabled: true\n"
|
||||
for _, tc := range []struct {
|
||||
name, text string
|
||||
bad bool
|
||||
interval time.Duration
|
||||
}{
|
||||
{"default", valid, false, 24 * time.Hour},
|
||||
{"hourly", valid + " refresh_interval: 1h\n", false, time.Hour},
|
||||
{"too often", valid + " refresh_interval: 59m\n", true, 0},
|
||||
{"too rare", valid + " refresh_interval: 169h\n", true, 0},
|
||||
{"no regions", "meshmapper:\n zones:\n enabled: true\n", true, 0},
|
||||
{"disabled", "meshmapper:\n zones:\n enabled: false\n", false, 24 * time.Hour},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
p := filepath.Join(t.TempDir(), "config.yaml")
|
||||
if err := os.WriteFile(p, []byte(tc.text), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg, err := Load(p)
|
||||
if (err != nil) != tc.bad {
|
||||
t.Fatalf("error=%v", err)
|
||||
}
|
||||
if err == nil && cfg.MeshMapper.Zones.Interval() != tc.interval {
|
||||
t.Fatal(cfg.MeshMapper.Zones.Interval())
|
||||
}
|
||||
})
|
||||
}
|
||||
cfg := &Config{Regions: []RegionConfig{{IATAs: []string{"YYZ", "YOW"}}, {IATAs: []string{"YOW"}}}}
|
||||
if got := strings.Join(cfg.RegionIATAs(), ","); got != "YOW,YYZ" {
|
||||
t.Fatal(got)
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,7 @@ func TestForeignNodeUpdates(t *testing.T) {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
w.cfg.LocalBorders = nil
|
||||
if tc.enabled {
|
||||
w.cfg.LocalBorders = local
|
||||
w.cfg.LocalBorders = borders.NewLive(local)
|
||||
}
|
||||
data := []byte{tc.role}
|
||||
if tc.position {
|
||||
|
||||
@@ -58,8 +58,8 @@ import (
|
||||
|
||||
// Config holds the connection parameters for one broker.
|
||||
type Config struct {
|
||||
// LocalBorders is the same immutable classifier used by node API reads.
|
||||
LocalBorders *borders.Local
|
||||
// LocalBorders is the same live classifier used by node API reads.
|
||||
LocalBorders *borders.Live
|
||||
// BrokerName is a short human-readable label ("mqtt1", "mqtt2") used in
|
||||
// log messages and stored in packet_observations.source_broker.
|
||||
BrokerName string
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
// Package meshmapper imports the published regional scope catalogue, off the ingest path.
|
||||
// Package meshmapper imports published regional scope catalogues and boundaries, off the ingest path.
|
||||
package meshmapper
|
||||
|
||||
import (
|
||||
|
||||
@@ -0,0 +1,443 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package meshmapper
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/iatadb"
|
||||
)
|
||||
|
||||
const (
|
||||
ZonesURL = "https://meshmapper.net/get_zones.php"
|
||||
MaxZoneList = 1 << 20
|
||||
MaxBoundaryBody = 8 << 20 // outlines are never simplified
|
||||
|
||||
zoneListFresh = time.Hour // the API's own cache lifetime
|
||||
zoneFailureRetry = time.Hour // fair use: no more than one poll an hour
|
||||
)
|
||||
|
||||
var zoneSite = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*\.meshmapper\.net$`)
|
||||
|
||||
// Boundary is one IATA's imported MeshMapper outline and its fetch state.
|
||||
type Boundary struct {
|
||||
IATA, URL string
|
||||
Feature json.RawMessage
|
||||
ETag string
|
||||
CheckedAt, AttemptedAt, NextAttempt time.Time
|
||||
LastError string
|
||||
}
|
||||
|
||||
type ZoneStore interface {
|
||||
PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error)
|
||||
ListZoneBoundaries(ctx context.Context) ([]Boundary, error)
|
||||
SaveZoneBoundary(ctx context.Context, b Boundary) error
|
||||
}
|
||||
|
||||
type zoneRegion struct {
|
||||
country string
|
||||
b Boundary
|
||||
}
|
||||
|
||||
type zoneEntry struct {
|
||||
url string
|
||||
hasBoundary bool
|
||||
}
|
||||
|
||||
type zoneList struct {
|
||||
zones map[string]zoneEntry
|
||||
etag string
|
||||
fetchedAt, nextAttempt time.Time
|
||||
}
|
||||
|
||||
// Zones is owned by one background task; listeners are wired before Restore.
|
||||
type Zones struct {
|
||||
store ZoneStore
|
||||
enabled bool
|
||||
iatas []string
|
||||
interval time.Duration
|
||||
client *http.Client
|
||||
listURL string
|
||||
boundsURL func(site string) (string, bool)
|
||||
regions []*zoneRegion
|
||||
lists map[string]*zoneList
|
||||
retryAfter time.Time
|
||||
onChange func(ctx context.Context, iata string)
|
||||
onUpdate func(imported map[string]json.RawMessage)
|
||||
}
|
||||
|
||||
func NewZones(cfg config.MeshMapperZonesConfig, iatas []string, store ZoneStore) *Zones {
|
||||
return &Zones{store: store, enabled: cfg.Enabled, iatas: iatas, interval: cfg.Interval(), listURL: ZonesURL,
|
||||
boundsURL: boundaryEndpoint, lists: map[string]*zoneList{}, client: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
}}
|
||||
}
|
||||
|
||||
// OnChange receives each IATA whose served border changed, for cache invalidation.
|
||||
func (z *Zones) OnChange(fn func(ctx context.Context, iata string)) { z.onChange = fn }
|
||||
|
||||
// OnUpdate receives every imported boundary whenever the set changes.
|
||||
func (z *Zones) OnUpdate(fn func(imported map[string]json.RawMessage)) { z.onUpdate = fn }
|
||||
|
||||
// Restore prunes imports for unconfigured IATAs (all of them when disabled) and
|
||||
// loads saved boundaries, without making HTTP requests.
|
||||
func (z *Zones) Restore(ctx context.Context) error {
|
||||
keep := []string{}
|
||||
if z.enabled {
|
||||
keep = z.iatas
|
||||
}
|
||||
pruned, err := z.store.PruneZoneBoundaries(ctx, keep)
|
||||
if err != nil {
|
||||
return fmt.Errorf("prune MeshMapper boundaries: %w", err)
|
||||
}
|
||||
for _, iata := range pruned {
|
||||
slog.Info("MeshMapper boundary removed", "component", "meshmapper.zones", "iata", iata)
|
||||
z.changed(ctx, iata)
|
||||
}
|
||||
if !z.enabled {
|
||||
return nil
|
||||
}
|
||||
saved, err := z.store.ListZoneBoundaries(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("restore MeshMapper boundaries: %w", err)
|
||||
}
|
||||
byIATA := map[string]Boundary{}
|
||||
for _, b := range saved {
|
||||
byIATA[b.IATA] = b
|
||||
}
|
||||
for _, iata := range z.iatas {
|
||||
country := iatadb.CountryFor(iata)
|
||||
if country == "" {
|
||||
slog.Warn("MeshMapper boundary skipped: IATA has no known country", "component", "meshmapper.zones", "iata", iata)
|
||||
continue
|
||||
}
|
||||
b, ok := byIATA[iata]
|
||||
if !ok {
|
||||
b = Boundary{IATA: iata}
|
||||
}
|
||||
if b.LastError == "HTTP 429" && b.NextAttempt.After(z.retryAfter) {
|
||||
z.retryAfter = b.NextAttempt
|
||||
}
|
||||
z.regions = append(z.regions, &zoneRegion{country: country, b: b})
|
||||
z.log(b, "restored")
|
||||
}
|
||||
z.publish()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Refresh makes at most one request. The scheduler serializes calls every 15s.
|
||||
func (z *Zones) Refresh(ctx context.Context) (err error) {
|
||||
parent := ctx
|
||||
defer func() {
|
||||
if parent.Err() != nil {
|
||||
err = nil
|
||||
}
|
||||
}()
|
||||
ctx, cancel := context.WithTimeout(ctx, time.Minute)
|
||||
defer cancel()
|
||||
now := time.Now().UTC()
|
||||
if now.Before(z.retryAfter) {
|
||||
return nil
|
||||
}
|
||||
for _, r := range z.regions {
|
||||
if now.Before(r.b.NextAttempt) {
|
||||
continue
|
||||
}
|
||||
list := z.lists[r.country]
|
||||
if list == nil || now.Sub(list.fetchedAt) >= zoneListFresh {
|
||||
if list != nil && now.Before(list.nextAttempt) {
|
||||
continue
|
||||
}
|
||||
return z.fetchList(ctx, r.country, now)
|
||||
}
|
||||
return z.refresh(ctx, r, list, now)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (z *Zones) fetchList(ctx context.Context, country string, now time.Time) error {
|
||||
list := z.lists[country]
|
||||
if list == nil {
|
||||
list = &zoneList{}
|
||||
z.lists[country] = list
|
||||
}
|
||||
endpoint := z.listURL + "?country=" + url.QueryEscape(country)
|
||||
etag := ""
|
||||
if list.zones != nil {
|
||||
etag = list.etag
|
||||
}
|
||||
status, body, header, err := z.get(ctx, endpoint, etag, MaxZoneList)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
problem := ""
|
||||
switch status {
|
||||
case http.StatusOK:
|
||||
zones, decodeErr := decodeZones(body, country)
|
||||
if decodeErr != nil {
|
||||
problem = "invalid response"
|
||||
} else {
|
||||
list.zones, list.etag = zones, header.Get("ETag")
|
||||
}
|
||||
case http.StatusNotModified:
|
||||
if list.zones == nil {
|
||||
problem = "304 without cached list"
|
||||
}
|
||||
default:
|
||||
problem = z.statusProblem(status, header, now, &list.nextAttempt)
|
||||
}
|
||||
if problem == "" {
|
||||
list.fetchedAt = now
|
||||
} else if list.nextAttempt.Before(now.Add(zoneFailureRetry)) {
|
||||
list.nextAttempt = now.Add(zoneFailureRetry)
|
||||
}
|
||||
level := slog.LevelInfo
|
||||
if problem != "" {
|
||||
level = slog.LevelWarn
|
||||
}
|
||||
slog.Log(ctx, level, "MeshMapper zone list checked", "component", "meshmapper.zones", "country", country,
|
||||
"zones", len(list.zones), "last_error", problem)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (z *Zones) refresh(ctx context.Context, r *zoneRegion, list *zoneList, now time.Time) error {
|
||||
update := Boundary{IATA: r.b.IATA, URL: r.b.URL, AttemptedAt: now, NextAttempt: now.Add(z.interval)}
|
||||
var retryAt time.Time
|
||||
entry, listed := list.zones[r.b.IATA]
|
||||
endpoint, valid := "", false
|
||||
if listed {
|
||||
update.URL = entry.url
|
||||
endpoint, valid = z.boundsURL(entry.url)
|
||||
}
|
||||
switch {
|
||||
case !listed:
|
||||
update.LastError = "not listed"
|
||||
case !entry.hasBoundary:
|
||||
update.LastError = "no boundary"
|
||||
case !valid:
|
||||
update.LastError = "invalid site URL"
|
||||
default:
|
||||
etag := ""
|
||||
if r.b.Feature != nil {
|
||||
etag = r.b.ETag
|
||||
}
|
||||
status, body, header, err := z.get(ctx, endpoint, etag, MaxBoundaryBody)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch status {
|
||||
case http.StatusOK:
|
||||
feature, decodeErr := decodeBoundary(body, r.b.IATA)
|
||||
switch {
|
||||
case decodeErr != nil:
|
||||
update.LastError = "invalid response" // includes truncated bodies; retry
|
||||
case feature == nil:
|
||||
update.LastError = "no boundary"
|
||||
default:
|
||||
update.Feature = feature
|
||||
}
|
||||
case http.StatusNotModified:
|
||||
if r.b.Feature == nil {
|
||||
update.LastError = "304 without cached boundary"
|
||||
}
|
||||
default:
|
||||
update.LastError = z.statusProblem(status, header, now, &retryAt)
|
||||
}
|
||||
if update.LastError == "" {
|
||||
etag := header.Get("ETag")
|
||||
if len(etag) > 256 || strings.ContainsAny(etag, "\r\n") {
|
||||
update.LastError, update.Feature = "invalid ETag", nil
|
||||
} else {
|
||||
if status == http.StatusNotModified && etag == "" {
|
||||
etag = r.b.ETag
|
||||
}
|
||||
update.ETag, update.CheckedAt = etag, now
|
||||
}
|
||||
}
|
||||
if update.LastError != "" && update.LastError != "no boundary" {
|
||||
update.NextAttempt = now.Add(min(z.interval, zoneFailureRetry))
|
||||
if retryAt.After(update.NextAttempt) {
|
||||
update.NextAttempt = retryAt
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := z.store.SaveZoneBoundary(ctx, update); err != nil {
|
||||
r.b.NextAttempt = update.NextAttempt // avoid retrying every tick during a DB outage
|
||||
return fmt.Errorf("persist MeshMapper boundary %s: %w", r.b.IATA, err)
|
||||
}
|
||||
changed := update.Feature != nil && !bytes.Equal(update.Feature, r.b.Feature)
|
||||
if update.Feature != nil {
|
||||
r.b.Feature = update.Feature
|
||||
}
|
||||
if !update.CheckedAt.IsZero() {
|
||||
r.b.CheckedAt, r.b.ETag = update.CheckedAt, update.ETag
|
||||
}
|
||||
r.b.URL, r.b.AttemptedAt, r.b.NextAttempt, r.b.LastError = update.URL, now, update.NextAttempt, update.LastError
|
||||
if changed {
|
||||
z.changed(ctx, r.b.IATA)
|
||||
z.publish()
|
||||
}
|
||||
z.log(r.b, "checked")
|
||||
return nil
|
||||
}
|
||||
|
||||
// get returns transport errors only when ctx ended; other failures become status 0.
|
||||
func (z *Zones) get(ctx context.Context, endpoint, etag string, limit int64) (int, []byte, http.Header, error) {
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
||||
if err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
request.Header.Set("Accept", "application/json")
|
||||
request.Header.Set("User-Agent", "Beacon-MeshMapper-Zones/1")
|
||||
if etag != "" {
|
||||
request.Header.Set("If-None-Match", etag)
|
||||
}
|
||||
response, err := z.client.Do(request)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return 0, nil, nil, ctx.Err()
|
||||
}
|
||||
return 0, nil, http.Header{}, nil
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return response.StatusCode, nil, response.Header, nil
|
||||
}
|
||||
body, err := io.ReadAll(io.LimitReader(response.Body, limit+1))
|
||||
if err != nil || int64(len(body)) > limit {
|
||||
return response.StatusCode, nil, response.Header, nil
|
||||
}
|
||||
return response.StatusCode, body, response.Header, nil
|
||||
}
|
||||
|
||||
// statusProblem describes a failed response and honours Retry-After on 429/503.
|
||||
func (z *Zones) statusProblem(status int, header http.Header, now time.Time, retryAt *time.Time) string {
|
||||
if status == 0 {
|
||||
return "request failed" // don't persist untrusted error text
|
||||
}
|
||||
if status == http.StatusTooManyRequests || status == http.StatusServiceUnavailable {
|
||||
if seconds, err := strconv.ParseInt(header.Get("Retry-After"), 10, 64); err == nil && seconds > 0 && seconds <= int64((1<<63-1)/time.Second) {
|
||||
*retryAt = now.Add(time.Duration(seconds) * time.Second)
|
||||
}
|
||||
// The rate limit is shared by both endpoints, so pause every request.
|
||||
if status == http.StatusTooManyRequests {
|
||||
z.retryAfter = *retryAt
|
||||
if retryAt.IsZero() {
|
||||
z.retryAfter = now.Add(time.Minute)
|
||||
}
|
||||
}
|
||||
}
|
||||
return fmt.Sprintf("HTTP %d", status)
|
||||
}
|
||||
|
||||
func (z *Zones) changed(ctx context.Context, iata string) {
|
||||
if z.onChange != nil {
|
||||
z.onChange(ctx, iata)
|
||||
}
|
||||
}
|
||||
|
||||
func (z *Zones) publish() {
|
||||
if z.onUpdate == nil {
|
||||
return
|
||||
}
|
||||
imported := map[string]json.RawMessage{}
|
||||
for _, r := range z.regions {
|
||||
if r.b.Feature != nil {
|
||||
imported[r.b.IATA] = r.b.Feature
|
||||
}
|
||||
}
|
||||
z.onUpdate(imported)
|
||||
}
|
||||
|
||||
func (z *Zones) log(b Boundary, action string) {
|
||||
level := slog.LevelInfo
|
||||
if b.LastError != "" && b.LastError != "no boundary" && b.LastError != "not listed" {
|
||||
level = slog.LevelWarn
|
||||
}
|
||||
slog.Log(context.Background(), level, "MeshMapper boundary "+action, "component", "meshmapper.zones", "iata", b.IATA,
|
||||
"source", b.URL, "imported", b.Feature != nil, "checked_at", b.CheckedAt, "next_attempt", b.NextAttempt, "last_error", b.LastError)
|
||||
}
|
||||
|
||||
// boundaryEndpoint accepts only a published region site root.
|
||||
func boundaryEndpoint(site string) (string, bool) {
|
||||
u, err := url.Parse(site)
|
||||
if err != nil || u.Scheme != "https" || u.User != nil || !zoneSite.MatchString(u.Host) ||
|
||||
(u.Path != "" && u.Path != "/") || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" {
|
||||
return "", false
|
||||
}
|
||||
return "https://" + u.Host + "/get_geojson.php", true
|
||||
}
|
||||
|
||||
func decodeZones(body []byte, country string) (map[string]zoneEntry, error) {
|
||||
var document struct {
|
||||
Country string `json:"country"`
|
||||
Zones *[]struct {
|
||||
Code string `json:"code"`
|
||||
URL string `json:"url"`
|
||||
HasBoundary bool `json:"has_boundary"`
|
||||
} `json:"zones"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &document); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !strings.EqualFold(document.Country, country) || document.Zones == nil {
|
||||
return nil, fmt.Errorf("invalid zone list")
|
||||
}
|
||||
zones := make(map[string]zoneEntry, len(*document.Zones))
|
||||
for _, zone := range *document.Zones {
|
||||
zones[strings.ToUpper(zone.Code)] = zoneEntry{url: zone.URL, hasBoundary: zone.HasBoundary}
|
||||
}
|
||||
return zones, nil
|
||||
}
|
||||
|
||||
// decodeBoundary returns nil when MeshMapper has no usable outline for the region.
|
||||
func decodeBoundary(body []byte, iata string) (json.RawMessage, error) {
|
||||
var collection struct {
|
||||
Type string `json:"type"`
|
||||
Features []json.RawMessage `json:"features"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &collection); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if collection.Type != "FeatureCollection" || len(collection.Features) != 1 {
|
||||
return nil, fmt.Errorf("expected one region feature")
|
||||
}
|
||||
var feature struct {
|
||||
Geometry json.RawMessage `json:"geometry"`
|
||||
Properties struct {
|
||||
Code string `json:"code"`
|
||||
} `json:"properties"`
|
||||
}
|
||||
if err := json.Unmarshal(collection.Features[0], &feature); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !strings.EqualFold(feature.Properties.Code, iata) {
|
||||
return nil, fmt.Errorf("feature is for another region")
|
||||
}
|
||||
if len(feature.Geometry) == 0 || string(feature.Geometry) == "null" {
|
||||
return nil, nil
|
||||
}
|
||||
validated, err := config.ValidateBorder(collection.Features[0])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Foreign marking must accept anything the map serves.
|
||||
if _, err := config.BuildLocalBorders(nil, map[string]json.RawMessage{iata: validated}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return validated, nil
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package meshmapper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
)
|
||||
|
||||
const square = `{"type":"Polygon","coordinates":[[[-76,45],[-75,45],[-75,46],[-76,46],[-76,45]]]}`
|
||||
|
||||
func boundaryBody(code, geometry string) string {
|
||||
return `{"type":"FeatureCollection","generated_at":"2026-09-25T15:00:00Z","features":[{"type":"Feature","geometry":` + geometry +
|
||||
`,"properties":{"code":"` + code + `","name":"Ottawa, CA","center":[-75.7,45.4],"radius_km":42,"has_boundary":true}}]}`
|
||||
}
|
||||
|
||||
type zoneMemoryStore struct {
|
||||
rows map[string]Boundary
|
||||
pruned []string
|
||||
fail bool
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) PruneZoneBoundaries(_ context.Context, keep []string) ([]string, error) {
|
||||
var removed []string
|
||||
for iata := range s.rows {
|
||||
if !slices.Contains(keep, iata) {
|
||||
delete(s.rows, iata)
|
||||
removed = append(removed, iata)
|
||||
}
|
||||
}
|
||||
s.pruned = removed
|
||||
return removed, nil
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) ListZoneBoundaries(context.Context) ([]Boundary, error) {
|
||||
var out []Boundary
|
||||
for _, b := range s.rows {
|
||||
out = append(out, b)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) SaveZoneBoundary(_ context.Context, b Boundary) error {
|
||||
if s.fail {
|
||||
return errors.New("offline")
|
||||
}
|
||||
old := s.rows[b.IATA]
|
||||
if b.Feature == nil {
|
||||
b.Feature = old.Feature
|
||||
}
|
||||
if b.CheckedAt.IsZero() {
|
||||
b.CheckedAt, b.ETag = old.CheckedAt, old.ETag
|
||||
}
|
||||
s.rows[b.IATA] = b
|
||||
return nil
|
||||
}
|
||||
|
||||
type fakeMeshMapper struct {
|
||||
*httptest.Server
|
||||
list, boundary string
|
||||
listStatus, status int
|
||||
etag, retryAfter string
|
||||
listCalls, calls int
|
||||
ifNoneMatch []string
|
||||
}
|
||||
|
||||
func newFakeMeshMapper(t *testing.T) *fakeMeshMapper {
|
||||
f := &fakeMeshMapper{listStatus: 200, status: 200, etag: `"v1"`, boundary: boundaryBody("YOW", square)}
|
||||
f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/get_zones.php":
|
||||
f.listCalls++
|
||||
if r.URL.Query().Get("country") != "CA" {
|
||||
t.Errorf("country query: %s", r.URL.RawQuery)
|
||||
}
|
||||
w.WriteHeader(f.listStatus)
|
||||
_, _ = w.Write([]byte(f.list))
|
||||
case "/get_geojson.php":
|
||||
f.calls++
|
||||
f.ifNoneMatch = append(f.ifNoneMatch, r.Header.Get("If-None-Match"))
|
||||
w.Header().Set("ETag", f.etag)
|
||||
w.Header().Set("Retry-After", f.retryAfter)
|
||||
w.WriteHeader(f.status)
|
||||
if f.status == 200 {
|
||||
_, _ = w.Write([]byte(f.boundary))
|
||||
}
|
||||
default:
|
||||
t.Errorf("unexpected path %s", r.URL.Path)
|
||||
}
|
||||
}))
|
||||
f.list = `{"country":"CA","count":1,"zones":[{"code":"YOW","url":"` + f.URL + `/","has_boundary":true,"group":null}],"groups":[]}`
|
||||
t.Cleanup(f.Close)
|
||||
return f
|
||||
}
|
||||
|
||||
type zoneHarness struct {
|
||||
z *Zones
|
||||
store *zoneMemoryStore
|
||||
changed []string
|
||||
imported map[string]json.RawMessage
|
||||
}
|
||||
|
||||
func newZoneHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, enabled bool) *zoneHarness {
|
||||
t.Helper()
|
||||
h := &zoneHarness{store: store}
|
||||
h.z = NewZones(config.MeshMapperZonesConfig{Enabled: enabled}, []string{"YOW"}, store)
|
||||
h.z.listURL = f.URL + "/get_zones.php"
|
||||
h.z.boundsURL = func(site string) (string, bool) { return site + "get_geojson.php", strings.HasPrefix(site, f.URL) }
|
||||
h.z.OnChange(func(_ context.Context, iata string) { h.changed = append(h.changed, iata) })
|
||||
h.z.OnUpdate(func(imported map[string]json.RawMessage) { h.imported = imported })
|
||||
if err := h.z.Restore(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// tick forces the region due and runs refreshes until the boundary request is made.
|
||||
func (h *zoneHarness) tick(t *testing.T) Boundary {
|
||||
t.Helper()
|
||||
h.z.regions[0].b.NextAttempt = time.Time{}
|
||||
for range 2 {
|
||||
if err := h.z.Refresh(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return h.store.rows["YOW"]
|
||||
}
|
||||
|
||||
func TestZonesImportAndConditionalRefresh(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true)
|
||||
b := h.tick(t)
|
||||
if f.listCalls != 1 || f.calls != 1 || b.Feature == nil || b.ETag != `"v1"` || b.LastError != "" || b.URL != f.URL+"/" {
|
||||
t.Fatalf("import failed: %+v list=%d calls=%d", b, f.listCalls, f.calls)
|
||||
}
|
||||
if !slices.Equal(h.changed, []string{"YOW"}) || h.imported["YOW"] == nil {
|
||||
t.Fatal("listeners not told", h.changed, h.imported)
|
||||
}
|
||||
if !strings.Contains(string(b.Feature), `"bbox"`) || !strings.Contains(string(b.Feature), `"code":"YOW"`) {
|
||||
t.Fatal("feature not validated with bbox and properties", string(b.Feature))
|
||||
}
|
||||
if got := b.NextAttempt.Sub(b.AttemptedAt); got != 24*time.Hour {
|
||||
t.Fatal("refresh interval", got)
|
||||
}
|
||||
|
||||
f.status = 304
|
||||
b = h.tick(t)
|
||||
if f.listCalls != 1 || f.ifNoneMatch[1] != `"v1"` || b.Feature == nil || b.LastError != "" || len(h.changed) != 1 {
|
||||
t.Fatalf("304 mishandled: %+v list=%d changed=%v", b, f.listCalls, h.changed)
|
||||
}
|
||||
|
||||
f.status, f.etag = 200, `"v2"`
|
||||
f.boundary = boundaryBody("YOW", `{"type":"Polygon","coordinates":[[[-77,45],[-75,45],[-75,46],[-77,46],[-77,45]]]}`)
|
||||
b = h.tick(t)
|
||||
if b.ETag != `"v2"` || len(h.changed) != 2 || !strings.Contains(string(h.imported["YOW"]), "-77") {
|
||||
t.Fatal("changed boundary not published", b.ETag, h.changed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesKeepLastGoodBoundary(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, boundary string
|
||||
status int
|
||||
wantErr string
|
||||
wantRetry time.Duration
|
||||
}{
|
||||
{"null geometry", boundaryBody("YOW", "null"), 200, "no boundary", 24 * time.Hour},
|
||||
{"not found", "", 404, "HTTP 404", time.Hour},
|
||||
{"unavailable", "", 503, "HTTP 503", time.Hour},
|
||||
{"truncated", boundaryBody("YOW", square)[:80], 200, "invalid response", time.Hour},
|
||||
{"other region", boundaryBody("YVR", square), 200, "invalid response", time.Hour},
|
||||
{"oversized", `{"type":"FeatureCollection","features":[],"pad":"` + strings.Repeat("x", MaxBoundaryBody) + `"}`, 200, "invalid response", time.Hour},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true)
|
||||
good := h.tick(t).Feature
|
||||
f.status, f.boundary = tc.status, tc.boundary
|
||||
b := h.tick(t)
|
||||
if b.LastError != tc.wantErr || string(b.Feature) != string(good) || string(h.imported["YOW"]) != string(good) || len(h.changed) != 1 {
|
||||
t.Fatalf("got %q, feature kept=%v", b.LastError, string(b.Feature) == string(good))
|
||||
}
|
||||
if got := b.NextAttempt.Sub(b.AttemptedAt); got != tc.wantRetry {
|
||||
t.Fatal("retry", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesSkipRegionsWithoutBoundary(t *testing.T) {
|
||||
for _, tc := range []struct{ name, zones, want string }{
|
||||
{"not listed", `[]`, "not listed"},
|
||||
{"no boundary", `[{"code":"YOW","url":"URL/","has_boundary":false}]`, "no boundary"},
|
||||
{"foreign site", `[{"code":"yow","url":"https://example.com/","has_boundary":true}]`, "invalid site URL"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
f.list = `{"country":"CA","count":1,"zones":` + strings.ReplaceAll(tc.zones, "URL", f.URL) + `,"groups":[]}`
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true)
|
||||
b := h.tick(t)
|
||||
if f.calls != 0 || b.LastError != tc.want || b.Feature != nil || len(h.changed) != 0 {
|
||||
t.Fatalf("got %+v calls=%d", b, f.calls)
|
||||
}
|
||||
if got := b.NextAttempt.Sub(b.AttemptedAt); got != 24*time.Hour {
|
||||
t.Fatal("authoritative answer retried early", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesListFailureBacksOff(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
f.listStatus = 500
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true)
|
||||
for range 3 {
|
||||
if err := h.z.Refresh(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if f.listCalls != 1 || f.calls != 0 || len(h.store.rows) != 0 {
|
||||
t.Fatal("failed list retried early", f.listCalls, f.calls)
|
||||
}
|
||||
f.listStatus, f.list = 200, `{"country":"US","zones":[]}`
|
||||
h.z.lists["CA"].nextAttempt, h.z.lists["CA"].fetchedAt = time.Time{}, time.Time{}
|
||||
_ = h.z.Refresh(context.Background())
|
||||
if h.z.lists["CA"].zones != nil {
|
||||
t.Fatal("list for another country accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesRateLimitPausesEveryRequest(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true)
|
||||
f.status, f.retryAfter = 429, "120"
|
||||
b := h.tick(t)
|
||||
if b.LastError != "HTTP 429" || b.NextAttempt.Sub(b.AttemptedAt) != time.Hour || h.z.retryAfter.Sub(b.AttemptedAt) != 2*time.Minute {
|
||||
t.Fatalf("429 not honoured: %+v retryAfter=%v", b, h.z.retryAfter)
|
||||
}
|
||||
h.z.regions[0].b.NextAttempt = time.Time{}
|
||||
_ = h.z.Refresh(context.Background())
|
||||
if f.calls != 1 {
|
||||
t.Fatal("requested during Retry-After")
|
||||
}
|
||||
restarted := newZoneHarness(t, f, h.store, true)
|
||||
if !restarted.z.retryAfter.Equal(b.NextAttempt) {
|
||||
t.Fatal("persisted 429 not restored")
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesRestoreAndPrune(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
store := &zoneMemoryStore{rows: map[string]Boundary{}}
|
||||
first := newZoneHarness(t, f, store, true)
|
||||
first.tick(t)
|
||||
store.rows["OLD"] = Boundary{IATA: "OLD", Feature: json.RawMessage(`{}`)}
|
||||
|
||||
restarted := newZoneHarness(t, f, store, true)
|
||||
if !slices.Equal(store.pruned, []string{"OLD"}) || !slices.Equal(restarted.changed, []string{"OLD"}) {
|
||||
t.Fatal("unconfigured IATA not pruned", store.pruned, restarted.changed)
|
||||
}
|
||||
if restarted.imported["YOW"] == nil || f.calls != 1 {
|
||||
t.Fatal("saved boundary not restored without a request")
|
||||
}
|
||||
if err := restarted.z.Refresh(context.Background()); err != nil || f.calls != 1 || f.listCalls != 1 {
|
||||
t.Fatal("refreshed before due", err, f.calls)
|
||||
}
|
||||
|
||||
disabled := newZoneHarness(t, f, store, false)
|
||||
if len(store.rows) != 0 || !slices.Equal(disabled.changed, []string{"YOW"}) || len(disabled.z.regions) != 0 {
|
||||
t.Fatal("disabling did not remove imports")
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesPersistFailureWaitsForNextAttempt(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}, fail: true}, true)
|
||||
h.z.regions[0].b.NextAttempt = time.Time{}
|
||||
_ = h.z.Refresh(context.Background())
|
||||
if err := h.z.Refresh(context.Background()); err == nil {
|
||||
t.Fatal("persist error hidden")
|
||||
}
|
||||
if h.z.regions[0].b.NextAttempt.IsZero() || h.imported["YOW"] != nil || len(h.changed) != 0 {
|
||||
t.Fatal("uncommitted boundary published")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBoundaryEndpoint(t *testing.T) {
|
||||
for site, want := range map[string]string{
|
||||
"https://yow.meshmapper.net/": "https://yow.meshmapper.net/get_geojson.php",
|
||||
"https://yow.meshmapper.net": "https://yow.meshmapper.net/get_geojson.php",
|
||||
"http://yow.meshmapper.net/": "",
|
||||
"https://meshmapper.net.evil.com/": "",
|
||||
"https://user@yow.meshmapper.net/": "",
|
||||
"https://yow.meshmapper.net/x/": "",
|
||||
"https://yow.meshmapper.net/?a=b": "",
|
||||
"https://yow.meshmapper.net:8443/": "",
|
||||
"https://yow.meshmapper.net/#fragment": "",
|
||||
} {
|
||||
got, ok := boundaryEndpoint(site)
|
||||
if got != want || ok != (want != "") {
|
||||
t.Errorf("%s: %q %v", site, got, ok)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeBoundaryRejectsUnusableOutlines(t *testing.T) {
|
||||
for name, body := range map[string]string{
|
||||
"two features": `{"type":"FeatureCollection","features":[{"type":"Feature","geometry":null,"properties":{"code":"YOW"}},{"type":"Feature","geometry":null,"properties":{"code":"YOW"}}]}`,
|
||||
"not a polygon": boundaryBody("YOW", `{"type":"Point","coordinates":[-75,45]}`),
|
||||
"open ring": boundaryBody("YOW", `{"type":"Polygon","coordinates":[[[-76,45],[-75,45],[-75,46],[-76,46]]]}`),
|
||||
"antimeridian": boundaryBody("YOW", `{"type":"Polygon","coordinates":[[[179,10],[-179,10],[-179,20],[179,20],[179,10]]]}`),
|
||||
"zero area": boundaryBody("YOW", `{"type":"Polygon","coordinates":[[[-76,45],[-75,45],[-74,45],[-76,45]]]}`),
|
||||
} {
|
||||
if _, err := decodeBoundary([]byte(body), "YOW"); err == nil {
|
||||
t.Errorf("%s accepted", name)
|
||||
}
|
||||
}
|
||||
if feature, err := decodeBoundary([]byte(boundaryBody("yow", square)), "YOW"); err != nil || feature == nil {
|
||||
t.Fatal("case-insensitive code rejected", err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user