From 38b1a206b1f487f5b714ff371b6062fe09bd8e7e Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 22:56:04 -0400 Subject: [PATCH] fix(meshmapper): respect MeshMapper's per-endpoint rate limits --- cmd/beacon/main.go | 11 +++- config.yaml.example | 7 ++- db/meshmapper.go | 30 +++++++++++ db/meshmapper_integration_test.go | 20 +++++-- db/migrations/001_baseline.sql | 13 +++++ db/queries/queries.sql | 16 ++++++ db/sqlc/mock/querier.go | 29 +++++++++++ db/sqlc/models.go | 10 ++++ db/sqlc/querier.go | 3 ++ db/sqlc/queries.sql.go | 69 ++++++++++++++++++++++++ internal/config/meshmapper.go | 16 ++++-- internal/config/meshmapper_test.go | 8 ++- internal/meshmapper/directory.go | 78 ++++++++++++++++++--------- internal/meshmapper/groups_test.go | 2 +- internal/meshmapper/scopes.go | 34 +++++------- internal/meshmapper/scopes_test.go | 45 ++++++++-------- internal/meshmapper/zones.go | 45 +++++----------- internal/meshmapper/zones_test.go | 84 +++++++++++++++++++++++------- 18 files changed, 386 insertions(+), 134 deletions(-) diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index fa4eab1..2194636 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -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) diff --git a/config.yaml.example b/config.yaml.example index 014c9d2..e315c9e 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -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: diff --git a/db/meshmapper.go b/db/meshmapper.go index 7b4f3cb..6ff3fa0 100644 --- a/db/meshmapper.go +++ b/db/meshmapper.go @@ -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 { diff --git a/db/meshmapper_integration_test.go b/db/meshmapper_integration_test.go index 5db0161..7dd8c50 100644 --- a/db/meshmapper_integration_test.go +++ b/db/meshmapper_integration_test.go @@ -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) { diff --git a/db/migrations/001_baseline.sql b/db/migrations/001_baseline.sql index ca09fe3..515df56 100644 --- a/db/migrations/001_baseline.sql +++ b/db/migrations/001_baseline.sql @@ -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); diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 902baca..2e99933 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 12e4e2b..81e4ff4 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -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() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index eba458a..5ddfc8b 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -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"` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index f319a90..00d07da 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -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) diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index e16f105..d7ee331 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go index b1c7094..f76d3bd 100644 --- a/internal/config/meshmapper.go +++ b/internal/config/meshmapper.go @@ -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") diff --git a/internal/config/meshmapper_test.go b/internal/config/meshmapper_test.go index 43c8c22..5127959 100644 --- a/internal/config/meshmapper_test.go +++ b/internal/config/meshmapper_test.go @@ -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}, diff --git a/internal/meshmapper/directory.go b/internal/meshmapper/directory.go index c4a9d85..e429b92 100644 --- a/internal/meshmapper/directory.go +++ b/internal/meshmapper/directory.go @@ -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 } diff --git a/internal/meshmapper/groups_test.go b/internal/meshmapper/groups_test.go index 81bd67a..e632888 100644 --- a/internal/meshmapper/groups_test.go +++ b/internal/meshmapper/groups_test.go @@ -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 } diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 907f427..a4171be 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -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 diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go index d12dfc0..48104f5 100644 --- a/internal/meshmapper/scopes_test.go +++ b/internal/meshmapper/scopes_test.go @@ -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 { diff --git a/internal/meshmapper/zones.go b/internal/meshmapper/zones.go index b37b49b..d14ab4c 100644 --- a/internal/meshmapper/zones.go +++ b/internal/meshmapper/zones.go @@ -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) diff --git a/internal/meshmapper/zones_test.go b/internal/meshmapper/zones_test.go index 6dc4cf6..bf14120 100644 --- a/internal/meshmapper/zones_test.go +++ b/internal/meshmapper/zones_test.go @@ -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) } }