uber · sbalabanov · Jul 23, 2026
@@ -120,6 +120,8 @@ When the consumer is wired with `errs.AlwaysRetryableProcessor` (DLQ reconciliat
 
 The consumer records controller operations with `process.start` and `process.finish`. The finish histogram records both latency and completion count with `result=success|error|cancel`; error and cancellation series also include `origin=infra|infra_retryable|user` and `dependency=yes|no`. These dimensions are added after error processing, so they describe the classified error that drives ack, nack, or reject behavior rather than the controller's raw return value. The lifecycle histogram count replaces separate received, processed, and controller-error counters.
 
+The consumer also owns lifecycle metrics for the resulting `ack`, `nack`, or `reject` transport operation. Queue controllers should emit only domain-specific event counters; they must not duplicate the consumer-owned `process` lifecycle metrics.
+
 ## Lifecycle
 
 1. **Register** controllers before starting.

@@ -83,12 +83,14 @@ type activeSubscription struct {
 // consumers (per-node classifier walk that preserves controller-attached
 // framework wraps), or errs.AlwaysRetryableProcessor for narrowly-scoped
 // consumers such as DLQ reconciliation that must redeliver on any failure.
-// processor must not be nil; callers that genuinely want no transformation
-// can pass errs.NewClassifierProcessor() with no classifiers.
+// scope is used as provided so wiring can distinguish primary and DLQ consumers
+// without introducing duplicate consumer sub-scopes. processor must not be nil;
+// callers that genuinely want no transformation can pass
+// errs.NewClassifierProcessor() with no classifiers.
 func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor) Consumer {
 	return &consumer{
 		logger:        logger,
-		metricsScope:  scope.SubScope("consumer"),
+		metricsScope:  scope,
 		registry:      registry,
 		processor:     processor,
 		subscriptions: make(map[TopicKey]*activeSubscription),
@@ -394,14 +396,16 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			)
 
 			// Reject moves to DLQ (or acks if DLQ disabled)
-			if rejectErr := delivery.Reject(ctx, err.Error()); rejectErr != nil {
+			rejectOp := metrics.Begin(controllerScope, "reject", metrics.StorageLatencyBuckets)
+			rejectErr := delivery.Reject(ctx, err.Error())
+			rejectOp.Complete(rejectErr)
+			if rejectErr != nil {
 				m.logger.Errorw("failed to reject non-retryable message",
 					"controller", controller.Name(),
 					"topic_key", controller.TopicKey(),
 					"message_id", msg.ID,
 					"error", rejectErr,
 				)
-				metrics.NamedCounter(controllerScope, opName, "reject_errors", 1)
 			}
 			return
 		}
@@ -424,48 +428,34 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		)
 
 		// Nack with no delay - let visibility timeout handle retry delay
-		nackStart := time.Now()
-		if nackErr := delivery.Nack(ctx, 0); nackErr != nil {
+		nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets)
+		nackErr := delivery.Nack(ctx, 0)
+		nackOp.Complete(nackErr)
+		if nackErr != nil {
 			m.logger.Errorw("failed to nack message",
 				"controller", controller.Name(),
 				"topic_key", topicKey,
 				"message_id", msg.ID,
 				"error", nackErr,
 			)
-			metrics.NamedCounter(controllerScope, opName, "nack_errors", 1)
-		} else {
-			metrics.NamedCounter(controllerScope, opName, "nack_count", 1)
-			metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets,
-				metrics.NewTag("operation", "nack"),
-				metrics.NewTag("success", "true"),
-			).RecordDuration(time.Since(nackStart))
 		}
 		return
 	}
 
 	// Controller succeeded - ack message
-	ackStart := time.Now()
-	if ackErr := delivery.Ack(ctx); ackErr != nil {
+	ackOp := metrics.Begin(controllerScope, "ack", metrics.StorageLatencyBuckets)
+	ackErr := delivery.Ack(ctx)
+	ackOp.Complete(ackErr)
+	if ackErr != nil {
 		m.logger.Errorw("failed to ack message",
 			"controller", controller.Name(),
 			"topic_key", topicKey,
 			"message_id", msg.ID,
 			"error", ackErr,
 		)
-		metrics.NamedCounter(controllerScope, opName, "ack_errors", 1)
-		metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets,
-			metrics.NewTag("operation", "ack"),
-			metrics.NewTag("success", "false"),
-		).RecordDuration(time.Since(ackStart))
 		return
 	}
 
-	metrics.NamedCounter(controllerScope, opName, "ack_count", 1)
-	metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets,
-		metrics.NewTag("operation", "ack"),
-		metrics.NewTag("success", "true"),
-	).RecordDuration(time.Since(ackStart))
-
 	m.logger.Debugw("message processed successfully",
 		"controller", controller.Name(),
 		"topic_key", topicKey,

@@ -117,6 +117,10 @@ func setupDelivery(del *queuemock.MockDelivery, msg entityqueue.Message, ackErr,
 		close(done)
 		return nackErr
 	}).MaxTimes(1)
+	del.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, reason string) error {
+		close(done)
+		return nil
+	}).MaxTimes(1)
 	return done
 }
 
@@ -436,20 +440,20 @@ func TestConsumer_Stop(t *testing.T) {
 
 func TestConsumer_ObservabilityTags(t *testing.T) {
 	tests := []struct {
-		name           string
-		handlerError   error
-		nackError      error
-		processor      errs.ErrorProcessor
-		expectedTags   map[string]string
-		expectAckCount bool
+		name         string
+		handlerError error
+		nackError    error
+		processor    errs.ErrorProcessor
+		expectedTags map[string]string
+		transport    string
 	}{
 		{
-			name:           "success with ack",
-			handlerError:   nil,
-			nackError:      nil,
-			processor:      errs.NewClassifierProcessor(),
-			expectedTags:   map[string]string{"result": "success"},
-			expectAckCount: true,
+			name:         "success with ack",
+			handlerError: nil,
+			nackError:    nil,
+			processor:    errs.NewClassifierProcessor(),
+			expectedTags: map[string]string{"result": "success"},
+			transport:    "ack",
 		},
 		{
 			name:         "classified failure with nack",
@@ -463,7 +467,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 				"origin":     "infra_retryable",
 				"dependency": "yes",
 			},
-			expectAckCount: false,
+			transport: "nack",
 		},
 		{
 			name:         "classified cancellation with nack",
@@ -477,7 +481,18 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 				"origin":     "infra_retryable",
 				"dependency": "no",
 			},
-			expectAckCount: false,
+			transport: "nack",
+		},
+		{
+			name:         "user failure with reject",
+			handlerError: errs.NewUserError(fmt.Errorf("invalid request")),
+			processor:    errs.NewClassifierProcessor(),
+			expectedTags: map[string]string{
+				"result":     "error",
+				"origin":     "user",
+				"dependency": "no",
+			},
+			transport: "reject",
 		},
 	}
 
@@ -528,6 +543,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 
 			var foundLatency bool
 			for _, histogram := range histograms {
+				assert.NotContains(t, histogram.Name(), "consumer.consumer")
 				if strings.Contains(histogram.Name(), "process.finish") {
 					foundLatency = true
 					tags := histogram.Tags()
@@ -538,27 +554,32 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 			}
 			assert.True(t, foundLatency, "Should have process.finish metric")
 
+			var foundTransport bool
+			for _, histogram := range histograms {
+				if strings.Contains(histogram.Name(), tt.transport+".finish") {
+					foundTransport = true
+					assert.Equal(t, "success", histogram.Tags()["result"])
+				}
+				assert.NotContains(t, histogram.Name(), "ack_nack_latency")
+			}
+			assert.True(t, foundTransport, "Should have %s.finish metric", tt.transport)
+
 			counters := snapshot.Counters()
 			for _, duplicate := range []string{
 				"messages_received",
 				"messages_processed",
 				"non_retryable_errors",
 				"controller_errors",
+				"ack_count",
+				"ack_errors",
+				"nack_count",
+				"nack_errors",
+				"reject_errors",
 			} {
 				for _, counter := range counters {
 					assert.NotContains(t, counter.Name(), duplicate)
 				}
 			}
-			if tt.expectAckCount {
-				var foundAck bool
-				for _, counter := range counters {
-					if strings.Contains(counter.Name(), "ack_count") {
-						foundAck = true
-						assert.Greater(t, counter.Value(), int64(0))
-					}
-				}
-				assert.True(t, foundAck, "Should have ack_count metric")
-			}
 
 			_ = testC.Stop(30000)
 		})
@@ -616,7 +637,7 @@ func TestControllerClassificationTags(t *testing.T) {
 	}
 }
 
-func TestConsumer_AckNackLatencyTracking(t *testing.T) {
+func TestConsumer_AckLifecycleMetrics(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	logger := zaptest.NewLogger(t).Sugar()
 	scope := tally.NewTestScope("consumer", nil)
@@ -654,14 +675,21 @@ func TestConsumer_AckNackLatencyTracking(t *testing.T) {
 	<-done
 
 	snapshot := scope.Snapshot()
-	assert.NotEmpty(t, snapshot.Histograms(), "Should have histogram metrics for latency tracking")
-	assert.NotEmpty(t, snapshot.Counters(), "Should have counter metrics")
+	histograms := snapshot.Histograms()
+	var foundAck bool
+	for _, histogram := range histograms {
+		if strings.Contains(histogram.Name(), "ack.finish") {
+			foundAck = true
+			assert.Equal(t, "success", histogram.Tags()["result"])
+		}
+	}
+	assert.True(t, foundAck, "Should have successful ack.finish metric")
 
 	err = c.Stop(30000)
 	require.NoError(t, err)
 }
 
-func TestConsumer_ErrorMetrics(t *testing.T) {
+func TestConsumer_NackLifecycleMetrics(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	logger := zaptest.NewLogger(t).Sugar()
 	scope := tally.NewTestScope("consumer", nil)
@@ -701,16 +729,15 @@ func TestConsumer_ErrorMetrics(t *testing.T) {
 	<-done
 
 	snapshot := scope.Snapshot()
-	counters := snapshot.Counters()
-
-	var hasErrorMetrics bool
-	for _, counter := range counters {
-		if strings.Contains(counter.Name(), "errors") {
-			hasErrorMetrics = true
-			break
+	histograms := snapshot.Histograms()
+	var foundNackError bool
+	for _, histogram := range histograms {
+		if strings.Contains(histogram.Name(), "nack.finish") {
+			foundNackError = true
+			assert.Equal(t, "error", histogram.Tags()["result"])
 		}
 	}
-	assert.True(t, hasErrorMetrics, "Should track error metrics")
+	assert.True(t, foundNackError, "Should have failed nack.finish metric")
 
 	err = c.Stop(30000)
 	require.NoError(t, err)

@@ -89,6 +89,7 @@ func (d *deliveryWrapper) Metadata() map[string]string {
 // The Controller interface enables clean separation of concerns:
 // - Controller focuses on business logic (deserialize, process, return error status)
 // - Consumer handles infrastructure (subscription, ack/nack, metrics, lifecycle)
+// Controllers may emit domain event counters, but must not duplicate the consumer-owned Process lifecycle metrics.
 // The implementation of the controller should be idempotent and stateless. The controller is expected to be retried for the same message multiple times and should process side effects gracefully.
 // The implementation must be thread-safe.
 type Controller interface {

@@ -747,13 +747,18 @@ func (w *partitionWorker) run(ctx context.Context) {
 // Partition leasing guarantees a single writer, so the TOCTOU gap between
 // GetDeliveryState and MarkDelivered cannot cause incorrect behavior — no other
 // worker can mutate the same (consumer_group, topic, partition_key, offset).
-func (w *partitionWorker) pollAndDeliver(ctx context.Context) error {
-	start := time.Now()
+func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) {
 	s := w.subscriber
 	sub := w.sub
 	cfg := sub.config
 	partitionKey := w.partitionKey
 
+	op := metrics.Begin(s.scope, "poll", metrics.StorageLatencyBuckets,
+		metrics.NewTag("topic", sub.topic),
+		metrics.NewTag("partition_key", partitionKey),
+	)
+	defer func() { op.Complete(retErr) }()
+
 	// Initialize offset for this partition once per worker lifetime
 	if !w.offsetInitialized {
 		if err := s.offsetStore.Initialize(ctx, sub.topic, partitionKey, cfg.ConsumerGroup); err != nil {
@@ -912,15 +917,10 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) error {
 
 	// Record poll metrics
 	if messageCount > 0 {
-		elapsed := time.Since(start)
 		metrics.NamedCounter(s.scope, "poll", "messages_delivered", int64(messageCount),
 			metrics.NewTag("topic", sub.topic),
 			metrics.NewTag("partition_key", partitionKey),
 		)
-		metrics.NamedHistogram(s.scope, "poll", "latency", metrics.StorageLatencyBuckets,
-			metrics.NewTag("topic", sub.topic),
-			metrics.NewTag("partition_key", partitionKey),
-		).RecordDuration(elapsed)
 	}
 
 	return nil

@@ -460,10 +460,11 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) {
 	mockOffsetStore := NewMockoffsetStore(ctrl)
 	mockLeaseStore := NewMockpartitionLeaseStore(ctrl)
 	mockDeliveryState := NewMockdeliveryStateStore(ctrl)
+	metricsScope := tally.NewTestScope("test", nil)
 
 	s := NewSubscriber(
 		zaptest.NewLogger(t).Sugar(),
-		tally.NoopScope,
+		metricsScope,
 		mockMessageStore,
 		mockOffsetStore,
 		mockLeaseStore,
@@ -512,7 +513,7 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) {
 		done:         make(chan struct{}),
 	}
 
-	w.pollAndDeliver(ctx)
+	require.NoError(t, w.pollAndDeliver(ctx))
 
 	// Verify message was delivered
 	select {
@@ -524,6 +525,29 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) {
 
 	// Verify offset was initialized only once
 	assert.True(t, w.offsetInitialized)
+
+	snapshot := metricsScope.Snapshot()
+	var foundStart bool
+	for _, counter := range snapshot.Counters() {
+		if counter.Name() == "test.subscriber.poll.start" {
+			foundStart = true
+			assert.Equal(t, "test_topic", counter.Tags()["topic"])
+			assert.Equal(t, "part-1", counter.Tags()["partition_key"])
+		}
+	}
+	assert.True(t, foundStart, "expected poll.start counter")
+
+	var foundFinish bool
+	for _, histogram := range snapshot.Histograms() {
+		if histogram.Name() == "test.subscriber.poll.finish" {
+			foundFinish = true
+			assert.Equal(t, "success", histogram.Tags()["result"])
+			assert.Equal(t, "test_topic", histogram.Tags()["topic"])
+			assert.Equal(t, "part-1", histogram.Tags()["partition_key"])
+		}
+		assert.NotContains(t, histogram.Name(), "poll.latency")
+	}
+	assert.True(t, foundFinish, "expected poll.finish histogram")
 }
 
 // TestSubscriber_StopAllWorkers tests that all workers are stopped gracefully.

@@ -57,11 +57,7 @@ h := metrics.NamedHistogram(c.scope, "process", "duration", metrics.FastLatencyB
 h.RecordDuration(elapsed)
 ```
 
-Use `tally.Scope` directly for gauges:
-
-```go
-c.scope.SubScope("consumer").Gauge("pending_messages").Update(float64(len(pending)))
-```
+Do not emit gauges or timers. Represent operation latency and completion count with lifecycle histograms, and represent instantaneous quantities as sampled histogram values when needed.
 
 ### Why histograms, not timers