From e449a5d5acf3012fa4323b4692b30f475c713655 Mon Sep 17 00:00:00 2001
From: sergeyb <sergeyb@uber.com>
Date: Thu, 16 Jul 2026 20:13:25 +0000
Subject: [PATCH] refactor(metrics): replace timers with histograms

Remove the NamedTimer helper and stop recording any duration as a
tally timer. Durations are now histograms throughout.

Addressing review feedback:
- Drop the NamedDurationHistogram convenience helper; callers use
  NamedHistogram with explicit buckets (+ RecordDuration for one-shots).
- Op.Complete now takes a required buckets parameter (no baked-in
  default) since operations differ widely in expected latency; every
  callsite passes an appropriate set.
- Export three common duration bucket sets from the metrics package:
  FastLatencyBuckets, StorageLatencyBuckets, LongLatencyBuckets.

Timers are unsafe in a distributed system: percentiles are computed
per time series and cannot be combined, so any rollup across series
(regions, zones, or any unpinned tag) is imprecise for everything but
the mean. Bucketed histograms merge exactly by summing buckets.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 platform/consumer/consumer.go                 |  14 +-
 platform/consumer/consumer_test.go            |  12 +-
 platform/extension/counter/mysql/counter.go   |   2 +-
 .../mysql/delivery_state_store.go             |  12 +-
 .../messagequeue/mysql/message_store.go       |  12 +-
 .../messagequeue/mysql/offset_store.go        |   8 +-
 .../mysql/partition_lease_store.go            |  10 +-
 .../extension/messagequeue/mysql/publisher.go |   4 +-
 .../messagequeue/mysql/subscriber.go          |  12 +-
 .../mysql/subscriber_heartbeat_store.go       |   6 +-
 platform/metrics/README.md                    |  49 ++++---
 platform/metrics/metrics.go                   | 131 ++++++++++++------
 platform/metrics/metrics_test.go              |  64 +++------
 runway/controller/merge/merge.go              |   2 +-
 .../mergeconflictcheck/mergeconflictcheck.go  |   2 +-
 runway/controller/ping.go                     |   2 +-
 stovepipe/controller/build/build.go           |   2 +-
 .../controller/buildsignal/buildsignal.go     |   2 +-
 stovepipe/controller/dlq/request.go           |   2 +-
 stovepipe/controller/ingest.go                |   2 +-
 stovepipe/controller/ping.go                  |   2 +-
 stovepipe/controller/process/process.go       |   2 +-
 .../extension/storage/mysql/build_store.go    |   6 +-
 .../extension/storage/mysql/queue_store.go    |   6 +-
 .../extension/storage/mysql/request_store.go  |   6 +-
 .../storage/mysql/request_uri_store.go        |   4 +-
 .../changeprovider/github/provider.go         |   2 +-
 .../changeprovider/phabricator/provider.go    |   2 +-
 .../extension/mergechecker/github/checker.go  |   2 +-
 .../extension/pusher/git/git_pusher.go        |   2 +-
 .../extension/scorer/composite/scorer.go      |   2 +-
 .../extension/scorer/heuristic/scorer.go      |   2 +-
 .../storage/mysql/batch_dependent_store.go    |   6 +-
 .../extension/storage/mysql/batch_store.go    |  10 +-
 .../extension/storage/mysql/build_store.go    |   6 +-
 .../extension/storage/mysql/change_store.go   |   4 +-
 .../storage/mysql/request_log_store.go        |   4 +-
 .../mysql/request_queue_summary_store.go      |   8 +-
 .../extension/storage/mysql/request_store.go  |   6 +-
 .../storage/mysql/request_summary_store.go    |   6 +-
 .../storage/mysql/request_uri_store.go        |   4 +-
 submitqueue/gateway/controller/land.go        |   2 +-
 submitqueue/gateway/controller/list.go        |   2 +-
 submitqueue/gateway/controller/log/log.go     |   2 +-
 .../gateway/controller/request_history.go     |   4 +-
 .../gateway/controller/request_summary.go     |   4 +-
 .../orchestrator/controller/batch/batch.go    |   2 +-
 .../orchestrator/controller/build/build.go    |   2 +-
 .../controller/buildsignal/buildsignal.go     |   2 +-
 .../controller/conclude/conclude.go           |   2 +-
 .../orchestrator/controller/dlq/batch.go      |   2 +-
 .../controller/dlq/buildsignal.go             |   2 +-
 .../orchestrator/controller/dlq/log.go        |   2 +-
 .../controller/dlq/mergeconflictsignal.go     |   2 +-
 .../controller/dlq/mergesignal.go             |   2 +-
 .../orchestrator/controller/dlq/request.go    |   2 +-
 .../orchestrator/controller/merge/merge.go    |   2 +-
 .../mergeconflictsignal.go                    |   2 +-
 .../controller/mergesignal/mergesignal.go     |   2 +-
 submitqueue/orchestrator/controller/ping.go   |   2 +-
 .../orchestrator/controller/score/score.go    |   2 +-
 .../controller/speculate/speculate.go         |   2 +-
 .../orchestrator/controller/start/start.go    |   2 +-
 .../controller/validate/validate.go           |   2 +-
 64 files changed, 261 insertions(+), 231 deletions(-)

diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go
index 0569ba2de..f5d4c773a 100644
--- a/platform/consumer/consumer.go
+++ b/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(),
diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go
index 1bf4682fa..f897fa085 100644
--- a/platform/consumer/consumer_test.go
+++ b/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)
diff --git a/platform/extension/counter/mysql/counter.go b/platform/extension/counter/mysql/counter.go
index ab839ce33..6693e1cf1 100644
--- a/platform/extension/counter/mysql/counter.go
+++ b/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,
diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go
index 9981beac7..063aee6bf 100644
--- a/platform/extension/messagequeue/mysql/delivery_state_store.go
+++ b/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
diff --git a/platform/extension/messagequeue/mysql/message_store.go b/platform/extension/messagequeue/mysql/message_store.go
index df3ae0c60..1d378a1da 100644
--- a/platform/extension/messagequeue/mysql/message_store.go
+++ b/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
diff --git a/platform/extension/messagequeue/mysql/offset_store.go b/platform/extension/messagequeue/mysql/offset_store.go
index a71116fe6..5d37fa82e 100644
--- a/platform/extension/messagequeue/mysql/offset_store.go
+++ b/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(`
diff --git a/platform/extension/messagequeue/mysql/partition_lease_store.go b/platform/extension/messagequeue/mysql/partition_lease_store.go
index 6ea259e36..fc2f5458f 100644
--- a/platform/extension/messagequeue/mysql/partition_lease_store.go
+++ b/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
diff --git a/platform/extension/messagequeue/mysql/publisher.go b/platform/extension/messagequeue/mysql/publisher.go
index 1719a2633..07dbd0830 100644
--- a/platform/extension/messagequeue/mysql/publisher.go
+++ b/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
diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go
index 9cc12f291..63182f1b4 100644
--- a/platform/extension/messagequeue/mysql/subscriber.go
+++ b/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()
diff --git a/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go b/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go
index b7967f96a..5a86a0e3c 100644
--- a/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go
+++ b/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()
 
diff --git a/platform/metrics/README.md b/platform/metrics/README.md
index 16f8fe0f5..c960a89fb 100644
--- a/platform/metrics/README.md
+++ b/platform/metrics/README.md
@@ -1,12 +1,12 @@
 # Metrics Utilities (`platform/metrics`)
 
-The `metrics` package provides reusable helpers for emitting counters, timers, histograms, and gauges on a `tally.Scope`. It standardizes metric names across controllers and integrates with `platform/errs` for automatic error classification tags.
+The `metrics` package provides reusable helpers for emitting counters, histograms, and gauges on a `tally.Scope`. It standardizes metric names across controllers and integrates with `platform/errs` for automatic error classification tags.
 
 ## Design
 
 **Free functions on `tally.Scope`** — no wrapper types. Existing constructors accept `tally.Scope` and don't need to change.
 
-**Operation lifecycle** — `Begin` and `Complete` tie the full metrics lifecycle together. `Begin` captures the start time and emits `{name}.called`; `Complete` emits succeeded/failed counters, a latency timer, and a latency histogram. This prevents mismatched or forgotten metrics calls.
+**Operation lifecycle** — `Begin` and `Complete` tie the full metrics lifecycle together. `Begin` captures the start time and emits `{name}.called`; `Complete` emits succeeded/failed counters and a latency histogram. This prevents mismatched or forgotten metrics calls.
 
 **Error-aware tagging** — `ErrorTags` integrates with `platform/errs` to produce `error_origin=user|infra`, `retryable=true|false`, and `dependency=true` tags automatically. `Complete` uses these to tag latency metrics on failure.
 
@@ -19,13 +19,15 @@ For any operation with a clear start/end, use `Begin`/`Complete`:
 | Function | Emits |
 |----------|-------|
 | `Begin(scope, name, ...tags)` | `{name}.called` counter +1, returns `Op` |
-| `op.Complete(err)` | `{name}.succeeded` or `{name}.failed` counter, `{name}.latency` timer, `{name}.latency_histogram` histogram — all tagged with `result=success\|error` and error classification tags on failure |
+| `op.Complete(err, buckets)` | `{name}.succeeded` or `{name}.failed` counter, `{name}.latency` histogram (recorded with the given `buckets`) — tagged with `result=success\|error` and error classification tags on failure |
+
+`buckets` is required — there is no default. Operations differ widely in expected latency, so the caller passes the bucket set (see [Latency Buckets](#latency-buckets)) that matches the operation.
 
 ```go
 // RPC controller
 func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *pb.LandResponse, retErr error) {
     op := metrics.Begin(c.scope, "land")
-    defer func() { op.Complete(retErr) }()
+    defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
     // ... business logic ...
     return &pb.LandResponse{Sqid: request.ID}, nil
@@ -34,7 +36,7 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p
 // Queue controller
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
     op := metrics.Begin(c.scope, "process")
-    defer func() { op.Complete(retErr) }()
+    defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
     // ... business logic ...
     return nil
@@ -43,13 +45,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 
 On success, `Complete` emits:
 - `{name}.succeeded` counter +1
-- `{name}.latency` timer tagged `result=success`
-- `{name}.latency_histogram` histogram tagged `result=success`
+- `{name}.latency` histogram tagged `result=success`
 
 On failure, `Complete` emits:
 - `{name}.failed` counter +1
-- `{name}.latency` timer tagged `result=error`, `error_origin=user|infra`, `retryable=true|false`, and optionally `dependency=true`
-- `{name}.latency_histogram` histogram with the same tags
+- `{name}.latency` histogram tagged `result=error`, `error_origin=user|infra`, `retryable=true|false`, and optionally `dependency=true`
 
 ## Named Helpers
 
@@ -58,7 +58,6 @@ For ad-hoc metrics that don't fit the Begin/Complete lifecycle. All follow the `
 | Function | Emits | Example |
 |----------|-------|---------|
 | `NamedCounter(scope, name, counter, value, ...tags)` | `{name}.{counter}` counter | `publish.attempts` |
-| `NamedTimer(scope, name, timer, duration, ...tags)` | `{name}.{timer}` timer | `publish.queue_latency` |
 | `NamedHistogram(scope, name, histogram, buckets, ...tags)` | `{name}.{histogram}` histogram | `process.duration` |
 | `NamedGauge(scope, name, gauge, value, ...tags)` | `{name}.{gauge}` gauge | `consumer.pending_messages` |
 
@@ -66,17 +65,21 @@ For ad-hoc metrics that don't fit the Begin/Complete lifecycle. All follow the `
 // Count a specific sub-event
 metrics.NamedCounter(c.scope, "publish", "attempts", 1)
 
-// Record a specific sub-latency
-metrics.NamedTimer(c.scope, "publish", "queue_latency", elapsed)
+// Record a one-shot sub-latency as a histogram (pass the bucket set that fits)
+metrics.NamedHistogram(c.scope, "publish", "queue_latency", metrics.StorageLatencyBuckets).RecordDuration(elapsed)
 
 // Track current queue depth (goes up and down)
 metrics.NamedGauge(c.scope, "consumer", "pending_messages", float64(len(pending)))
 
-// Create a reusable histogram (store on struct, call RecordDuration per invocation)
-h := metrics.NamedHistogram(c.scope, "process", "duration", tally.DurationBuckets{...})
+// Reuse a histogram on a hot path (store on struct, call RecordDuration per invocation)
+h := metrics.NamedHistogram(c.scope, "process", "duration", metrics.FastLatencyBuckets)
 h.RecordDuration(elapsed)
 ```
 
+### Why histograms, not timers
+
+Durations are recorded as **histograms**, never timers. A timer ships raw durations and the monitoring backend derives percentiles (p50/p99/max) **per time series** — one series per unique combination of metric name and tag values, so each distinct tag value (region, zone, …) multiplies the series count. The moment a dashboard or alert spans more than one series — rolling up a tag you didn't pin to a single value — the backend has to combine already-aggregated per-series statistics, and timer percentiles don't combine: the p99 across N series is not the average, max, or any function of each series' p99. Only the (count-weighted) mean survives, so precision degrades as the number of aggregated series grows — and high-cardinality tags make it worse. The rollups you reach for during an incident ("p99 across the whole region") are exactly the imprecise ones. Bucketed histograms merge exactly: summing per-series bucket counts reconstructs the true combined distribution, so every percentile stays accurate at any aggregation level and over any time window.
+
 ## Error Tags
 
 `ErrorTags` classifies errors using `platform/errs` and returns tags for dimensional filtering:
@@ -101,21 +104,25 @@ Use `NewTag` to pass additional dimensional tags to any helper:
 
 ```go
 op := metrics.Begin(c.scope, "process", metrics.NewTag("queue", req.Queue))
-defer func() { op.Complete(retErr) }()
+defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 metrics.NamedCounter(c.scope, "publish", "attempts", 1, metrics.NewTag("topic", c.topic))
 ```
 
 ## Latency Buckets
 
-`Complete` uses default latency buckets (5ms to 4h) automatically, suitable for both fast RPCs and long-running operations like builds and merges:
+There is **no default** bucket set: operations range from sub-millisecond in-memory work to multi-hour builds, and one set can't serve all of them well. Both `Op.Complete` and `NamedHistogram` require the caller to pass buckets, so resolution concentrates where the operation's latency actually lands and buckets far outside that range don't waste series cardinality. The package exports three common sets:
 
-```
-5ms, 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s, 30s, 1m, 2m, 5m, 10m, 30m, 1h, 2h, 4h
-```
+| Set | Range | Use for |
+|-----|-------|---------|
+| `FastLatencyBuckets` | ~100µs – 5s | Fast in-process work: scoring, cache lookups, CPU-bound operations |
+| `StorageLatencyBuckets` | ~1ms – 1m | Storage and message-queue round-trips: DB reads/writes, publish/consume, RPC handlers |
+| `LongLatencyBuckets` | ~5ms – 4h | Long-running pipeline work and external calls: builds, merges, git pushes, provider calls |
 
-For custom histograms, pass your own buckets to `NamedHistogram`:
+Pass one of these, or your own `tally.DurationBuckets` when none fits:
 
 ```go
-h := metrics.NamedHistogram(c.scope, "build", "duration", tally.DurationBuckets{...})
+defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+
+h := metrics.NamedHistogram(c.scope, "build", "duration", tally.DurationBuckets{ /* custom */ })
 ```
diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go
index 473e3e27a..858ebf1ba 100644
--- a/platform/metrics/metrics.go
+++ b/platform/metrics/metrics.go
@@ -34,41 +34,88 @@ func NewTag(key, value string) Tag {
 	return Tag{Key: key, Value: value}
 }
 
-// defaultLatencyBuckets provides pre-defined duration buckets for common latency histograms.
-// Covers sub-millisecond to multi-hour ranges suitable for RPC calls, queue processing,
-// and long-running operations like builds and merges.
-var defaultLatencyBuckets = tally.DurationBuckets{
-	5 * time.Millisecond,
-	10 * time.Millisecond,
-	25 * time.Millisecond,
-	50 * time.Millisecond,
-	100 * time.Millisecond,
-	250 * time.Millisecond,
-	500 * time.Millisecond,
-	1 * time.Second,
-	2500 * time.Millisecond,
-	5 * time.Second,
-	10 * time.Second,
-	30 * time.Second,
-	1 * time.Minute,
-	2 * time.Minute,
-	5 * time.Minute,
-	10 * time.Minute,
-	30 * time.Minute,
-	1 * time.Hour,
-	2 * time.Hour,
-	4 * time.Hour,
-}
+// Common duration bucket sets for latency histograms. Operations differ widely
+// in expected latency, so there is no single default — pick the set whose range
+// matches the operation and pass it to Op.Complete or NamedHistogram. Buckets
+// far outside an operation's real latency waste series cardinality and lose
+// resolution where the data actually lands.
+var (
+	// FastLatencyBuckets suits fast in-process operations (~microseconds to
+	// seconds): scoring, cache lookups, and other CPU-bound work.
+	FastLatencyBuckets = tally.DurationBuckets{
+		100 * time.Microsecond,
+		250 * time.Microsecond,
+		500 * time.Microsecond,
+		1 * time.Millisecond,
+		2500 * time.Microsecond,
+		5 * time.Millisecond,
+		10 * time.Millisecond,
+		25 * time.Millisecond,
+		50 * time.Millisecond,
+		100 * time.Millisecond,
+		250 * time.Millisecond,
+		500 * time.Millisecond,
+		1 * time.Second,
+		2500 * time.Millisecond,
+		5 * time.Second,
+	}
+
+	// StorageLatencyBuckets suits storage and message-queue round-trips
+	// (~1ms to a minute): database reads/writes, publish/consume, and RPC
+	// handlers whose latency is dominated by such calls.
+	StorageLatencyBuckets = tally.DurationBuckets{
+		1 * time.Millisecond,
+		2500 * time.Microsecond,
+		5 * time.Millisecond,
+		10 * time.Millisecond,
+		25 * time.Millisecond,
+		50 * time.Millisecond,
+		100 * time.Millisecond,
+		250 * time.Millisecond,
+		500 * time.Millisecond,
+		1 * time.Second,
+		2500 * time.Millisecond,
+		5 * time.Second,
+		10 * time.Second,
+		30 * time.Second,
+		1 * time.Minute,
+	}
+
+	// LongLatencyBuckets suits long-running pipeline work and external calls
+	// (~5ms to hours): builds, merges, git pushes, and external provider calls.
+	LongLatencyBuckets = tally.DurationBuckets{
+		5 * time.Millisecond,
+		10 * time.Millisecond,
+		25 * time.Millisecond,
+		50 * time.Millisecond,
+		100 * time.Millisecond,
+		250 * time.Millisecond,
+		500 * time.Millisecond,
+		1 * time.Second,
+		2500 * time.Millisecond,
+		5 * time.Second,
+		10 * time.Second,
+		30 * time.Second,
+		1 * time.Minute,
+		2 * time.Minute,
+		5 * time.Minute,
+		10 * time.Minute,
+		30 * time.Minute,
+		1 * time.Hour,
+		2 * time.Hour,
+		4 * time.Hour,
+	}
+)
 
 // Op tracks the lifecycle of a named operation. It captures the start time on
 // creation, emits a {name}.called counter, and records the outcome (succeeded/failed
-// counters + latency timer with error classification tags) when Complete is called.
+// counters + latency histogram with error classification tags) when Complete is called.
 //
 // Usage:
 //
 //	func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 //	    op := metrics.Begin(c.scope, "process")
-//	    defer func() { op.Complete(retErr) }()
+//	    defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 //	    // ... business logic ...
 //	}
 type Op struct {
@@ -87,19 +134,25 @@ func Begin(scope tally.Scope, name string, tags ...Tag) Op {
 }
 
 // Complete records the outcome of the operation. It emits a {name}.succeeded or
-// {name}.failed counter based on err, and records elapsed time on both
-// {name}.latency (timer) and {name}.latency_histogram (histogram with
-// defaultLatencyBuckets for percentile distributions), tagged with result=success|error.
-// On failure, error classification tags (error_origin, retryable, dependency)
-// are added to both the timer and histogram.
-func (o Op) Complete(err error) {
+// {name}.failed counter based on err, and records elapsed time on the
+// {name}.latency histogram (using the given buckets for percentile
+// distributions), tagged with result=success|error. On failure, error
+// classification tags (error_origin, retryable, dependency) are added to the
+// histogram.
+//
+// buckets is required and has no default: operations differ widely in expected
+// latency, so the caller picks a set (e.g. FastLatencyBuckets,
+// StorageLatencyBuckets, LongLatencyBuckets) matching the operation.
+//
+// Latency is recorded as a histogram rather than a timer because timer
+// percentiles cannot be combined across time series (see the package README).
+func (o Op) Complete(err error, buckets tally.Buckets) {
 	elapsed := time.Since(o.start)
 
 	if err == nil {
 		o.scope.Counter("succeeded").Inc(1)
 		s := o.scope.Tagged(map[string]string{"result": "success"})
-		s.Timer("latency").Record(elapsed)
-		s.Histogram("latency_histogram", defaultLatencyBuckets).RecordDuration(elapsed)
+		s.Histogram("latency", buckets).RecordDuration(elapsed)
 		return
 	}
 
@@ -110,8 +163,7 @@ func (o Op) Complete(err error) {
 		latencyTags[t.Key] = t.Value
 	}
 	s := o.scope.Tagged(latencyTags)
-	s.Timer("latency").Record(elapsed)
-	s.Histogram("latency_histogram", defaultLatencyBuckets).RecordDuration(elapsed)
+	s.Histogram("latency", buckets).RecordDuration(elapsed)
 }
 
 // NamedCounter increments the {name}.{counter} counter by value.
@@ -119,11 +171,6 @@ func NamedCounter(scope tally.Scope, name string, counter string, value int64, t
 	tagged(scope, tags).SubScope(name).Counter(counter).Inc(value)
 }
 
-// NamedTimer records a duration on the {name}.{timer} timer.
-func NamedTimer(scope tally.Scope, name string, timer string, d time.Duration, tags ...Tag) {
-	tagged(scope, tags).SubScope(name).Timer(timer).Record(d)
-}
-
 // NamedHistogram returns a tally.Histogram at {name}.{histogram} with the given
 // bucket configuration. Store the returned histogram and call RecordDuration or
 // RecordValue on each invocation.
diff --git a/platform/metrics/metrics_test.go b/platform/metrics/metrics_test.go
index e0ea63d7b..7318e4918 100644
--- a/platform/metrics/metrics_test.go
+++ b/platform/metrics/metrics_test.go
@@ -90,7 +90,7 @@ func TestComplete(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			scope := tally.NewTestScope("", nil)
 			op := Begin(scope, "process")
-			op.Complete(tt.err)
+			op.Complete(tt.err, FastLatencyBuckets)
 
 			snapshot := scope.Snapshot()
 			counters := snapshot.Counters()
@@ -105,14 +105,9 @@ func TestComplete(t *testing.T) {
 				assert.True(t, ok, "expected process.succeeded counter")
 				assert.Equal(t, int64(1), c.Value())
 
-				timers := snapshot.Timers()
-				timer, ok := timers["process.latency+result=success"]
-				assert.True(t, ok, "expected process.latency timer with result=success")
-				assert.NotEmpty(t, timer.Values())
-
 				histograms := snapshot.Histograms()
-				_, ok = histograms["process.latency_histogram+result=success"]
-				assert.True(t, ok, "expected process.latency_histogram with result=success")
+				_, ok = histograms["process.latency+result=success"]
+				assert.True(t, ok, "expected process.latency histogram with result=success")
 			} else {
 				c, ok := counters["process.failed+"]
 				assert.True(t, ok, "expected process.failed counter")
@@ -127,15 +122,7 @@ func TestComplete(t *testing.T) {
 				tagSuffix += ",result=" + tt.expectResultTag
 				tagSuffix += ",retryable=" + tt.expectRetryable
 
-				timerKey := "process.latency+" + tagSuffix
-				timers := snapshot.Timers()
-				timer, ok := timers[timerKey]
-				assert.True(t, ok, "expected timer key %s, got keys: %v", timerKey, timerKeys(timers))
-				if ok {
-					assert.NotEmpty(t, timer.Values())
-				}
-
-				histogramKey := "process.latency_histogram+" + tagSuffix
+				histogramKey := "process.latency+" + tagSuffix
 				histograms := snapshot.Histograms()
 				_, ok = histograms[histogramKey]
 				assert.True(t, ok, "expected histogram key %s, got keys: %v", histogramKey, histogramKeys(histograms))
@@ -147,7 +134,7 @@ func TestComplete(t *testing.T) {
 func TestBegin_WithTags(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
 	op := Begin(scope, "process", NewTag("env", "prod"))
-	op.Complete(nil)
+	op.Complete(nil, FastLatencyBuckets)
 
 	snapshot := scope.Snapshot()
 	counters := snapshot.Counters()
@@ -172,20 +159,9 @@ func TestNamedCounter(t *testing.T) {
 	assert.Equal(t, int64(5), c.Value())
 }
 
-func TestNamedTimer(t *testing.T) {
-	scope := tally.NewTestScope("", nil)
-	NamedTimer(scope, "publish", "queue_latency", 42*time.Millisecond)
-
-	snapshot := scope.Snapshot()
-	timers := snapshot.Timers()
-	timer, ok := timers["publish.queue_latency+"]
-	assert.True(t, ok, "expected publish.queue_latency timer")
-	assert.Equal(t, []time.Duration{42 * time.Millisecond}, timer.Values())
-}
-
 func TestNamedHistogram(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
-	h := NamedHistogram(scope, "process", "duration", defaultLatencyBuckets)
+	h := NamedHistogram(scope, "process", "duration", StorageLatencyBuckets)
 	assert.NotNil(t, h)
 
 	h.RecordDuration(50 * time.Millisecond)
@@ -207,11 +183,20 @@ func TestNamedGauge(t *testing.T) {
 	assert.Equal(t, float64(42), g.Value())
 }
 
-func TestDefaultLatencyBuckets_Sorted(t *testing.T) {
-	for i := 1; i < len(defaultLatencyBuckets); i++ {
-		assert.Greater(t, defaultLatencyBuckets[i], defaultLatencyBuckets[i-1],
-			"defaultLatencyBuckets[%d] (%v) must be greater than defaultLatencyBuckets[%d] (%v)",
-			i, defaultLatencyBuckets[i], i-1, defaultLatencyBuckets[i-1])
+func TestLatencyBuckets_Sorted(t *testing.T) {
+	sets := map[string]tally.DurationBuckets{
+		"FastLatencyBuckets":    FastLatencyBuckets,
+		"StorageLatencyBuckets": StorageLatencyBuckets,
+		"LongLatencyBuckets":    LongLatencyBuckets,
+	}
+	for name, buckets := range sets {
+		t.Run(name, func(t *testing.T) {
+			for i := 1; i < len(buckets); i++ {
+				assert.Greater(t, buckets[i], buckets[i-1],
+					"%s[%d] (%v) must be greater than %s[%d] (%v)",
+					name, i, buckets[i], name, i-1, buckets[i-1])
+			}
+		})
 	}
 }
 
@@ -269,15 +254,6 @@ func TestErrorTags(t *testing.T) {
 	}
 }
 
-// timerKeys extracts map keys for error messages.
-func timerKeys(m map[string]tally.TimerSnapshot) []string {
-	keys := make([]string, 0, len(m))
-	for k := range m {
-		keys = append(keys, k)
-	}
-	return keys
-}
-
 // counterKeys extracts map keys for error messages.
 func counterKeys(m map[string]tally.CounterSnapshot) []string {
 	keys := make([]string, 0, len(m))
diff --git a/runway/controller/merge/merge.go b/runway/controller/merge/merge.go
index dfba98016..49712e724 100644
--- a/runway/controller/merge/merge.go
+++ b/runway/controller/merge/merge.go
@@ -80,7 +80,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/runway/controller/mergeconflictcheck/mergeconflictcheck.go b/runway/controller/mergeconflictcheck/mergeconflictcheck.go
index a08112351..d5cb176ec 100644
--- a/runway/controller/mergeconflictcheck/mergeconflictcheck.go
+++ b/runway/controller/mergeconflictcheck/mergeconflictcheck.go
@@ -80,7 +80,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/runway/controller/ping.go b/runway/controller/ping.go
index b7b59f5e9..c2416edbb 100644
--- a/runway/controller/ping.go
+++ b/runway/controller/ping.go
@@ -44,7 +44,7 @@ func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *p
 	const opName = "ping"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
 
 	message := "pong!"
 	isEcho := false
diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go
index 11395870c..5d8d666f0 100644
--- a/stovepipe/controller/build/build.go
+++ b/stovepipe/controller/build/build.go
@@ -81,7 +81,7 @@ func NewController(
 // ack (success) or an error to nack (retry) / reject (DLQ).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, _opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go
index bdfdb229c..de026b6e7 100644
--- a/stovepipe/controller/buildsignal/buildsignal.go
+++ b/stovepipe/controller/buildsignal/buildsignal.go
@@ -96,7 +96,7 @@ func NewController(
 // ack (success) or an error to nack (retry) / reject (DLQ).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, _opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go
index 8e0b420e3..56261c449 100644
--- a/stovepipe/controller/dlq/request.go
+++ b/stovepipe/controller/dlq/request.go
@@ -68,7 +68,7 @@ func NewController(
 // instead of dead-lettering the DLQ message itself.
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, _opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go
index 7e009d4aa..dd60e493d 100644
--- a/stovepipe/controller/ingest.go
+++ b/stovepipe/controller/ingest.go
@@ -90,7 +90,7 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
 	const opName = "ingest"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	if req.Queue == "" {
 		return entity.IngestResult{}, fmt.Errorf("requires the request to have a queue name specified: %w", ErrInvalidRequest)
diff --git a/stovepipe/controller/ping.go b/stovepipe/controller/ping.go
index f43586d0d..6347f363d 100644
--- a/stovepipe/controller/ping.go
+++ b/stovepipe/controller/ping.go
@@ -44,7 +44,7 @@ func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *p
 	const opName = "ping"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
 
 	message := "pong!"
 	isEcho := false
diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go
index 12d8fd1a2..c0b7d72a4 100644
--- a/stovepipe/controller/process/process.go
+++ b/stovepipe/controller/process/process.go
@@ -85,7 +85,7 @@ func NewController(
 // and admits the latest when a slot is open. Returns nil to ack (success) or an error to nack (retry).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, _opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/extension/storage/mysql/build_store.go b/stovepipe/extension/storage/mysql/build_store.go
index 45dd5aefe..58eab1cb3 100644
--- a/stovepipe/extension/storage/mysql/build_store.go
+++ b/stovepipe/extension/storage/mysql/build_store.go
@@ -40,7 +40,7 @@ func NewBuildStore(db *sql.DB, scope tally.Scope) storage.BuildStore {
 // Create persists a new build. Returns ErrAlreadyExists if the build ID already exists.
 func (b *buildStore) Create(ctx context.Context, build entity.Build) (retErr error) {
 	op := metrics.Begin(b.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := b.db.ExecContext(ctx,
 		`INSERT INTO build (id, request_id, status, version)
@@ -63,7 +63,7 @@ func (b *buildStore) Create(ctx context.Context, build entity.Build) (retErr err
 // Get retrieves a build by ID. Returns ErrNotFound if the build is not found.
 func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retErr error) {
 	op := metrics.Begin(b.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var build entity.Build
 	err := b.db.QueryRowContext(ctx,
@@ -93,7 +93,7 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 // caller owns version arithmetic.
 func (b *buildStore) Update(ctx context.Context, build entity.Build, oldVersion, newVersion int32) (retErr error) {
 	op := metrics.Begin(b.scope, "update")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := b.db.ExecContext(ctx,
 		`UPDATE build
diff --git a/stovepipe/extension/storage/mysql/queue_store.go b/stovepipe/extension/storage/mysql/queue_store.go
index ee922d319..63492e247 100644
--- a/stovepipe/extension/storage/mysql/queue_store.go
+++ b/stovepipe/extension/storage/mysql/queue_store.go
@@ -39,7 +39,7 @@ func NewQueueStore(db *sql.DB, scope tally.Scope) storage.QueueStore {
 // Create persists a new queue row. Returns ErrAlreadyExists if the name already exists.
 func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr error) {
 	op := metrics.Begin(q.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := q.db.ExecContext(ctx,
 		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
@@ -62,7 +62,7 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err
 // Get retrieves a queue by name. Returns ErrNotFound if the queue is not found.
 func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, retErr error) {
 	op := metrics.Begin(q.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var queue entity.Queue
 	err := q.db.QueryRowContext(ctx,
@@ -90,7 +90,7 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re
 // writing newVersion. Returns ErrVersionMismatch if the stored version does not match.
 func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion, newVersion int32) (retErr error) {
 	op := metrics.Begin(q.scope, "update")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := q.db.ExecContext(ctx,
 		`UPDATE queue
diff --git a/stovepipe/extension/storage/mysql/request_store.go b/stovepipe/extension/storage/mysql/request_store.go
index 7e7d112e7..38a2b12d7 100644
--- a/stovepipe/extension/storage/mysql/request_store.go
+++ b/stovepipe/extension/storage/mysql/request_store.go
@@ -45,7 +45,7 @@ func NewRequestStore(db *sql.DB, scope tally.Scope) storage.RequestStore {
 // Create persists a new request. Returns ErrAlreadyExists if the request ID already exists.
 func (r *requestStore) Create(ctx context.Context, request entity.Request) (retErr error) {
 	op := metrics.Begin(r.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := r.db.ExecContext(ctx,
 		`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
@@ -71,7 +71,7 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
 // Get retrieves a request by ID. Returns ErrNotFound if the request is not found.
 func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, retErr error) {
 	op := metrics.Begin(r.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var req entity.Request
 	err := r.db.QueryRowContext(ctx,
@@ -104,7 +104,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 // version arithmetic.
 func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVersion, newVersion int32) (retErr error) {
 	op := metrics.Begin(r.scope, "update")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := r.db.ExecContext(ctx,
 		`UPDATE request
diff --git a/stovepipe/extension/storage/mysql/request_uri_store.go b/stovepipe/extension/storage/mysql/request_uri_store.go
index 1f5c5aea4..c63f33bb8 100644
--- a/stovepipe/extension/storage/mysql/request_uri_store.go
+++ b/stovepipe/extension/storage/mysql/request_uri_store.go
@@ -45,7 +45,7 @@ func NewRequestURIStore(db *sql.DB, scope tally.Scope) storage.RequestURIStore {
 // is already mapped to a request.
 func (r *requestURIStore) Create(ctx context.Context, queue, uri, id string) (retErr error) {
 	op := metrics.Begin(r.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := r.db.ExecContext(ctx,
 		"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?)",
@@ -64,7 +64,7 @@ func (r *requestURIStore) Create(ctx context.Context, queue, uri, id string) (re
 // GetIDByURI returns the id of the request validating (queue, uri). Returns ErrNotFound if absent.
 func (r *requestURIStore) GetIDByURI(ctx context.Context, queue, uri string) (ret string, retErr error) {
 	op := metrics.Begin(r.scope, "get_id_by_uri")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var id string
 	err := r.db.QueryRowContext(ctx,
diff --git a/submitqueue/extension/changeprovider/github/provider.go b/submitqueue/extension/changeprovider/github/provider.go
index 67907debb..58beb85ee 100644
--- a/submitqueue/extension/changeprovider/github/provider.go
+++ b/submitqueue/extension/changeprovider/github/provider.go
@@ -46,7 +46,7 @@ func NewProvider(params Params) changeprovider.ChangeProvider {
 // Returns one ChangeInfo per URI (one per PR in stacked changes).
 func (p *provider) Get(ctx context.Context, request entity.Request) (_ []entity.ChangeInfo, retErr error) {
 	op := coremetrics.Begin(p.metricsScope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, coremetrics.LongLatencyBuckets) }()
 
 	change := request.Change
 
diff --git a/submitqueue/extension/changeprovider/phabricator/provider.go b/submitqueue/extension/changeprovider/phabricator/provider.go
index 89d68a799..a5553e479 100644
--- a/submitqueue/extension/changeprovider/phabricator/provider.go
+++ b/submitqueue/extension/changeprovider/phabricator/provider.go
@@ -49,7 +49,7 @@ func NewProvider(params Params) changeprovider.ChangeProvider {
 // Returns one ChangeInfo per URI.
 func (p *provider) Get(ctx context.Context, request entity.Request) (_ []entity.ChangeInfo, retErr error) {
 	op := coremetrics.Begin(p.metricsScope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, coremetrics.LongLatencyBuckets) }()
 
 	change := request.Change
 
diff --git a/submitqueue/extension/mergechecker/github/checker.go b/submitqueue/extension/mergechecker/github/checker.go
index 1104893aa..4fcb76a56 100644
--- a/submitqueue/extension/mergechecker/github/checker.go
+++ b/submitqueue/extension/mergechecker/github/checker.go
@@ -65,7 +65,7 @@ func (c *mergeChecker) Check(ctx context.Context, request entity.Request) (resul
 	const opName = "check"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	change := request.Change
 
diff --git a/submitqueue/extension/pusher/git/git_pusher.go b/submitqueue/extension/pusher/git/git_pusher.go
index a281b669e..97dde484d 100644
--- a/submitqueue/extension/pusher/git/git_pusher.go
+++ b/submitqueue/extension/pusher/git/git_pusher.go
@@ -136,7 +136,7 @@ func NewPusher(params Params) pusher.Pusher {
 // Push fulfils the pusher.Pusher contract.
 func (p *gitPusher) Push(ctx context.Context, batches []entity.Batch) (ret entity.PushResult, retErr error) {
 	op := coremetrics.Begin(p.metricsScope, "push")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, coremetrics.LongLatencyBuckets) }()
 
 	// Resolve each batch's changes, keeping per-batch counts so the flat
 	// outcomes can be regrouped per batch on success.
diff --git a/submitqueue/extension/scorer/composite/scorer.go b/submitqueue/extension/scorer/composite/scorer.go
index ed342c4d6..c82471dce 100644
--- a/submitqueue/extension/scorer/composite/scorer.go
+++ b/submitqueue/extension/scorer/composite/scorer.go
@@ -92,7 +92,7 @@ func New(scorers map[string]scorer.Scorer, reduce ReduceFunc, scope tally.Scope)
 // reduce function. If any child scorer returns an error, that error is returned immediately.
 func (c *compositeScorer) Score(ctx context.Context, batch entity.Batch) (ret float64, retErr error) {
 	op := metrics.Begin(c.scope, "score")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
 
 	scores := make(map[string]float64, len(c.scorers))
 	for name, s := range c.scorers {
diff --git a/submitqueue/extension/scorer/heuristic/scorer.go b/submitqueue/extension/scorer/heuristic/scorer.go
index e37e2d4bd..3cdd8928d 100644
--- a/submitqueue/extension/scorer/heuristic/scorer.go
+++ b/submitqueue/extension/scorer/heuristic/scorer.go
@@ -70,7 +70,7 @@ func New(resolver changeset.Resolver, buckets []Bucket, valueFunc ValueFunc, sco
 // if no bucket matches.
 func (s *heuristicScorer) Score(ctx context.Context, batch entity.Batch) (ret float64, retErr error) {
 	op := metrics.Begin(s.scope, "score")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
 	changes, err := s.resolver.DetailedForBatch(ctx, batch)
 	if err != nil {
 		return 0, err
diff --git a/submitqueue/extension/storage/mysql/batch_dependent_store.go b/submitqueue/extension/storage/mysql/batch_dependent_store.go
index 64de64315..e7f3cd7b3 100644
--- a/submitqueue/extension/storage/mysql/batch_dependent_store.go
+++ b/submitqueue/extension/storage/mysql/batch_dependent_store.go
@@ -42,7 +42,7 @@ func NewBatchDependentStore(db *sql.DB, scope tally.Scope) storage.BatchDependen
 // Get retrieves the batch dependent by batch ID. Returns ErrNotFound if the batch dependent is not found.
 func (s *batchDependentStore) Get(ctx context.Context, batchID string) (ret entity.BatchDependent, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var bd entity.BatchDependent
 	var dependentsJSON []byte
@@ -69,7 +69,7 @@ func (s *batchDependentStore) Get(ctx context.Context, batchID string) (ret enti
 // Create creates a new batch dependent. Returns ErrAlreadyExists if the entry already exists.
 func (s *batchDependentStore) Create(ctx context.Context, batchDependent entity.BatchDependent) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	dependentsJSON, err := json.Marshal(batchDependent.Dependents)
 	if err != nil {
@@ -96,7 +96,7 @@ func (s *batchDependentStore) Create(ctx context.Context, batchDependent entity.
 // Version arithmetic is owned by the caller; this is a pure conditional write.
 func (s *batchDependentStore) UpdateDependents(ctx context.Context, batchID string, oldVersion, newVersion int32, dependents []string) (retErr error) {
 	op := metrics.Begin(s.scope, "update_dependents")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	dependentsJSON, err := json.Marshal(dependents)
 	if err != nil {
diff --git a/submitqueue/extension/storage/mysql/batch_store.go b/submitqueue/extension/storage/mysql/batch_store.go
index 07020dac2..170490150 100644
--- a/submitqueue/extension/storage/mysql/batch_store.go
+++ b/submitqueue/extension/storage/mysql/batch_store.go
@@ -43,7 +43,7 @@ func NewBatchStore(db *sql.DB, scope tally.Scope) storage.BatchStore {
 // Get retrieves a batch by ID. Returns ErrNotFound if the batch is not found.
 func (s *batchStore) Get(ctx context.Context, id string) (ret entity.Batch, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var batch entity.Batch
 	var containsJSON []byte
@@ -75,7 +75,7 @@ func (s *batchStore) Get(ctx context.Context, id string) (ret entity.Batch, retE
 // Create creates a new batch. The batch must have a unique ID already assigned. Returns ErrAlreadyExists if the batch ID already exists.
 func (s *batchStore) Create(ctx context.Context, batch entity.Batch) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	containsJSON, err := json.Marshal(batch.Contains)
 	if err != nil {
@@ -107,7 +107,7 @@ func (s *batchStore) Create(ctx context.Context, batch entity.Batch) (retErr err
 // Version arithmetic is owned by the caller; this is a pure conditional write.
 func (s *batchStore) UpdateState(ctx context.Context, id string, oldVersion, newVersion int32, newState entity.BatchState) (retErr error) {
 	op := metrics.Begin(s.scope, "update_state")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := s.db.ExecContext(ctx,
 		"UPDATE batch SET state = ?, version = ? WHERE id = ? AND version = ?",
@@ -143,7 +143,7 @@ func (s *batchStore) UpdateState(ctx context.Context, id string, oldVersion, new
 // Version arithmetic is owned by the caller; this is a pure conditional write.
 func (s *batchStore) UpdateScoreAndState(ctx context.Context, id string, oldVersion, newVersion int32, score float64, newState entity.BatchState) (retErr error) {
 	op := metrics.Begin(s.scope, "update_score_and_state")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := s.db.ExecContext(ctx,
 		"UPDATE batch SET score = ?, state = ?, version = ? WHERE id = ? AND version = ?",
@@ -177,7 +177,7 @@ func (s *batchStore) UpdateScoreAndState(ctx context.Context, id string, oldVers
 // GetByQueueAndStates retrieves all batches that belong to the given queue and are in the given states.
 func (s *batchStore) GetByQueueAndStates(ctx context.Context, queue string, states []entity.BatchState) (ret []entity.Batch, retErr error) {
 	op := metrics.Begin(s.scope, "get_by_queue_and_states")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if len(states) == 0 {
 		return nil, nil
diff --git a/submitqueue/extension/storage/mysql/build_store.go b/submitqueue/extension/storage/mysql/build_store.go
index 6d1b01849..250d59581 100644
--- a/submitqueue/extension/storage/mysql/build_store.go
+++ b/submitqueue/extension/storage/mysql/build_store.go
@@ -41,7 +41,7 @@ func NewBuildStore(db *sql.DB, scope tally.Scope) storage.BuildStore {
 // Get retrieves a build by ID. Returns ErrNotFound if the build is not found.
 func (s *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var build entity.Build
 
@@ -63,7 +63,7 @@ func (s *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 // Create creates a new build. The build must have a unique ID already assigned. Returns ErrAlreadyExists if the build ID already exists.
 func (s *buildStore) Create(ctx context.Context, build entity.Build) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := s.db.ExecContext(ctx,
 		"INSERT INTO build (id, batch_id, status) VALUES (?, ?, ?)",
@@ -83,7 +83,7 @@ func (s *buildStore) Create(ctx context.Context, build entity.Build) (retErr err
 // UpdateStatus updates the status of a build. Returns ErrNotFound if the build is not found.
 func (s *buildStore) UpdateStatus(ctx context.Context, id string, newStatus entity.BuildStatus) (retErr error) {
 	op := metrics.Begin(s.scope, "update_status")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := s.db.ExecContext(ctx,
 		"UPDATE build SET status = ? WHERE id = ?",
diff --git a/submitqueue/extension/storage/mysql/change_store.go b/submitqueue/extension/storage/mysql/change_store.go
index 436b8b20d..dd7755cc2 100644
--- a/submitqueue/extension/storage/mysql/change_store.go
+++ b/submitqueue/extension/storage/mysql/change_store.go
@@ -42,7 +42,7 @@ func NewChangeStore(db *sql.DB, scope tally.Scope) storage.ChangeStore {
 // queue-redelivery of the same request is a no-op.
 func (s *changeStore) Create(ctx context.Context, record entity.ChangeRecord) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	detailsJSON, err := marshalDetails(record.Details)
 	if err != nil {
@@ -62,7 +62,7 @@ func (s *changeStore) Create(ctx context.Context, record entity.ChangeRecord) (r
 // clause to align with the (queue, uri, request_id) PK so this is a PK-prefix scan.
 func (s *changeStore) GetByURI(ctx context.Context, queue string, uri string) (ret []entity.ChangeRecord, retErr error) {
 	op := metrics.Begin(s.scope, "get_by_uri")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	const query = "SELECT uri, request_id, queue, details, created_at, updated_at, version FROM `change` WHERE queue = ? AND uri = ?"
 	rows, err := s.db.QueryContext(ctx, query, queue, uri)
diff --git a/submitqueue/extension/storage/mysql/request_log_store.go b/submitqueue/extension/storage/mysql/request_log_store.go
index b19b21199..afdb7cffe 100644
--- a/submitqueue/extension/storage/mysql/request_log_store.go
+++ b/submitqueue/extension/storage/mysql/request_log_store.go
@@ -44,7 +44,7 @@ func NewRequestLogStore(db *sql.DB, scope tally.Scope) storage.RequestLogStore {
 // without requiring the caller to manage deduplication.
 func (r *requestLogStore) Insert(ctx context.Context, log entity.RequestLog) (retErr error) {
 	op := metrics.Begin(r.scope, "insert")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	metadataJSON, err := json.Marshal(log.Metadata)
 	if err != nil {
@@ -72,7 +72,7 @@ func (r *requestLogStore) Insert(ctx context.Context, log entity.RequestLog) (re
 // timestamp, but it is not included in the SELECT columns and never returned to callers.
 func (r *requestLogStore) List(ctx context.Context, requestID string) (ret []entity.RequestLog, retErr error) {
 	op := metrics.Begin(r.scope, "list")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	rows, err := r.db.QueryContext(ctx,
 		"SELECT request_id, timestamp_ms, status, request_version, last_error, metadata FROM request_log WHERE request_id = ? ORDER BY timestamp_ms ASC, salt ASC",
diff --git a/submitqueue/extension/storage/mysql/request_queue_summary_store.go b/submitqueue/extension/storage/mysql/request_queue_summary_store.go
index 8a292fe04..97871d152 100644
--- a/submitqueue/extension/storage/mysql/request_queue_summary_store.go
+++ b/submitqueue/extension/storage/mysql/request_queue_summary_store.go
@@ -41,7 +41,7 @@ func NewRequestQueueSummaryStore(db *sql.DB, scope tally.Scope) storage.RequestQ
 
 func (s *requestQueueSummaryStore) Create(ctx context.Context, summary entity.RequestQueueSummary) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata)
 	if err != nil {
@@ -67,7 +67,7 @@ func (s *requestQueueSummaryStore) Create(ctx context.Context, summary entity.Re
 
 func (s *requestQueueSummaryStore) Get(ctx context.Context, queue string, receivedAtMs int64, requestID string) (ret entity.RequestQueueSummary, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var changeURIsJSON []byte
 	var metadataJSON []byte
@@ -91,7 +91,7 @@ func (s *requestQueueSummaryStore) Get(ctx context.Context, queue string, receiv
 
 func (s *requestQueueSummaryStore) Update(ctx context.Context, summary entity.RequestQueueSummary, oldVersion, newVersion int32) (retErr error) {
 	op := metrics.Begin(s.scope, "update")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	metadataJSON, err := json.Marshal(normalizeMetadata(summary.Metadata))
 	if err != nil {
@@ -119,7 +119,7 @@ func (s *requestQueueSummaryStore) Update(ctx context.Context, summary entity.Re
 
 func (s *requestQueueSummaryStore) List(ctx context.Context, query storage.RequestQueueSummaryQuery) (ret []entity.RequestQueueSummary, retErr error) {
 	op := metrics.Begin(s.scope, "list")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	statement := `
 		SELECT queue, received_at_ms, request_id, change_uris, status,
diff --git a/submitqueue/extension/storage/mysql/request_store.go b/submitqueue/extension/storage/mysql/request_store.go
index fd542c9d7..12fdaeeb8 100644
--- a/submitqueue/extension/storage/mysql/request_store.go
+++ b/submitqueue/extension/storage/mysql/request_store.go
@@ -42,7 +42,7 @@ func NewRequestStore(db *sql.DB, scope tally.Scope) storage.RequestStore {
 // Get retrieves a land request by ID. Returns ErrNotFound if the request is not found.
 func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, retErr error) {
 	op := metrics.Begin(r.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var req entity.Request
 	var changeURIsJSON []byte
@@ -70,7 +70,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 // Create creates a new land request. The request must have a unique ID already assigned. Returns ErrAlreadyExists if the request ID already exists.
 func (r *requestStore) Create(ctx context.Context, request entity.Request) (retErr error) {
 	op := metrics.Begin(r.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	// Marshal the change URIs to JSON
 	changeURIsJSON, err := json.Marshal(request.Change.URIs)
@@ -98,7 +98,7 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
 // Version arithmetic is owned by the caller; this is a pure conditional write.
 func (r *requestStore) UpdateState(ctx context.Context, id string, oldVersion, newVersion int32, newState entity.RequestState) (retErr error) {
 	op := metrics.Begin(r.scope, "update_state")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	result, err := r.db.ExecContext(ctx,
 		"UPDATE request SET state = ?, version = ? WHERE id = ? AND version = ?",
diff --git a/submitqueue/extension/storage/mysql/request_summary_store.go b/submitqueue/extension/storage/mysql/request_summary_store.go
index 36849dea4..f94f1a285 100644
--- a/submitqueue/extension/storage/mysql/request_summary_store.go
+++ b/submitqueue/extension/storage/mysql/request_summary_store.go
@@ -41,7 +41,7 @@ func NewRequestSummaryStore(db *sql.DB, scope tally.Scope) storage.RequestSummar
 
 func (s *requestSummaryStore) Create(ctx context.Context, summary entity.RequestSummary) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata)
 	if err != nil {
@@ -71,7 +71,7 @@ func (s *requestSummaryStore) Create(ctx context.Context, summary entity.Request
 
 func (s *requestSummaryStore) Get(ctx context.Context, requestID string) (ret entity.RequestSummary, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var changeURIsJSON []byte
 	var metadataJSON []byte
@@ -101,7 +101,7 @@ func (s *requestSummaryStore) Get(ctx context.Context, requestID string) (ret en
 
 func (s *requestSummaryStore) Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) (retErr error) {
 	op := metrics.Begin(s.scope, "update")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	metadata := normalizeMetadata(summary.Metadata)
 	metadataJSON, err := json.Marshal(metadata)
diff --git a/submitqueue/extension/storage/mysql/request_uri_store.go b/submitqueue/extension/storage/mysql/request_uri_store.go
index 35375d5ce..bda76569a 100644
--- a/submitqueue/extension/storage/mysql/request_uri_store.go
+++ b/submitqueue/extension/storage/mysql/request_uri_store.go
@@ -40,7 +40,7 @@ func NewRequestURIStore(db *sql.DB, scope tally.Scope) storage.RequestURIStore {
 
 func (s *requestURIStore) Create(ctx context.Context, mapping entity.RequestURI) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	_, err := s.db.ExecContext(ctx,
 		"INSERT INTO change_uri_request_mapping (change_uri, received_at_ms, request_id) VALUES (?, ?, ?)",
@@ -58,7 +58,7 @@ func (s *requestURIStore) Create(ctx context.Context, mapping entity.RequestURI)
 
 func (s *requestURIStore) ListByURI(ctx context.Context, changeURI string, limit int) (ret []entity.RequestURI, retErr error) {
 	op := metrics.Begin(s.scope, "list_by_uri")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	rows, err := s.db.QueryContext(ctx, `
 		SELECT change_uri, received_at_ms, request_id
diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go
index 5e4ec6e3b..8b753b7a8 100644
--- a/submitqueue/gateway/controller/land.go
+++ b/submitqueue/gateway/controller/land.go
@@ -92,7 +92,7 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu
 	const opName = "land"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	// Validate provider-agnostic request constraints before allocating an sqid.
 	if err := validateQueueIdentifier(req.Queue); err != nil {
diff --git a/submitqueue/gateway/controller/list.go b/submitqueue/gateway/controller/list.go
index 187d94394..f0d28d0ef 100644
--- a/submitqueue/gateway/controller/list.go
+++ b/submitqueue/gateway/controller/list.go
@@ -65,7 +65,7 @@ func NewListController(logger *zap.SugaredLogger, scope tally.Scope, requestQueu
 // List returns one page of requests received for a queue in the supplied half-open time range.
 func (c *ListController) List(ctx context.Context, req entity.ListRequest) (result entity.ListResult, retErr error) {
 	op := metrics.Begin(c.metricsScope, "list")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if err := validateStoredIdentifier("queue", req.Queue); err != nil {
 		return entity.ListResult{}, fmt.Errorf("invalid queue: %w", err)
diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go
index 730add0d2..8b10fefa7 100644
--- a/submitqueue/gateway/controller/log/log.go
+++ b/submitqueue/gateway/controller/log/log.go
@@ -69,7 +69,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/gateway/controller/request_history.go b/submitqueue/gateway/controller/request_history.go
index d306d4a5e..4311fa251 100644
--- a/submitqueue/gateway/controller/request_history.go
+++ b/submitqueue/gateway/controller/request_history.go
@@ -50,7 +50,7 @@ func NewRequestHistoryController(logger *zap.SugaredLogger, scope tally.Scope, r
 // GetRequestHistoryByID returns every retained request-log event for one sqid.
 func (c *RequestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) (logs []entity.RequestLog, retErr error) {
 	op := metrics.Begin(c.metricsScope, "get_by_id")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if err := validateStoredIdentifier("sqid", req.ID); err != nil {
 		return nil, fmt.Errorf("GetRequestHistoryByID invalid request: %w", err)
@@ -74,7 +74,7 @@ func (c *RequestHistoryController) GetRequestHistoryByID(ctx context.Context, re
 // GetRequestHistoryByChangeURI returns retained histories for an exact pinned change URI.
 func (c *RequestHistoryController) GetRequestHistoryByChangeURI(ctx context.Context, req entity.GetRequestHistoryByChangeURIRequest) (result []entity.RequestHistory, retErr error) {
 	op := metrics.Begin(c.metricsScope, "get_by_change_uri")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if err := validateStoredIdentifier("change URI", req.ChangeURI); err != nil {
 		return nil, fmt.Errorf("GetRequestHistoryByChangeURI invalid request: %w", err)
diff --git a/submitqueue/gateway/controller/request_summary.go b/submitqueue/gateway/controller/request_summary.go
index ff604b196..0cf0eb26c 100644
--- a/submitqueue/gateway/controller/request_summary.go
+++ b/submitqueue/gateway/controller/request_summary.go
@@ -47,7 +47,7 @@ func NewRequestSummaryController(logger *zap.SugaredLogger, scope tally.Scope, r
 // GetRequestSummaryByID returns the current materialized view of one request.
 func (c *RequestSummaryController) GetRequestSummaryByID(ctx context.Context, req entity.GetRequestSummaryByIDRequest) (summary entity.RequestSummary, retErr error) {
 	op := metrics.Begin(c.metricsScope, "get_by_id")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if err := validateStoredIdentifier("sqid", req.ID); err != nil {
 		return entity.RequestSummary{}, fmt.Errorf("GetRequestSummaryByID invalid request: %w", err)
@@ -74,7 +74,7 @@ func (c *RequestSummaryController) GetRequestSummaryByID(ctx context.Context, re
 // GetRequestSummaryByChangeURI returns current materialized views for an exact pinned change URI.
 func (c *RequestSummaryController) GetRequestSummaryByChangeURI(ctx context.Context, req entity.GetRequestSummaryByChangeURIRequest) (summaries []entity.RequestSummary, retErr error) {
 	op := metrics.Begin(c.metricsScope, "get_by_change_uri")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	if err := validateStoredIdentifier("change URI", req.ChangeURI); err != nil {
 		return nil, fmt.Errorf("GetRequestSummaryByChangeURI invalid request: %w", err)
diff --git a/submitqueue/orchestrator/controller/batch/batch.go b/submitqueue/orchestrator/controller/batch/batch.go
index 9bb13e851..931b5da54 100644
--- a/submitqueue/orchestrator/controller/batch/batch.go
+++ b/submitqueue/orchestrator/controller/batch/batch.go
@@ -79,7 +79,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go
index 8745f772d..1e583035b 100644
--- a/submitqueue/orchestrator/controller/build/build.go
+++ b/submitqueue/orchestrator/controller/build/build.go
@@ -74,7 +74,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
index 9be2d4842..e8744d356 100644
--- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
+++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
@@ -105,7 +105,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go
index 4f7ed3ae9..5f9bad6f7 100644
--- a/submitqueue/orchestrator/controller/conclude/conclude.go
+++ b/submitqueue/orchestrator/controller/conclude/conclude.go
@@ -66,7 +66,7 @@ func NewController(
 // Returns nil to ack (success), or error to nack (retry).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, "process")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go
index 41309a488..09853effd 100644
--- a/submitqueue/orchestrator/controller/dlq/batch.go
+++ b/submitqueue/orchestrator/controller/dlq/batch.go
@@ -74,7 +74,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go
index 58a6c5f23..b77fdce2f 100644
--- a/submitqueue/orchestrator/controller/dlq/buildsignal.go
+++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go
@@ -72,7 +72,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/log.go b/submitqueue/orchestrator/controller/dlq/log.go
index 11ecd1488..726ceca5d 100644
--- a/submitqueue/orchestrator/controller/dlq/log.go
+++ b/submitqueue/orchestrator/controller/dlq/log.go
@@ -61,7 +61,7 @@ func (c *logController) Process(_ context.Context, delivery consumer.Delivery) (
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 	dmeta := delivery.Metadata()
diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
index 26f0fa7f5..9e2f878a3 100644
--- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
+++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
@@ -68,7 +68,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go
index fff67ce13..0a15951c9 100644
--- a/submitqueue/orchestrator/controller/dlq/mergesignal.go
+++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go
@@ -67,7 +67,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go
index b00244be8..a319d4d24 100644
--- a/submitqueue/orchestrator/controller/dlq/request.go
+++ b/submitqueue/orchestrator/controller/dlq/request.go
@@ -110,7 +110,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go
index d1d886ce1..7e8b2257e 100644
--- a/submitqueue/orchestrator/controller/merge/merge.go
+++ b/submitqueue/orchestrator/controller/merge/merge.go
@@ -94,7 +94,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
index 66bbc9bdb..c2f54e63a 100644
--- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
+++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
@@ -80,7 +80,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
index de0e59c71..e584f1fde 100644
--- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
+++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
@@ -81,7 +81,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/ping.go b/submitqueue/orchestrator/controller/ping.go
index b27835bcd..5370a048d 100644
--- a/submitqueue/orchestrator/controller/ping.go
+++ b/submitqueue/orchestrator/controller/ping.go
@@ -44,7 +44,7 @@ func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *p
 	const opName = "ping"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
 
 	message := "pong!"
 	isEcho := false
diff --git a/submitqueue/orchestrator/controller/score/score.go b/submitqueue/orchestrator/controller/score/score.go
index 525a7947b..b0191f6aa 100644
--- a/submitqueue/orchestrator/controller/score/score.go
+++ b/submitqueue/orchestrator/controller/score/score.go
@@ -77,7 +77,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go
index c180b862b..cd69dd34e 100644
--- a/submitqueue/orchestrator/controller/speculate/speculate.go
+++ b/submitqueue/orchestrator/controller/speculate/speculate.go
@@ -91,7 +91,7 @@ func NewController(
 // Returns nil to ack (success), or error to nack (retry).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go
index 69266610d..fd8d0493c 100644
--- a/submitqueue/orchestrator/controller/start/start.go
+++ b/submitqueue/orchestrator/controller/start/start.go
@@ -74,7 +74,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	const opName = "process"
 
 	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go
index 390969d27..3c0a65e25 100644
--- a/submitqueue/orchestrator/controller/validate/validate.go
+++ b/submitqueue/orchestrator/controller/validate/validate.go
@@ -90,7 +90,7 @@ func NewController(
 // Returns nil to ack (success or non-retryable rejection), error to nack (retry).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := coremetrics.Begin(c.metricsScope, "process")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, coremetrics.LongLatencyBuckets) }()
 
 	msg := delivery.Message()