mirror of
https://github.com/AmanTahiliani/box-box.git
synced 2026-08-07 11:54:59 -04:00
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
107
internal/api/pacing_test.go
Normal file
107
internal/api/pacing_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user