metrics: do not count postponed tasks as failed

deploy/chorus/README.md

@@ -120,8 +120,8 @@ Worker:
 
 | Metric | Type | Labels | Description |
 |--------|------|--------|-------------|
-| `worker_processed_tasks_total` | counter | `queue`, `task_type` | Tasks processed (both successful and failed) |
-| `worker_failed_tasks_total` | counter | `queue`, `task_type` | Tasks whose processing returned an error |
+| `worker_processed_tasks_total` | counter | `queue`, `task_type` | Tasks processed (both successful and failed); tasks postponed by a rate limit or a busy object lock are not counted |
+| `worker_failed_tasks_total` | counter | `queue`, `task_type` | Tasks whose processing returned an error, except tasks postponed by a rate limit or a busy object lock |
 | `worker_in_progress_tasks` | gauge | `queue`, `task_type` | Tasks currently being processed |
 | `worker_task_duration_seconds` | histogram | `queue`, `task_type` | Task processing time (exponential buckets, 0.1s–600s) |
 | `copy_in_progress_obj_bytes` | gauge | `flow`, `user`, `bucket` | Total size of objects currently being copied |

go.mod

@@ -52,6 +52,7 @@ require (
 	github.com/moby/moby/api v1.55.0
 	github.com/onsi/ginkgo/v2 v2.32.1
 	github.com/onsi/gomega v1.42.1
+	github.com/prometheus/client_model v0.6.2
 	github.com/testcontainers/testcontainers-go v0.44.0
 	golang.org/x/crypto v0.57.0
 )
@@ -283,7 +284,6 @@ require (
 	github.com/philhofer/fwd v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6 // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.70.1 // indirect
 	github.com/prometheus/procfs v0.21.1 // indirect
 	github.com/prometheus/prom2json v1.5.0 // indirect

pkg/metrics/worker_middleware.go

@@ -18,10 +18,13 @@ package metrics
 
 import (
 	"context"
+	"errors"
 
 	"github.com/hibiken/asynq"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/clyso/chorus/pkg/dom"
 )
 
 var (
@@ -66,6 +69,12 @@ func WorkerMiddleware() asynq.MiddlewareFunc {
 			err := next.ProcessTask(ctx, t)
 			timer.ObserveDuration()
 			inProgressGauge.WithLabelValues(queue, t.Type()).Dec()
+			// A rate limit or a busy object lock postpones the task (dom.ErrRateLimitExceeded,
+			// logged at debug level by the worker error handler): it is neither processed nor failed yet.
+			var rlErr *dom.ErrRateLimitExceeded
+			if errors.As(err, &rlErr) {
+				return err
+			}
 			if err != nil {
 				failedCounter.WithLabelValues(queue, t.Type()).Inc()
 			}

pkg/metrics/worker_middleware_test.go

@@ -0,0 +1,46 @@
+package metrics
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/clyso/chorus/pkg/dom"
+)
+
+func TestWorkerMiddleware_FailedTasks(t *testing.T) {
+	r := require.New(t)
+	taskType := "test:failed-tasks"
+	value := func(c *prometheus.CounterVec) float64 {
+		m := &dto.Metric{}
+		r.NoError(c.WithLabelValues("", taskType).Write(m))
+		return m.GetCounter().GetValue()
+	}
+	failed := func() float64 { return value(failedCounter) }
+	processed := func() float64 { return value(processedCounter) }
+	run := func(err error) error {
+		h := WorkerMiddleware()(asynq.HandlerFunc(func(context.Context, *asynq.Task) error { return err }))
+		return h.ProcessTask(context.Background(), asynq.NewTask(taskType, nil))
+	}
+
+	r.NoError(run(nil))
+	r.EqualValues(0, failed())
+
+	r.Error(run(errors.New("copy failed")))
+	r.EqualValues(1, failed())
+
+	r.EqualValues(2, processed())
+
+	// postponed by a rate limit or a busy lock: neither processed nor failed, it will be retried
+	postponed := fmt.Errorf("%w: lock not obtained", &dom.ErrRateLimitExceeded{RetryIn: time.Second})
+	r.ErrorIs(run(postponed), postponed)
+	r.EqualValues(1, failed())
+	r.EqualValues(2, processed())
+}