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) + } +}