@@ -95,7 +95,9 @@ func (s *Service) HydrateThread(ctx context.Context, repo contracts.RepoRef, num
9595 }
9696 cleanupCtx , cancel := context .WithTimeout (context .WithoutCancel (ctx ), 5 * time .Second )
9797 defer cancel ()
98- _ = c .FailRun (cleanupCtx , run .ID , hydrateErr .Error ())
98+ _ = corpus .RetryBusy (cleanupCtx , func (ctx context.Context ) error {
99+ return c .FailRun (ctx , run .ID , hydrateErr .Error ())
100+ })
99101 }()
100102
101103 repoProjection , err := c .GetRepository (ctx , ref .Owner , ref .Repo )
@@ -194,7 +196,9 @@ func (s *Service) HydrateThread(ctx context.Context, repo contracts.RepoRef, num
194196 "pages" : result .Pages ,
195197 "requests" : result .Requests ,
196198 })
197- if err := c .FinishRun (ctx , run .ID , string (statsPayload )); err != nil {
199+ if err := corpus .RetryBusy (ctx , func (ctx context.Context ) error {
200+ return c .FinishRun (ctx , run .ID , string (statsPayload ))
201+ }); err != nil {
198202 hydrateErr = err
199203 return nil , hydrateErr
200204 }
@@ -285,13 +289,17 @@ func (f *facetRunner) hydrateIssueTimeline() (HydratedFacet, error) {
285289 opts .Page = res .Page .NextPage
286290 }
287291 if ! complete {
288- if _ , err := f .c .AdvanceFacetCAS (f .ctx , f .repoID , & f .threadID , FacetIssueTimeline , sourceUpdatedAt , false , f .runID , expectedSequence ); err != nil {
292+ if _ , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (bool , error ) {
293+ return f .c .AdvanceFacetCAS (ctx , f .repoID , & f .threadID , FacetIssueTimeline , sourceUpdatedAt , false , f .runID , expectedSequence )
294+ }); err != nil {
289295 return HydratedFacet {}, err
290296 }
291297 return HydratedFacet {Facet : FacetIssueTimeline , Count : total , Pages : pages , Complete : false }, nil
292298 }
293299 collapseFacetSearchText (pageObservations )
294- applied , err := f .c .ApplyFacetObservationSetCAS (f .ctx , f .repoID , & f .threadID , FacetIssueTimeline , sourceUpdatedAt , pageObservations , true , f .runID , expectedSequence )
300+ applied , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (bool , error ) {
301+ return f .c .ApplyFacetObservationSetCAS (ctx , f .repoID , & f .threadID , FacetIssueTimeline , sourceUpdatedAt , pageObservations , true , f .runID , expectedSequence )
302+ })
295303 if err != nil {
296304 return HydratedFacet {}, err
297305 }
@@ -356,7 +364,9 @@ func (f *facetRunner) persistTimelineResolution(events []github.IssueTimelineEve
356364 return errors .New ("closing commit timeline observation is unavailable" )
357365 }
358366 }
359- _ , err := f .c .SaveResolutionRecord (f .ctx , corpus.ResolutionRecord {ThreadID : f .threadID , Kind : kind , Summary : summary , RuleVersion : "resolution.v1" , SourceUpdatedAt : sourceUpdatedAt , SourceObservationRefs : refs })
367+ _ , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (* corpus.ResolutionRecord , error ) {
368+ return f .c .SaveResolutionRecord (ctx , corpus.ResolutionRecord {ThreadID : f .threadID , Kind : kind , Summary : summary , RuleVersion : "resolution.v1" , SourceUpdatedAt : sourceUpdatedAt , SourceObservationRefs : refs })
369+ })
360370 return err
361371}
362372
@@ -427,13 +437,17 @@ func hydratePaginatedFacet[T any](f *facetRunner, spec paginatedFacetSpec[T]) (H
427437 return HydratedFacet {}, err
428438 }
429439 if ! complete {
430- if _ , err := f .c .AdvanceFacetCAS (f .ctx , f .repoID , & f .threadID , spec .facet , sourceUpdatedAt , false , f .runID , expectedSequence ); err != nil {
440+ if _ , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (bool , error ) {
441+ return f .c .AdvanceFacetCAS (ctx , f .repoID , & f .threadID , spec .facet , sourceUpdatedAt , false , f .runID , expectedSequence )
442+ }); err != nil {
431443 return HydratedFacet {}, err
432444 }
433445 return HydratedFacet {Facet : spec .facet , Count : total , Pages : pages , Complete : false }, nil
434446 }
435447 collapseFacetSearchText (pageObservations )
436- if _ , err := f .c .ApplyFacetObservationSetCAS (f .ctx , f .repoID , & f .threadID , spec .facet , sourceUpdatedAt , pageObservations , true , f .runID , expectedSequence ); err != nil {
448+ if _ , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (bool , error ) {
449+ return f .c .ApplyFacetObservationSetCAS (ctx , f .repoID , & f .threadID , spec .facet , sourceUpdatedAt , pageObservations , true , f .runID , expectedSequence )
450+ }); err != nil {
437451 return HydratedFacet {}, err
438452 }
439453 return HydratedFacet {Facet : spec .facet , Count : total , Pages : pages , Complete : true }, nil
@@ -482,7 +496,9 @@ func (f *facetRunner) hydratePullRequestDetails() (HydratedFacet, error) {
482496 }
483497
484498 pages := []corpus.FacetObservationInput {{SourceUpdatedAt : updatedAt , Payload : string (payload )}}
485- applied , err := f .c .ApplyFacetObservationSetCAS (f .ctx , f .repoID , & f .threadID , FacetPRDetails , updatedAt , pages , true , f .runID , expectedSequence )
499+ applied , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (bool , error ) {
500+ return f .c .ApplyFacetObservationSetCAS (ctx , f .repoID , & f .threadID , FacetPRDetails , updatedAt , pages , true , f .runID , expectedSequence )
501+ })
486502 if err != nil {
487503 return HydratedFacet {}, err
488504 }
@@ -516,7 +532,9 @@ func (f *facetRunner) hydratePullRequestDetails() (HydratedFacet, error) {
516532 } else {
517533 projection .MergedAt = time.Time {}
518534 }
519- stored , err := f .c .UpsertThread (f .ctx , projection , string (payload ))
535+ stored , err := corpus .RetryBusyValue (f .ctx , func (ctx context.Context ) (* corpus.Thread , error ) {
536+ return f .c .UpsertThread (ctx , projection , string (payload ))
537+ })
520538 if err != nil {
521539 return HydratedFacet {}, fmt .Errorf ("project pr details: %w" , err )
522540 }
0 commit comments