worker/server: update metrics on requeue

When requeuing a job the next worker requesting the job would decrement
pending counter, but the pending counter only ever got incremented once,
when the job was first enqueued. Thus make sure to increment the pending
counter when a job is requeued.

internal/prometheus/job_metrics.go

@@ -68,6 +68,11 @@ func DequeueJobMetrics(pending time.Time, started time.Time, jobType, tenant, ar
 	}
 }
 
+func RequeueJobMetrics(jobType, tenant string) {
+	PendingJobs.WithLabelValues(jobType, tenant).Inc()
+	RunningJobs.WithLabelValues(jobType, tenant).Dec()
+}
+
 func CancelJobMetrics(started time.Time, jobType, tenant string) {
 	if !started.IsZero() {
 		RunningJobs.WithLabelValues(jobType, tenant).Dec()

internal/worker/server.go

@@ -708,7 +708,7 @@ func (s *Server) RequeueOrFinishJob(token uuid.UUID, maxRetries uint64, result j
 		}
 	}
 
-	err = s.jobs.RequeueOrFinishJob(jobId, maxRetries, result)
+	requeued, err := s.jobs.RequeueOrFinishJob(jobId, maxRetries, result)
 	if err != nil {
 		switch err {
 		case jobqueue.ErrNotRunning:
@@ -718,6 +718,14 @@ func (s *Server) RequeueOrFinishJob(token uuid.UUID, maxRetries uint64, result j
 		}
 	}
 
+	if requeued {
+		jobInfo, err := s.jobInfo(jobId, nil)
+		if err != nil {
+			return fmt.Errorf("error requeueing job: %w", err)
+		}
+		prometheus.RequeueJobMetrics(jobInfo.JobType, jobInfo.Channel)
+	}
+
 	jobType, err := s.JobType(jobId)
 	if err != nil {
 		return err