From 4069696c4cba9db551db14f29b70924b1622e100 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/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 +-
 .../mysql/speculation_path_build_store.go     |   4 +-
 .../storage/mysql/speculation_tree_store.go   |   6 +-
 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/queue.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 +-
 .../controller/prioritize/prioritize.go       |   2 +-
 .../orchestrator/controller/score/score.go    |   2 +-
 .../controller/speculate/speculate.go         |   2 +-
 .../orchestrator/controller/start/start.go    |   2 +-
 .../controller/validate/validate.go           |   2 +-
 66 files changed, 266 insertions(+), 236 deletions(-)

diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go
index 0569ba2d..f5d4c773 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 1bf4682f..f897fa08 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 ab839ce3..6693e1cf 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 9981beac..063aee6b 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 df3ae0c6..1d378a1d 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 a71116fe..5d37fa82 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 6ea259e3..fc2f5458 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 1719a263..07dbd083 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 9cc12f29..63182f1b 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 b7967f96..5a86a0e3 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 16f8fe0f..c960a89f 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 473e3e27..858ebf1b 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 e0ea63d7..7318e491 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 dfba9801..49712e72 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 a0811235..d5cb176e 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 b7b59f5e..c2416edb 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/dlq/request.go b/stovepipe/controller/dlq/request.go
index 8e0b420e..56261c44 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 db97bfd2..53ddebb8 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("IngestController requires the request to have a queue name specified: %w", ErrInvalidRequest)
diff --git a/stovepipe/controller/ping.go b/stovepipe/controller/ping.go
index f43586d0..6347f363 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 a99f6503..cce3a543 100644
--- a/stovepipe/controller/process/process.go
+++ b/stovepipe/controller/process/process.go
@@ -83,7 +83,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 45dd5aef..58eab1cb 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 ee922d31..63492e24 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 7e7d112e..38a2b12d 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 1f5c5aea..c63f33bb 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 67907deb..58beb85e 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 89d68a79..a5553e47 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 1104893a..4fcb76a5 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 a281b669..97dde484 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 ed342c4d..c82471dc 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 e37e2d4b..3cdd8928 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 64de6431..e7f3cd7b 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 07020dac..17049015 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 cfcf8d99..e34e744e 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, speculation_path_id) 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 436b8b20..dd7755cc 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 b19b2119..afdb7cff 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 8a292fe0..97871d15 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 fd542c9d..12fdaeeb 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 36849dea..f94f1a28 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 35375d5c..bda76569 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/extension/storage/mysql/speculation_path_build_store.go b/submitqueue/extension/storage/mysql/speculation_path_build_store.go
index edb9d7e8..f1df5258 100644
--- a/submitqueue/extension/storage/mysql/speculation_path_build_store.go
+++ b/submitqueue/extension/storage/mysql/speculation_path_build_store.go
@@ -43,7 +43,7 @@ func NewSpeculationPathBuildStore(db *sql.DB, scope tally.Scope) storage.Specula
 // persisted as-is; version arithmetic is owned by the controller, not the store.
 func (s *speculationPathBuildStore) Create(ctx context.Context, pathBuild entity.SpeculationPathBuild) (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 speculation_path_build (path_id, build_id, batch_id, version, created_at) VALUES (?, ?, ?, ?, ?)",
@@ -64,7 +64,7 @@ func (s *speculationPathBuildStore) Create(ctx context.Context, pathBuild entity
 // ErrNotFound if no mapping exists for pathID.
 func (s *speculationPathBuildStore) Get(ctx context.Context, pathID string) (ret entity.SpeculationPathBuild, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var pb entity.SpeculationPathBuild
 
diff --git a/submitqueue/extension/storage/mysql/speculation_tree_store.go b/submitqueue/extension/storage/mysql/speculation_tree_store.go
index 3245eef5..dd05f7cc 100644
--- a/submitqueue/extension/storage/mysql/speculation_tree_store.go
+++ b/submitqueue/extension/storage/mysql/speculation_tree_store.go
@@ -42,7 +42,7 @@ func NewSpeculationTreeStore(db *sql.DB, scope tally.Scope) storage.SpeculationT
 // Get retrieves the speculation tree by batch ID. Returns ErrNotFound if the speculation tree is not found.
 func (s *speculationTreeStore) Get(ctx context.Context, batchID string) (ret entity.SpeculationTree, retErr error) {
 	op := metrics.Begin(s.scope, "get")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	var st entity.SpeculationTree
 	var pathsJSON []byte
@@ -69,7 +69,7 @@ func (s *speculationTreeStore) Get(ctx context.Context, batchID string) (ret ent
 // Create creates a new speculation tree. Returns ErrAlreadyExists if the entry already exists.
 func (s *speculationTreeStore) Create(ctx context.Context, speculationTree entity.SpeculationTree) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	pathsJSON, err := json.Marshal(speculationTree.Paths)
 	if err != nil {
@@ -98,7 +98,7 @@ func (s *speculationTreeStore) Create(ctx context.Context, speculationTree entit
 // pure conditional write.
 func (s *speculationTreeStore) Update(ctx context.Context, batchID string, oldVersion, newVersion int32, paths []entity.SpeculationPathInfo) (retErr error) {
 	op := metrics.Begin(s.scope, "update")
-	defer func() { op.Complete(retErr) }()
+	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
 
 	pathsJSON, err := json.Marshal(paths)
 	if err != nil {
diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go
index 6175fcc0..97f7e9a5 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 c629777e..4236b17a 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("ListController invalid queue: %w", err)
diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go
index 730add0d..8b10fefa 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 d306d4a5..4311fa25 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 ff604b19..0cf0eb26 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 9bb13e85..931b5da5 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 8745f772..1e583035 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 9be2d484..e8744d35 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 4f7ed3ae..5f9bad6f 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 41309a48..09853eff 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 58a6c5f2..b77fdce2 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 11ecd148..726ceca5 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 26f0fa7f..9e2f878a 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 fff67ce1..0a15951c 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/queue.go b/submitqueue/orchestrator/controller/dlq/queue.go
index 2b17cbdb..079ac73d 100644
--- a/submitqueue/orchestrator/controller/dlq/queue.go
+++ b/submitqueue/orchestrator/controller/dlq/queue.go
@@ -92,7 +92,7 @@ func (c *queueController) 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/request.go b/submitqueue/orchestrator/controller/dlq/request.go
index b00244be..a319d4d2 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 d1d886ce..7e8b2257 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 d04aaa26..9a95200d 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 de0e59c7..e584f1fd 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 b27835bc..5370a048 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/prioritize/prioritize.go b/submitqueue/orchestrator/controller/prioritize/prioritize.go
index 6bb53293..e6cbad22 100644
--- a/submitqueue/orchestrator/controller/prioritize/prioritize.go
+++ b/submitqueue/orchestrator/controller/prioritize/prioritize.go
@@ -104,7 +104,7 @@ func NewController(
 // recomputes it.
 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/score/score.go b/submitqueue/orchestrator/controller/score/score.go
index 257ae86d..5c70a234 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 f2d2b21b..dd903321 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 69266610..fd8d0493 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 390969d2..3c0a65e2 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()