refactor: standardize metrics instrumentation

Author: sbalabanov-zz

Makefile

@@ -52,7 +52,7 @@ define assert_clean
 	fi
 endef
 
-.PHONY: build build-all-linux build-runway-linux build-submitqueue-gateway-linux build-submitqueue-orchestrator-linux build-stovepipe-linux build-stovepipe-linux-debug check-gazelle check-mocks check-tidy clean clean-proto deps e2e-test fmt gazelle integration-test integration-test-submitqueue-consumer integration-test-extensions integration-test-submitqueue-gateway integration-test-submitqueue-orchestrator license-fix lint lint-fmt lint-license local-init-runway-queue-schema local-init-stovepipe-schemas local-runway-start local-runway-stop local-submitqueue-clean local-submitqueue-gateway-start local-submitqueue-gateway-stop local-init-submitqueue-schemas local-submitqueue-logs local-submitqueue-orchestrator-start local-submitqueue-orchestrator-stop local-submitqueue-ps local-submitqueue-restart local-submitqueue-start local-stop local-stovepipe-debug-start local-stovepipe-logs local-stovepipe-start local-stovepipe-stop mocks proto query-deps query-targets run-client-runway run-client-submitqueue-gateway run-client-submitqueue-orchestrator run-client-stovepipe run-queue-admin test test-no-cache tidy tidy-bazel tidy-go help
+.PHONY: build build-all-linux build-runway-linux build-submitqueue-gateway-linux build-submitqueue-orchestrator-linux build-stovepipe-linux build-stovepipe-linux-debug check-gazelle check-mocks check-tidy clean clean-proto deps e2e-test fmt gazelle integration-test integration-test-submitqueue-consumer integration-test-extensions integration-test-submitqueue-gateway integration-test-submitqueue-orchestrator license-fix lint lint-fmt lint-license lint-metrics local-init-runway-queue-schema local-init-stovepipe-schemas local-runway-start local-runway-stop local-submitqueue-clean local-submitqueue-gateway-start local-submitqueue-gateway-stop local-init-submitqueue-schemas local-submitqueue-logs local-submitqueue-orchestrator-start local-submitqueue-orchestrator-stop local-submitqueue-ps local-submitqueue-restart local-submitqueue-start local-stop local-stovepipe-debug-start local-stovepipe-logs local-stovepipe-start local-stovepipe-stop mocks proto query-deps query-targets run-client-runway run-client-submitqueue-gateway run-client-submitqueue-orchestrator run-client-stovepipe run-queue-admin test test-no-cache tidy tidy-bazel tidy-go help
 
 
 build: ## Build all services and examples
@@ -170,7 +170,7 @@ integration-test-submitqueue-orchestrator: build-submitqueue-orchestrator-linux
 license-fix: ## Add missing license headers to source files
 	@$(BAZEL) run //tool/linter/licenseheader -- --fix
 
-lint: lint-fmt lint-license ## Run all linters
+lint: lint-fmt lint-license lint-metrics ## Run all linters
 	@echo "All lint checks passed."
 
 lint-fmt: fmt ## Check code formatting (fails if unformatted)
@@ -180,6 +180,14 @@ lint-fmt: fmt ## Check code formatting (fails if unformatted)
 lint-license: ## Check license headers on all source files
 	@$(BAZEL) run //tool/linter/licenseheader -- --check
 
+lint-metrics: ## Check metrics use the platform framework
+	@violations="$$(git grep -n -E '\.(Counter|Gauge|Histogram|Timer)\(' -- '*.go' ':(exclude)platform/metrics/metrics.go' || true)"; \
+	if [ -n "$$violations" ]; then \
+		echo "Direct Tally metric construction is not allowed outside platform/metrics:" >&2; \
+		echo "$$violations" >&2; \
+		exit 1; \
+	fi
+
 local-submitqueue-clean: ## Stop and remove all local services, volumes, and images
 	@echo "Cleaning all services and data..."
 	@$(COMPOSE) -f $(COMPOSE_FILE) -p $(SUBMITQUEUE_LOCAL_PROJECT) down -v --rmi local

platform/consumer/README.md

@@ -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.

platform/consumer/consumer.go

@@ -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,

platform/consumer/consumer_test.go

@@ -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)

platform/consumer/controller.go

@@ -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 {

platform/extension/messagequeue/mysql/subscriber.go

@@ -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