refactor(metrics): replace NamedTimer with histograms

platform/consumer/consumer.go

@@ -376,7 +376,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		}
 	}
 
-	metrics.NamedTimer(controllerScope, opName, "controller_latency", elapsed, metrics.NewTag("success", successTag))
+	metrics.NamedHistogram(controllerScope, opName, "controller_latency", metrics.LongLatencyBuckets, metrics.NewTag("success", successTag)).RecordDuration(elapsed)
 
 	if err != nil {
 		// Single explicit classification pass through the configured
@@ -443,10 +443,10 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			metrics.NamedCounter(controllerScope, opName, "nack_errors", 1)
 		} else {
 			metrics.NamedCounter(controllerScope, opName, "nack_count", 1)
-			metrics.NamedTimer(controllerScope, opName, "ack_nack_latency", time.Since(nackStart),
+			metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets,
 				metrics.NewTag("operation", "nack"),
 				metrics.NewTag("success", "true"),
-			)
+			).RecordDuration(time.Since(nackStart))
 		}
 		return
 	}
@@ -461,19 +461,19 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			"error", ackErr,
 		)
 		metrics.NamedCounter(controllerScope, opName, "ack_errors", 1)
-		metrics.NamedTimer(controllerScope, opName, "ack_nack_latency", time.Since(ackStart),
+		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, "messages_processed", 1)
 	metrics.NamedCounter(controllerScope, opName, "ack_count", 1)
-	metrics.NamedTimer(controllerScope, opName, "ack_nack_latency", time.Since(ackStart),
+	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(),

platform/consumer/consumer_test.go

@@ -470,14 +470,14 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 
 			snapshot := testScope.Snapshot()
 
-			timers := snapshot.Timers()
-			assert.NotEmpty(t, timers, "Should have timer metrics")
+			histograms := snapshot.Histograms()
+			assert.NotEmpty(t, histograms, "Should have histogram metrics")
 
 			var foundLatency bool
-			for _, timer := range timers {
-				if strings.Contains(timer.Name(), "controller_latency") {
+			for _, histogram := range histograms {
+				if strings.Contains(histogram.Name(), "controller_latency") {
 					foundLatency = true
-					tags := timer.Tags()
+					tags := histogram.Tags()
 					if tt.expectSuccess {
 						assert.Equal(t, "true", tags["success"])
 					} else {
@@ -542,7 +542,7 @@ func TestConsumer_AckNackLatencyTracking(t *testing.T) {
 	<-done
 
 	snapshot := scope.Snapshot()
-	assert.NotEmpty(t, snapshot.Timers(), "Should have timer metrics for latency tracking")
+	assert.NotEmpty(t, snapshot.Histograms(), "Should have histogram metrics for latency tracking")
 	assert.NotEmpty(t, snapshot.Counters(), "Should have counter metrics")
 
 	err = c.Stop(30000)

platform/extension/counter/mysql/counter.go

@@ -38,7 +38,7 @@ func NewCounter(db *sql.DB, scope tally.Scope) counter.Counter {
 // Uses MySQL's LAST_INSERT_ID() to set the value atomically and read the incremented value.
 func (c *mysqlCounter) Next(ctx context.Context, domain string) (ret int64, retErr error) {
 	op := metrics.Begin(c.scope, "next")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 	result, err := c.db.ExecContext(ctx,
 		"INSERT INTO counter (domain, value) VALUES (?, LAST_INSERT_ID(1)) ON DUPLICATE KEY UPDATE value = LAST_INSERT_ID(value + 1)",
 		domain,

platform/extension/messagequeue/mysql/delivery_state_store.go

@@ -53,7 +53,7 @@ func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := time.Now().UnixMilli()
 	invisibleUntil := now + visibilityTimeoutMs
@@ -94,7 +94,7 @@ func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGr
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := time.Now().UnixMilli()
 	invisibleUntil := now + visibilityTimeoutMs
@@ -128,7 +128,7 @@ func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, to
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
 		INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count)
@@ -151,7 +151,7 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := time.Now().UnixMilli()
 	invisibleUntil := now + delayMs
@@ -178,7 +178,7 @@ func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGr
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var state DeliveryState
 	err := s.db.QueryRowContext(ctx, fmt.Sprintf(`
@@ -205,7 +205,7 @@ func (s *sqldeliveryStateStore) AdvanceWatermark(ctx context.Context, consumerGr
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if len(offsets) == 0 {
 		return currentWatermark, nil

platform/extension/messagequeue/mysql/message_store.go

@@ -63,7 +63,7 @@ func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []e
 // errors.
 func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messages []entityqueue.Message, visibleAfterMs int64) (retErr error) {
 	op := metrics.Begin(s.scope, "insert", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if len(messages) == 0 {
 		return nil
@@ -133,7 +133,7 @@ func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messa
 // Delete deletes a message by topic, partition key, and ID
 func (s *sqlmessageStore) Delete(ctx context.Context, topic string, partitionKey string, messageID string) (retErr error) {
 	op := metrics.Begin(s.scope, "delete", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
 		DELETE FROM %s WHERE topic = ? AND partition_key = ? AND id = ?
@@ -152,7 +152,7 @@ func (s *sqlmessageStore) Delete(ctx context.Context, topic string, partitionKey
 // Messages are fetched from the immutable log; no per-message mutation occurs.
 func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, nowMs int64, limit int) (_ []messageRow, retErr error) {
 	op := metrics.Begin(s.scope, "fetch", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
 		SELECT offset, id, payload, metadata, partition_key, published_at, failed_at, failure_count, last_error, original_topic
@@ -228,7 +228,7 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti
 // This allows DLQ messages to be consumed using the normal subscriber
 func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partitionKey string, messageID string, failureCount int, lastError string, dlqTopicSuffix string) (retErr error) {
 	op := metrics.Begin(s.scope, "move_to_dlq", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	// Construct DLQ topic name
 	dlqTopic := topic + dlqTopicSuffix
@@ -301,7 +301,7 @@ func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partition
 // Returns the number of rows deleted.
 func (s *sqlmessageStore) GarbageCollect(ctx context.Context, topic string, partitionKey string, minAckedOffset int64) (_ int64, retErr error) {
 	op := metrics.Begin(s.scope, "gc", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if minAckedOffset == 0 {
 		return 0, nil
@@ -343,7 +343,7 @@ func (s *sqlmessageStore) GarbageCollect(ctx context.Context, topic string, part
 // GetOffsetsAbove returns message offsets above afterOffset for a partition, ordered ascending.
 func (s *sqlmessageStore) GetOffsetsAbove(ctx context.Context, topic string, partitionKey string, afterOffset int64, limit int) (_ []int64, retErr error) {
 	op := metrics.Begin(s.scope, "get_offsets_above", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
 		SELECT offset FROM %s

platform/extension/messagequeue/mysql/offset_store.go

@@ -44,7 +44,7 @@ func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partition
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey),
 		metrics.NewTag("consumer_group", consumerGroup))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := time.Now().UnixMilli()
 
@@ -67,7 +67,7 @@ func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, parti
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey),
 		metrics.NewTag("consumer_group", consumerGroup))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var offset int64
 	err := s.db.QueryRowContext(ctx, fmt.Sprintf(`
@@ -92,7 +92,7 @@ func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, pa
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey),
 		metrics.NewTag("consumer_group", consumerGroup))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := time.Now().UnixMilli()
 
@@ -115,7 +115,7 @@ func (s *sqloffsetStore) GetMinAckedOffset(ctx context.Context, topic string, pa
 	op := metrics.Begin(s.scope, "get_min_acked_offset",
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var minOffset int64
 	err := s.db.QueryRowContext(ctx, fmt.Sprintf(`

platform/extension/messagequeue/mysql/partition_lease_store.go

@@ -45,7 +45,7 @@ func newPartitionLeaseStore(db *sql.DB, logger *zap.SugaredLogger, scope tally.S
 // TryAcquireLease attempts to acquire or renew a lease for a partition
 func (s *sqlpartitionLeaseStore) TryAcquireLease(ctx context.Context, topic string, partitionKey string, subscriberName string, consumerGroup string, leaseDurationMs int64) (_ bool, retErr error) {
 	op := metrics.Begin(s.scope, "try_acquire_lease", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := currentTimeMillis()
 	staleThreshold := now - leaseDurationMs
@@ -94,7 +94,7 @@ func (s *sqlpartitionLeaseStore) TryAcquireLease(ctx context.Context, topic stri
 // RenewLease renews the lease for a partition owned by this worker
 func (s *sqlpartitionLeaseStore) RenewLease(ctx context.Context, topic string, partitionKey string, subscriberName string, consumerGroup string, leaseDurationMs int64) (retErr error) {
 	op := metrics.Begin(s.scope, "renew_lease", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := currentTimeMillis()
 
@@ -128,7 +128,7 @@ func (s *sqlpartitionLeaseStore) RenewLease(ctx context.Context, topic string, p
 // ReleaseLease releases the lease for a partition owned by this worker
 func (s *sqlpartitionLeaseStore) ReleaseLease(ctx context.Context, topic string, partitionKey string, subscriberName string, consumerGroup string) (retErr error) {
 	op := metrics.Begin(s.scope, "release_lease", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := s.db.ExecContext(ctx, fmt.Sprintf(`
 		DELETE FROM %s
@@ -163,7 +163,7 @@ func (s *sqlpartitionLeaseStore) ReleaseLease(ctx context.Context, topic string,
 // GetLeasedPartitions returns all partitions currently leased by this worker
 func (s *sqlpartitionLeaseStore) GetLeasedPartitions(ctx context.Context, topic string, subscriberName string, consumerGroup string) (_ []string, retErr error) {
 	op := metrics.Begin(s.scope, "get_leased_partitions", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
 		SELECT partition_key FROM %s
@@ -201,7 +201,7 @@ func (s *sqlpartitionLeaseStore) GetLeasedPartitions(ctx context.Context, topic
 // maxPartitions limits how many total partitions this subscriber can own (0 = unlimited)
 func (s *sqlpartitionLeaseStore) DiscoverAndAcquirePartitions(ctx context.Context, topic string, subscriberName string, consumerGroup string, leaseDurationMs int64, maxPartitions int) (_ int, _ []string, retErr error) {
 	op := metrics.Begin(s.scope, "discover_and_acquire", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	// Query distinct partition_keys from messages table.
 	// No LIMIT is applied because all partitions must be discoverable for fair

platform/extension/messagequeue/mysql/publisher.go

@@ -47,7 +47,7 @@ func NewPublisher(logger *zap.SugaredLogger, scope tally.Scope, messageStore mes
 // Publish sends a message to the specified topic
 func (p *publisher) Publish(ctx context.Context, topic string, message entityqueue.Message) (retErr error) {
 	op := metrics.Begin(p.scope, "publish", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	// Check if closed (under lock)
 	p.mu.RLock()
@@ -73,7 +73,7 @@ func (p *publisher) Publish(ctx context.Context, topic string, message entityque
 // delayMs <= 0 is equivalent to Publish.
 func (p *publisher) PublishAfter(ctx context.Context, topic string, message entityqueue.Message, delayMs int64) (retErr error) {
 	op := metrics.Begin(p.scope, "publish_after", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	p.mu.RLock()
 	closed := p.closed

platform/extension/messagequeue/mysql/subscriber.go

@@ -367,7 +367,7 @@ func (s *subscriber) advanceWatermark(ctx context.Context, consumerGroup, topic,
 // Subscribe starts consuming messages from the specified topic
 func (s *subscriber) Subscribe(ctx context.Context, topic string, config extqueue.SubscriptionConfig) (_ <-chan extqueue.Delivery, retErr error) {
 	op := metrics.Begin(s.scope, "subscribe", metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	s.mu.RLock()
 	closed := s.closed
@@ -835,10 +835,10 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) error {
 
 		// Calculate message age for metrics
 		messageAge := time.Duration(time.Now().UnixMilli()-row.PublishedAt) * time.Millisecond
-		metrics.NamedTimer(s.scope, "poll", "message_age", messageAge,
+		metrics.NamedHistogram(s.scope, "poll", "message_age", metrics.LongLatencyBuckets,
 			metrics.NewTag("topic", sub.topic),
 			metrics.NewTag("partition_key", partitionKey),
-		)
+		).RecordDuration(messageAge)
 
 		// Create delivery ID from offset
 		deliveryID := strconv.FormatInt(row.Offset, 10)
@@ -920,10 +920,10 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) error {
 			metrics.NewTag("topic", sub.topic),
 			metrics.NewTag("partition_key", partitionKey),
 		)
-		metrics.NamedTimer(s.scope, "poll", "latency", elapsed,
+		metrics.NamedHistogram(s.scope, "poll", "latency", metrics.StorageLatencyBuckets,
 			metrics.NewTag("topic", sub.topic),
 			metrics.NewTag("partition_key", partitionKey),
-		)
+		).RecordDuration(elapsed)
 	}
 
 	return nil
@@ -1089,7 +1089,7 @@ func (s *subscriber) fairShareCap(ctx context.Context, sub *subscription, owned
 //     (see managePartitions shutdown sequence)
 func (s *subscriber) Close() (retErr error) {
 	op := metrics.Begin(s.scope, "close")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	s.mu.Lock()
 	defer s.mu.Unlock()

platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go

@@ -46,7 +46,7 @@ func newSubscriberHeartbeatStore(db *sql.DB, logger *zap.SugaredLogger, scope ta
 // Heartbeat registers or renews a subscriber's heartbeat.
 func (s *sqlSubscriberHeartbeatStore) Heartbeat(ctx context.Context, topic string, subscriberName string, consumerGroup string) (retErr error) {
 	op := metrics.Begin(s.scope, "heartbeat")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := s.nowFunc().UnixMilli()
 
@@ -66,7 +66,7 @@ func (s *sqlSubscriberHeartbeatStore) Heartbeat(ctx context.Context, topic strin
 // ActiveSubscribers returns the names of subscribers with a heartbeat newer than the stale threshold.
 func (s *sqlSubscriberHeartbeatStore) ActiveSubscribers(ctx context.Context, topic string, consumerGroup string, staleDurationMs int64) (_ []string, retErr error) {
 	op := metrics.Begin(s.scope, "active_subscribers")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	staleThreshold := s.nowFunc().UnixMilli() - staleDurationMs
 
@@ -105,7 +105,7 @@ func (s *sqlSubscriberHeartbeatStore) ActiveSubscribers(ctx context.Context, top
 // Idempotent: no-op if already deregistered.
 func (s *sqlSubscriberHeartbeatStore) Deregister(ctx context.Context, topic string, subscriberName string, consumerGroup string) (retErr error) {
 	op := metrics.Begin(s.scope, "deregister")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	now := s.nowFunc().UnixMilli()