From 7f0c338a57312fa2d57a3a182fd96cfca117c958 Mon Sep 17 00:00:00 2001 From: AmanTahiliani Date: Fri, 3 Jul 2026 00:47:52 -0400 Subject: [PATCH] Add OpenF1 request pacing, 429 retry, and hub completeness guard Concurrent fan-outs could burst past the free-tier rate limit; callers swallowed the 429s as missing data, and the championship hub cached the resulting partial season for its full TTL (observed live: 8 of 22 rounds, zero wins/podiums/form). - requestPacer spaces live requests (350ms anonymous, 100ms with key); cache hits never wait - get() retries 429s up to 3 times honouring Retry-After (capped 10s) - hub tracks per-meeting fetch failures and caches incomplete aggregates for only 2 minutes instead of 15min/24h Verified against the live API: 2026 hub now reports all 22 rounds with wins/podiums/form populated. Co-Authored-By: Claude Fable 5 --- internal/api/client.go | 39 ++++++++++ internal/api/openf1.go | 51 +++++++++--- internal/api/pacing_test.go | 107 ++++++++++++++++++++++++++ internal/web/api.go | 26 +++++-- internal/web/championship_hub_test.go | 15 +++- 5 files changed, 221 insertions(+), 17 deletions(-) create mode 100644 internal/api/pacing_test.go diff --git a/internal/api/client.go b/internal/api/client.go index 712d8a9..fe04625 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -2,15 +2,52 @@ package api import ( "net/http" + "sync" "sync/atomic" "time" ) +// Minimum spacing between live OpenF1 requests. The free tier throttles +// bursts of more than ~3 requests/second, so anonymous clients are paced +// conservatively; authenticated (paid tier) clients get a higher rate. +const ( + anonRequestInterval = 350 * time.Millisecond + authRequestInterval = 100 * time.Millisecond +) + +// requestPacer spaces network requests evenly so concurrent callers +// (e.g. the championship hub fan-out) cannot burst past the API rate limit. +// Cache hits never touch the pacer. +type requestPacer struct { + mu sync.Mutex + interval time.Duration + next time.Time +} + +// wait blocks until this caller's reserved slot arrives. +func (p *requestPacer) wait() { + if p == nil || p.interval <= 0 { + return + } + p.mu.Lock() + now := time.Now() + if p.next.Before(now) { + p.next = now + } + sleep := p.next.Sub(now) + p.next = p.next.Add(p.interval) + p.mu.Unlock() + if sleep > 0 { + time.Sleep(sleep) + } +} + type OpenF1Client struct { url string apiKey string httpClient *http.Client cache *Cache + pacer *requestPacer // staleFlag is set to 1 atomically whenever a request falls back to stale // cached data (e.g. because the API is locked during a live session). @@ -24,6 +61,7 @@ func NewOpenF1Client(url string, timeout time.Duration) *OpenF1Client { url: url, httpClient: &http.Client{Timeout: timeout}, cache: NewCache(), + pacer: &requestPacer{interval: anonRequestInterval}, } } @@ -35,6 +73,7 @@ func NewOpenF1ClientWithKey(url string, timeout time.Duration, apiKey string) *O apiKey: apiKey, httpClient: &http.Client{Timeout: timeout}, cache: NewCache(), + pacer: &requestPacer{interval: authRequestInterval}, } } diff --git a/internal/api/openf1.go b/internal/api/openf1.go index 1d36234..c55ef79 100644 --- a/internal/api/openf1.go +++ b/internal/api/openf1.go @@ -35,6 +35,45 @@ func IsLiveSessionError(err error) bool { return errors.Is(err, ErrLiveSessionLocked) } +// max429Retries is how many times doPaced re-attempts a request that came +// back 429 before giving up and letting the caller fall back to stale data. +const max429Retries = 3 + +// retryAfter429 extracts the server-requested backoff from a 429 response, +// with a sane default and cap so a hostile header can't stall the app. +func retryAfter429(resp *http.Response) time.Duration { + delay := time.Second + if header := resp.Header.Get("Retry-After"); header != "" { + if seconds, err := strconv.Atoi(header); err == nil && seconds > 0 { + delay = time.Duration(seconds) * time.Second + } + } + if delay > 10*time.Second { + delay = 10 * time.Second + } + return delay +} + +// doPaced executes req through the client's request pacer and transparently +// retries 429 responses (honouring Retry-After) up to max429Retries times. +// Without this, concurrent fan-outs (championship hub, track prefetch) burst +// past the free-tier limit and callers silently treat 429s as missing data. +func (c *OpenF1Client) doPaced(req *http.Request) (*http.Response, error) { + for attempt := 0; ; attempt++ { + c.pacer.wait() + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusTooManyRequests || attempt >= max429Retries { + return resp, nil + } + delay := retryAfter429(resp) + resp.Body.Close() + time.Sleep(delay) + } +} + // get performs a GET request and returns the response body, or an error if the // status code is not 200 OK. It checks the SQLite cache before making a // network request. @@ -59,7 +98,7 @@ func (c *OpenF1Client) get(url string) (io.ReadCloser, error) { req.Header.Set("Authorization", "Bearer "+c.apiKey) } - resp, err := c.httpClient.Do(req) + resp, err := c.doPaced(req) if err != nil { return c.tryStale(url, err) } @@ -114,20 +153,14 @@ func (c *OpenF1Client) FetchStrict(url string) ([]byte, error) { req.Header.Set("Authorization", "Bearer "+c.apiKey) } - resp, err := c.httpClient.Do(req) + resp, err := c.doPaced(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode == http.StatusTooManyRequests { - retryAfterDur := 500 * time.Millisecond - if retryAfterHeader := resp.Header.Get("Retry-After"); retryAfterHeader != "" { - if seconds, err := strconv.Atoi(retryAfterHeader); err == nil { - retryAfterDur = time.Duration(seconds) * time.Second - } - } - return nil, &RateLimitError{RetryAfter: retryAfterDur} + return nil, &RateLimitError{RetryAfter: retryAfter429(resp)} } data, err := io.ReadAll(resp.Body) diff --git a/internal/api/pacing_test.go b/internal/api/pacing_test.go new file mode 100644 index 0000000..284f819 --- /dev/null +++ b/internal/api/pacing_test.go @@ -0,0 +1,107 @@ +package api + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" +) + +// newPacedTestClient builds a client against a test server with an isolated cache +// (HOME is pointed at a temp dir so the SQLite cache never touches the real one). +func newPacedTestClient(t *testing.T, srvURL string, interval time.Duration) *OpenF1Client { + t.Helper() + t.Setenv("HOME", t.TempDir()) + t.Setenv("XDG_CACHE_HOME", t.TempDir()) + c := NewOpenF1Client(srvURL, 5*time.Second) + c.pacer = &requestPacer{interval: interval} + t.Cleanup(func() { _ = c.Close() }) + return c +} + +func TestRequestPacerSpacesConcurrentCallers(t *testing.T) { + p := &requestPacer{interval: 20 * time.Millisecond} + const callers = 5 + + start := time.Now() + var wg sync.WaitGroup + for i := 0; i < callers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + p.wait() + }() + } + wg.Wait() + + // 5 callers at 20ms spacing: the last slot is 80ms after the first. + if elapsed := time.Since(start); elapsed < 4*20*time.Millisecond { + t.Fatalf("pacer did not space callers: %d finished in %v", callers, elapsed) + } +} + +func TestRequestPacerNilSafe(t *testing.T) { + var p *requestPacer + p.wait() // must not panic +} + +func TestGetRetriesOn429(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) <= 2 { + w.Header().Set("Retry-After", "0") + w.WriteHeader(http.StatusTooManyRequests) + return + } + _ = json.NewEncoder(w).Encode([]map[string]any{{"meeting_key": 1}}) + })) + defer srv.Close() + + c := newPacedTestClient(t, srv.URL, time.Millisecond) + body, err := c.get(srv.URL + "/v1/meetings") + if err != nil { + t.Fatalf("get after 429s should succeed, got %v", err) + } + data, _ := io.ReadAll(body) + body.Close() + if len(data) == 0 { + t.Fatal("expected response body after retries") + } + if got := calls.Load(); got != 3 { + t.Fatalf("expected 3 attempts (2×429 + 1×200), got %d", got) + } +} + +func TestGetGivesUpAfterMaxRetries(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Header().Set("Retry-After", "0") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + + c := newPacedTestClient(t, srv.URL, time.Millisecond) + _, err := c.get(srv.URL + "/v1/meetings") + if err == nil { + t.Fatal("expected error when server keeps returning 429") + } + if got := calls.Load(); got != int32(max429Retries)+1 { + t.Fatalf("expected %d attempts, got %d", max429Retries+1, got) + } +} + +func TestRetryAfter429Cap(t *testing.T) { + resp := &http.Response{Header: http.Header{"Retry-After": []string{"3600"}}} + if got := retryAfter429(resp); got != 10*time.Second { + t.Fatalf("expected 10s cap, got %v", got) + } + resp = &http.Response{Header: http.Header{}} + if got := retryAfter429(resp); got != time.Second { + t.Fatalf("expected 1s default, got %v", got) + } +} diff --git a/internal/web/api.go b/internal/web/api.go index f7507d1..57cd5c1 100644 --- a/internal/web/api.go +++ b/internal/web/api.go @@ -11,6 +11,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" readability "codeberg.org/readeck/go-readability/v2" @@ -655,6 +656,9 @@ const champHubWorkers = 5 const ( champHubCurrentTTL = 15 * time.Minute champHubPastTTL = 24 * time.Hour + // champHubIncompleteTTL keeps a partially-fetched aggregate around just + // long enough to absorb page-load bursts while retrying soon after. + champHubIncompleteTTL = 2 * time.Minute ) // champHubTTL returns the in-memory cache TTL for a season's hub response. @@ -687,13 +691,13 @@ func (c *champHubCache) get(year int, now time.Time) (champHubResponse, bool) { return e.resp, true } -func (c *champHubCache) put(year int, resp champHubResponse, now time.Time) { +func (c *champHubCache) put(year int, resp champHubResponse, now time.Time, ttl time.Duration) { c.mu.Lock() defer c.mu.Unlock() if c.entries == nil { c.entries = map[int]champHubEntry{} } - c.entries[year] = champHubEntry{resp: resp, expires: now.Add(champHubTTL(year, now))} + c.entries[year] = champHubEntry{resp: resp, expires: now.Add(ttl)} } // fetchMeetingRaces fans fetch out across meetings with bounded concurrency. @@ -763,9 +767,14 @@ func (s *Server) handleChampionshipHub(w http.ResponseWriter, r *http.Request) { sort.Slice(meetings, func(i, j int) bool { return meetings[i].DateStart < meetings[j].DateStart }) + // Any per-meeting fetch failure (network, rate limit) yields an incomplete + // aggregate: serve it so the page still renders, but cache it only briefly + // so a partial view of the season doesn't stick around for the full TTL. + var incomplete atomic.Bool races := fetchMeetingRaces(meetings, champHubWorkers, func(m models.Meeting) (meetingRace, bool) { sessions, serr := s.client.GetSessionsForMeeting(int(m.MeetingKey)) if serr != nil { + incomplete.Store(true) return meetingRace{}, false } raceKey := 0 @@ -778,13 +787,20 @@ func (s *Server) handleChampionshipHub(w http.ResponseWriter, r *http.Request) { if raceKey == 0 { return meetingRace{}, false // not a GP meeting (e.g. pre-season testing) } - results, _ := s.client.GetSessionResult(raceKey) - grid, _ := s.client.GetStartingGrid(raceKey) + results, rerr := s.client.GetSessionResult(raceKey) + grid, gerr := s.client.GetStartingGrid(raceKey) + if rerr != nil || gerr != nil { + incomplete.Store(true) + } return meetingRace{Meeting: m, RaceSessionKey: raceKey, Results: results, Grid: grid}, true }) resp := aggregateChampionshipHub(year, races, champ, teams, driverInfo) - s.hubCache.put(year, resp, time.Now()) + ttl := champHubTTL(year, time.Now()) + if incomplete.Load() { + ttl = champHubIncompleteTTL + } + s.hubCache.put(year, resp, time.Now(), ttl) writeJSON(w, resp) } diff --git a/internal/web/championship_hub_test.go b/internal/web/championship_hub_test.go index 516753f..f1721e6 100644 --- a/internal/web/championship_hub_test.go +++ b/internal/web/championship_hub_test.go @@ -230,8 +230,8 @@ func TestChampHubCache(t *testing.T) { t.Fatal("empty cache should miss") } - c.put(2026, champHubResponse{Season: 2026, Round: 10}, now) - c.put(2024, champHubResponse{Season: 2024, Round: 24}, now) + c.put(2026, champHubResponse{Season: 2026, Round: 10}, now, champHubTTL(2026, now)) + c.put(2024, champHubResponse{Season: 2024, Round: 24}, now, champHubTTL(2024, now)) // Current-year entry: hit within 15 min, miss after. if resp, ok := c.get(2026, now.Add(14*time.Minute)); !ok || resp.Round != 10 { @@ -250,8 +250,17 @@ func TestChampHubCache(t *testing.T) { } // Re-put refreshes the entry. - c.put(2026, champHubResponse{Season: 2026, Round: 11}, now.Add(20*time.Minute)) + c.put(2026, champHubResponse{Season: 2026, Round: 11}, now.Add(20*time.Minute), champHubTTL(2026, now)) if resp, ok := c.get(2026, now.Add(30*time.Minute)); !ok || resp.Round != 11 { t.Errorf("refreshed entry = (%+v, %v), want hit with Round 11", resp, ok) } + + // Incomplete aggregates get the short TTL: hit right away, gone in 3 min. + c.put(2023, champHubResponse{Season: 2023, Round: 5}, now, champHubIncompleteTTL) + if _, ok := c.get(2023, now.Add(time.Minute)); !ok { + t.Error("incomplete entry should hit within its short TTL") + } + if _, ok := c.get(2023, now.Add(3*time.Minute)); ok { + t.Error("incomplete entry should expire after champHubIncompleteTTL") + } }