mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-16 03:00:29 +00:00
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.
This commit is contained in:
@@ -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{}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user