From 3b994e6510551d56ece5d3be47e398155a750b6d 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 e462401a55..1a27bfd502 100644 --- a/services/postprocessing/pkg/postprocessing/postprocessing.go +++ b/services/postprocessing/pkg/postprocessing/postprocessing.go @@ -22,6 +22,7 @@ type Postprocessing struct { Status Status Failures int InitiatorID string + Finished bool config config.Postprocessing } diff --git a/services/postprocessing/pkg/service/service.go b/services/postprocessing/pkg/service/service.go index 2be92d6fc8..95430e6189 100644 --- a/services/postprocessing/pkg/service/service.go +++ b/services/postprocessing/pkg/service/service.go @@ -150,8 +150,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 @@ -261,6 +267,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()) }