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

Preserve optional analytics partial ingestion

Browse Source
This commit is contained in:
Aman Tahiliani
2026-05-25 02:51:50 -04:00
parent 95060b07a5
commit 5eb0983ffb
2 changed files with 136 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
internal/ingest/ingest.go
View File

@@ -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) {

95
internal/ingest/ingest_test.go
View File

@@ -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"}