diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 7d02900..c7fdca5 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -214,6 +214,7 @@ func (s *Service) IngestMeeting(meetingKey int) (Summary, error) { } sessionFailures := 0 + 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) @@ -230,11 +231,19 @@ func (s *Service) IngestMeeting(meetingKey int) (Summary, error) { "session %d (%s): %v", sess.SessionKey, sess.SessionName, err, )) } + if err == nil && sessSummary.Status == "partial" { + partialSessions++ + for _, partialErr := range sessSummary.Errors { + summary.Errors = append(summary.Errors, fmt.Sprintf( + "session %d (%s): %s", sess.SessionKey, sess.SessionName, partialErr, + )) + } + } summary.SessionSummaries = append(summary.SessionSummaries, ss) summary.mergeCounts(sessSummary) } - summary.Status = meetingStatus(sessionFailures, len(sessions), s.opts.DryRun) + summary.Status = meetingStatus(sessionFailures, partialSessions, len(sessions), s.opts.DryRun) s.finishRun(runID, summary) s.opts.Progress.Summary(summary) @@ -325,11 +334,12 @@ func (s *Service) IngestSession(sessionKey int) (Summary, error) { datasetSummary, err := s.ingestSessionDatasets(sessionKey, meetingKey) summary.mergeCounts(datasetSummary) + summary.Errors = append(summary.Errors, datasetSummary.Errors...) if err != nil { return s.finishFailed(runID, summary, err) } - summary.Status = statusForDryRun(s.opts.DryRun) + summary.Status = datasetSummary.Status s.finishRun(runID, summary) s.opts.Progress.Summary(summary) return summary, nil @@ -427,26 +437,24 @@ func (s *Service) ingestSessionDatasets(sessionKey, meetingKey int) (Summary, er } s.delay() - if err := s.ingestStints(&summary, meetingKey, sk); err != nil { - return summary, err + optionalIngests := []struct { + name string + run func(*Summary, int, int) error + }{ + {name: "stints", run: s.ingestStints}, + {name: "pit_stops", run: s.ingestPitStops}, + {name: "positions", run: s.ingestPositions}, + {name: "race_control", run: s.ingestRaceControl}, + {name: "weather", run: s.ingestWeather}, + {name: "laps", run: s.ingestLaps}, } - if err := s.ingestPitStops(&summary, meetingKey, sk); err != nil { - return summary, err - } - if err := s.ingestPositions(&summary, meetingKey, sk); err != nil { - return summary, err - } - if err := s.ingestRaceControl(&summary, meetingKey, sk); err != nil { - return summary, err - } - if err := s.ingestWeather(&summary, meetingKey, sk); err != nil { - return summary, err - } - if err := s.ingestLaps(&summary, meetingKey, sk); err != nil { - return summary, err + for _, optional := range optionalIngests { + if err := optional.run(&summary, meetingKey, sk); err != nil { + summary.Errors = append(summary.Errors, fmt.Sprintf("%s: %v", optional.name, err)) + } } - summary.Status = statusForDryRun(s.opts.DryRun) + summary.Status = statusForErrors(s.opts.DryRun, summary.Errors) return summary, nil } @@ -464,11 +472,14 @@ func (s *Summary) mergeCounts(other Summary) { s.RawInserted += other.RawInserted } -func meetingStatus(sessionFailures, sessionTotal int, dryRun bool) string { +func meetingStatus(sessionFailures, partialSessions, sessionTotal int, dryRun bool) string { if dryRun { return "dry_run" } if sessionFailures == 0 { + if partialSessions > 0 { + return "partial" + } return "completed" } if sessionFailures == sessionTotal { @@ -674,6 +685,16 @@ func statusForDryRun(dryRun bool) string { return "completed" } +func statusForErrors(dryRun bool, errs []string) string { + if dryRun { + return "dry_run" + } + if len(errs) > 0 { + return "partial" + } + return "completed" +} + type fetchFunc[T any] func() (FetchResult, T, error) func fetchWithRetry[T any](s *Service, fn fetchFunc[T]) (FetchResult, T, error) { diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 951bb59..a0f5354 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -476,6 +476,51 @@ func TestSourceErrorStopsRun(t *testing.T) { } } +func TestOptionalAnalyticsErrorMakesSessionPartialAndContinues(t *testing.T) { + _, sessionKey, src := testSessionFixtures() + src.failOn = "stints" + 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 optional error", summary.Errors) + } + if summary.Drivers != 2 || summary.SessionResults != 2 || summary.StartingGrid != 2 { + t.Fatalf("core counts = %+v, want drivers/results/grid preserved", summary) + } + if summary.Stints != 0 { + t.Fatalf("summary.Stints = %d, want 0 for failed optional dataset", summary.Stints) + } + if summary.PitStops != 1 || summary.Positions != 3 || summary.RaceControl != 1 || summary.Weather != 1 || summary.Laps != 1 { + t.Fatalf("optional counts after stints failure = %+v, want remaining analytics preserved", summary) + } + if summary.RawPayloads != 10 { + t.Fatalf("summary.RawPayloads = %d, want 10 successful payloads", summary.RawPayloads) + } + + stints, err := st.ListStints(sessionKey) + if err != nil { + t.Fatalf("ListStints() error = %v", err) + } + if len(stints) != 0 { + t.Fatalf("stints after failure = %d, want 0", len(stints)) + } + positions, err := st.ListPositionSamples(sessionKey) + if err != nil || len(positions) != 3 { + t.Fatalf("positions = %+v, err = %v, want 3 preserved", positions, err) + } +} + func TestLiveSessionLockoutSurfacesControlledFailure(t *testing.T) { _, sessionKey, src := testSessionFixtures() src.liveLockout = true @@ -567,6 +612,56 @@ func TestIngestMeetingWeekendIngestsAllSessions(t *testing.T) { } } +func TestIngestMeetingOptionalAnalyticsPartialDoesNotFailMeeting(t *testing.T) { + meetingKey, qualiKey, raceKey, src := testWeekendFixtures() + src.failSession = map[int]string{raceKey: "weather"} + st := openTestStore(t) + + opts := DefaultOptions() + opts.RequestDelay = 0 + svc := NewService(st, src, opts) + + summary, err := svc.IngestMeeting(meetingKey) + if err != nil { + t.Fatalf("IngestMeeting() error = %v, want nil for optional partial", 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 optional error", summary.Errors) + } + if len(summary.SessionSummaries) != 2 { + t.Fatalf("session summaries = %d, want 2", len(summary.SessionSummaries)) + } + + var raceSummary Summary + for _, ss := range summary.SessionSummaries { + if ss.SessionKey == raceKey { + raceSummary = ss.Summary + } + } + if raceSummary.Status != "partial" { + t.Fatalf("race session summary status = %q, want partial", raceSummary.Status) + } + if len(raceSummary.Errors) != 1 { + t.Fatalf("race session errors = %v, want 1 optional error", raceSummary.Errors) + } + + qualiDrivers, err := st.ListSessionDrivers(qualiKey) + if err != nil || len(qualiDrivers) != 2 { + t.Fatalf("quali drivers = %+v, err = %v, want 2", qualiDrivers, err) + } + raceDrivers, err := st.ListSessionDrivers(raceKey) + if err != nil || len(raceDrivers) != 2 { + t.Fatalf("race drivers = %+v, err = %v, want 2", raceDrivers, err) + } + positions, err := st.ListPositionSamples(raceKey) + if err != nil || len(positions) != 3 { + t.Fatalf("race positions = %+v, err = %v, want 3 preserved", positions, err) + } +} + func TestIngestMeetingPartialFailurePreservesSuccessfulSessions(t *testing.T) { meetingKey, qualiKey, raceKey, src := testWeekendFixtures() src.failSession = map[int]string{qualiKey: "session_result"}