diff --git a/README.md b/README.md index 5b07c46..119c329 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 7e54751..c6f93a1 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -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) } diff --git a/config.yaml.example b/config.yaml.example index 39160c2..cc863fc 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -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 diff --git a/db/meshmapper.go b/db/meshmapper.go index 884b550..02e0de2 100644 --- a/db/meshmapper.go +++ b/db/meshmapper.go @@ -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 diff --git a/db/meshmapper_integration_test.go b/db/meshmapper_integration_test.go index 7d8c2a1..e1589a6 100644 --- a/db/meshmapper_integration_test.go +++ b/db/meshmapper_integration_test.go @@ -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) } diff --git a/db/queries/queries.sql b/db/queries/queries.sql index afcd8c5..f215ce0 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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 diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index a0aca7b..5ab8a15 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -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() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 7a64a31..1b3bbc9 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -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 // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index e8b52ff..bc83377 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 ( diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go index e0bc107..d4fa642 100644 --- a/internal/config/meshmapper.go +++ b/internal/config/meshmapper.go @@ -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://.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 } diff --git a/internal/config/meshmapper_test.go b/internal/config/meshmapper_test.go index cbaf0b3..8cd3264 100644 --- a/internal/config/meshmapper_test.go +++ b/internal/config/meshmapper_test.go @@ -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") } } diff --git a/internal/meshmapper/directory.go b/internal/meshmapper/directory.go new file mode 100644 index 0000000..0ead347 --- /dev/null +++ b/internal/meshmapper/directory.go @@ -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 +} diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 1c26767..907f427 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -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"` diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go index 95ac0b9..d12dfc0 100644 --- a/internal/meshmapper/scopes_test.go +++ b/internal/meshmapper/scopes_test.go @@ -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) + } +} diff --git a/internal/meshmapper/zones.go b/internal/meshmapper/zones.go index 37e6bc0..55d2398 100644 --- a/internal/meshmapper/zones.go +++ b/internal/meshmapper/zones.go @@ -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. diff --git a/internal/meshmapper/zones_test.go b/internal/meshmapper/zones_test.go index 9798e3f..f93152e 100644 --- a/internal/meshmapper/zones_test.go +++ b/internal/meshmapper/zones_test.go @@ -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") } }