From caa83ab952b5d4694282536dd7694184276d0c6f Mon Sep 17 00:00:00 2001 From: jkoberg Date: Wed, 24 Apr 2024 11:56:30 +0200 Subject: [PATCH] feat(postprocessing): dont retry finished uploads Signed-off-by: jkoberg --- .../pkg/postprocessing/postprocessing.go | 1 + services/postprocessing/pkg/service/service.go | 15 +++++++++++++-- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/services/postprocessing/pkg/postprocessing/postprocessing.go b/services/postprocessing/pkg/postprocessing/postprocessing.go index 40b94d56f..d1efcc2bb 100644 --- a/services/postprocessing/pkg/postprocessing/postprocessing.go +++ b/services/postprocessing/pkg/postprocessing/postprocessing.go @@ -21,6 +21,7 @@ type Postprocessing struct { Steps []events.Postprocessingstep Status Status Failures int + Finished bool config config.Postprocessing } diff --git a/services/postprocessing/pkg/service/service.go b/services/postprocessing/pkg/service/service.go index 70e1221a5..4b38e6f3f 100644 --- a/services/postprocessing/pkg/service/service.go +++ b/services/postprocessing/pkg/service/service.go @@ -148,8 +148,14 @@ func (pps *PostprocessingService) processEvent(e events.Event) error { next = pp.Delay() case events.UploadReady: if ev.Failed { - // the upload failed - let's keep it around for a while - return nil + // the upload failed - let's keep it around for a while - but mark it as finished + pp, err = pps.getPP(pps.store, ev.UploadID) + if err != nil { + pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot get upload") + return fmt.Errorf("%w: cannot get upload", ErrEvent) + } + pp.Finished = true + return storePP(pps.store, pp) } // the storage provider thinks the upload is done - so no need to keep it any more @@ -256,6 +262,11 @@ func (pps *PostprocessingService) resumePP(ctx context.Context, uploadID string) return fmt.Errorf("cannot get upload: %w", err) } + if pp.Finished { + // dont retry finished uploads + return nil + } + return events.Publish(ctx, pps.pub, pp.CurrentStep()) }