mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-23 09:52:32 +00:00
feat(meshmapper): discover scope sources from the zone list
This commit is contained in:
1 parent
7267b8bca3
commit
bfda0d5281
16 files changed
+518
-258
No files matched your search
@@ -194,13 +194,11 @@ even when the same bearer-authenticated request works with curl.
|
||||
```yaml
|
||||
# MeshMapper integrations (optional, no API key). Import transport scopes and
|
||||
# IATA border outlines from MeshMapper instead of maintaining scopes: and
|
||||
# iatas.*.borderFile by hand. Neither needs the IATAs to be in a configured
|
||||
# region. See config.yaml.example for limits and behaviour.
|
||||
# iatas.*.borderFile by hand. Both cover every known IATA that MeshMapper lists;
|
||||
# no sources to configure. See config.yaml.example for limits and behaviour.
|
||||
#meshmapper:
|
||||
# scopes:
|
||||
# enabled: true
|
||||
# sources:
|
||||
# YVR: https://yvr.meshmapper.net/get_scopes.php
|
||||
# zones:
|
||||
# enabled: true
|
||||
|
||||
|
||||
+3
-2
@@ -198,10 +198,11 @@ func main() {
|
||||
scopes.Load(scopeEntries)
|
||||
scopes.SetManualMembers(cfg.ManualScopeMembers())
|
||||
slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup")
|
||||
directory := meshmapper.NewDirectory()
|
||||
var scopeImporter *meshmapper.Importer
|
||||
if cfg.MeshMapper.Scopes.Enabled {
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
scopeImporter, err = meshmapper.New(restoreCtx, cfg.MeshMapper.Scopes, store, scopes, scopeEntries)
|
||||
scopeImporter, err = meshmapper.New(restoreCtx, cfg.MeshMapper.Scopes, store, directory, scopes, scopeEntries)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper scope catalogues", "component", "startup", "error", err)
|
||||
@@ -210,7 +211,7 @@ func main() {
|
||||
}
|
||||
|
||||
// Always restored, so disabling the import prunes it and file borders return.
|
||||
zones := meshmapper.NewZones(cfg.MeshMapper.Zones, store)
|
||||
zones := meshmapper.NewZones(cfg.MeshMapper.Zones, store, directory)
|
||||
if cr, ok := reader.(*cache.CachedReader); ok {
|
||||
zones.OnChange(cr.InvalidateIATABorder)
|
||||
}
|
||||
|
||||
+8
-7
@@ -46,21 +46,22 @@ ratelimit:
|
||||
# iatas.*.borderFile by hand.
|
||||
#
|
||||
# Scope-name discovery. Manual scopes remain authoritative.
|
||||
# Sources are keyed by IATA code; the IATA need not be in a configured region.
|
||||
# Group endpoints are not supported: their names cannot be attributed to each IATA.
|
||||
# At most 16 sources, 64 names per source, 64 KiB per response; no page scraping.
|
||||
# Every known IATA, including ones auto-created from traffic, is looked up in its
|
||||
# country's zone list (shared with zones below) and its site's get_scopes.php
|
||||
# catalogue is imported; IATAs MeshMapper doesn't list are skipped. No sources
|
||||
# to configure; a legacy sources: list is ignored with a warning.
|
||||
# Catalogues covering more than one IATA are rejected.
|
||||
# 64 names per catalogue, 64 KiB per response; no page scraping.
|
||||
# Catalogues and ETags persist in PostgreSQL. Failed refreshes retain known names.
|
||||
# Refresh success/failure/freshness is logged under component=meshmapper.scopes.
|
||||
# Removing/turning off a source deactivates its imported matching keys on restart;
|
||||
# Turning scopes off deactivates imported matching keys on restart;
|
||||
# historical scope identities and recorded evidence are retained.
|
||||
# Region-filtered scope lists and scope stats show imported names only under the
|
||||
# IATAs whose catalogue lists them; regions without a source show manual scopes only.
|
||||
# IATAs whose catalogue lists them; IATAs without a catalogue show manual scopes only.
|
||||
#meshmapper:
|
||||
# scopes:
|
||||
# enabled: false
|
||||
# refresh_interval: 1h # 5m-24h; one source checked per 15s tick
|
||||
# sources:
|
||||
# YOW: https://yow.meshmapper.net/get_scopes.php
|
||||
#
|
||||
# Region boundaries (Zones API). Every known IATA, including ones auto-created
|
||||
# from traffic, is looked up in its country's zone list, then its get_geojson.php
|
||||
|
||||
+18
-1
@@ -29,6 +29,23 @@ func (s *Store) GetScopeCatalogue(ctx context.Context, iata, url string) (*meshm
|
||||
return cache, nil
|
||||
}
|
||||
|
||||
func (s *Store) ListScopeCatalogues(ctx context.Context) ([]meshmapper.Catalogue, error) {
|
||||
rows, err := s.q.ListScopeCatalogues(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]meshmapper.Catalogue, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
c := meshmapper.Catalogue{IATA: row.Iata, URL: row.Url, Cache: meshmapper.Cache{Payload: row.Payload, CheckedAt: row.CheckedAt.Time,
|
||||
AttemptedAt: row.AttemptedAt.Time, NextAttempt: row.NextAttempt.Time, LastError: row.LastError}}
|
||||
if row.Etag != nil {
|
||||
c.ETag = *row.Etag
|
||||
}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *Store) SaveScopeCatalogue(ctx context.Context, iata, url string, cache meshmapper.Cache, entries []scopestore.Entry) error {
|
||||
params := sqlc.SaveScopeCatalogueParams{
|
||||
Iata: iata, Url: url, Payload: cache.Payload, LastError: cache.LastError,
|
||||
@@ -48,7 +65,7 @@ func (s *Store) SaveScopeCatalogue(ctx context.Context, iata, url string, cache
|
||||
return s.q.SaveScopeCatalogue(ctx, params)
|
||||
}
|
||||
|
||||
func (s *Store) ListZoneIATAs(ctx context.Context) ([]string, error) {
|
||||
func (s *Store) ListKnownIATAs(ctx context.Context) ([]string, error) {
|
||||
rows, err := s.q.ListIATAs(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -44,7 +44,7 @@ func TestMeshMapperCataloguePostgres(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
for _, table := range []string{"transport_scopes", "meshmapper_scope_catalogues"} {
|
||||
for _, table := range []string{"iata_codes", "transport_scopes", "meshmapper_scope_catalogues"} {
|
||||
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -55,6 +55,9 @@ func TestMeshMapperCataloguePostgres(t *testing.T) {
|
||||
if err := store.UpsertTransportScope(ctx, manual.Name, "Operator label", manual.TransportKey, manual.KeyFingerprint); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.UpsertIATA(ctx, "YOW"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
url := "https://yow.meshmapper.net/get_scopes.php"
|
||||
if row, err := store.GetScopeCatalogue(ctx, "YOW", url); err != nil || row != nil {
|
||||
t.Fatal(row, err)
|
||||
@@ -79,7 +82,7 @@ func TestMeshMapperCataloguePostgres(t *testing.T) {
|
||||
t.Fatal(row, err)
|
||||
}
|
||||
scopes := scopestore.New()
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": url}}, store, scopes, manualRows)
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(), scopes, manualRows)
|
||||
if err != nil || len(scopes.Entries()) != 2 {
|
||||
t.Fatal("restart did not restore imported membership", err)
|
||||
}
|
||||
@@ -102,7 +105,7 @@ func TestMeshMapperCataloguePostgres(t *testing.T) {
|
||||
if err := store.SaveScopeCatalogue(ctx, "YOW", url, update, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": url}}, store, scopes, manualRows)
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(), scopes, manualRows)
|
||||
if err != nil || len(scopes.Entries()) != 1 {
|
||||
t.Fatal("removed imported membership remained active", err)
|
||||
}
|
||||
|
||||
@@ -1576,6 +1576,9 @@ SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at,
|
||||
-- name: GetScopeCatalogue :one
|
||||
SELECT * FROM meshmapper_scope_catalogues WHERE iata = $1 AND url = $2;
|
||||
|
||||
-- name: ListScopeCatalogues :many
|
||||
SELECT * FROM meshmapper_scope_catalogues ORDER BY iata, attempted_at;
|
||||
|
||||
-- name: SaveScopeCatalogue :exec
|
||||
-- One statement commits the validated snapshot and its lookup identities together.
|
||||
-- Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data
|
||||
|
||||
@@ -1254,6 +1254,21 @@ func (mr *MockQuerierMockRecorder) ListRouteEvidence(ctx, arg any) *gomock.Call
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListRouteEvidence", reflect.TypeOf((*MockQuerier)(nil).ListRouteEvidence), ctx, arg)
|
||||
}
|
||||
|
||||
// ListScopeCatalogues mocks base method.
|
||||
func (m *MockQuerier) ListScopeCatalogues(ctx context.Context) ([]db.MeshmapperScopeCatalogue, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "ListScopeCatalogues", ctx)
|
||||
ret0, _ := ret[0].([]db.MeshmapperScopeCatalogue)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// ListScopeCatalogues indicates an expected call of ListScopeCatalogues.
|
||||
func (mr *MockQuerierMockRecorder) ListScopeCatalogues(ctx any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListScopeCatalogues", reflect.TypeOf((*MockQuerier)(nil).ListScopeCatalogues), ctx)
|
||||
}
|
||||
|
||||
// ListTraceTags mocks base method.
|
||||
func (m *MockQuerier) ListTraceTags(ctx context.Context, arg db.ListTraceTagsParams) ([]db.ListTraceTagsRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@@ -207,6 +207,7 @@ type Querier interface {
|
||||
// Full-byte equality is required even when the compact digest matches. TRACE path bytes
|
||||
// carry readings; unclassified legacy observations cannot be safely called ordinary paths.
|
||||
ListRouteEvidence(ctx context.Context, arg ListRouteEvidenceParams) ([]ListRouteEvidenceRow, error)
|
||||
ListScopeCatalogues(ctx context.Context) ([]MeshmapperScopeCatalogue, error)
|
||||
// ============================================================
|
||||
// TRACES
|
||||
// ============================================================
|
||||
|
||||
@@ -3723,6 +3723,39 @@ func (q *Queries) ListRegions(ctx context.Context) ([]ListRegionsRow, error) {
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listScopeCatalogues = `-- name: ListScopeCatalogues :many
|
||||
SELECT iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_scope_catalogues ORDER BY iata, attempted_at
|
||||
`
|
||||
|
||||
func (q *Queries) ListScopeCatalogues(ctx context.Context) ([]MeshmapperScopeCatalogue, error) {
|
||||
rows, err := q.db.Query(ctx, listScopeCatalogues)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []MeshmapperScopeCatalogue{}
|
||||
for rows.Next() {
|
||||
var i MeshmapperScopeCatalogue
|
||||
if err := rows.Scan(
|
||||
&i.Iata,
|
||||
&i.Url,
|
||||
&i.Payload,
|
||||
&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 listTraceTags = `-- name: ListTraceTags :many
|
||||
|
||||
WITH tags AS (
|
||||
|
||||
@@ -5,14 +5,10 @@ package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MaxScopeSources bounds both network refresh work and imported matching keys.
|
||||
const MaxScopeSources = 16
|
||||
|
||||
type MeshMapperConfig struct {
|
||||
Scopes MeshMapperScopesConfig `yaml:"scopes"`
|
||||
Zones MeshMapperZonesConfig `yaml:"zones"`
|
||||
@@ -32,9 +28,10 @@ func (c MeshMapperZonesConfig) Interval() time.Duration {
|
||||
}
|
||||
|
||||
// MeshMapperScopesConfig augments, but never replaces, the manual scopes list.
|
||||
// Sources are discovered per known IATA from the zone list.
|
||||
type MeshMapperScopesConfig struct {
|
||||
Enabled bool `yaml:"enabled"`
|
||||
Sources map[string]string `yaml:"sources"` // configured IATA -> published regional endpoint
|
||||
Sources map[string]string `yaml:"sources"` // deprecated: ignored, kept to warn
|
||||
RefreshInterval duration `yaml:"refresh_interval"`
|
||||
}
|
||||
|
||||
@@ -45,11 +42,6 @@ func (c MeshMapperScopesConfig) Interval() time.Duration {
|
||||
return c.RefreshInterval.Duration
|
||||
}
|
||||
|
||||
var (
|
||||
meshMapperHost = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*\.meshmapper\.net$`)
|
||||
iataCode = regexp.MustCompile(`^[A-Z]{3}$`)
|
||||
)
|
||||
|
||||
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.
|
||||
@@ -61,22 +53,11 @@ func (c *Config) validateMeshMapper() error {
|
||||
if !s.Enabled {
|
||||
return nil
|
||||
}
|
||||
if len(s.Sources) == 0 || len(s.Sources) > MaxScopeSources {
|
||||
return fmt.Errorf("meshmapper.scopes.sources must contain 1-%d regional endpoints", MaxScopeSources)
|
||||
}
|
||||
if s.Interval() < 5*time.Minute || s.Interval() > 24*time.Hour {
|
||||
return fmt.Errorf("meshmapper.scopes.refresh_interval must be between 5m and 24h")
|
||||
}
|
||||
for iata, endpoint := range s.Sources {
|
||||
// Catalogues must name the source IATA exactly, so reject keys that could never match.
|
||||
if !iataCode.MatchString(iata) {
|
||||
return fmt.Errorf("meshmapper.scopes source %q must be a three-letter uppercase IATA code", iata)
|
||||
}
|
||||
u, err := url.Parse(endpoint)
|
||||
if err != nil || u.Scheme != "https" || u.User != nil || !meshMapperHost.MatchString(u.Host) ||
|
||||
u.Path != "/get_scopes.php" || u.RawPath != "" || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" {
|
||||
return fmt.Errorf("meshmapper.scopes source %q must be a published https://<region>.meshmapper.net/get_scopes.php endpoint without credentials or parameters", iata)
|
||||
}
|
||||
if len(s.Sources) > 0 {
|
||||
slog.Warn("meshmapper.scopes.sources is ignored; sources are discovered from the MeshMapper zone list", "component", "config")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -12,22 +12,16 @@ import (
|
||||
)
|
||||
|
||||
func TestMeshMapperConfig(t *testing.T) {
|
||||
valid := "regions:\n - slug: ottawa\n iatas: [YOW]\nmeshmapper:\n scopes:\n enabled: true\n sources:\n YOW: https://yow.meshmapper.net/get_scopes.php\n"
|
||||
valid := "meshmapper:\n scopes:\n enabled: true\n"
|
||||
for _, tc := range []struct {
|
||||
name, text string
|
||||
bad bool
|
||||
}{
|
||||
{"valid", valid, false},
|
||||
{"outside any region", strings.Replace(valid, "iatas: [YOW]", "iatas: [YVR]", 1), false},
|
||||
{"no regions", valid[strings.Index(valid, "meshmapper:"):], false},
|
||||
{"lowercase IATA", strings.Replace(valid, " YOW:", " yow:", 1), true},
|
||||
{"long IATA", strings.Replace(valid, " YOW:", " YOWX:", 1), true},
|
||||
{"http", strings.Replace(valid, "https:", "http:", 1), true},
|
||||
{"private host", strings.Replace(valid, "yow.meshmapper.net", "127.0.0.1", 1), true},
|
||||
{"credentials", strings.Replace(valid, "https://", "https://secret@", 1), true},
|
||||
{"query", strings.Replace(valid, "get_scopes.php", "get_scopes.php?key=secret", 1), true},
|
||||
{"non API", strings.Replace(valid, "get_scopes.php", "index.php", 1), true},
|
||||
{"daily", valid + " refresh_interval: 24h\n", false},
|
||||
{"legacy sources ignored", valid + " sources:\n yow: http://127.0.0.1/\n", false},
|
||||
{"fast", valid + " refresh_interval: 1s\n", true},
|
||||
{"slow", valid + " refresh_interval: 25h\n", true},
|
||||
{"negative", valid + " refresh_interval: -1h\n", true},
|
||||
{"disabled", strings.Replace(valid, "enabled: true", "enabled: false", 1), false},
|
||||
} {
|
||||
@@ -36,18 +30,14 @@ func TestMeshMapperConfig(t *testing.T) {
|
||||
if err := os.WriteFile(p, []byte(tc.text), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg, err := Load(p)
|
||||
_, err := Load(p)
|
||||
if (err != nil) != tc.bad {
|
||||
t.Fatalf("error=%v", err)
|
||||
}
|
||||
if err == nil && cfg.MeshMapper.Scopes.Interval() != time.Hour {
|
||||
t.Fatal("wrong default")
|
||||
}
|
||||
})
|
||||
}
|
||||
cfg := Config{MeshMapper: MeshMapperConfig{Scopes: MeshMapperScopesConfig{Enabled: true}}}
|
||||
if err := cfg.validateMeshMapper(); err == nil {
|
||||
t.Fatal("empty sources accepted")
|
||||
if (MeshMapperScopesConfig{}).Interval() != time.Hour {
|
||||
t.Fatal("wrong default")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package meshmapper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type zoneEntry struct {
|
||||
url string
|
||||
hasBoundary bool
|
||||
}
|
||||
|
||||
type zoneList struct {
|
||||
zones map[string]zoneEntry
|
||||
etag string
|
||||
fetchedAt, nextAttempt time.Time
|
||||
}
|
||||
|
||||
// Directory caches MeshMapper's per-country zone lists, shared by the zones and scopes imports.
|
||||
type Directory struct {
|
||||
mu sync.Mutex
|
||||
client *http.Client
|
||||
listURL string
|
||||
lists map[string]*zoneList
|
||||
retryAfter time.Time
|
||||
}
|
||||
|
||||
func NewDirectory() *Directory {
|
||||
return &Directory{listURL: ZonesURL, lists: map[string]*zoneList{}, client: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
}}
|
||||
}
|
||||
|
||||
// pausedUntil is when a list 429 stops pausing every MeshMapper request.
|
||||
func (d *Directory) pausedUntil() time.Time {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.retryAfter
|
||||
}
|
||||
|
||||
// List returns the country's fresh zone list. fetched reports that this call
|
||||
// spent the caller's one request; nil zones without fetched means back off.
|
||||
func (d *Directory) List(ctx context.Context, country string, now time.Time) (zones map[string]zoneEntry, fetched bool, err error) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
list := d.lists[country]
|
||||
if list != nil && list.zones != nil && now.Sub(list.fetchedAt) < zoneListFresh {
|
||||
return list.zones, false, nil
|
||||
}
|
||||
if now.Before(d.retryAfter) || (list != nil && now.Before(list.nextAttempt)) {
|
||||
return nil, false, nil
|
||||
}
|
||||
return nil, true, d.fetch(ctx, country, now)
|
||||
}
|
||||
|
||||
func (d *Directory) fetch(ctx context.Context, country string, now time.Time) error {
|
||||
list := d.lists[country]
|
||||
if list == nil {
|
||||
list = &zoneList{}
|
||||
d.lists[country] = list
|
||||
}
|
||||
endpoint := d.listURL + "?country=" + url.QueryEscape(country)
|
||||
etag := ""
|
||||
if list.zones != nil {
|
||||
etag = list.etag
|
||||
}
|
||||
status, body, header, err := get(ctx, d.client, "Beacon-MeshMapper-Zones/1", 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:
|
||||
var retryAt time.Time
|
||||
problem = statusProblem(status, header, now, &retryAt)
|
||||
if retryAt.After(list.nextAttempt) {
|
||||
list.nextAttempt = retryAt
|
||||
}
|
||||
if until, ok := rateLimited(status, retryAt, now); ok {
|
||||
d.retryAfter = until
|
||||
}
|
||||
}
|
||||
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 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
|
||||
}
|
||||
@@ -19,11 +19,12 @@ import (
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/iatadb"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
|
||||
)
|
||||
|
||||
const (
|
||||
MaxScopes = 64 // per source; at most 16 configured sources (1024 imported identities)
|
||||
MaxScopes = 64 // per source
|
||||
MaxBody = 64 << 10
|
||||
PollInterval = 15 * time.Second // one source per tick, at most four requests/minute
|
||||
|
||||
@@ -38,16 +39,23 @@ type Cache struct {
|
||||
LastError string
|
||||
}
|
||||
|
||||
// Catalogue is one saved source snapshot.
|
||||
type Catalogue struct {
|
||||
IATA, URL string
|
||||
Cache
|
||||
}
|
||||
|
||||
type Store interface {
|
||||
GetScopeCatalogue(context.Context, string, string) (*Cache, error)
|
||||
ListKnownIATAs(context.Context) ([]string, error)
|
||||
ListScopeCatalogues(context.Context) ([]Catalogue, error)
|
||||
SaveScopeCatalogue(context.Context, string, string, Cache, []scopestore.Entry) error
|
||||
}
|
||||
|
||||
type source struct {
|
||||
iata, url string
|
||||
cache Cache
|
||||
entries []scopestore.Entry
|
||||
generated time.Time
|
||||
iata, country, url string
|
||||
cache Cache
|
||||
entries []scopestore.Entry
|
||||
generated time.Time
|
||||
}
|
||||
|
||||
// Importer is owned by one background task; ScopeStore synchronizes its consumers.
|
||||
@@ -55,6 +63,9 @@ 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
|
||||
@@ -66,44 +77,65 @@ type Importer struct {
|
||||
func (i *Importer) SetCacheInvalidator(fn func(context.Context)) { i.onChange = fn }
|
||||
|
||||
// New restores validated snapshots before ingestion, without making HTTP requests.
|
||||
func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, scopes *scopestore.ScopeStore, manual []scopestore.Entry) (*Importer, error) {
|
||||
i := &Importer{store: store, scopes: scopes, manual: manual, interval: cfg.Interval(), client: &http.Client{
|
||||
Timeout: 10 * time.Second,
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
}}
|
||||
func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, dir *Directory, scopes *scopestore.ScopeStore, manual []scopestore.Entry) (*Importer, error) {
|
||||
i := &Importer{store: store, scopes: scopes, manual: manual, dir: dir, scopesURL: scopesEndpoint, seen: map[string]bool{},
|
||||
interval: cfg.Interval(), client: &http.Client{
|
||||
Timeout: 10 * time.Second,
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
}}
|
||||
if !cfg.Enabled {
|
||||
return i, nil
|
||||
}
|
||||
keys := make([]string, 0, len(cfg.Sources))
|
||||
for key := range cfg.Sources {
|
||||
keys = append(keys, key)
|
||||
saved, err := store.ListScopeCatalogues(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("restore scope catalogues: %w", err)
|
||||
}
|
||||
slices.Sort(keys)
|
||||
for _, key := range keys {
|
||||
s := source{iata: key, url: cfg.Sources[key]}
|
||||
cached, err := store.GetScopeCatalogue(ctx, s.iata, s.url)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("restore scope catalogue %s: %w", key, err)
|
||||
// A site that moved leaves its old row behind; the latest attempt wins.
|
||||
latest := map[string]Catalogue{}
|
||||
for _, c := range saved {
|
||||
if old, ok := latest[c.IATA]; !ok || c.AttemptedAt.After(old.AttemptedAt) {
|
||||
latest[c.IATA] = c
|
||||
}
|
||||
if cached != nil {
|
||||
s.cache = *cached
|
||||
if len(cached.Payload) > 0 {
|
||||
s.entries, s.generated, err = decode(cached.Payload, key)
|
||||
}
|
||||
iatas, err := store.ListKnownIATAs(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list IATAs for MeshMapper scopes: %w", err)
|
||||
}
|
||||
i.track(iatas, latest)
|
||||
i.publish()
|
||||
return i, nil
|
||||
}
|
||||
|
||||
// track adds a source for each IATA not seen before.
|
||||
func (i *Importer) track(iatas []string, saved map[string]Catalogue) {
|
||||
for _, iata := range iatas {
|
||||
if i.seen[iata] {
|
||||
continue
|
||||
}
|
||||
i.seen[iata] = true
|
||||
country := iatadb.CountryFor(iata)
|
||||
if country == "" {
|
||||
continue
|
||||
}
|
||||
s := source{iata: iata, country: country}
|
||||
if c, ok := saved[iata]; ok {
|
||||
s.url, s.cache = c.URL, c.Cache
|
||||
if len(c.Payload) > 0 {
|
||||
var err error
|
||||
s.entries, s.generated, err = decode(c.Payload, iata)
|
||||
if err != nil {
|
||||
s.entries = nil
|
||||
s.cache.Payload, s.cache.ETag = nil, ""
|
||||
s.cache.LastError = "invalid saved catalogue; awaiting refresh"
|
||||
}
|
||||
}
|
||||
if cached.LastError == "HTTP 429" && cached.NextAttempt.After(i.retryAfter) {
|
||||
i.retryAfter = cached.NextAttempt
|
||||
if c.LastError == "HTTP 429" && c.NextAttempt.After(i.retryAfter) {
|
||||
i.retryAfter = c.NextAttempt
|
||||
}
|
||||
i.log(s, "restored")
|
||||
}
|
||||
i.sources = append(i.sources, s)
|
||||
i.log(s, "restored")
|
||||
}
|
||||
i.publish()
|
||||
return i, nil
|
||||
}
|
||||
|
||||
// Refresh checks only one due source. The scheduler serializes calls every 15s.
|
||||
@@ -117,14 +149,43 @@ func (i *Importer) Refresh(ctx context.Context) (err error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
||||
defer cancel()
|
||||
now := time.Now().UTC()
|
||||
if now.Before(i.retryAfter) {
|
||||
if now.Before(i.retryAfter) || now.Before(i.dir.pausedUntil()) {
|
||||
return nil
|
||||
}
|
||||
iatas, err := i.store.ListKnownIATAs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list IATAs for MeshMapper scopes: %w", err)
|
||||
}
|
||||
i.track(iatas, nil)
|
||||
for n := range i.sources {
|
||||
s := &i.sources[n]
|
||||
if now.Before(s.cache.NextAttempt) {
|
||||
continue
|
||||
}
|
||||
zones, fetched, err := i.dir.List(ctx, s.country, now)
|
||||
if err != nil || fetched {
|
||||
return err
|
||||
}
|
||||
if zones == nil {
|
||||
continue
|
||||
}
|
||||
entry, listed := zones[s.iata]
|
||||
endpoint, valid := "", false
|
||||
if listed {
|
||||
endpoint, valid = i.scopesURL(entry.url)
|
||||
}
|
||||
if !valid {
|
||||
// No request to make; names already imported stay until a catalogue replaces them.
|
||||
s.cache.NextAttempt, s.cache.LastError = now.Add(i.interval), "not listed"
|
||||
if listed {
|
||||
s.cache.LastError = "invalid site URL"
|
||||
}
|
||||
i.log(*s, "skipped")
|
||||
continue
|
||||
}
|
||||
if endpoint != s.url {
|
||||
s.url, s.cache.ETag = endpoint, ""
|
||||
}
|
||||
return i.refresh(ctx, s, now)
|
||||
}
|
||||
return nil
|
||||
@@ -263,13 +324,15 @@ func (i *Importer) publish() {
|
||||
|
||||
func (i *Importer) log(s source, action string) {
|
||||
level := slog.LevelInfo
|
||||
if s.cache.LastError != "" {
|
||||
if s.cache.LastError != "" && s.cache.LastError != "not listed" {
|
||||
level = slog.LevelWarn
|
||||
}
|
||||
slog.Log(context.Background(), level, "MeshMapper scopes "+action, "component", "meshmapper.scopes", "iata", s.iata, "source", s.url,
|
||||
"names", len(s.entries), "generated_at", s.generated, "checked_at", s.cache.CheckedAt, "next_attempt", s.cache.NextAttempt, "last_error", s.cache.LastError)
|
||||
}
|
||||
|
||||
func scopesEndpoint(site string) (string, bool) { return siteEndpoint(site, "get_scopes.php") }
|
||||
|
||||
func decode(body []byte, iata string) ([]scopestore.Entry, time.Time, error) {
|
||||
var document struct {
|
||||
GeneratedAt time.Time `json:"generated_at"`
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
@@ -29,24 +30,36 @@ func catalogue(names ...string) []byte {
|
||||
return b
|
||||
}
|
||||
|
||||
type sourceKey struct{ iata, url string }
|
||||
|
||||
type memoryStore struct {
|
||||
rows map[string]Cache
|
||||
iatas []string
|
||||
rows map[sourceKey]Cache
|
||||
fail bool
|
||||
saved []scopestore.Entry
|
||||
}
|
||||
|
||||
func (s *memoryStore) GetScopeCatalogue(_ context.Context, iata, url string) (*Cache, error) {
|
||||
row, ok := s.rows[iata+url]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return &row, nil
|
||||
func newMemoryStore(iatas ...string) *memoryStore {
|
||||
return &memoryStore{iatas: iatas, rows: map[sourceKey]Cache{}}
|
||||
}
|
||||
|
||||
func (s *memoryStore) ListKnownIATAs(context.Context) ([]string, error) {
|
||||
return slices.Clone(s.iatas), nil
|
||||
}
|
||||
|
||||
func (s *memoryStore) ListScopeCatalogues(context.Context) ([]Catalogue, error) {
|
||||
var out []Catalogue
|
||||
for key, c := range s.rows {
|
||||
out = append(out, Catalogue{IATA: key.iata, URL: key.url, Cache: c})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *memoryStore) SaveScopeCatalogue(_ context.Context, iata, url string, next Cache, entries []scopestore.Entry) error {
|
||||
if s.fail {
|
||||
return errors.New("offline")
|
||||
}
|
||||
old := s.rows[iata+url]
|
||||
old := s.rows[sourceKey{iata, url}]
|
||||
if next.Payload == nil {
|
||||
next.Payload = old.Payload
|
||||
}
|
||||
@@ -54,11 +67,25 @@ func (s *memoryStore) SaveScopeCatalogue(_ context.Context, iata, url string, ne
|
||||
next.CheckedAt = old.CheckedAt
|
||||
next.ETag = old.ETag
|
||||
}
|
||||
s.rows[iata+url] = next
|
||||
s.rows[sourceKey{iata, url}] = next
|
||||
s.saved = append(s.saved, entries...)
|
||||
return nil
|
||||
}
|
||||
|
||||
// newImporter restores an importer for YOW whose source points at url.
|
||||
func newImporter(t *testing.T, cfg config.MeshMapperScopesConfig, store *memoryStore, url string, scopes *scopestore.ScopeStore, manual ...scopestore.Entry) *Importer {
|
||||
t.Helper()
|
||||
imp, err := New(context.Background(), cfg, store, NewDirectory(), scopes, manual)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(imp.sources) != 1 {
|
||||
t.Fatal("expected one source", imp.sources)
|
||||
}
|
||||
imp.sources[0].url = url
|
||||
return imp
|
||||
}
|
||||
|
||||
func names(store *scopestore.ScopeStore) []string {
|
||||
out := []string{}
|
||||
for _, e := range store.Entries() {
|
||||
@@ -84,15 +111,12 @@ func TestCatalogueRefreshFallbackAndRestart(t *testing.T) {
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}}
|
||||
store := &memoryStore{rows: map[string]Cache{}}
|
||||
cfg := config.MeshMapperScopesConfig{Enabled: true}
|
||||
store := newMemoryStore("YOW")
|
||||
scopes := scopestore.New()
|
||||
manual := scopestore.FromName("manual")
|
||||
manual.TransportKey = []byte("manual wins")
|
||||
imp, err := New(ctx, cfg, store, scopes, []scopestore.Entry{manual})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
imp := newImporter(t, cfg, store, server.URL, scopes, manual)
|
||||
now := time.Now().UTC()
|
||||
check := func() {
|
||||
t.Helper()
|
||||
@@ -134,10 +158,7 @@ func TestCatalogueRefreshFallbackAndRestart(t *testing.T) {
|
||||
t.Fatal("Retry-After ignored")
|
||||
}
|
||||
restored := scopestore.New()
|
||||
restart, err := New(ctx, cfg, store, restored, []scopestore.Entry{manual})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
restart := newImporter(t, cfg, store, server.URL, restored, manual)
|
||||
if !reflect.DeepEqual(names(restored), names(scopes)) {
|
||||
t.Fatal("outage restart lost cached names")
|
||||
}
|
||||
@@ -215,11 +236,7 @@ func TestNoRedirectAndResponseBounds(t *testing.T) {
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
store := &memoryStore{rows: map[string]Cache{}}
|
||||
imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}}, store, scopestore.New(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
imp := newImporter(t, config.MeshMapperScopesConfig{Enabled: true}, newMemoryStore("YOW"), server.URL, scopestore.New())
|
||||
imp.client.Timeout = 20 * time.Millisecond
|
||||
for _, value := range []string{"redirect", "oversize", "timeout", "304"} {
|
||||
mode.Store(value)
|
||||
@@ -237,25 +254,22 @@ func TestNoRedirectAndResponseBounds(t *testing.T) {
|
||||
|
||||
func TestImportedSourcesOverlapWithoutGlobalMembership(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := &memoryStore{rows: map[string]Cache{}}
|
||||
store := newMemoryStore("YOW", "YVR")
|
||||
for _, iata := range []string{"YOW", "YVR"} {
|
||||
store.rows[iata+iata] = Cache{Payload: []byte(strings.ReplaceAll(string(catalogue("can")), "YOW", iata))}
|
||||
store.rows[sourceKey{iata, "old-" + iata}] = Cache{Payload: []byte(`{}`)}
|
||||
store.rows[sourceKey{iata, iata}] = Cache{Payload: []byte(strings.ReplaceAll(string(catalogue("can")), "YOW", iata)), AttemptedAt: time.Now()}
|
||||
}
|
||||
scopes := scopestore.New()
|
||||
_, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": "YOW", "YVR": "YVR"}}, store, scopes, nil)
|
||||
_, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, NewDirectory(), scopes, nil)
|
||||
if err != nil || len(scopes.Entries()) != 1 || !reflect.DeepEqual(scopes.Entries()[0].IATAs, []string{"YOW", "YVR"}) {
|
||||
t.Fatal(scopes.Entries(), err)
|
||||
t.Fatal("latest saved catalogue per IATA not restored", scopes.Entries(), err)
|
||||
}
|
||||
if got := scopes.NamesForIATAs([]string{"YVR"}); !reflect.DeepEqual(got, []string{"#can"}) {
|
||||
t.Fatal(got)
|
||||
}
|
||||
_, err = New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": "changed-source"}}, store, scopes, nil)
|
||||
if err != nil || len(scopes.Entries()) != 0 || len(scopes.NamesForIATAs([]string{"YOW", "YVR"})) != 0 {
|
||||
t.Fatal("removed/changed source remained active", err)
|
||||
}
|
||||
// Disabled mode never calls the store, even if URLs remain configured.
|
||||
_, err = New(ctx, config.MeshMapperScopesConfig{Sources: map[string]string{"YOW": "YOW"}}, nil, scopes, nil)
|
||||
if err != nil {
|
||||
// Disabled mode never calls the store.
|
||||
disabled := scopestore.New()
|
||||
if _, err = New(ctx, config.MeshMapperScopesConfig{}, nil, NewDirectory(), disabled, nil); err != nil || len(disabled.Entries()) != 0 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -268,19 +282,20 @@ func TestInvalidSavedCatalogueDoesNotBlockStartupOrRefresh(t *testing.T) {
|
||||
_, _ = w.Write(catalogue("yow"))
|
||||
}))
|
||||
defer server.Close()
|
||||
store := &memoryStore{rows: map[string]Cache{"YOW" + server.URL: {Payload: []byte(`{"invalid":true}`), ETag: `"old"`}}}
|
||||
store := newMemoryStore("YOW")
|
||||
store.rows[sourceKey{"YOW", server.URL}] = Cache{Payload: []byte(`{"invalid":true}`), ETag: `"old"`}
|
||||
scopes := scopestore.New()
|
||||
imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}}, store, scopes, []scopestore.Entry{scopestore.FromName("manual")})
|
||||
if err != nil || !reflect.DeepEqual(names(scopes), []string{"#manual"}) {
|
||||
t.Fatal("invalid optional source blocked manual startup", err)
|
||||
imp := newImporter(t, config.MeshMapperScopesConfig{Enabled: true}, store, server.URL, scopes, scopestore.FromName("manual"))
|
||||
if !reflect.DeepEqual(names(scopes), []string{"#manual"}) {
|
||||
t.Fatal("invalid optional source blocked manual startup")
|
||||
}
|
||||
if err = imp.Refresh(context.Background()); err != nil || !reflect.DeepEqual(names(scopes), []string{"#manual", "#yow"}) {
|
||||
if err := imp.refresh(context.Background(), &imp.sources[0], time.Now()); err != nil || !reflect.DeepEqual(names(scopes), []string{"#manual", "#yow"}) {
|
||||
t.Fatal("source did not recover on refresh", err, names(scopes))
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
imp.sources[0].cache.NextAttempt = time.Time{}
|
||||
if err = imp.Refresh(ctx); err != nil {
|
||||
if err := imp.Refresh(ctx); err != nil {
|
||||
t.Fatalf("normal shutdown is a task failure: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -292,11 +307,7 @@ func TestRetryAfterShorterThanIntervalIsHonoured(t *testing.T) {
|
||||
w.WriteHeader(http.StatusTooManyRequests)
|
||||
}))
|
||||
defer server.Close()
|
||||
cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}}
|
||||
imp, err := New(ctx, cfg, &memoryStore{rows: map[string]Cache{}}, scopestore.New(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
imp := newImporter(t, config.MeshMapperScopesConfig{Enabled: true}, newMemoryStore("YOW"), server.URL, scopestore.New())
|
||||
now := time.Now().UTC()
|
||||
if err := imp.refresh(ctx, &imp.sources[0], now); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -310,11 +321,7 @@ func TestFailedRefreshRetriesSoon(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(500) }))
|
||||
defer server.Close()
|
||||
cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}}
|
||||
imp, err := New(ctx, cfg, &memoryStore{rows: map[string]Cache{}}, scopestore.New(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
imp := newImporter(t, config.MeshMapperScopesConfig{Enabled: true}, newMemoryStore("YOW"), server.URL, scopestore.New())
|
||||
now := time.Now().UTC()
|
||||
if err := imp.refresh(ctx, &imp.sources[0], now); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -323,3 +330,80 @@ func TestFailedRefreshRetriesSoon(t *testing.T) {
|
||||
t.Fatalf("next attempt after failure in %v, want %v", got, failureRetry)
|
||||
}
|
||||
}
|
||||
|
||||
// fakeSites serves a CA zone list and YOW's scope catalogue.
|
||||
type fakeSites struct {
|
||||
*httptest.Server
|
||||
list string
|
||||
listCalls, scopeHits int
|
||||
}
|
||||
|
||||
func newFakeSites(t *testing.T) *fakeSites {
|
||||
f := &fakeSites{}
|
||||
f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/get_zones.php":
|
||||
f.listCalls++
|
||||
_, _ = w.Write([]byte(f.list))
|
||||
case "/get_scopes.php":
|
||||
f.scopeHits++
|
||||
_, _ = w.Write(catalogue("yow"))
|
||||
default:
|
||||
t.Errorf("unexpected path %s", r.URL.Path)
|
||||
}
|
||||
}))
|
||||
f.list = `{"country":"CA","zones":[{"code":"YOW","url":"` + f.URL + `/","has_boundary":true,"group":null}]}`
|
||||
t.Cleanup(f.Close)
|
||||
return f
|
||||
}
|
||||
|
||||
func newDiscoveringImporter(t *testing.T, f *fakeSites, store *memoryStore, scopes *scopestore.ScopeStore) *Importer {
|
||||
t.Helper()
|
||||
dir := NewDirectory()
|
||||
dir.listURL = f.URL + "/get_zones.php"
|
||||
imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopes, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
imp.scopesURL = func(site string) (string, bool) { return site + "get_scopes.php", strings.HasPrefix(site, f.URL) }
|
||||
return imp
|
||||
}
|
||||
|
||||
func refreshN(t *testing.T, imp *Importer, n int) {
|
||||
t.Helper()
|
||||
for range n {
|
||||
if err := imp.Refresh(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestScopesDiscoverSourcesFromZoneList(t *testing.T) {
|
||||
f := newFakeSites(t)
|
||||
store := newMemoryStore()
|
||||
scopes := scopestore.New()
|
||||
imp := newDiscoveringImporter(t, f, store, scopes)
|
||||
refreshN(t, imp, 2)
|
||||
if f.listCalls != 0 || f.scopeHits != 0 {
|
||||
t.Fatal("requested with no IATAs", f.listCalls, f.scopeHits)
|
||||
}
|
||||
store.iatas = []string{"YOW", "ZZZ"}
|
||||
refreshN(t, imp, 3)
|
||||
if f.listCalls != 1 || f.scopeHits != 1 || !reflect.DeepEqual(names(scopes), []string{"#yow"}) {
|
||||
t.Fatal("source not discovered", f.listCalls, f.scopeHits, names(scopes))
|
||||
}
|
||||
if _, ok := store.rows[sourceKey{"YOW", f.URL + "/get_scopes.php"}]; !ok {
|
||||
t.Fatal("catalogue not saved under the discovered URL", store.rows)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScopesSkipUnlistedIATAs(t *testing.T) {
|
||||
f := newFakeSites(t)
|
||||
f.list = `{"country":"CA","zones":[]}`
|
||||
imp := newDiscoveringImporter(t, f, newMemoryStore("YOW"), scopestore.New())
|
||||
refreshN(t, imp, 3)
|
||||
s := imp.sources[0]
|
||||
if f.scopeHits != 0 || s.cache.LastError != "not listed" || s.cache.NextAttempt.Sub(time.Now()) < 59*time.Minute {
|
||||
t.Fatalf("unlisted IATA requested or retried early: hits=%d %+v", f.scopeHits, s.cache)
|
||||
}
|
||||
}
|
||||
+43
-113
@@ -42,7 +42,7 @@ type Boundary struct {
|
||||
}
|
||||
|
||||
type ZoneStore interface {
|
||||
ListZoneIATAs(ctx context.Context) ([]string, error)
|
||||
ListKnownIATAs(ctx context.Context) ([]string, error)
|
||||
PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error)
|
||||
ListZoneBoundaries(ctx context.Context) ([]Boundary, error)
|
||||
SaveZoneBoundary(ctx context.Context, b Boundary) error
|
||||
@@ -53,17 +53,6 @@ type zoneRegion struct {
|
||||
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
|
||||
@@ -71,18 +60,17 @@ type Zones struct {
|
||||
seen map[string]bool
|
||||
interval time.Duration
|
||||
client *http.Client
|
||||
listURL string
|
||||
dir *Directory
|
||||
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, store ZoneStore) *Zones {
|
||||
return &Zones{store: store, enabled: cfg.Enabled, seen: map[string]bool{}, interval: cfg.Interval(), listURL: ZonesURL,
|
||||
boundsURL: boundaryEndpoint, lists: map[string]*zoneList{}, client: &http.Client{
|
||||
func NewZones(cfg config.MeshMapperZonesConfig, store ZoneStore, dir *Directory) *Zones {
|
||||
return &Zones{store: store, enabled: cfg.Enabled, seen: map[string]bool{}, interval: cfg.Interval(), dir: dir,
|
||||
boundsURL: boundaryEndpoint, client: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
}}
|
||||
@@ -97,7 +85,7 @@ func (z *Zones) OnUpdate(fn func(imported map[string]json.RawMessage)) { z.onUpd
|
||||
// Restore prunes imports for unknown IATAs (all of them when disabled) and
|
||||
// loads saved boundaries, without making HTTP requests.
|
||||
func (z *Zones) Restore(ctx context.Context) error {
|
||||
iatas, err := z.store.ListZoneIATAs(ctx)
|
||||
iatas, err := z.store.ListKnownIATAs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list IATAs for MeshMapper boundaries: %w", err)
|
||||
}
|
||||
@@ -166,11 +154,11 @@ func (z *Zones) Refresh(ctx context.Context) (err error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, time.Minute)
|
||||
defer cancel()
|
||||
now := time.Now().UTC()
|
||||
if now.Before(z.retryAfter) {
|
||||
if now.Before(z.retryAfter) || now.Before(z.dir.pausedUntil()) {
|
||||
return nil
|
||||
}
|
||||
// IATAs are created from traffic, so pick up new ones every tick.
|
||||
iatas, err := z.store.ListZoneIATAs(ctx)
|
||||
iatas, err := z.store.ListKnownIATAs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list IATAs for MeshMapper boundaries: %w", err)
|
||||
}
|
||||
@@ -179,67 +167,22 @@ func (z *Zones) Refresh(ctx context.Context) (err error) {
|
||||
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)
|
||||
zones, fetched, err := z.dir.List(ctx, r.country, now)
|
||||
if err != nil || fetched {
|
||||
return err
|
||||
}
|
||||
return z.refresh(ctx, r, list, now)
|
||||
if zones == nil {
|
||||
continue
|
||||
}
|
||||
return z.refresh(ctx, r, zones, 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 {
|
||||
func (z *Zones) refresh(ctx context.Context, r *zoneRegion, zones map[string]zoneEntry, 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]
|
||||
entry, listed := zones[r.b.IATA]
|
||||
endpoint, valid := "", false
|
||||
if listed {
|
||||
update.URL = entry.url
|
||||
@@ -257,7 +200,7 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, list *zoneList, now
|
||||
if r.b.Feature != nil {
|
||||
etag = r.b.ETag
|
||||
}
|
||||
status, body, header, err := z.get(ctx, endpoint, etag, MaxBoundaryBody)
|
||||
status, body, header, err := get(ctx, z.client, "Beacon-MeshMapper-Zones/1", endpoint, etag, MaxBoundaryBody)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -277,7 +220,10 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, list *zoneList, now
|
||||
update.LastError = "304 without cached boundary"
|
||||
}
|
||||
default:
|
||||
update.LastError = z.statusProblem(status, header, now, &retryAt)
|
||||
update.LastError = statusProblem(status, header, now, &retryAt)
|
||||
if until, ok := rateLimited(status, retryAt, now); ok {
|
||||
z.retryAfter = until // the rate limit is shared by every endpoint
|
||||
}
|
||||
}
|
||||
if update.LastError == "" {
|
||||
etag := header.Get("ETag")
|
||||
@@ -318,17 +264,17 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, list *zoneList, now
|
||||
}
|
||||
|
||||
// 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) {
|
||||
func get(ctx context.Context, client *http.Client, agent, 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")
|
||||
request.Header.Set("User-Agent", agent)
|
||||
if etag != "" {
|
||||
request.Header.Set("If-None-Match", etag)
|
||||
}
|
||||
response, err := z.client.Do(request)
|
||||
response, err := client.Do(request)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return 0, nil, nil, ctx.Err()
|
||||
@@ -346,8 +292,8 @@ func (z *Zones) get(ctx context.Context, endpoint, etag string, limit int64) (in
|
||||
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 {
|
||||
// statusProblem describes a failed response and reads Retry-After on 429/503.
|
||||
func 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
|
||||
}
|
||||
@@ -355,17 +301,21 @@ func (z *Zones) statusProblem(status int, header http.Header, now time.Time, ret
|
||||
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)
|
||||
}
|
||||
|
||||
// rateLimited returns how long a 429 pauses every request.
|
||||
func rateLimited(status int, retryAt, now time.Time) (time.Time, bool) {
|
||||
if status != http.StatusTooManyRequests {
|
||||
return time.Time{}, false
|
||||
}
|
||||
if retryAt.IsZero() {
|
||||
return now.Add(time.Minute), true
|
||||
}
|
||||
return retryAt, true
|
||||
}
|
||||
|
||||
func (z *Zones) changed(ctx context.Context, iata string) {
|
||||
if z.onChange != nil {
|
||||
z.onChange(ctx, iata)
|
||||
@@ -394,36 +344,16 @@ func (z *Zones) log(b Boundary, action string) {
|
||||
"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) {
|
||||
func boundaryEndpoint(site string) (string, bool) { return siteEndpoint(site, "get_geojson.php") }
|
||||
|
||||
// siteEndpoint accepts only a published region site root.
|
||||
func siteEndpoint(site, file 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
|
||||
return "https://" + u.Host + "/" + file, true
|
||||
}
|
||||
|
||||
// decodeBoundary returns nil when MeshMapper has no usable outline for the region.
|
||||
|
||||
@@ -31,7 +31,7 @@ type zoneMemoryStore struct {
|
||||
fail bool
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) ListZoneIATAs(context.Context) ([]string, error) {
|
||||
func (s *zoneMemoryStore) ListKnownIATAs(context.Context) ([]string, error) {
|
||||
return slices.Clone(s.iatas), nil
|
||||
}
|
||||
|
||||
@@ -121,8 +121,9 @@ func newZoneHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, ena
|
||||
if store.iatas == nil {
|
||||
store.iatas = []string{"YOW"}
|
||||
}
|
||||
h.z = NewZones(config.MeshMapperZonesConfig{Enabled: enabled}, store)
|
||||
h.z.listURL = f.URL + "/get_zones.php"
|
||||
dir := NewDirectory()
|
||||
dir.listURL = f.URL + "/get_zones.php"
|
||||
h.z = NewZones(config.MeshMapperZonesConfig{Enabled: enabled}, store, dir)
|
||||
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 })
|
||||
@@ -239,9 +240,9 @@ func TestZonesListFailureBacksOff(t *testing.T) {
|
||||
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.dir.lists["CA"].nextAttempt, h.z.dir.lists["CA"].fetchedAt = time.Time{}, time.Time{}
|
||||
_ = h.z.Refresh(context.Background())
|
||||
if h.z.lists["CA"].zones != nil {
|
||||
if h.z.dir.lists["CA"].zones != nil {
|
||||
t.Fatal("list for another country accepted")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user