mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-16 09:29:52 +00:00
fix(meshmapper): respect MeshMapper's per-endpoint rate limits
This commit is contained in:
+10
-1
@@ -198,7 +198,16 @@ func main() {
|
||||
scopes.Load(scopeEntries)
|
||||
scopes.SetManualMembers(cfg.ManualScopeMembers())
|
||||
slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup")
|
||||
directory := meshmapper.NewDirectory()
|
||||
directory := meshmapper.NewDirectory(store)
|
||||
if cfg.MeshMapper.Scopes.Enabled || cfg.MeshMapper.Zones.Enabled {
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
err = directory.Restore(restoreCtx)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper zone lists", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
var scopeImporter *meshmapper.Importer
|
||||
if cfg.MeshMapper.Scopes.Enabled {
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
|
||||
+5
-2
@@ -61,7 +61,7 @@ ratelimit:
|
||||
#meshmapper:
|
||||
# scopes:
|
||||
# enabled: false
|
||||
# refresh_interval: 1h # 5m-24h; one source checked per 15s tick
|
||||
# refresh_interval: 1h # 1h-24h (MeshMapper allows one get_scopes.php per region per 55m); one source per 15s tick
|
||||
#
|
||||
# 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
|
||||
@@ -70,6 +70,9 @@ ratelimit:
|
||||
# A missing or failed boundary keeps the last good import, or the borderFile when
|
||||
# there is none. Disabling drops every import on restart. Logged under
|
||||
# component=meshmapper.zones.
|
||||
# Each country's zone list is fetched at most once a day (MeshMapper allows one
|
||||
# get_zones.php per country per 23.5h) and saved, so restarts don't refetch it.
|
||||
# Failed or rate-limited requests also wait out the limit, or Retry-After if longer.
|
||||
# import_groups turns MeshMapper zone groups (e.g. ONQC: YOW, YQB, YUL) into
|
||||
# regions, from the zone lists already fetched. A group is imported when at least
|
||||
# one member is a known IATA, with its full member list; slug = lowercased group
|
||||
@@ -79,7 +82,7 @@ ratelimit:
|
||||
# zones:
|
||||
# enabled: false
|
||||
# import_groups: false
|
||||
# refresh_interval: 24h # 1h-168h; one request per 15s tick, ETag-conditional
|
||||
# refresh_interval: 24h # 24h-168h (one get_geojson.php per region per 23.5h); one request per 15s tick, ETag-conditional
|
||||
|
||||
iatas:
|
||||
YVR:
|
||||
|
||||
@@ -114,6 +114,36 @@ func (s *Store) SaveZoneBoundary(ctx context.Context, b meshmapper.Boundary) err
|
||||
return s.q.SaveZoneBoundary(ctx, params)
|
||||
}
|
||||
|
||||
func (s *Store) ListZoneLists(ctx context.Context) ([]meshmapper.ZoneList, error) {
|
||||
rows, err := s.q.ListZoneLists(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]meshmapper.ZoneList, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
l := meshmapper.ZoneList{Country: row.Country, Payload: row.Payload, FetchedAt: row.FetchedAt.Time,
|
||||
AttemptedAt: row.AttemptedAt.Time, NextAttempt: row.NextAttempt.Time, LastError: row.LastError}
|
||||
if row.Etag != nil {
|
||||
l.ETag = *row.Etag
|
||||
}
|
||||
out = append(out, l)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *Store) SaveZoneList(ctx context.Context, l meshmapper.ZoneList) error {
|
||||
params := sqlc.SaveZoneListParams{
|
||||
Country: l.Country, Payload: l.Payload, LastError: l.LastError,
|
||||
FetchedAt: pgtype.Timestamptz{Time: l.FetchedAt, Valid: !l.FetchedAt.IsZero()},
|
||||
AttemptedAt: pgtype.Timestamptz{Time: l.AttemptedAt, Valid: true},
|
||||
NextAttempt: pgtype.Timestamptz{Time: l.NextAttempt, Valid: true},
|
||||
}
|
||||
if !l.FetchedAt.IsZero() {
|
||||
params.Etag = &l.ETag
|
||||
}
|
||||
return s.q.SaveZoneList(ctx, params)
|
||||
}
|
||||
|
||||
func (s *Store) ListRegionState(ctx context.Context) ([]meshmapper.RegionState, error) {
|
||||
rows, err := s.q.ListRegionState(ctx)
|
||||
if err != nil {
|
||||
|
||||
@@ -82,7 +82,7 @@ func TestMeshMapperCataloguePostgres(t *testing.T) {
|
||||
t.Fatal(row, err)
|
||||
}
|
||||
scopes := scopestore.New()
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(), scopes, manualRows)
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(store), scopes, manualRows)
|
||||
if err != nil || len(scopes.Entries()) != 2 {
|
||||
t.Fatal("restart did not restore imported membership", err)
|
||||
}
|
||||
@@ -105,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}, store, meshmapper.NewDirectory(), scopes, manualRows)
|
||||
_, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(store), scopes, manualRows)
|
||||
if err != nil || len(scopes.Entries()) != 1 {
|
||||
t.Fatal("removed imported membership remained active", err)
|
||||
}
|
||||
@@ -168,7 +168,7 @@ func TestMeshMapperZoneBoundariesPostgres(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
for _, table := range []string{"iata_codes", "meshmapper_zone_boundaries"} {
|
||||
for _, table := range []string{"iata_codes", "meshmapper_zone_boundaries", "meshmapper_zone_lists"} {
|
||||
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -225,6 +225,20 @@ func TestMeshMapperZoneBoundariesPostgres(t *testing.T) {
|
||||
if !strings.Contains(border("YOW"), `"file"`) {
|
||||
t.Fatal("manual border did not return after pruning")
|
||||
}
|
||||
list := meshmapper.ZoneList{Country: "CA", Payload: []byte(`{"country":"CA","zones":[]}`), ETag: `"z1"`,
|
||||
FetchedAt: now, AttemptedAt: now, NextAttempt: now.Add(24 * time.Hour)}
|
||||
if err := store.SaveZoneList(ctx, list); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
later := now.Add(25 * time.Hour)
|
||||
if err := store.SaveZoneList(ctx, meshmapper.ZoneList{Country: "CA", AttemptedAt: later, NextAttempt: later.Add(48 * time.Hour), LastError: "HTTP 429"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lists, err := store.ListZoneLists(ctx)
|
||||
if err != nil || len(lists) != 1 || len(lists[0].Payload) == 0 || lists[0].ETag != `"z1"` || !lists[0].FetchedAt.Equal(now) ||
|
||||
lists[0].LastError != "HTTP 429" || !lists[0].NextAttempt.Equal(later.Add(48*time.Hour)) {
|
||||
t.Fatal("failed fetch lost the last good zone list", lists, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMeshMapperRegionsPostgres(t *testing.T) {
|
||||
|
||||
@@ -665,6 +665,16 @@ CREATE TABLE meshmapper_zone_boundaries (
|
||||
last_error text DEFAULT ''::text NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE meshmapper_zone_lists (
|
||||
country text NOT NULL,
|
||||
payload jsonb,
|
||||
etag text,
|
||||
fetched_at timestamp with time zone,
|
||||
attempted_at timestamp with time zone NOT NULL,
|
||||
next_attempt timestamp with time zone NOT NULL,
|
||||
last_error text DEFAULT ''::text NOT NULL
|
||||
);
|
||||
|
||||
CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS
|
||||
WITH combined AS (
|
||||
SELECT iata, hour, observation_count, unique_packets FROM analytics_live_hourly_iata_stats
|
||||
@@ -1023,6 +1033,9 @@ ALTER TABLE ONLY meshmapper_scope_catalogues
|
||||
ALTER TABLE ONLY meshmapper_zone_boundaries
|
||||
ADD CONSTRAINT meshmapper_zone_boundaries_pkey PRIMARY KEY (iata);
|
||||
|
||||
ALTER TABLE ONLY meshmapper_zone_lists
|
||||
ADD CONSTRAINT meshmapper_zone_lists_pkey PRIMARY KEY (country);
|
||||
|
||||
ALTER TABLE ONLY node_iatas
|
||||
ADD CONSTRAINT node_iatas_pkey PRIMARY KEY (node_id, iata);
|
||||
|
||||
|
||||
@@ -1645,6 +1645,22 @@ ON CONFLICT (iata) DO UPDATE SET
|
||||
next_attempt = EXCLUDED.next_attempt,
|
||||
last_error = EXCLUDED.last_error;
|
||||
|
||||
-- name: ListZoneLists :many
|
||||
SELECT * FROM meshmapper_zone_lists ORDER BY country;
|
||||
|
||||
-- name: SaveZoneList :exec
|
||||
-- NULL payload/etag/fetched_at retain the last good list after an error or 304.
|
||||
INSERT INTO meshmapper_zone_lists (country, payload, etag, fetched_at, attempted_at, next_attempt, last_error)
|
||||
VALUES (@country, sqlc.narg(payload)::jsonb, sqlc.narg(etag)::text,
|
||||
sqlc.narg(fetched_at)::timestamptz, @attempted_at, @next_attempt, @last_error)
|
||||
ON CONFLICT (country) DO UPDATE SET
|
||||
payload = COALESCE(EXCLUDED.payload, meshmapper_zone_lists.payload),
|
||||
etag = COALESCE(EXCLUDED.etag, meshmapper_zone_lists.etag),
|
||||
fetched_at = COALESCE(EXCLUDED.fetched_at, meshmapper_zone_lists.fetched_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;
|
||||
|
||||
@@ -1357,6 +1357,21 @@ func (mr *MockQuerierMockRecorder) ListZoneBoundaries(ctx any) *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListZoneBoundaries", reflect.TypeOf((*MockQuerier)(nil).ListZoneBoundaries), ctx)
|
||||
}
|
||||
|
||||
// ListZoneLists mocks base method.
|
||||
func (m *MockQuerier) ListZoneLists(ctx context.Context) ([]db.MeshmapperZoneList, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "ListZoneLists", ctx)
|
||||
ret0, _ := ret[0].([]db.MeshmapperZoneList)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// ListZoneLists indicates an expected call of ListZoneLists.
|
||||
func (mr *MockQuerierMockRecorder) ListZoneLists(ctx any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListZoneLists", reflect.TypeOf((*MockQuerier)(nil).ListZoneLists), ctx)
|
||||
}
|
||||
|
||||
// PruneImportedRegions mocks base method.
|
||||
func (m *MockQuerier) PruneImportedRegions(ctx context.Context, keep []string) ([]string, error) {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1674,6 +1689,20 @@ func (mr *MockQuerierMockRecorder) SaveZoneBoundary(ctx, arg any) *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveZoneBoundary", reflect.TypeOf((*MockQuerier)(nil).SaveZoneBoundary), ctx, arg)
|
||||
}
|
||||
|
||||
// SaveZoneList mocks base method.
|
||||
func (m *MockQuerier) SaveZoneList(ctx context.Context, arg db.SaveZoneListParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "SaveZoneList", ctx, arg)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// SaveZoneList indicates an expected call of SaveZoneList.
|
||||
func (mr *MockQuerierMockRecorder) SaveZoneList(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveZoneList", reflect.TypeOf((*MockQuerier)(nil).SaveZoneList), ctx, arg)
|
||||
}
|
||||
|
||||
// SearchKnownRoutes mocks base method.
|
||||
func (m *MockQuerier) SearchKnownRoutes(ctx context.Context, arg db.SearchKnownRoutesParams) ([]db.SearchKnownRoutesRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@@ -256,6 +256,16 @@ type MeshmapperZoneBoundary struct {
|
||||
LastError string `json:"last_error"`
|
||||
}
|
||||
|
||||
type MeshmapperZoneList struct {
|
||||
Country string `json:"country"`
|
||||
Payload []byte `json:"payload"`
|
||||
Etag *string `json:"etag"`
|
||||
FetchedAt pgtype.Timestamptz `json:"fetched_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"`
|
||||
|
||||
@@ -225,6 +225,7 @@ type Querier interface {
|
||||
// internal/ingest.BackfillChannelMessages.
|
||||
ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error)
|
||||
ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBoundary, error)
|
||||
ListZoneLists(ctx context.Context) ([]MeshmapperZoneList, error)
|
||||
PruneImportedRegions(ctx context.Context, keep []string) ([]string, error)
|
||||
// Drops imports for IATAs no longer configured, so their manual border returns.
|
||||
PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error)
|
||||
@@ -270,6 +271,8 @@ type Querier interface {
|
||||
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
|
||||
// NULL payload/etag/fetched_at retain the last good list after an error or 304.
|
||||
SaveZoneList(ctx context.Context, arg SaveZoneListParams) 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)
|
||||
|
||||
@@ -4000,6 +4000,38 @@ func (q *Queries) ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBound
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listZoneLists = `-- name: ListZoneLists :many
|
||||
SELECT country, payload, etag, fetched_at, attempted_at, next_attempt, last_error FROM meshmapper_zone_lists ORDER BY country
|
||||
`
|
||||
|
||||
func (q *Queries) ListZoneLists(ctx context.Context) ([]MeshmapperZoneList, error) {
|
||||
rows, err := q.db.Query(ctx, listZoneLists)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []MeshmapperZoneList{}
|
||||
for rows.Next() {
|
||||
var i MeshmapperZoneList
|
||||
if err := rows.Scan(
|
||||
&i.Country,
|
||||
&i.Payload,
|
||||
&i.Etag,
|
||||
&i.FetchedAt,
|
||||
&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 pruneImportedRegions = `-- name: PruneImportedRegions :many
|
||||
DELETE FROM regions WHERE imported AND NOT (slug = ANY($1::text[])) RETURNING slug
|
||||
`
|
||||
@@ -4636,6 +4668,43 @@ func (q *Queries) SaveZoneBoundary(ctx context.Context, arg SaveZoneBoundaryPara
|
||||
return err
|
||||
}
|
||||
|
||||
const saveZoneList = `-- name: SaveZoneList :exec
|
||||
INSERT INTO meshmapper_zone_lists (country, payload, etag, fetched_at, attempted_at, next_attempt, last_error)
|
||||
VALUES ($1, $2::jsonb, $3::text,
|
||||
$4::timestamptz, $5, $6, $7)
|
||||
ON CONFLICT (country) DO UPDATE SET
|
||||
payload = COALESCE(EXCLUDED.payload, meshmapper_zone_lists.payload),
|
||||
etag = COALESCE(EXCLUDED.etag, meshmapper_zone_lists.etag),
|
||||
fetched_at = COALESCE(EXCLUDED.fetched_at, meshmapper_zone_lists.fetched_at),
|
||||
attempted_at = EXCLUDED.attempted_at,
|
||||
next_attempt = EXCLUDED.next_attempt,
|
||||
last_error = EXCLUDED.last_error
|
||||
`
|
||||
|
||||
type SaveZoneListParams struct {
|
||||
Country string `json:"country"`
|
||||
Payload []byte `json:"payload"`
|
||||
Etag *string `json:"etag"`
|
||||
FetchedAt pgtype.Timestamptz `json:"fetched_at"`
|
||||
AttemptedAt pgtype.Timestamptz `json:"attempted_at"`
|
||||
NextAttempt pgtype.Timestamptz `json:"next_attempt"`
|
||||
LastError string `json:"last_error"`
|
||||
}
|
||||
|
||||
// NULL payload/etag/fetched_at retain the last good list after an error or 304.
|
||||
func (q *Queries) SaveZoneList(ctx context.Context, arg SaveZoneListParams) error {
|
||||
_, err := q.db.Exec(ctx, saveZoneList,
|
||||
arg.Country,
|
||||
arg.Payload,
|
||||
arg.Etag,
|
||||
arg.FetchedAt,
|
||||
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
|
||||
|
||||
@@ -9,6 +9,13 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// MeshMapper rate-limits get_zones/get_geojson to one request per target per
|
||||
// 23.5h and get_scopes to one per region per 55m; minimums round up.
|
||||
const (
|
||||
MinZonesRefresh = 24 * time.Hour
|
||||
MinScopesRefresh = time.Hour
|
||||
)
|
||||
|
||||
type MeshMapperConfig struct {
|
||||
Scopes MeshMapperScopesConfig `yaml:"scopes"`
|
||||
Zones MeshMapperZonesConfig `yaml:"zones"`
|
||||
@@ -48,17 +55,16 @@ func (c *Config) validateMeshMapper() error {
|
||||
return fmt.Errorf("meshmapper.zones.import_groups requires meshmapper.zones.enabled")
|
||||
}
|
||||
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 z.Interval() < MinZonesRefresh || z.Interval() > 7*24*time.Hour {
|
||||
return fmt.Errorf("meshmapper.zones.refresh_interval must be between 24h and 168h")
|
||||
}
|
||||
}
|
||||
s := c.MeshMapper.Scopes
|
||||
if !s.Enabled {
|
||||
return nil
|
||||
}
|
||||
if s.Interval() < 5*time.Minute || s.Interval() > 24*time.Hour {
|
||||
return fmt.Errorf("meshmapper.scopes.refresh_interval must be between 5m and 24h")
|
||||
if s.Interval() < MinScopesRefresh || s.Interval() > 24*time.Hour {
|
||||
return fmt.Errorf("meshmapper.scopes.refresh_interval must be between 1h and 24h")
|
||||
}
|
||||
if len(s.Sources) > 0 {
|
||||
slog.Warn("meshmapper.scopes.sources is ignored; sources are discovered from the MeshMapper zone list", "component", "config")
|
||||
|
||||
@@ -20,6 +20,8 @@ func TestMeshMapperConfig(t *testing.T) {
|
||||
{"valid", valid, false},
|
||||
{"daily", valid + " refresh_interval: 24h\n", false},
|
||||
{"legacy sources ignored", valid + " sources:\n yow: http://127.0.0.1/\n", false},
|
||||
{"hourly", valid + " refresh_interval: 1h\n", false},
|
||||
{"under the rate limit", valid + " refresh_interval: 55m\n", true},
|
||||
{"fast", valid + " refresh_interval: 1s\n", true},
|
||||
{"slow", valid + " refresh_interval: 25h\n", true},
|
||||
{"negative", valid + " refresh_interval: -1h\n", true},
|
||||
@@ -49,8 +51,10 @@ func TestMeshMapperZonesConfig(t *testing.T) {
|
||||
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},
|
||||
{"daily", valid + " refresh_interval: 24h\n", false, 24 * time.Hour},
|
||||
{"weekly", valid + " refresh_interval: 168h\n", false, 168 * time.Hour},
|
||||
{"under the rate limit", valid + " refresh_interval: 23h30m\n", true, 0},
|
||||
{"hourly", valid + " refresh_interval: 1h\n", true, 0},
|
||||
{"too rare", valid + " refresh_interval: 169h\n", true, 0},
|
||||
{"no regions", "meshmapper:\n zones:\n enabled: true\n", false, 24 * time.Hour},
|
||||
{"groups", valid + " import_groups: true\n", false, 24 * time.Hour},
|
||||
|
||||
@@ -35,28 +35,54 @@ type zoneList struct {
|
||||
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
|
||||
version int // bumped whenever a list is replaced
|
||||
// ZoneList is one country's saved get_zones.php response and fetch state.
|
||||
type ZoneList struct {
|
||||
Country string
|
||||
Payload json.RawMessage
|
||||
ETag string
|
||||
FetchedAt, AttemptedAt, NextAttempt time.Time
|
||||
LastError string
|
||||
}
|
||||
|
||||
func NewDirectory() *Directory {
|
||||
return &Directory{listURL: ZonesURL, lists: map[string]*zoneList{}, client: &http.Client{
|
||||
type DirectoryStore interface {
|
||||
ListZoneLists(ctx context.Context) ([]ZoneList, error)
|
||||
SaveZoneList(ctx context.Context, l ZoneList) error
|
||||
}
|
||||
|
||||
// Directory caches MeshMapper's per-country zone lists, shared by the zones and scopes imports.
|
||||
type Directory struct {
|
||||
mu sync.Mutex
|
||||
store DirectoryStore
|
||||
client *http.Client
|
||||
listURL string
|
||||
lists map[string]*zoneList
|
||||
version int // bumped whenever a list is replaced
|
||||
}
|
||||
|
||||
func NewDirectory(store DirectoryStore) *Directory {
|
||||
return &Directory{store: store, 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 {
|
||||
// Restore loads saved lists so a restart doesn't spend each country's daily request.
|
||||
func (d *Directory) Restore(ctx context.Context) error {
|
||||
saved, err := d.store.ListZoneLists(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("restore MeshMapper zone lists: %w", err)
|
||||
}
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.retryAfter
|
||||
for _, l := range saved {
|
||||
list := &zoneList{nextAttempt: l.NextAttempt}
|
||||
if zones, groups, err := decodeZones(l.Payload, l.Country); err == nil {
|
||||
list.zones, list.groups, list.etag, list.fetchedAt = zones, groups, l.ETag, l.FetchedAt
|
||||
d.version++
|
||||
}
|
||||
d.lists[l.Country] = list
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// List returns the country's fresh zone list. fetched reports that this call
|
||||
@@ -68,7 +94,7 @@ func (d *Directory) List(ctx context.Context, country string, now time.Time) (zo
|
||||
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)) {
|
||||
if list != nil && now.Before(list.nextAttempt) {
|
||||
return nil, false, nil
|
||||
}
|
||||
return nil, true, d.fetch(ctx, country, now)
|
||||
@@ -90,6 +116,8 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er
|
||||
return err
|
||||
}
|
||||
problem := ""
|
||||
var payload json.RawMessage
|
||||
var retryAt time.Time
|
||||
switch status {
|
||||
case http.StatusOK:
|
||||
zones, groups, decodeErr := decodeZones(body, country)
|
||||
@@ -97,6 +125,7 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er
|
||||
problem = "invalid response"
|
||||
} else {
|
||||
list.zones, list.groups, list.etag = zones, groups, header.Get("ETag")
|
||||
payload = body
|
||||
d.version++
|
||||
}
|
||||
case http.StatusNotModified:
|
||||
@@ -104,26 +133,27 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er
|
||||
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
|
||||
}
|
||||
}
|
||||
// Every request counts against the country's 23.5h allowance, failed or not.
|
||||
list.nextAttempt = now.Add(zoneListFresh)
|
||||
if retryAt.After(list.nextAttempt) {
|
||||
list.nextAttempt = retryAt
|
||||
}
|
||||
saved := ZoneList{Country: country, Payload: payload, AttemptedAt: now, NextAttempt: list.nextAttempt, LastError: problem}
|
||||
if problem == "" {
|
||||
list.fetchedAt = now
|
||||
} else if list.nextAttempt.Before(now.Add(zoneFailureRetry)) {
|
||||
list.nextAttempt = now.Add(zoneFailureRetry)
|
||||
saved.FetchedAt, saved.ETag = now, list.etag
|
||||
}
|
||||
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)
|
||||
"zones", len(list.zones), "groups", len(list.groups), "next_attempt", list.nextAttempt, "last_error", problem)
|
||||
if err := d.store.SaveZoneList(ctx, saved); err != nil {
|
||||
return fmt.Errorf("persist MeshMapper zone list %s: %w", country, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ func newGroupHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, im
|
||||
store.rows = map[string]Boundary{}
|
||||
}
|
||||
h := &groupHarness{f: f, store: store}
|
||||
dir := NewDirectory()
|
||||
dir := NewDirectory(newZoneListMemory())
|
||||
dir.listURL = f.URL + "/get_zones.php"
|
||||
h.z = NewZones(config.MeshMapperZonesConfig{Enabled: true, ImportGroups: importGroups}, store, dir)
|
||||
h.z.boundsURL = func(site string) (string, bool) { return site + "get_geojson.php", true }
|
||||
|
||||
@@ -28,7 +28,7 @@ const (
|
||||
MaxBody = 64 << 10
|
||||
PollInterval = 15 * time.Second // one source per tick, at most four requests/minute
|
||||
|
||||
failureRetry = 5 * time.Minute // transient failures shouldn't cost a whole refresh interval
|
||||
failureRetry = config.MinScopesRefresh // get_scopes.php allows one request per region per 55m
|
||||
)
|
||||
|
||||
// Cache stores source provenance and freshness separately from packet evidence.
|
||||
@@ -60,17 +60,16 @@ type source struct {
|
||||
|
||||
// Importer is owned by one background task; ScopeStore synchronizes its consumers.
|
||||
type Importer struct {
|
||||
store Store
|
||||
scopes *scopestore.ScopeStore
|
||||
manual []scopestore.Entry
|
||||
dir *Directory
|
||||
scopesURL func(site string) (string, bool)
|
||||
seen map[string]bool
|
||||
sources []source
|
||||
interval time.Duration
|
||||
client *http.Client
|
||||
retryAfter time.Time
|
||||
onChange func(context.Context)
|
||||
store Store
|
||||
scopes *scopestore.ScopeStore
|
||||
manual []scopestore.Entry
|
||||
dir *Directory
|
||||
scopesURL func(site string) (string, bool)
|
||||
seen map[string]bool
|
||||
sources []source
|
||||
interval time.Duration
|
||||
client *http.Client
|
||||
onChange func(context.Context)
|
||||
}
|
||||
|
||||
// SetCacheInvalidator is wired once at startup, before the background task starts.
|
||||
@@ -129,9 +128,6 @@ func (i *Importer) track(iatas []string, saved map[string]Catalogue) {
|
||||
s.cache.LastError = "invalid saved catalogue; awaiting refresh"
|
||||
}
|
||||
}
|
||||
if c.LastError == "HTTP 429" && c.NextAttempt.After(i.retryAfter) {
|
||||
i.retryAfter = c.NextAttempt
|
||||
}
|
||||
i.log(s, "restored")
|
||||
}
|
||||
i.sources = append(i.sources, s)
|
||||
@@ -149,9 +145,6 @@ 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) || 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)
|
||||
@@ -253,15 +246,12 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error
|
||||
}
|
||||
}
|
||||
if update.LastError != "" {
|
||||
// Transient failures shouldn't cost a whole refresh interval, but never beat a Retry-After.
|
||||
// Failures retry before the full interval, but never sooner than the rate limit or a Retry-After.
|
||||
update.NextAttempt = now.Add(min(i.interval, failureRetry))
|
||||
if retryAfter.After(update.NextAttempt) {
|
||||
update.NextAttempt = retryAfter
|
||||
}
|
||||
}
|
||||
if update.LastError == "HTTP 429" {
|
||||
i.retryAfter = update.NextAttempt
|
||||
}
|
||||
if err := i.store.SaveScopeCatalogue(ctx, s.iata, s.url, update, entries); err != nil {
|
||||
// Avoid a retry every scheduler tick during a DB outage; never publish uncommitted keys.
|
||||
s.cache.NextAttempt = update.NextAttempt
|
||||
|
||||
@@ -75,7 +75,7 @@ func (s *memoryStore) SaveScopeCatalogue(_ context.Context, iata, url string, ne
|
||||
// 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)
|
||||
imp, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory()), scopes, manual)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -154,7 +154,7 @@ func TestCatalogueRefreshFallbackAndRestart(t *testing.T) {
|
||||
}
|
||||
status, after = 429, "7200"
|
||||
check()
|
||||
if !imp.retryAfter.Equal(now.Add(2 * time.Hour)) {
|
||||
if !imp.sources[0].cache.NextAttempt.Equal(now.Add(2 * time.Hour)) {
|
||||
t.Fatal("Retry-After ignored")
|
||||
}
|
||||
restored := scopestore.New()
|
||||
@@ -167,7 +167,7 @@ func TestCatalogueRefreshFallbackAndRestart(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if calls != before {
|
||||
t.Fatal("restart ignored shared rate-limit cooldown")
|
||||
t.Fatal("restart retried before Retry-After")
|
||||
}
|
||||
status, after, body = 200, "", catalogue()
|
||||
check()
|
||||
@@ -260,7 +260,7 @@ func TestImportedSourcesOverlapWithoutGlobalMembership(t *testing.T) {
|
||||
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}, store, NewDirectory(), scopes, nil)
|
||||
_, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, NewDirectory(newZoneListMemory()), scopes, nil)
|
||||
if err != nil || len(scopes.Entries()) != 1 || !reflect.DeepEqual(scopes.Entries()[0].IATAs, []string{"YOW", "YVR"}) {
|
||||
t.Fatal("latest saved catalogue per IATA not restored", scopes.Entries(), err)
|
||||
}
|
||||
@@ -269,7 +269,7 @@ func TestImportedSourcesOverlapWithoutGlobalMembership(t *testing.T) {
|
||||
}
|
||||
// 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 {
|
||||
if _, err = New(ctx, config.MeshMapperScopesConfig{}, nil, NewDirectory(newZoneListMemory()), disabled, nil); err != nil || len(disabled.Entries()) != 0 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -300,20 +300,21 @@ func TestInvalidSavedCatalogueDoesNotBlockStartupOrRefresh(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetryAfterShorterThanIntervalIsHonoured(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Retry-After", "1800")
|
||||
w.WriteHeader(http.StatusTooManyRequests)
|
||||
}))
|
||||
defer server.Close()
|
||||
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)
|
||||
}
|
||||
if got := imp.sources[0].cache.NextAttempt.Sub(now); got != 30*time.Minute {
|
||||
t.Fatalf("next attempt after 429 in %v, want %v (server Retry-After must not be overridden by the 1h interval)", got, 30*time.Minute)
|
||||
func TestRetryAfterNeverBeatsTheRateLimit(t *testing.T) {
|
||||
for retryAfter, want := range map[string]time.Duration{"1800": time.Hour, "7200": 2 * time.Hour} {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Retry-After", retryAfter)
|
||||
w.WriteHeader(http.StatusTooManyRequests)
|
||||
}))
|
||||
imp := newImporter(t, config.MeshMapperScopesConfig{Enabled: true}, newMemoryStore("YOW"), server.URL, scopestore.New())
|
||||
now := time.Now().UTC()
|
||||
if err := imp.refresh(context.Background(), &imp.sources[0], now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
server.Close()
|
||||
if got := imp.sources[0].cache.NextAttempt.Sub(now); got != want {
|
||||
t.Fatalf("Retry-After %ss: next attempt in %v, want %v", retryAfter, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -326,8 +327,8 @@ func TestFailedRefreshRetriesSoon(t *testing.T) {
|
||||
if err := imp.refresh(ctx, &imp.sources[0], now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := imp.sources[0].cache.NextAttempt.Sub(now); got != failureRetry {
|
||||
t.Fatalf("next attempt after failure in %v, want %v", got, failureRetry)
|
||||
if got := imp.sources[0].cache.NextAttempt.Sub(now); got != time.Hour {
|
||||
t.Fatalf("next attempt after failure in %v, want 1h", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -359,7 +360,7 @@ func newFakeSites(t *testing.T) *fakeSites {
|
||||
|
||||
func newDiscoveringImporter(t *testing.T, f *fakeSites, store *memoryStore, scopes *scopestore.ScopeStore) *Importer {
|
||||
t.Helper()
|
||||
dir := NewDirectory()
|
||||
dir := NewDirectory(newZoneListMemory())
|
||||
dir.listURL = f.URL + "/get_zones.php"
|
||||
imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopes, nil)
|
||||
if err != nil {
|
||||
|
||||
@@ -26,8 +26,8 @@ const (
|
||||
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
|
||||
zoneListFresh = config.MinZonesRefresh // get_zones.php allows one request per country per 23.5h
|
||||
zoneFailureRetry = config.MinZonesRefresh // failed requests count too
|
||||
)
|
||||
|
||||
var zoneSite = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*\.meshmapper\.net$`)
|
||||
@@ -58,17 +58,16 @@ type zoneRegion struct {
|
||||
|
||||
// Zones is owned by one background task; listeners are wired before Restore.
|
||||
type Zones struct {
|
||||
store ZoneStore
|
||||
enabled bool
|
||||
seen map[string]bool
|
||||
interval time.Duration
|
||||
client *http.Client
|
||||
dir *Directory
|
||||
boundsURL func(site string) (string, bool)
|
||||
regions []*zoneRegion
|
||||
retryAfter time.Time
|
||||
onChange func(ctx context.Context, iata string)
|
||||
onUpdate func(imported map[string]json.RawMessage)
|
||||
store ZoneStore
|
||||
enabled bool
|
||||
seen map[string]bool
|
||||
interval time.Duration
|
||||
client *http.Client
|
||||
dir *Directory
|
||||
boundsURL func(site string) (string, bool)
|
||||
regions []*zoneRegion
|
||||
onChange func(ctx context.Context, iata string)
|
||||
onUpdate func(imported map[string]json.RawMessage)
|
||||
|
||||
importGroups bool
|
||||
groupsSynced bool
|
||||
@@ -147,9 +146,6 @@ func (z *Zones) track(iatas []string, saved map[string]Boundary) {
|
||||
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})
|
||||
if ok {
|
||||
z.log(b, "restored")
|
||||
@@ -168,9 +164,6 @@ 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) || now.Before(z.dir.pausedUntil()) {
|
||||
return nil
|
||||
}
|
||||
// IATAs are created from traffic, so pick up new ones every tick.
|
||||
iatas, err := z.store.ListKnownIATAs(ctx)
|
||||
if err != nil {
|
||||
@@ -238,9 +231,6 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, zones map[string]zon
|
||||
}
|
||||
default:
|
||||
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")
|
||||
@@ -322,17 +312,6 @@ func statusProblem(status int, header http.Header, now time.Time, retryAt *time.
|
||||
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)
|
||||
|
||||
@@ -30,6 +30,7 @@ type zoneMemoryStore struct {
|
||||
pruned []string
|
||||
fail bool
|
||||
regions map[string]RegionState
|
||||
lists *zoneListMemory
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) ListRegionState(context.Context) ([]RegionState, error) {
|
||||
@@ -102,6 +103,30 @@ func (s *zoneMemoryStore) SaveZoneBoundary(_ context.Context, b Boundary) error
|
||||
return nil
|
||||
}
|
||||
|
||||
type zoneListMemory struct{ rows map[string]ZoneList }
|
||||
|
||||
func newZoneListMemory() *zoneListMemory { return &zoneListMemory{rows: map[string]ZoneList{}} }
|
||||
|
||||
func (s *zoneListMemory) ListZoneLists(context.Context) ([]ZoneList, error) {
|
||||
var out []ZoneList
|
||||
for _, l := range s.rows {
|
||||
out = append(out, l)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *zoneListMemory) SaveZoneList(_ context.Context, l ZoneList) error {
|
||||
old := s.rows[l.Country]
|
||||
if l.Payload == nil {
|
||||
l.Payload = old.Payload
|
||||
}
|
||||
if l.FetchedAt.IsZero() {
|
||||
l.FetchedAt, l.ETag = old.FetchedAt, old.ETag
|
||||
}
|
||||
s.rows[l.Country] = l
|
||||
return nil
|
||||
}
|
||||
|
||||
type fakeMeshMapper struct {
|
||||
*httptest.Server
|
||||
list, boundary string
|
||||
@@ -153,8 +178,14 @@ func newZoneHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, ena
|
||||
if store.iatas == nil {
|
||||
store.iatas = []string{"YOW"}
|
||||
}
|
||||
dir := NewDirectory()
|
||||
if store.lists == nil {
|
||||
store.lists = newZoneListMemory()
|
||||
}
|
||||
dir := NewDirectory(store.lists)
|
||||
dir.listURL = f.URL + "/get_zones.php"
|
||||
if err := dir.Restore(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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) })
|
||||
@@ -216,11 +247,11 @@ func TestZonesKeepLastGoodBoundary(t *testing.T) {
|
||||
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},
|
||||
{"not found", "", 404, "HTTP 404", 24 * time.Hour},
|
||||
{"unavailable", "", 503, "HTTP 503", 24 * time.Hour},
|
||||
{"truncated", boundaryBody("YOW", square)[:80], 200, "invalid response", 24 * time.Hour},
|
||||
{"other region", boundaryBody("YVR", square), 200, "invalid response", 24 * time.Hour},
|
||||
{"oversized", `{"type":"FeatureCollection","features":[],"pad":"` + strings.Repeat("x", MaxBoundaryBody) + `"}`, 200, "invalid response", 24 * time.Hour},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newFakeMeshMapper(t)
|
||||
@@ -279,22 +310,37 @@ func TestZonesListFailureBacksOff(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestZonesRateLimitPausesEveryRequest(t *testing.T) {
|
||||
func TestZonesRateLimitWaitsOnlyForThatRegion(t *testing.T) {
|
||||
for retryAfter, want := range map[string]time.Duration{"120": 24 * time.Hour, "172800": 48 * time.Hour} {
|
||||
f := newFakeMeshMapper(t)
|
||||
h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true)
|
||||
f.status, f.retryAfter = 429, retryAfter
|
||||
b := h.tick(t)
|
||||
if b.LastError != "HTTP 429" || b.NextAttempt.Sub(b.AttemptedAt) != want {
|
||||
t.Fatalf("Retry-After %ss: next attempt in %v, want %v", retryAfter, b.NextAttempt.Sub(b.AttemptedAt), want)
|
||||
}
|
||||
restarted := newZoneHarness(t, f, h.store, true)
|
||||
if err := restarted.z.Refresh(context.Background()); err != nil || f.calls != 1 {
|
||||
t.Fatal("requested again before the rate limit allows", err, f.calls)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestZoneListSurvivesRestart(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)
|
||||
store := &zoneMemoryStore{rows: map[string]Boundary{}}
|
||||
h := newZoneHarness(t, f, store, true)
|
||||
h.tick(t)
|
||||
if f.listCalls != 1 {
|
||||
t.Fatal("list not fetched", f.listCalls)
|
||||
}
|
||||
h.z.regions[0].b.NextAttempt = time.Time{}
|
||||
_ = h.z.Refresh(context.Background())
|
||||
if f.calls != 1 {
|
||||
t.Fatal("requested during Retry-After")
|
||||
if got := h.z.dir.lists["CA"].nextAttempt.Sub(h.z.dir.lists["CA"].fetchedAt); got != 24*time.Hour {
|
||||
t.Fatal("zone list refetched sooner than the rate limit allows", got)
|
||||
}
|
||||
restarted := newZoneHarness(t, f, h.store, true)
|
||||
if !restarted.z.retryAfter.Equal(b.NextAttempt) {
|
||||
t.Fatal("persisted 429 not restored")
|
||||
restarted := newZoneHarness(t, f, store, true)
|
||||
restarted.tick(t)
|
||||
if f.listCalls != 1 || f.calls != 2 {
|
||||
t.Fatal("restart refetched the zone list", f.listCalls, f.calls)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user