Merge branch 'master' into echatman/skipped-as-duplicate-tracking

Author: echatman

CHANGELOG.md

@@ -7,10 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-### Added
-
 - Added `unique_skipped_as_duplicate` attributes to otel `insert_many` spans and `insert_count` metric. [PR #58](https://github.com/riverqueue/rivercontrib/pull/58).
 
+### Changed
+
+- Record snoozed jobs with status `ok` instead of `error` in `otelriver` middleware. Add new `snooze.duration` span attribute. [PR #59](https://github.com/riverqueue/rivercontrib/pull/59).
+
 ## [0.8.0] - 2026-05-15
 
 ### Added

otelriver/middleware.go

@@ -230,6 +230,11 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 	defer func() {
 		duration := m.durationInPreferredUnit(time.Since(begin))
 
+		var (
+			cancelErr *river.JobCancelError
+			snoozeErr *river.JobSnoozeError
+		)
+
 		if err != nil {
 			var batchResult interface { // To be superseded if riverbatch.MultiError is moved to rivertype.
 				ErrorsByID() map[int64]error
@@ -238,11 +243,6 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 				err = batchResult.ErrorsByID()[job.ID]
 			}
 
-			var (
-				cancelErr *river.JobCancelError
-				snoozeErr *river.JobSnoozeError
-			)
-
 			switch {
 			case errors.As(err, &cancelErr):
 				attrs = append(attrs, attribute.Bool("cancel", true))
@@ -253,16 +253,17 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 
 		setStatus(attrs, statusIndex, span, panicked, err)
 
-		{
-			// Add some higher cardinality attributes to spans, but keep them
-			// out of metrics given it's been traditional wisdom that metric
-			// attribute sets shouldn't be too large.
-			attrs := append(attrs,
-				attribute.Int64("id", job.ID),
-				attribute.String("created_at", job.CreatedAt.Format(time.RFC3339)),
-				attribute.String("scheduled_at", job.ScheduledAt.Format(time.RFC3339)),
-			)
-			span.SetAttributes(attrs...) // set after finalizing status
+		// Add some higher cardinality attributes to spans, but keep them
+		// out of metrics given it's been traditional wisdom that metric
+		// attribute sets shouldn't be too large.
+		span.SetAttributes(
+			attribute.Int64("id", job.ID),
+			attribute.String("created_at", job.CreatedAt.Format(time.RFC3339)),
+			attribute.String("scheduled_at", job.ScheduledAt.Format(time.RFC3339)),
+		)
+		span.SetAttributes(attrs...) // set after finalizing status
+		if snoozeErr != nil {
+			span.SetAttributes(attribute.String("snooze.duration", snoozeErr.Duration.String()))
 		}
 
 		// This allocates a new slice, so make sure to do it as few times as possible.
@@ -336,6 +337,13 @@ func setStatus(attrs []attribute.KeyValue, statusIndex int, span trace.Span, pan
 	case panicked:
 		attrs[statusIndex] = attribute.String("status", "panic")
 		span.SetStatus(codes.Error, "panic")
+	case errors.Is(err, &river.JobSnoozeError{}):
+		// Snooze is flow control, not failure: the job will be retried
+		// later. Record as ok so it doesn't pollute error rates; the
+		// snooze:true span/metric attribute (set by the caller) keeps
+		// snoozes queryable as a dimension.
+		attrs[statusIndex] = attribute.String("status", "ok")
+		span.SetStatus(codes.Ok, "")
 	case err != nil:
 		attrs[statusIndex] = attribute.String("status", "error")
 		span.SetStatus(codes.Error, err.Error())

otelriver/middleware_test.go

@@ -469,19 +469,37 @@ func TestMiddleware(t *testing.T) {
 		middleware, bundle := setup(t)
 
 		doInner := func(ctx context.Context) error {
-			return fmt.Errorf("wrapped job snooze: %w", &rivertype.JobSnoozeError{})
+			return fmt.Errorf("wrapped job snooze: %w", &rivertype.JobSnoozeError{Duration: 5 * time.Second})
 		}
 
 		err := middleware.Work(ctx, &rivertype.JobRow{
 			Kind: "no_op",
 		}, doInner)
-		require.EqualError(t, err, "wrapped job snooze: JobSnoozeError: 0s")
+		require.EqualError(t, err, "wrapped job snooze: JobSnoozeError: 5s")
 
 		spans := bundle.traceExporter.GetSpans()
 		require.Len(t, spans, 1)
 
 		span := spans[0]
 		require.True(t, getAttribute(t, span.Attributes, "snooze").AsBool())
+		// Snooze is flow control, not failure: span status is ok and the
+		// snooze.duration is recorded as a span-only attribute.
+		require.Equal(t, "ok", getAttribute(t, span.Attributes, "status").AsString())
+		require.Equal(t, codes.Ok, span.Status.Code)
+		require.Equal(t, "5s", getAttribute(t, span.Attributes, "snooze.duration").AsString())
+
+		// The metric records status:ok with snooze:true so snoozes remain
+		// queryable as a metric dimension without counting against the
+		// error rate.
+		var (
+			expectedAttrs = []attribute.KeyValue{
+				attribute.String("status", "ok"),
+				attribute.Bool("snooze", true),
+			}
+			metrics metricdata.ResourceMetrics
+		)
+		require.NoError(t, bundle.metricReader.Collect(ctx, &metrics))
+		requireSum(t, metrics, "river.work_count", 1, expectedAttrs...)
 	})
 
 	t.Run("WorkBatchResultWithJobError", func(t *testing.T) {
@@ -558,7 +576,7 @@ func TestMiddleware(t *testing.T) {
 
 		doInner := func(ctx context.Context) error {
 			return &fakeBatchError{errorsByID: map[int64]error{
-				123: &rivertype.JobSnoozeError{},
+				123: &rivertype.JobSnoozeError{Duration: 7 * time.Second},
 			}}
 		}
 
@@ -570,6 +588,19 @@ func TestMiddleware(t *testing.T) {
 
 		span := spans[0]
 		require.True(t, getAttribute(t, span.Attributes, "snooze").AsBool())
+		require.Equal(t, "ok", getAttribute(t, span.Attributes, "status").AsString())
+		require.Equal(t, codes.Ok, span.Status.Code)
+		require.Equal(t, "7s", getAttribute(t, span.Attributes, "snooze.duration").AsString())
+
+		var (
+			expectedAttrs = []attribute.KeyValue{
+				attribute.String("status", "ok"),
+				attribute.Bool("snooze", true),
+			}
+			metrics metricdata.ResourceMetrics
+		)
+		require.NoError(t, bundle.metricReader.Collect(ctx, &metrics))
+		requireSum(t, metrics, "river.work_count", 1, expectedAttrs...)
 	})
 
 	t.Run("WorkPanic", func(t *testing.T) {