Method running

modules/backendworker/backendworker.go:183–228

(ctx context.Context)

}

func (w *BackendWorker) running(ctx context.Context) error {
	level.Info(log.Logger).Log("msg", "backend worker running")

	b := backoff.New(ctx, w.cfg.Backoff)

	jobCtx := ctx
	if w.cfg.FinishOnShutdownTimeout > 0 {
		var jobsCancel context.CancelFunc
		jobCtx, jobsCancel = createShutdownContext(ctx, w.cfg.FinishOnShutdownTimeout)
		defer jobsCancel()
	}

	if w.subservices != nil {
		for {
			select {
			case <-ctx.Done():
				return nil
			case err := <-w.subservicesWatcher.Chan():
				return fmt.Errorf("worker subservices failed: %w", err)
			default:
				if err := w.processJobs(jobCtx); err != nil {
					level.Error(log.Logger).Log("msg", "error processing jobs", "err", err, "backoff", b.NextDelay())
					b.Wait()
					continue
				}

				b.Reset()
			}
		}
	} else {
		for {
			select {
			case <-ctx.Done():
				return nil
			default:
				if err := w.processJobs(jobCtx); err != nil {
					level.Error(log.Logger).Log("msg", "error processing jobs", "err", err, "backoff", b.NextDelay())
					b.Wait()
					continue
				}

				b.Reset()
			}
		}
	}
}

func (w *BackendWorker) processJobs(ctx context.Context) error {
	var (