From ec0037860c0332e479a8fbce57605e3035b8b134 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Fri, 2 Oct 2026 16:48:56 -0400 Subject: [PATCH] fix(meshmapper): count cut-short requests and qualify groups by heard IATAs Record each attempt before sending so a request lost to a deadline still waits out MeshMapper's limit, and stop holding the directory lock across HTTP. Groups now qualify only through heard or configured IATAs, and one missing country list only skips pruning. --- db/meshmapper.go | 9 +++ internal/meshmapper/channels.go | 7 ++ internal/meshmapper/channels_test.go | 22 +++++++ internal/meshmapper/directory.go | 69 +++++++++++++------- internal/meshmapper/groups.go | 54 ++++++++++++---- internal/meshmapper/groups_test.go | 52 +++++++++++++++ internal/meshmapper/scopes.go | 7 ++ internal/meshmapper/scopes_test.go | 17 +++++ internal/meshmapper/zones.go | 23 +++++-- internal/meshmapper/zones_test.go | 95 +++++++++++++++++++++++++++- 10 files changed, 309 insertions(+), 46 deletions(-) diff --git a/db/meshmapper.go b/db/meshmapper.go index 8ce90d1..82e2d59 100644 --- a/db/meshmapper.go +++ b/db/meshmapper.go @@ -77,6 +77,15 @@ func (s *Store) ListKnownIATAs(ctx context.Context) ([]string, error) { return iatas, nil } +// ListHeardIATAs returns the IATAs observers last reported from. +func (s *Store) ListHeardIATAs(ctx context.Context) ([]string, error) { + rows, err := s.pool.Query(ctx, `SELECT DISTINCT last_iata::text FROM observers WHERE last_iata IS NOT NULL ORDER BY 1`) + if err != nil { + return nil, err + } + return pgx.CollectRows(rows, pgx.RowTo[string]) +} + func (s *Store) PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error) { if keep == nil { keep = []string{} diff --git a/internal/meshmapper/channels.go b/internal/meshmapper/channels.go index 4ead380..33a07e3 100644 --- a/internal/meshmapper/channels.go +++ b/internal/meshmapper/channels.go @@ -178,6 +178,13 @@ func (c *Channels) refresh(ctx context.Context, s *channelSource, now time.Time) if len(s.cache.Payload) > 0 { etag = s.cache.ETag } + // Recorded first: an abandoned request may still have used the region's call. + attempt := Cache{AttemptedAt: now, NextAttempt: now.Add(min(c.interval, channelRetry)), LastError: "no response"} + s.cache.NextAttempt = attempt.NextAttempt + if err := c.store.SaveChannelCatalogue(ctx, s.iata, s.url, attempt, nil); err != nil { + return fmt.Errorf("persist channel catalogue %s: %w", s.iata, err) + } + s.cache.AttemptedAt, s.cache.LastError = now, attempt.LastError status, body, header, err := get(ctx, c.client, "Beacon-MeshMapper-Channels/1", s.url, etag, MaxChannelBody) if err != nil { return err diff --git a/internal/meshmapper/channels_test.go b/internal/meshmapper/channels_test.go index 5a3def5..29773aa 100644 --- a/internal/meshmapper/channels_test.go +++ b/internal/meshmapper/channels_test.go @@ -232,3 +232,25 @@ func TestChannelsDisabledClearsMembers(t *testing.T) { t.Fatal("disabled import kept membership or sources") } } + +func TestChannelsRequestCutShortStillCounts(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() })) + defer server.Close() + store := newChannelMemoryStore("YOW") + c, err := NewChannels(context.Background(), config.MeshMapperChannelsConfig{Enabled: true}, store, NewDirectory(newZoneListMemory()), keystore.NewMapKeyStore(nil)) + if err != nil { + t.Fatal(err) + } + s := &c.sources[0] + s.url = server.URL + now := time.Now().UTC() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = c.refresh(ctx, s, now) + if got := s.cache.NextAttempt.Sub(now); got != channelRetry { + t.Fatal("abandoned request retried inside the rate limit", got) + } + if row := store.rows[sourceKey{"YOW", server.URL}]; row.NextAttempt.Sub(now) != channelRetry { + t.Fatal("abandoned request not persisted", row) + } +} diff --git a/internal/meshmapper/directory.go b/internal/meshmapper/directory.go index 20f7503..5113f02 100644 --- a/internal/meshmapper/directory.go +++ b/internal/meshmapper/directory.go @@ -36,6 +36,7 @@ type zoneList struct { groups map[string]zoneGroup etag string fetchedAt, nextAttempt time.Time + fetching bool } // ZoneList is one country's saved get_zones.php response and fetch state. @@ -89,31 +90,45 @@ func (d *Directory) Restore(ctx context.Context) error { } // 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. +// spent the caller's one request; nil zones without fetched means back off, +// including while another task's request for the country is in flight. 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 { + d.mu.Unlock() return list.zones, false, nil } - if list != nil && now.Before(list.nextAttempt) { + if list != nil && (list.fetching || now.Before(list.nextAttempt)) { + d.mu.Unlock() 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 { + // Every request counts against the country's 23.5h allowance, failed, abandoned or not. + list.fetching, list.nextAttempt = true, now.Add(zoneListFresh) + cached, etag := list.zones != nil, "" + if cached { etag = list.etag } + d.mu.Unlock() + defer func() { + d.mu.Lock() + list.fetching = false + d.mu.Unlock() + }() + return nil, true, d.fetch(ctx, list, country, cached, etag, now) +} + +// fetch runs without d.mu so other tasks never wait out this request. +func (d *Directory) fetch(ctx context.Context, list *zoneList, country string, cached bool, etag string, now time.Time) error { + attempt := ZoneList{Country: country, AttemptedAt: now, NextAttempt: now.Add(zoneListFresh), LastError: "no response"} + if err := d.store.SaveZoneList(ctx, attempt); err != nil { + return fmt.Errorf("persist MeshMapper zone list %s: %w", country, err) + } + endpoint := d.listURL + "?country=" + url.QueryEscape(country) status, body, header, err := get(ctx, d.client, "Beacon-MeshMapper-Zones/1", endpoint, etag, MaxZoneList) if err != nil { return err @@ -121,25 +136,28 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er problem := "" var payload json.RawMessage var retryAt time.Time + var zones map[string]zoneEntry + var groups map[string]zoneGroup switch status { case http.StatusOK: - zones, groups, decodeErr := decodeZones(body, country) - if decodeErr != nil { + var decodeErr error + if zones, groups, decodeErr = decodeZones(body, country); decodeErr != nil { problem = "invalid response" } else { - list.zones, list.groups, list.etag = zones, groups, header.Get("ETag") payload = body - d.version++ } case http.StatusNotModified: - if list.zones == nil { + if !cached { problem = "304 without cached list" } default: problem = statusProblem(status, header, now, &retryAt) } - // Every request counts against the country's 23.5h allowance, failed or not. - list.nextAttempt = now.Add(zoneListFresh) + d.mu.Lock() + if payload != nil { + list.zones, list.groups, list.etag = zones, groups, header.Get("ETag") + d.version++ + } if retryAt.After(list.nextAttempt) { list.nextAttempt = retryAt } @@ -148,28 +166,31 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er list.fetchedAt = now saved.FetchedAt, saved.ETag = now, list.etag } + nZones, nGroups := len(list.zones), len(list.groups) + d.mu.Unlock() 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), "groups", len(list.groups), "next_attempt", list.nextAttempt, "last_error", problem) + "zones", nZones, "groups", nGroups, "next_attempt", saved.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 } -// snapshot merges every loaded list's zones and groups by code. complete is false until -// each country has a list, so a cold start can't look like every group disappearing. +// snapshot merges every loaded list's zones and groups by code. complete is false while +// a country has no list, so a cold start or failed list can't look like groups disappearing. func (d *Directory) snapshot(countries []string) (merged map[string]zoneGroup, zones map[string]zoneEntry, version int, complete bool) { d.mu.Lock() defer d.mu.Unlock() - merged, zones = map[string]zoneGroup{}, map[string]zoneEntry{} + merged, zones, complete = map[string]zoneGroup{}, map[string]zoneEntry{}, true for _, country := range countries { list := d.lists[country] if list == nil || list.zones == nil { - return nil, nil, d.version, false + complete = false + continue } for code, z := range list.zones { zones[code] = z @@ -184,7 +205,7 @@ func (d *Directory) snapshot(countries []string) (merged map[string]zoneGroup, z merged[code] = g } } - return merged, zones, d.version, true + return merged, zones, d.version, complete } var ( diff --git a/internal/meshmapper/groups.go b/internal/meshmapper/groups.go index 588e1ee..9a18154 100644 --- a/internal/meshmapper/groups.go +++ b/internal/meshmapper/groups.go @@ -8,6 +8,7 @@ import ( "context" "fmt" "log/slog" + "maps" "slices" "strings" ) @@ -55,9 +56,15 @@ func (z *Zones) pruneGroups(ctx context.Context) error { return nil } +// syncInputs is what syncDirectory last applied; it reruns when any of them changes. +type syncInputs struct { + version, known int + complete bool + heard string +} + // syncDirectory applies the loaded zone lists to IATA details and, when enabled, -// imported regions. It only runs once every tracked country has a list, and again -// when a list or the IATA set changes. +// imported regions. Groups are only pruned once every tracked country has a list. func (z *Zones) syncDirectory(ctx context.Context, iatas []string) error { var countries []string for _, r := range z.regions { @@ -65,18 +72,27 @@ func (z *Zones) syncDirectory(ctx context.Context, iatas []string) error { } slices.Sort(countries) groups, zones, version, complete := z.dir.snapshot(slices.Compact(countries)) - if !complete || (z.groupsSynced && version == z.groupsVersion && len(iatas) == z.groupsKnown) { + var heard []string + if z.importGroups { + var err error + if heard, err = z.store.ListHeardIATAs(ctx); err != nil { + return fmt.Errorf("list heard IATAs for MeshMapper groups: %w", err) + } + slices.Sort(heard) + } + inputs := syncInputs{version: version, known: len(iatas), complete: complete, heard: strings.Join(heard, ",")} + if z.groupsSynced && inputs == z.synced { return nil } if err := z.syncIATADetails(ctx, zones); err != nil { return err } if z.importGroups { - if err := z.syncGroups(ctx, iatas, groups, zones); err != nil { + if err := z.syncGroups(ctx, heard, groups, zones, complete); err != nil { return err } } - z.groupsSynced, z.groupsVersion, z.groupsKnown = true, version, len(iatas) + z.groupsSynced, z.synced = true, inputs return nil } @@ -111,10 +127,15 @@ func (z *Zones) syncIATADetails(ctx context.Context, zones map[string]zoneEntry) return nil } -func (z *Zones) syncGroups(ctx context.Context, iatas []string, groups map[string]zoneGroup, zones map[string]zoneEntry) error { - known := make(map[string]bool, len(iatas)) - for _, iata := range iatas { - known[iata] = true +// syncGroups imports groups with a member that is heard or configured, never one only +// its own import created. Without every country's list it adds and updates but never prunes. +func (z *Zones) syncGroups(ctx context.Context, heard []string, groups map[string]zoneGroup, zones map[string]zoneEntry, complete bool) error { + inUse := maps.Clone(z.configured) + if inUse == nil { + inUse = map[string]bool{} + } + for _, iata := range heard { + inUse[iata] = true } state, err := z.store.ListRegionState(ctx) if err != nil { @@ -125,11 +146,14 @@ func (z *Zones) syncGroups(ctx context.Context, iatas []string, groups map[strin stored[r.Slug] = r if !r.Imported { handOrder = max(handOrder, r.DisplayOrder) + for _, iata := range r.IATAs { + inUse[iata] = true + } } } var want []RegionState for code, g := range groups { - if slices.ContainsFunc(g.members, func(m string) bool { return known[m] }) { + if slices.ContainsFunc(g.members, func(m string) bool { return inUse[m] }) { lat, lng := center(g.members, zones) want = append(want, RegionState{Slug: strings.ToLower(code), Name: g.name, Imported: true, IATAs: g.members, CenterLat: lat, CenterLng: lng}) } @@ -157,11 +181,13 @@ func (z *Zones) syncGroups(ctx context.Context, iatas []string, groups map[strin keep, changed = append(keep, r.Slug), true slog.Info("MeshMapper region imported", "component", "meshmapper.zones", "slug", r.Slug, "name", r.Name, "iatas", r.IATAs) } - pruned, err := z.store.PruneImportedRegions(ctx, keep) - if err != nil { - return fmt.Errorf("prune MeshMapper regions: %w", err) + var pruned []string + if complete { + if pruned, err = z.store.PruneImportedRegions(ctx, keep); err != nil { + return fmt.Errorf("prune MeshMapper regions: %w", err) + } + z.regionsRemoved(pruned) } - z.regionsRemoved(pruned) if changed || len(pruned) > 0 { z.regionsChanged(ctx) } diff --git a/internal/meshmapper/groups_test.go b/internal/meshmapper/groups_test.go index 4caefa3..fa1d853 100644 --- a/internal/meshmapper/groups_test.go +++ b/internal/meshmapper/groups_test.go @@ -32,6 +32,9 @@ func newGroupHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, im if store.iatas == nil { store.iatas = []string{"YOW"} } + if store.heard == nil { + store.heard = slices.Clone(store.iatas) + } if store.rows == nil { store.rows = map[string]Boundary{} } @@ -182,3 +185,52 @@ func TestZoneListNamesAndLocatesIATAs(t *testing.T) { t.Fatal("unchanged details rewritten", h.store.writes, changed) } } + +func TestGroupStopsQualifyingOnImportedMembers(t *testing.T) { + h := newGroupHarness(t, groupedFake(t), &zoneMemoryStore{}, true) + h.refetch(t) + if _, ok := h.store.regions["onqc"]; !ok { + t.Fatal("group not imported") + } + without := strings.Replace(groupedList, `"members":["YUL","YOW","YQB"]`, `"members":["YUL","YQB"]`, 1) + h.f.list = strings.ReplaceAll(without, "URL", h.f.URL) + h.refetch(t) + if _, ok := h.store.regions["onqc"]; ok { + t.Fatal("group kept only by members its own import created") + } +} + +func TestGroupQualifiesThroughConfiguredRegion(t *testing.T) { + hand := RegionState{Slug: "east", Name: "East", IATAs: []string{"YYZ"}} + h := newGroupHarness(t, groupedFake(t), &zoneMemoryStore{regions: map[string]RegionState{"east": hand}}, true) + h.refetch(t) + if _, ok := h.store.regions["golm"]; !ok { + t.Fatal("group with a configured member not imported") + } +} + +func TestGroupImportsFromListsItHas(t *testing.T) { + old := RegionState{Slug: "old", Name: "Old", Imported: true, IATAs: []string{"YOW"}} + store := &zoneMemoryStore{iatas: []string{"YOW", "SEA"}, regions: map[string]RegionState{"old": old}} + f := groupedFake(t) + f.otherStatus = 500 + h := newGroupHarness(t, f, store, true) + h.refetch(t) + for range 2 { + if err := h.z.Refresh(context.Background()); err != nil { + t.Fatal(err) + } + } + if f.otherStatus != 500 || h.z.dir.lists["US"] == nil || h.z.dir.lists["US"].zones != nil { + t.Fatal("US list should have failed") + } + if _, ok := store.regions["onqc"]; !ok { + t.Fatal("one failed country blocked group import") + } + if store.details["YOW"].Name != "Ottawa" { + t.Fatal("one failed country blocked IATA details", store.details) + } + if _, ok := store.regions["old"]; !ok { + t.Fatal("pruned with a country list missing") + } +} diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 3ecdcbe..4099aae 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -195,6 +195,13 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error if len(s.cache.Payload) > 0 && s.cache.ETag != "" { request.Header.Set("If-None-Match", s.cache.ETag) } + // Recorded first: an abandoned request may still have used the region's call. + attempt := Cache{AttemptedAt: now, NextAttempt: now.Add(min(i.interval, failureRetry)), LastError: "no response"} + s.cache.NextAttempt = attempt.NextAttempt + if err := i.store.SaveScopeCatalogue(ctx, s.iata, s.url, attempt, nil); err != nil { + return fmt.Errorf("persist scope catalogue %s: %w", s.iata, err) + } + s.cache.AttemptedAt, s.cache.LastError = now, attempt.LastError response, err := i.client.Do(request) var entries []scopestore.Entry var generated time.Time diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go index 48104f5..cd64550 100644 --- a/internal/meshmapper/scopes_test.go +++ b/internal/meshmapper/scopes_test.go @@ -408,3 +408,20 @@ func TestScopesSkipUnlistedIATAs(t *testing.T) { t.Fatalf("unlisted IATA requested or retried early: hits=%d %+v", f.scopeHits, s.cache) } } + +func TestScopesRequestCutShortStillCounts(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() })) + defer server.Close() + store := newMemoryStore("YOW") + imp := newImporter(t, config.MeshMapperScopesConfig{Enabled: true}, store, server.URL, scopestore.New()) + now := time.Now().UTC() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = imp.refresh(ctx, &imp.sources[0], now) + if got := imp.sources[0].cache.NextAttempt.Sub(now); got != failureRetry { + t.Fatal("abandoned request retried inside the rate limit", got) + } + if row := store.rows[sourceKey{"YOW", server.URL}]; row.NextAttempt.Sub(now) != failureRetry { + t.Fatal("abandoned request not persisted", row) + } +} diff --git a/internal/meshmapper/zones.go b/internal/meshmapper/zones.go index 1310b71..4bec389 100644 --- a/internal/meshmapper/zones.go +++ b/internal/meshmapper/zones.go @@ -28,7 +28,7 @@ const ( // MeshMapper asks clients to allow 60-120s; a request given up on may still use the day's call. requestTimeout = 60 * time.Second - refreshTimeout = requestTimeout + 30*time.Second // room for the DB write after a slow request + refreshTimeout = requestTimeout + 30*time.Second // room for the DB writes around a slow request zoneListFresh = config.MinZonesRefresh // get_zones.php allows one request per country per 23.5h zoneFailureRetry = config.MinZonesRefresh // failed requests count too @@ -47,6 +47,8 @@ type Boundary struct { type ZoneStore interface { ListKnownIATAs(ctx context.Context) ([]string, error) + // ListHeardIATAs returns IATAs observers report from; group imports create IATAs too. + ListHeardIATAs(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 @@ -75,12 +77,12 @@ type Zones struct { onChange func(ctx context.Context, iata string) onUpdate func(imported map[string]json.RawMessage) - importGroups bool - groupsSynced bool - groupsVersion, groupsKnown int - onRegions func(ctx context.Context) - configured map[string]bool - onIATAs func(ctx context.Context) + importGroups bool + groupsSynced bool + synced syncInputs + onRegions func(ctx context.Context) + configured map[string]bool + onIATAs func(ctx context.Context) } func NewZones(cfg config.MeshMapperZonesConfig, store ZoneStore, dir *Directory) *Zones { @@ -218,6 +220,13 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, zones map[string]zon if r.b.Feature != nil { etag = r.b.ETag } + // Recorded first: an abandoned request may still have used the region's call. + attempt := Boundary{IATA: r.b.IATA, URL: update.URL, AttemptedAt: now, NextAttempt: now.Add(min(z.interval, zoneFailureRetry)), LastError: "no response"} + r.b.NextAttempt = attempt.NextAttempt + if err := z.store.SaveZoneBoundary(ctx, attempt); err != nil { + return fmt.Errorf("persist MeshMapper boundary %s: %w", r.b.IATA, err) + } + r.b.URL, r.b.AttemptedAt, r.b.LastError = attempt.URL, now, attempt.LastError status, body, header, err := get(ctx, z.client, "Beacon-MeshMapper-Zones/1", endpoint, etag, MaxBoundaryBody) if err != nil { return err diff --git a/internal/meshmapper/zones_test.go b/internal/meshmapper/zones_test.go index bdf4589..b4bb35f 100644 --- a/internal/meshmapper/zones_test.go +++ b/internal/meshmapper/zones_test.go @@ -11,6 +11,7 @@ import ( "net/http/httptest" "slices" "strings" + "sync/atomic" "testing" "time" @@ -28,6 +29,7 @@ func boundaryBody(code, geometry string) string { type zoneMemoryStore struct { iatas []string + heard []string rows map[string]Boundary pruned []string fail bool @@ -96,6 +98,10 @@ func (s *zoneMemoryStore) ListKnownIATAs(context.Context) ([]string, error) { return slices.Clone(s.iatas), nil } +func (s *zoneMemoryStore) ListHeardIATAs(context.Context) ([]string, error) { + return slices.Clone(s.heard), nil +} + func (s *zoneMemoryStore) PruneZoneBoundaries(_ context.Context, keep []string) ([]string, error) { var removed []string for iata := range s.rows { @@ -162,16 +168,28 @@ type fakeMeshMapper struct { etag, retryAfter string listCalls, calls int ifNoneMatch []string + otherStatus int // get_zones.php status for countries other than CA; 0 is unexpected + hang atomic.Bool // requests wait until the client gives up + hung atomic.Int32 } func newFakeMeshMapper(t *testing.T) *fakeMeshMapper { f := &fakeMeshMapper{listStatus: 200, status: 200, etag: `"v1"`, boundary: boundaryBody("YOW", square)} f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if f.hang.Load() { + f.hung.Add(1) + <-r.Context().Done() + return + } switch r.URL.Path { case "/get_zones.php": f.listCalls++ if r.URL.Query().Get("country") != "CA" { - t.Errorf("country query: %s", r.URL.RawQuery) + if f.otherStatus == 0 { + t.Errorf("country query: %s", r.URL.RawQuery) + } + w.WriteHeader(f.otherStatus) + return } w.WriteHeader(f.listStatus) _, _ = w.Write([]byte(f.list)) @@ -492,3 +510,78 @@ func TestClientsAllowMeshMapperTimeout(t *testing.T) { t.Fatal("a refresh must outlast its request") } } + +func TestZoneListCutShortStillCounts(t *testing.T) { + f := newFakeMeshMapper(t) + f.hang.Store(true) + lists := newZoneListMemory() + dir := NewDirectory(lists) + dir.listURL = f.URL + "/get_zones.php" + now := time.Now().UTC() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, fetched, _ := dir.List(ctx, "CA", now); !fetched { + t.Fatal("first call should spend the request") + } + f.hang.Store(false) + if zones, fetched, err := dir.List(context.Background(), "CA", now.Add(15*time.Second)); zones != nil || fetched || err != nil || f.listCalls != 0 || f.hung.Load() != 1 { + t.Fatal("abandoned request retried inside the rate limit", f.listCalls, fetched, err) + } + if got := lists.rows["CA"].NextAttempt.Sub(now); got != zoneListFresh { + t.Fatal("abandoned request not persisted", lists.rows["CA"]) + } +} + +func TestZoneListFetchDoesNotBlockOtherCallers(t *testing.T) { + release := make(chan struct{}) + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + <-release + _, _ = w.Write([]byte(`{"country":"CA","zones":[]}`)) + })) + defer server.Close() + defer close(release) + dir := NewDirectory(newZoneListMemory()) + dir.listURL = server.URL + now := time.Now().UTC() + go func() { _, _, _ = dir.List(context.Background(), "CA", now) }() + for calls.Load() == 0 { + time.Sleep(time.Millisecond) + } + done := make(chan struct{}) + go func() { + defer close(done) + if zones, fetched, err := dir.List(context.Background(), "CA", now); zones != nil || fetched || err != nil { + t.Error("second caller should back off while the list is in flight", fetched, err) + } + _, _, _, _ = dir.snapshot([]string{"CA"}) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("caller waited on another task's request") + } + if calls.Load() != 1 { + t.Fatal("duplicate list request", calls.Load()) + } +} + +func TestBoundaryCutShortStillCounts(t *testing.T) { + f := newFakeMeshMapper(t) + h := newZoneHarness(t, f, &zoneMemoryStore{rows: map[string]Boundary{}}, true) + if err := h.z.Refresh(context.Background()); err != nil || f.listCalls != 1 { + t.Fatal("list not fetched", err) + } + f.hang.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = h.z.Refresh(ctx) + f.hang.Store(false) + if err := h.z.Refresh(context.Background()); err != nil || f.calls != 0 || f.hung.Load() != 1 { + t.Fatal("abandoned boundary request retried inside the rate limit", err, f.calls) + } + if b := h.store.rows["YOW"]; b.NextAttempt.Sub(b.AttemptedAt) != 24*time.Hour { + t.Fatal("abandoned boundary request not persisted", b) + } +}