amantahiliani/box-box
1
0
Fork 0
mirror of https://github.com/AmanTahiliani/box-box.git synced 2026-08-07 19:56:18 -04:00
Code Issues Packages Projects Releases Wiki Activity

Tighten race result ingestion ordering

Browse Source
This commit is contained in:
Aman Tahiliani
2026-05-25 10:03:57 -04:00
parent 404029ff24
commit 17c94d83ad
9 changed files with 258 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
internal/ingest/ingest.go
View File

@@ -217,7 +217,7 @@ func (s *Service) IngestMeeting(meetingKey int) (Summary, error) {
partialSessions := 0
for _, sess := range sessions {
s.opts.Progress.Step("ingesting Race Hub datasets for session %d (%s)", sess.SessionKey, sess.SessionName)
sessSummary, err := s.ingestSessionDatasets(sess.SessionKey, meetingKey)
sessSummary, err := s.ingestSessionDatasets(sess)
ss := SessionSummary{
SessionKey: sess.SessionKey,
SessionName: sess.SessionName,
@@ -332,7 +332,7 @@ func (s *Service) IngestSession(sessionKey int) (Summary, error) {
summary.Sessions++
}
datasetSummary, err := s.ingestSessionDatasets(sessionKey, meetingKey)
datasetSummary, err := s.ingestSessionDatasets(sess)
summary.mergeCounts(datasetSummary)
summary.Errors = append(summary.Errors, datasetSummary.Errors...)
if err != nil {
@@ -345,7 +345,9 @@ func (s *Service) IngestSession(sessionKey int) (Summary, error) {
return summary, nil
}
func (s *Service) ingestSessionDatasets(sessionKey, meetingKey int) (Summary, error) {
func (s *Service) ingestSessionDatasets(sess models.Session) (Summary, error) {
sessionKey := sess.SessionKey
meetingKey := sess.MeetingKey
summary := Summary{
ScopeType: "session",
ScopeKey: fmt.Sprintf("%d", sessionKey),
@@ -410,37 +412,13 @@ func (s *Service) ingestSessionDatasets(sessionKey, meetingKey int) (Summary, er
}
s.delay()
s.opts.Progress.Step("fetching starting grid for session %d", sessionKey)
gridFetch, grid, err := fetchWithRetry(s, func() (FetchResult, []models.StartingGrid, error) {
return s.source.FetchStartingGrid(sessionKey)
})
if err != nil {
return summary, err
}
summary.RawPayloads++
if !s.opts.DryRun {
inserted, err := s.storeRaw(gridFetch, &meetingKey, &sk)
if err != nil {
return summary, err
}
if inserted {
summary.RawInserted++
}
for _, g := range grid {
if err := s.store.UpsertStartingGridEntry(startingGridToStore(g)); err != nil {
return summary, err
}
summary.StartingGrid++
}
} else {
summary.StartingGrid = len(grid)
}
s.delay()
optionalIngests := []struct {
name string
run func(*Summary, int, int) error
}{
{name: "starting_grid", run: func(summary *Summary, meetingKey, sessionKey int) error {
return s.ingestStartingGrid(summary, sess)
}},
{name: "stints", run: s.ingestStints},
{name: "pit_stops", run: s.ingestPitStops},
{name: "positions", run: s.ingestPositions},
@@ -534,6 +512,86 @@ func (s *Service) delay() {
}
}
func (s *Service) ingestStartingGrid(summary *Summary, sess models.Session) error {
sessionKey := sess.SessionKey
meetingKey := sess.MeetingKey
sourceSessionKey, err := s.startingGridSourceSessionKey(sess)
if err != nil {
return err
}
if sourceSessionKey == sessionKey {
s.opts.Progress.Step("fetching starting grid for session %d", sessionKey)
} else {
s.opts.Progress.Step("fetching starting grid for session %d from qualifying session %d", sessionKey, sourceSessionKey)
}
fetch, grid, err := fetchWithRetry(s, func() (FetchResult, []models.StartingGrid, error) {
return s.source.FetchStartingGrid(sourceSessionKey)
})
if err != nil {
return err
}
if sourceSessionKey != sessionKey {
fetch.RequestKey = fmt.Sprintf("%s;target_session_key=%d", fetch.RequestKey, sessionKey)
}
mk := meetingKey
sk := sessionKey
summary.RawPayloads++
if s.opts.DryRun {
summary.StartingGrid = len(grid)
s.delay()
return nil
}
inserted, err := s.storeRaw(fetch, &mk, &sk)
if err != nil {
return err
}
if inserted {
summary.RawInserted++
}
for _, g := range grid {
g.SessionKey = sessionKey
g.MeetingKey = meetingKey
if err := s.store.UpsertStartingGridEntry(startingGridToStore(g)); err != nil {
return err
}
summary.StartingGrid++
}
s.delay()
return nil
}
func (s *Service) startingGridSourceSessionKey(sess models.Session) (int, error) {
if !isRaceSession(sess) {
return sess.SessionKey, nil
}
_, sessions, err := fetchWithRetry(s, func() (FetchResult, []models.Session, error) {
return s.source.FetchSessionsForMeeting(sess.MeetingKey)
})
if err != nil {
return 0, err
}
s.delay()
for _, candidate := range sessions {
if isQualifyingSession(candidate) {
return candidate.SessionKey, nil
}
}
return 0, fmt.Errorf("no qualifying session found for meeting %d", sess.MeetingKey)
}
func isRaceSession(sess models.Session) bool {
return strings.EqualFold(sess.SessionType, "Race") || strings.EqualFold(sess.SessionName, "Race")
}
func isQualifyingSession(sess models.Session) bool {
return strings.EqualFold(sess.SessionType, "Qualifying") || strings.EqualFold(sess.SessionName, "Qualifying")
}
func (s *Service) ingestStints(summary *Summary, meetingKey, sessionKey int) error {
s.opts.Progress.Step("fetching stints for session %d", sessionKey)
fetch, stints, err := fetchWithRetry(s, func() (FetchResult, []models.Stint, error) {

66
internal/ingest/ingest_test.go
View File

@@ -199,6 +199,7 @@ func openTestStore(t *testing.T) *store.Store {
func testSessionFixtures() (int, int, *fakeSource) {
const meetingKey = 1229
const sessionKey = 9472
const qualifyingKey = 9471
src := newFakeSource()
src.meetingsByKey[meetingKey] = []models.Meeting{{
@@ -221,6 +222,16 @@ func testSessionFixtures() (int, int, *fakeSource) {
SessionType: "Race",
CircuitKey: 10,
}}
src.sessionsByMeeting[meetingKey] = []models.Session{
{
SessionKey: qualifyingKey,
MeetingKey: meetingKey,
SessionName: "Qualifying",
SessionType: "Qualifying",
CircuitKey: 10,
},
src.sessionsByKey[sessionKey][0],
}
src.drivers[sessionKey] = []models.Driver{
{
DriverNumber: 1,
@@ -243,9 +254,9 @@ func testSessionFixtures() (int, int, *fakeSource) {
{SessionKey: sessionKey, MeetingKey: meetingKey, DriverNumber: 1, Position: 1, Points: 25, NumberOfLaps: 78},
{SessionKey: sessionKey, MeetingKey: meetingKey, DriverNumber: 44, Position: 2, Points: 18, NumberOfLaps: 78, GapToLeader: 1.5},
}
src.grid[sessionKey] = []models.StartingGrid{
{SessionKey: sessionKey, MeetingKey: meetingKey, DriverNumber: 1, Position: 1, LapDuration: 71.234},
{SessionKey: sessionKey, MeetingKey: meetingKey, DriverNumber: 44, Position: 2, LapDuration: 71.456},
src.grid[qualifyingKey] = []models.StartingGrid{
{SessionKey: qualifyingKey, MeetingKey: meetingKey, DriverNumber: 1, Position: 1, LapDuration: 71.234},
{SessionKey: qualifyingKey, MeetingKey: meetingKey, DriverNumber: 44, Position: 2, LapDuration: 71.456},
}
src.stints[sessionKey] = []models.Stint{
{SessionKey: sessionKey, MeetingKey: meetingKey, DriverNumber: 1, StintNumber: 1, Compound: models.CompoundMedium, LapStart: 1, LapEnd: 30},
@@ -363,6 +374,11 @@ func TestIngestSessionWritesDomainAndRawRows(t *testing.T) {
if len(grid) != 2 {
t.Fatalf("starting grid = %d, want 2", len(grid))
}
for _, g := range grid {
if g.SessionKey != sessionKey {
t.Fatalf("starting grid row session_key = %d, want race session %d", g.SessionKey, sessionKey)
}
}
raw, err := st.ListRawPayloadsBySession(sessionKey)
if err != nil {
@@ -371,6 +387,18 @@ func TestIngestSessionWritesDomainAndRawRows(t *testing.T) {
if len(raw) != 11 {
t.Fatalf("raw payloads = %d, want 11", len(raw))
}
foundGridRaw := false
for _, p := range raw {
if p.Endpoint == "starting_grid" {
foundGridRaw = true
if p.RequestKey != "session_key=9471;target_session_key=9472" {
t.Fatalf("starting grid raw request key = %q, want qualifying source and race target", p.RequestKey)
}
}
}
if !foundGridRaw {
t.Fatal("raw payloads missing starting_grid")
}
stints, err := st.ListStints(sessionKey)
if err != nil || len(stints) != 2 {
@@ -521,6 +549,38 @@ func TestOptionalAnalyticsErrorMakesSessionPartialAndContinues(t *testing.T) {
}
}
func TestStartingGridErrorMakesSessionPartialAndContinues(t *testing.T) {
_, sessionKey, src := testSessionFixtures()
src.failOn = "starting_grid"
st := openTestStore(t)
opts := DefaultOptions()
opts.RequestDelay = 0
svc := NewService(st, src, opts)
summary, err := svc.IngestSession(sessionKey)
if err != nil {
t.Fatalf("IngestSession() error = %v, want nil partial result", err)
}
if summary.Status != "partial" {
t.Fatalf("summary.Status = %q, want partial", summary.Status)
}
if len(summary.Errors) != 1 {
t.Fatalf("summary.Errors = %v, want 1 starting_grid error", summary.Errors)
}
if summary.StartingGrid != 0 {
t.Fatalf("summary.StartingGrid = %d, want 0 for failed starting_grid", summary.StartingGrid)
}
if summary.Stints != 2 || summary.PitStops != 1 || summary.Positions != 3 || summary.RaceControl != 1 || summary.Weather != 1 || summary.Laps != 1 {
t.Fatalf("analytics counts after starting_grid failure = %+v, want remaining analytics preserved", summary)
}
stints, err := st.ListStints(sessionKey)
if err != nil || len(stints) != 2 {
t.Fatalf("stints = %+v, err = %v, want 2 preserved", stints, err)
}
}
func TestLiveSessionLockoutSurfacesControlledFailure(t *testing.T) {
_, sessionKey, src := testSessionFixtures()
src.liveLockout = true