mirror of
https://github.com/AmanTahiliani/box-box.git
synced 2026-08-07 19:56:18 -04:00
Parallelize championship hub fetches and cache aggregated response
Cold-cache hub loads previously issued ~70 sequential OpenF1 calls. Per-meeting fetches now fan out 5-wide (order preserved), and the aggregated response is cached in memory: 15min TTL for the current season, 24h for past seasons. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -646,12 +646,99 @@ type meetingRace struct {
|
|||||||
Grid []models.StartingGrid
|
Grid []models.StartingGrid
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// champHubWorkers bounds the concurrent per-meeting fetches for the hub.
|
||||||
|
const champHubWorkers = 5
|
||||||
|
|
||||||
|
// champHubCurrentTTL / champHubPastTTL control how long an aggregated hub
|
||||||
|
// response stays cached: short for the in-progress season, long for past
|
||||||
|
// seasons whose results are final.
|
||||||
|
const (
|
||||||
|
champHubCurrentTTL = 15 * time.Minute
|
||||||
|
champHubPastTTL = 24 * time.Hour
|
||||||
|
)
|
||||||
|
|
||||||
|
// champHubTTL returns the in-memory cache TTL for a season's hub response.
|
||||||
|
func champHubTTL(year int, now time.Time) time.Duration {
|
||||||
|
if year >= now.Year() {
|
||||||
|
return champHubCurrentTTL
|
||||||
|
}
|
||||||
|
return champHubPastTTL
|
||||||
|
}
|
||||||
|
|
||||||
|
type champHubEntry struct {
|
||||||
|
resp champHubResponse
|
||||||
|
expires time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// champHubCache is an in-memory cache of aggregated hub responses keyed by
|
||||||
|
// year. The zero value is ready to use.
|
||||||
|
type champHubCache struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
entries map[int]champHubEntry
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *champHubCache) get(year int, now time.Time) (champHubResponse, bool) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
e, ok := c.entries[year]
|
||||||
|
if !ok || now.After(e.expires) {
|
||||||
|
return champHubResponse{}, false
|
||||||
|
}
|
||||||
|
return e.resp, true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *champHubCache) put(year int, resp champHubResponse, now time.Time) {
|
||||||
|
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))}
|
||||||
|
}
|
||||||
|
|
||||||
|
// fetchMeetingRaces fans fetch out across meetings with bounded concurrency.
|
||||||
|
// The returned slice preserves the input meeting order regardless of
|
||||||
|
// completion order; meetings for which fetch reports ok=false are skipped.
|
||||||
|
func fetchMeetingRaces(meetings []models.Meeting, workers int, fetch func(models.Meeting) (meetingRace, bool)) []meetingRace {
|
||||||
|
if workers < 1 {
|
||||||
|
workers = 1
|
||||||
|
}
|
||||||
|
slots := make([]*meetingRace, len(meetings))
|
||||||
|
sem := make(chan struct{}, workers)
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i, m := range meetings {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
sem <- struct{}{}
|
||||||
|
defer func() { <-sem }()
|
||||||
|
if mr, ok := fetch(m); ok {
|
||||||
|
slots[i] = &mr
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
races := make([]meetingRace, 0, len(meetings))
|
||||||
|
for _, mr := range slots {
|
||||||
|
if mr != nil {
|
||||||
|
races = append(races, *mr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return races
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Server) handleChampionshipHub(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) handleChampionshipHub(w http.ResponseWriter, r *http.Request) {
|
||||||
year, _ := strconv.Atoi(r.URL.Query().Get("year"))
|
year, _ := strconv.Atoi(r.URL.Query().Get("year"))
|
||||||
if year == 0 {
|
if year == 0 {
|
||||||
year = time.Now().Year()
|
year = time.Now().Year()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if resp, ok := s.hubCache.get(year, time.Now()); ok {
|
||||||
|
writeJSON(w, resp)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
meetings, err := s.client.GetMeetingsForYear(year)
|
meetings, err := s.client.GetMeetingsForYear(year)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeError(w, err, http.StatusInternalServerError, s.client.LastResponseWasStale())
|
writeError(w, err, http.StatusInternalServerError, s.client.LastResponseWasStale())
|
||||||
@@ -676,11 +763,10 @@ 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 })
|
sort.Slice(meetings, func(i, j int) bool { return meetings[i].DateStart < meetings[j].DateStart })
|
||||||
|
|
||||||
races := make([]meetingRace, 0, len(meetings))
|
races := fetchMeetingRaces(meetings, champHubWorkers, func(m models.Meeting) (meetingRace, bool) {
|
||||||
for _, m := range meetings {
|
|
||||||
sessions, serr := s.client.GetSessionsForMeeting(int(m.MeetingKey))
|
sessions, serr := s.client.GetSessionsForMeeting(int(m.MeetingKey))
|
||||||
if serr != nil {
|
if serr != nil {
|
||||||
continue
|
return meetingRace{}, false
|
||||||
}
|
}
|
||||||
raceKey := 0
|
raceKey := 0
|
||||||
for _, sess := range sessions {
|
for _, sess := range sessions {
|
||||||
@@ -690,14 +776,16 @@ func (s *Server) handleChampionshipHub(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if raceKey == 0 {
|
if raceKey == 0 {
|
||||||
continue // not a GP meeting (e.g. pre-season testing)
|
return meetingRace{}, false // not a GP meeting (e.g. pre-season testing)
|
||||||
}
|
}
|
||||||
results, _ := s.client.GetSessionResult(raceKey)
|
results, _ := s.client.GetSessionResult(raceKey)
|
||||||
grid, _ := s.client.GetStartingGrid(raceKey)
|
grid, _ := s.client.GetStartingGrid(raceKey)
|
||||||
races = append(races, meetingRace{Meeting: m, RaceSessionKey: raceKey, Results: results, Grid: grid})
|
return meetingRace{Meeting: m, RaceSessionKey: raceKey, Results: results, Grid: grid}, true
|
||||||
}
|
})
|
||||||
|
|
||||||
writeJSON(w, aggregateChampionshipHub(year, races, champ, teams, driverInfo))
|
resp := aggregateChampionshipHub(year, races, champ, teams, driverInfo)
|
||||||
|
s.hubCache.put(year, resp, time.Now())
|
||||||
|
writeJSON(w, resp)
|
||||||
}
|
}
|
||||||
|
|
||||||
// aggregateChampionshipHub is the pure aggregation core (no network) so it can be
|
// aggregateChampionshipHub is the pure aggregation core (no network) so it can be
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
package web
|
package web
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/AmanTahiliani/box-box/internal/models"
|
"github.com/AmanTahiliani/box-box/internal/models"
|
||||||
)
|
)
|
||||||
@@ -140,3 +142,116 @@ func TestAggregateChampionshipHubEmpty(t *testing.T) {
|
|||||||
t.Errorf("empty aggregation should be zero-valued, got %+v", resp)
|
t.Errorf("empty aggregation should be zero-valued, got %+v", resp)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestFetchMeetingRacesPreservesOrderAndSkips(t *testing.T) {
|
||||||
|
const n = 12
|
||||||
|
meetings := make([]models.Meeting, n)
|
||||||
|
for i := range meetings {
|
||||||
|
meetings[i] = models.Meeting{MeetingKey: int32(i + 1)}
|
||||||
|
}
|
||||||
|
|
||||||
|
races := fetchMeetingRaces(meetings, 5, func(m models.Meeting) (meetingRace, bool) {
|
||||||
|
// Later meetings finish first to shuffle completion order.
|
||||||
|
time.Sleep(time.Duration(n-int(m.MeetingKey)) * time.Millisecond)
|
||||||
|
if m.MeetingKey%3 == 0 {
|
||||||
|
return meetingRace{}, false // simulate skip (fetch error / no Race session)
|
||||||
|
}
|
||||||
|
return meetingRace{Meeting: m, RaceSessionKey: int(m.MeetingKey) * 100}, true
|
||||||
|
})
|
||||||
|
|
||||||
|
want := 0
|
||||||
|
for i := 1; i <= n; i++ {
|
||||||
|
if i%3 != 0 {
|
||||||
|
want++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(races) != want {
|
||||||
|
t.Fatalf("races = %d, want %d", len(races), want)
|
||||||
|
}
|
||||||
|
prev := int32(0)
|
||||||
|
for _, r := range races {
|
||||||
|
if r.Meeting.MeetingKey <= prev {
|
||||||
|
t.Fatalf("races out of input order: key %d after %d", r.Meeting.MeetingKey, prev)
|
||||||
|
}
|
||||||
|
if r.Meeting.MeetingKey%3 == 0 {
|
||||||
|
t.Fatalf("skipped meeting %d present in output", r.Meeting.MeetingKey)
|
||||||
|
}
|
||||||
|
if r.RaceSessionKey != int(r.Meeting.MeetingKey)*100 {
|
||||||
|
t.Fatalf("meeting %d has mismatched race key %d", r.Meeting.MeetingKey, r.RaceSessionKey)
|
||||||
|
}
|
||||||
|
prev = r.Meeting.MeetingKey
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestFetchMeetingRacesBoundsConcurrency(t *testing.T) {
|
||||||
|
const workers = 3
|
||||||
|
var inFlight, peak atomic.Int32
|
||||||
|
meetings := make([]models.Meeting, 20)
|
||||||
|
for i := range meetings {
|
||||||
|
meetings[i] = models.Meeting{MeetingKey: int32(i + 1)}
|
||||||
|
}
|
||||||
|
|
||||||
|
fetchMeetingRaces(meetings, workers, func(m models.Meeting) (meetingRace, bool) {
|
||||||
|
cur := inFlight.Add(1)
|
||||||
|
for {
|
||||||
|
p := peak.Load()
|
||||||
|
if cur <= p || peak.CompareAndSwap(p, cur) {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
time.Sleep(5 * time.Millisecond)
|
||||||
|
inFlight.Add(-1)
|
||||||
|
return meetingRace{Meeting: m, RaceSessionKey: 1}, true
|
||||||
|
})
|
||||||
|
|
||||||
|
if p := peak.Load(); p > workers {
|
||||||
|
t.Errorf("peak concurrent fetches = %d, want <= %d", p, workers)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChampHubTTL(t *testing.T) {
|
||||||
|
now := time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC)
|
||||||
|
if got := champHubTTL(2026, now); got != champHubCurrentTTL {
|
||||||
|
t.Errorf("current year TTL = %v, want %v", got, champHubCurrentTTL)
|
||||||
|
}
|
||||||
|
if got := champHubTTL(2027, now); got != champHubCurrentTTL {
|
||||||
|
t.Errorf("future year TTL = %v, want %v", got, champHubCurrentTTL)
|
||||||
|
}
|
||||||
|
if got := champHubTTL(2024, now); got != champHubPastTTL {
|
||||||
|
t.Errorf("past year TTL = %v, want %v", got, champHubPastTTL)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChampHubCache(t *testing.T) {
|
||||||
|
var c champHubCache
|
||||||
|
now := time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC)
|
||||||
|
|
||||||
|
if _, ok := c.get(2026, now); ok {
|
||||||
|
t.Fatal("empty cache should miss")
|
||||||
|
}
|
||||||
|
|
||||||
|
c.put(2026, champHubResponse{Season: 2026, Round: 10}, now)
|
||||||
|
c.put(2024, champHubResponse{Season: 2024, Round: 24}, 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 {
|
||||||
|
t.Errorf("current-year get within TTL = (%+v, %v), want hit with Round 10", resp, ok)
|
||||||
|
}
|
||||||
|
if _, ok := c.get(2026, now.Add(16*time.Minute)); ok {
|
||||||
|
t.Error("current-year entry should expire after 15 minutes")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Past-year entry: hit well beyond 15 min, miss after 24 h.
|
||||||
|
if resp, ok := c.get(2024, now.Add(12*time.Hour)); !ok || resp.Round != 24 {
|
||||||
|
t.Errorf("past-year get within TTL = (%+v, %v), want hit with Round 24", resp, ok)
|
||||||
|
}
|
||||||
|
if _, ok := c.get(2024, now.Add(25*time.Hour)); ok {
|
||||||
|
t.Error("past-year entry should expire after 24 hours")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Re-put refreshes the entry.
|
||||||
|
c.put(2026, champHubResponse{Season: 2026, Round: 11}, now.Add(20*time.Minute))
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -20,10 +20,11 @@ var assetsFS embed.FS
|
|||||||
|
|
||||||
// Server is the box-box web companion HTTP server.
|
// Server is the box-box web companion HTTP server.
|
||||||
type Server struct {
|
type Server struct {
|
||||||
client *api.OpenF1Client
|
client *api.OpenF1Client
|
||||||
query *query.Service
|
query *query.Service
|
||||||
hub *SSEHub
|
hub *SSEHub
|
||||||
addr string
|
addr string
|
||||||
|
hubCache champHubCache // aggregated championship hub responses, keyed by year
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewServer creates a new Server. Call Start() to begin serving.
|
// NewServer creates a new Server. Call Start() to begin serving.
|
||||||
|
|||||||
Reference in New Issue
Block a user