feat: report classified controller metrics

platform/consumer/README.md

@@ -118,6 +118,8 @@ func (c *MyController) Process(ctx context.Context, delivery consumer.Delivery)
 
 When the consumer is wired with `errs.AlwaysRetryableProcessor` (DLQ reconciliation), the framework overrides this: every non-nil error is forced retryable so the DLQ message comes back for another attempt. See `submitqueue/orchestrator/controller/dlq/README.md`.
 
+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.
+
 ## Lifecycle
 
 1. **Register** controllers before starting.

platform/consumer/consumer.go

@@ -341,9 +341,6 @@ func (m *consumer) processPartition(ctx context.Context, controller Controller,
 func (m *consumer) processDelivery(ctx context.Context, controller Controller, delivery extqueue.Delivery, controllerScope tally.Scope) {
 	const opName = "process"
 
-	start := time.Now()
-	metrics.NamedCounter(controllerScope, opName, "messages_received", 1)
-
 	msg := delivery.Message()
 	topicKey := controller.TopicKey()
 
@@ -359,31 +356,30 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	wrapped := &deliveryWrapper{delivery: delivery}
 
 	// Call controller with wrapped delivery
+	start := time.Now()
+	op := metrics.Begin(controllerScope, opName, metrics.LongLatencyBuckets)
 	err := controller.Process(ctx, wrapped)
 
 	elapsed := time.Since(start)
 
-	// By convention, Controller can only return context.Canceled if it is cancelled by the context, i.e. when consumer is stopped or application is shutting down
-	isCanceled := errors.Is(err, context.Canceled)
-
-	// Track latency with success/failure tags
-	successTag := "true"
-	if err != nil {
-		if isCanceled {
-			successTag = "cancel"
-		} else {
-			successTag = "false"
-		}
-	}
-
-	metrics.NamedHistogram(controllerScope, opName, "controller_latency", metrics.LongLatencyBuckets, metrics.NewTag("success", successTag)).RecordDuration(elapsed)
-
+	var completionTags []metrics.Tag
 	if err != nil {
 		// Single explicit classification pass through the configured
 		// ErrorProcessor. Primary consumers use a classifier-based processor
 		// (preserves controller framework wraps); DLQ consumers use the
 		// always-retryable processor (forces redelivery on any error).
 		err = m.processor.Process(err)
+		completionTags = controllerClassificationTags(err)
+	}
+
+	// Complete only after classification so the finish histogram carries the
+	// verdict used for ack/nack/reject behavior.
+	op.Complete(err, completionTags...)
+
+	if err != nil {
+		// By convention, Controller can only return context.Canceled if it is
+		// cancelled by the processing context during shutdown.
+		isCanceled := errors.Is(err, context.Canceled)
 
 		// Check if the error is non-retryable (poison pill message)
 		if !errs.IsRetryable(err) {
@@ -397,8 +393,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 				"elapsed_ms", elapsed.Milliseconds(),
 			)
 
-			metrics.NamedCounter(controllerScope, opName, "non_retryable_errors", 1)
-
 			// Reject moves to DLQ (or acks if DLQ disabled)
 			if rejectErr := delivery.Reject(ctx, err.Error()); rejectErr != nil {
 				m.logger.Errorw("failed to reject non-retryable message",
@@ -429,8 +423,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			"elapsed_ms", elapsed.Milliseconds(),
 		)
 
-		metrics.NamedCounter(controllerScope, opName, "controller_errors", 1)
-
 		// Nack with no delay - let visibility timeout handle retry delay
 		nackStart := time.Now()
 		if nackErr := delivery.Nack(ctx, 0); nackErr != nil {
@@ -468,7 +460,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		return
 	}
 
-	metrics.NamedCounter(controllerScope, opName, "messages_processed", 1)
 	metrics.NamedCounter(controllerScope, opName, "ack_count", 1)
 	metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets,
 		metrics.NewTag("operation", "ack"),
@@ -485,6 +476,25 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	)
 }
 
+func controllerClassificationTags(err error) []metrics.Tag {
+	origin := "infra"
+	if errs.IsRetryable(err) {
+		origin = "infra_retryable"
+	} else if errs.IsUserError(err) {
+		origin = "user"
+	}
+
+	dependency := "no"
+	if errs.IsDependencyError(err) {
+		dependency = "yes"
+	}
+
+	return []metrics.Tag{
+		metrics.NewTag("origin", origin),
+		metrics.NewTag("dependency", dependency),
+	}
+}
+
 // Stop gracefully shuts down all handlers with the specified timeout.
 // Cancels all subscription contexts and waits for consumption goroutines to finish.
 // timeoutMs is the maximum time in milliseconds to wait for graceful shutdown.

platform/consumer/consumer_test.go

@@ -47,6 +47,14 @@ type testController struct {
 	processFunc   func(context.Context, Delivery) error
 }
 
+type testClassifier struct {
+	verdict errs.Verdict
+}
+
+func (c testClassifier) Classify(error) errs.Verdict {
+	return c.verdict
+}
+
 func (c *testController) Process(ctx context.Context, delivery Delivery) error {
 	if c.processFunc == nil {
 		return nil
@@ -431,21 +439,44 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 		name           string
 		handlerError   error
 		nackError      error
-		expectSuccess  bool
+		processor      errs.ErrorProcessor
+		expectedTags   map[string]string
 		expectAckCount bool
 	}{
 		{
 			name:           "success with ack",
 			handlerError:   nil,
 			nackError:      nil,
-			expectSuccess:  true,
+			processor:      errs.NewClassifierProcessor(),
+			expectedTags:   map[string]string{"result": "success"},
 			expectAckCount: true,
 		},
 		{
-			name:           "failure with nack",
-			handlerError:   errs.NewRetryableError(fmt.Errorf("handler failed")),
-			nackError:      nil,
-			expectSuccess:  false,
+			name:         "classified failure with nack",
+			handlerError: fmt.Errorf("handler failed"),
+			nackError:    nil,
+			processor: errs.NewClassifierProcessor(testClassifier{
+				verdict: errs.InfraDependencyRetryable,
+			}),
+			expectedTags: map[string]string{
+				"result":     "error",
+				"origin":     "infra_retryable",
+				"dependency": "yes",
+			},
+			expectAckCount: false,
+		},
+		{
+			name:         "classified cancellation with nack",
+			handlerError: context.Canceled,
+			nackError:    nil,
+			processor: errs.NewClassifierProcessor(testClassifier{
+				verdict: errs.InfraRetryable,
+			}),
+			expectedTags: map[string]string{
+				"result":     "cancel",
+				"origin":     "infra_retryable",
+				"dependency": "no",
+			},
 			expectAckCount: false,
 		},
 	}
@@ -465,7 +496,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 
 			reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group")
 
-			testC := New(logger, testScope, reg, errs.NewClassifierProcessor())
+			testC := New(logger, testScope, reg, tt.processor)
 
 			handler := &testController{}
 			setupController(handler, "test-handler", testTopicKeyStart, "test-group",
@@ -497,19 +528,27 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 
 			var foundLatency bool
 			for _, histogram := range histograms {
-				if strings.Contains(histogram.Name(), "controller_latency") {
+				if strings.Contains(histogram.Name(), "process.finish") {
 					foundLatency = true
 					tags := histogram.Tags()
-					if tt.expectSuccess {
-						assert.Equal(t, "true", tags["success"])
-					} else {
-						assert.Equal(t, "false", tags["success"])
+					for key, value := range tt.expectedTags {
+						assert.Equal(t, value, tags[key])
 					}
 				}
 			}
-			assert.True(t, foundLatency, "Should have controller_latency metric")
+			assert.True(t, foundLatency, "Should have process.finish metric")
 
 			counters := snapshot.Counters()
+			for _, duplicate := range []string{
+				"messages_received",
+				"messages_processed",
+				"non_retryable_errors",
+				"controller_errors",
+			} {
+				for _, counter := range counters {
+					assert.NotContains(t, counter.Name(), duplicate)
+				}
+			}
 			if tt.expectAckCount {
 				var foundAck bool
 				for _, counter := range counters {
@@ -526,6 +565,57 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 	}
 }
 
+func TestControllerClassificationTags(t *testing.T) {
+	tests := []struct {
+		name     string
+		err      error
+		expected map[string]string
+	}{
+		{
+			name: "infra error",
+			err:  fmt.Errorf("failed"),
+			expected: map[string]string{
+				"origin":     "infra",
+				"dependency": "no",
+			},
+		},
+		{
+			name: "user error",
+			err:  errs.NewUserError(fmt.Errorf("invalid request")),
+			expected: map[string]string{
+				"origin":     "user",
+				"dependency": "no",
+			},
+		},
+		{
+			name: "retryable dependency error",
+			err:  errs.NewRetryableDependencyError(fmt.Errorf("database unavailable")),
+			expected: map[string]string{
+				"origin":     "infra_retryable",
+				"dependency": "yes",
+			},
+		},
+		{
+			name: "cancellation",
+			err:  errs.NewRetryableError(context.Canceled),
+			expected: map[string]string{
+				"origin":     "infra_retryable",
+				"dependency": "no",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actual := make(map[string]string)
+			for _, tag := range controllerClassificationTags(tt.err) {
+				actual[tag.Key] = tag.Value
+			}
+			assert.Equal(t, tt.expected, actual)
+		})
+	}
+}
+
 func TestConsumer_AckNackLatencyTracking(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	logger := zaptest.NewLogger(t).Sugar()

platform/metrics/README.md

@@ -8,7 +8,7 @@ The `metrics` package provides reusable helpers for emitting counters and histog
 
 **Operation lifecycle** — `Begin` and `Complete` tie operation metrics together. `Begin` captures the start time and emits `{name}.start`; `Complete` records duration and count on `{name}.finish`.
 
-**Result tagging** — the finish histogram is tagged with `result=success`, `result=error`, or `result=cancel`. Cancellation is detected with `errors.Is(err, context.Canceled)`. Error classification tags are intentionally omitted because many call sites complete before classification occurs.
+**Result tagging** — the finish histogram is tagged with `result=success`, `result=error`, or `result=cancel`. Cancellation is detected with `errors.Is(err, context.Canceled)`. Callers that accumulate tags while the operation runs can pass them to `Complete`.
 
 **Consistent naming** — named helpers follow the `{name}.{sub}` sub-scope pattern, producing metric paths such as `process.start` and `publish.attempts`.
 
@@ -19,7 +19,7 @@ For any operation with a clear start and end, use `Begin` and `Complete`:
 | Function | Emits |
 |----------|-------|
 | `Begin(scope, name, buckets, ...tags)` | `{name}.start` counter +1 and returns an `Op` |
-| `op.Complete(err)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` |
+| `op.Complete(err, ...tags)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` and any completion tags |
 
 `buckets` is required at `Begin` because operations differ widely in expected latency. The finish histogram records both the duration distribution and the number of completed operations, so `Complete` does not emit a separate counter.
 
@@ -33,6 +33,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 }
 ```
 
+Tags passed to `Begin` apply to both lifecycle metrics. Tags known only after execution, such as an error classification, can be attached to the finish histogram:
+
+```go
+err := controller.Process(ctx, delivery)
+err = classifier.Process(err)
+op.Complete(err, metrics.NewTag("origin", "infra_retryable"))
+```
+
 ## Named Helpers
 
 For ad-hoc metrics that do not fit the operation lifecycle:

platform/metrics/metrics.go

@@ -141,9 +141,10 @@ func Begin(scope tally.Scope, name string, buckets tally.Buckets, tags ...Tag) O
 }
 
 // Complete records elapsed time on the {name}.finish histogram, tagged with
-// result=success|error|cancel. The histogram records both duration and count.
-// Cancellation is detected through the error chain.
-func (o Op) Complete(err error) {
+// result=success|error|cancel and any additional tags accumulated while the
+// operation ran. The histogram records both duration and count. Cancellation
+// is detected through the error chain.
+func (o Op) Complete(err error, tags ...Tag) {
 	result := "success"
 	if err != nil {
 		result = "error"
@@ -152,7 +153,7 @@ func (o Op) Complete(err error) {
 		}
 	}
 
-	o.scope.
+	tagged(o.scope, tags).
 		Tagged(map[string]string{"result": result}).
 		Histogram("finish", o.buckets).
 		RecordDuration(time.Since(o.start))

platform/metrics/metrics_test.go

@@ -106,6 +106,25 @@ func TestBegin_WithTags(t *testing.T) {
 	assert.True(t, ok, "expected tagged finish histogram, got keys: %v", histogramKeys(histograms))
 }
 
+func TestOp_CompleteWithTags(t *testing.T) {
+	scope := tally.NewTestScope("", nil)
+	op := Begin(scope, "process", FastLatencyBuckets)
+	op.Complete(
+		fmt.Errorf("failed"),
+		NewTag("origin", "infra_retryable"),
+		NewTag("dependency", "no"),
+	)
+
+	snapshot := scope.Snapshot()
+	counters := snapshot.Counters()
+	_, ok := counters["process.start+"]
+	assert.True(t, ok, "finish-only tags should not be applied to start")
+
+	histograms := snapshot.Histograms()
+	_, ok = histograms["process.finish+dependency=no,origin=infra_retryable,result=error"]
+	assert.True(t, ok, "expected finish-only tags, got keys: %v", histogramKeys(histograms))
+}
+
 func TestNamedCounter(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
 	NamedCounter(scope, "publish", "attempts", 5)