diff --git a/platform/extension/counter/mysql/counter.go b/platform/extension/counter/mysql/counter.go
index 6693e1cf1..db9ab9658 100644
--- a/platform/extension/counter/mysql/counter.go
+++ b/platform/extension/counter/mysql/counter.go
@@ -37,8 +37,8 @@ func NewCounter(db *sql.DB, scope tally.Scope) counter.Counter {
 // Next atomically increments the counter for the given domain and returns the new value.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.scope, "next", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 	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 063aee6bf..9828b476c 100644
--- a/platform/extension/messagequeue/mysql/delivery_state_store.go
+++ b/platform/extension/messagequeue/mysql/delivery_state_store.go
@@ -49,11 +49,11 @@ func newDeliveryStateStore(db *sql.DB, logger *zap.SugaredLogger, scope tally.Sc
 // — only the lease holder calls MarkDelivered for a given partition, so no concurrent
 // mutation can occur between the two statements.
 func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (_ int, retErr error) {
-	op := metrics.Begin(s.scope, "mark_delivered",
+	op := metrics.Begin(s.scope, "mark_delivered", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	now := time.Now().UnixMilli()
 	invisibleUntil := now + visibilityTimeoutMs
@@ -90,11 +90,11 @@ func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup
 // ExtendVisibility extends the visibility timeout for an in-flight message
 // without incrementing retry_count. Used by ExtendVisibilityTimeout.
 func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (retErr error) {
-	op := metrics.Begin(s.scope, "extend_visibility",
+	op := metrics.Begin(s.scope, "extend_visibility", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	now := time.Now().UnixMilli()
 	invisibleUntil := now + visibilityTimeoutMs
@@ -124,11 +124,11 @@ func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGr
 
 // MarkAcked sets acked = TRUE to indicate this group has processed the message.
 func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (retErr error) {
-	op := metrics.Begin(s.scope, "mark_acked",
+	op := metrics.Begin(s.scope, "mark_acked", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
 		INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count)
@@ -147,11 +147,11 @@ func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, to
 // MarkNacked sets invisible_until = now + delay to schedule redelivery.
 // retry_count is NOT incremented here — it is incremented by MarkDelivered on redelivery.
 func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) (retErr error) {
-	op := metrics.Begin(s.scope, "mark_nacked",
+	op := metrics.Begin(s.scope, "mark_nacked", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	now := time.Now().UnixMilli()
 	invisibleUntil := now + delayMs
@@ -174,11 +174,11 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t
 // GetDeliveryState returns the full delivery state for a message offset.
 // Returns (state, found, error). found=false means no row (never delivered).
 func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (_ DeliveryState, _ bool, retErr error) {
-	op := metrics.Begin(s.scope, "get_delivery_state",
+	op := metrics.Begin(s.scope, "get_delivery_state", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	var state DeliveryState
 	err := s.db.QueryRowContext(ctx, fmt.Sprintf(`
@@ -201,11 +201,11 @@ func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGr
 // offsets are the actual message offsets above the current watermark (from messageStore).
 // Returns the new watermark (highest contiguous acked offset from currentWatermark).
 func (s *sqldeliveryStateStore) AdvanceWatermark(ctx context.Context, consumerGroup, topic, partitionKey string, currentWatermark int64, offsets []int64) (_ int64, retErr error) {
-	op := metrics.Begin(s.scope, "advance_watermark",
+	op := metrics.Begin(s.scope, "advance_watermark", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("consumer_group", consumerGroup),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	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 1d378a1da..97bad2e99 100644
--- a/platform/extension/messagequeue/mysql/message_store.go
+++ b/platform/extension/messagequeue/mysql/message_store.go
@@ -62,8 +62,8 @@ func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []e
 // second Cancel RPC for the same request) without surfacing 1062 duplicate-key
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "insert", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	if len(messages) == 0 {
 		return nil
@@ -132,8 +132,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
 		DELETE FROM %s WHERE topic = ? AND partition_key = ? AND id = ?
@@ -151,8 +151,8 @@ func (s *sqlmessageStore) Delete(ctx context.Context, topic string, partitionKey
 // (published via InsertDelayed) that should not yet be surfaced to subscribers.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "fetch", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	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
@@ -227,8 +227,8 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti
 // The message is inserted back into queue_messages table with the DLQ topic (original + suffix)
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "move_to_dlq", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	// Construct DLQ topic name
 	dlqTopic := topic + dlqTopicSuffix
@@ -300,8 +300,8 @@ func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partition
 // free of cross-table queries.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "gc", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	if minAckedOffset == 0 {
 		return 0, nil
@@ -342,8 +342,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get_offsets_above", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	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 5d37fa82e..60963a818 100644
--- a/platform/extension/messagequeue/mysql/offset_store.go
+++ b/platform/extension/messagequeue/mysql/offset_store.go
@@ -40,11 +40,11 @@ func newOffsetStore(db *sql.DB, scope tally.Scope) offsetStore {
 
 // Initialize creates an offset entry for a topic+partition if it doesn't exist
 func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partitionKey string, consumerGroup string) (retErr error) {
-	op := metrics.Begin(s.scope, "initialize",
+	op := metrics.Begin(s.scope, "initialize", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey),
 		metrics.NewTag("consumer_group", consumerGroup))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	now := time.Now().UnixMilli()
 
@@ -63,11 +63,11 @@ func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partition
 
 // GetAckedOffset returns the current acked offset for a topic+partition
 func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, partitionKey string, consumerGroup string) (_ int64, retErr error) {
-	op := metrics.Begin(s.scope, "get_acked_offset",
+	op := metrics.Begin(s.scope, "get_acked_offset", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey),
 		metrics.NewTag("consumer_group", consumerGroup))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	var offset int64
 	err := s.db.QueryRowContext(ctx, fmt.Sprintf(`
@@ -88,11 +88,11 @@ func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, parti
 
 // UpdateAckedOffset updates the offset_acked for a topic+partition (only if new offset is greater)
 func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, partitionKey string, offset int64, consumerGroup string) (retErr error) {
-	op := metrics.Begin(s.scope, "update_acked_offset",
+	op := metrics.Begin(s.scope, "update_acked_offset", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey),
 		metrics.NewTag("consumer_group", consumerGroup))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	now := time.Now().UnixMilli()
 
@@ -112,10 +112,10 @@ func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, pa
 // GetMinAckedOffset returns the minimum offset_acked across all consumer groups
 // for a topic+partition. Returns (0, false, nil) if no offset rows exist.
 func (s *sqloffsetStore) GetMinAckedOffset(ctx context.Context, topic string, partitionKey string) (_ int64, _ bool, retErr error) {
-	op := metrics.Begin(s.scope, "get_min_acked_offset",
+	op := metrics.Begin(s.scope, "get_min_acked_offset", metrics.StorageLatencyBuckets,
 		metrics.NewTag("topic", topic),
 		metrics.NewTag("partition_key", partitionKey))
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	defer func() { op.Complete(retErr) }()
 
 	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 fc2f5458f..6e2fb227a 100644
--- a/platform/extension/messagequeue/mysql/partition_lease_store.go
+++ b/platform/extension/messagequeue/mysql/partition_lease_store.go
@@ -44,8 +44,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "try_acquire_lease", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	now := currentTimeMillis()
 	staleThreshold := now - leaseDurationMs
@@ -93,8 +93,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "renew_lease", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	now := currentTimeMillis()
 
@@ -127,8 +127,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "release_lease", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	result, err := s.db.ExecContext(ctx, fmt.Sprintf(`
 		DELETE FROM %s
@@ -162,8 +162,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get_leased_partitions", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
 		SELECT partition_key FROM %s
@@ -200,8 +200,8 @@ func (s *sqlpartitionLeaseStore) GetLeasedPartitions(ctx context.Context, topic
 // Returns the number of new leases acquired and the full list of discovered partitions.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "discover_and_acquire", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	// 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 07dbd0830..225c89064 100644
--- a/platform/extension/messagequeue/mysql/publisher.go
+++ b/platform/extension/messagequeue/mysql/publisher.go
@@ -46,8 +46,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(p.scope, "publish", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	// Check if closed (under lock)
 	p.mu.RLock()
@@ -72,8 +72,8 @@ func (p *publisher) Publish(ctx context.Context, topic string, message entityque
 // now + delayMs; FetchByOffset skips it until that timestamp.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(p.scope, "publish_after", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	p.mu.RLock()
 	closed := p.closed
diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go
index 63182f1b4..514bcced5 100644
--- a/platform/extension/messagequeue/mysql/subscriber.go
+++ b/platform/extension/messagequeue/mysql/subscriber.go
@@ -366,8 +366,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "subscribe", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
+	defer func() { op.Complete(retErr) }()
 
 	s.mu.RLock()
 	closed := s.closed
@@ -411,9 +411,6 @@ func (s *subscriber) Subscribe(ctx context.Context, topic string, config extqueu
 
 	s.subscriptions[subKey] = sub
 
-	// Track active subscription
-	metrics.NamedGauge(s.scope, "subscribe", "active_subscriptions", 1, metrics.NewTag("topic", topic))
-
 	// Start the supervisor goroutine. It will discover partitions, acquire
 	// leases, and spawn per-partition worker goroutines. The supervisor runs
 	// until the subscription context is cancelled (via Close or explicit cancel).
@@ -1088,8 +1085,8 @@ func (s *subscriber) fairShareCap(ctx context.Context, sub *subscription, owned
 //  3. managePartitions internally handles stopping workers and closing deliveryCh
 //     (see managePartitions shutdown sequence)
 func (s *subscriber) Close() (retErr error) {
-	op := metrics.Begin(s.scope, "close")
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "close", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -1130,9 +1127,6 @@ func (s *subscriber) Close() (retErr error) {
 				"consumer_group", sub.config.ConsumerGroup,
 			)
 		}
-
-		// Update metrics
-		metrics.NamedGauge(s.scope, "subscribe", "active_subscriptions", 0, metrics.NewTag("topic", sub.topic))
 	}
 
 	s.subscriptions = make(map[string]*subscription)
diff --git a/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go b/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go
index 5a86a0e3c..7a2c18fff 100644
--- a/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go
+++ b/platform/extension/messagequeue/mysql/subscriber_heartbeat_store.go
@@ -45,8 +45,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "heartbeat", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	now := s.nowFunc().UnixMilli()
 
@@ -65,8 +65,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "active_subscribers", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	staleThreshold := s.nowFunc().UnixMilli() - staleDurationMs
 
@@ -104,8 +104,8 @@ func (s *sqlSubscriberHeartbeatStore) ActiveSubscribers(ctx context.Context, top
 // Deregister soft-deletes a subscriber's heartbeat entry by setting deregistered_at.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "deregister", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	now := s.nowFunc().UnixMilli()
 
diff --git a/platform/metrics/BUILD.bazel b/platform/metrics/BUILD.bazel
index 5bd7ed56c..081403509 100644
--- a/platform/metrics/BUILD.bazel
+++ b/platform/metrics/BUILD.bazel
@@ -5,10 +5,7 @@ go_library(
     srcs = ["metrics.go"],
     importpath = "github.com/uber/submitqueue/platform/metrics",
     visibility = ["//visibility:public"],
-    deps = [
-        "//platform/errs:go_default_library",
-        "@com_github_uber_go_tally//:go_default_library",
-    ],
+    deps = ["@com_github_uber_go_tally//:go_default_library"],
 )
 
 go_test(
@@ -16,7 +13,6 @@ go_test(
     srcs = ["metrics_test.go"],
     embed = [":go_default_library"],
     deps = [
-        "//platform/errs:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
     ],
diff --git a/platform/metrics/README.md b/platform/metrics/README.md
index c960a89fb..4ea6f2680 100644
--- a/platform/metrics/README.md
+++ b/platform/metrics/README.md
@@ -1,128 +1,83 @@
 # Metrics Utilities (`platform/metrics`)
 
-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.
+The `metrics` package provides reusable helpers for emitting counters and histograms on a `tally.Scope`.
 
 ## Design
 
-**Free functions on `tally.Scope`** — no wrapper types. Existing constructors accept `tally.Scope` and don't need to change.
+**Free functions on `tally.Scope`** — no wrapper types. Existing constructors accept `tally.Scope` and do not 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 and a latency histogram. This prevents mismatched or forgotten metrics calls.
+**Operation lifecycle** — `Begin` and `Complete` tie operation metrics together. `Begin` captures the start time and emits `{name}.start`; `Complete` records duration and count on `{name}.finish`.
 
-**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.
+**Result tagging** — the finish histogram is tagged with `result=success`, `result=error`, or `result=cancel`. Cancellation is detected with `errors.Is(err, context.Canceled)`. Error classification tags are intentionally omitted because many call sites complete before classification occurs.
 
-**Consistent naming** — all Named helpers follow the `{name}.{sub}` sub-scope pattern, producing structured metric paths like `process.called`, `publish.attempts`, `consumer.pending_messages`.
+**Consistent naming** — named helpers follow the `{name}.{sub}` sub-scope pattern, producing metric paths such as `process.start` and `publish.attempts`.
 
 ## Operation Lifecycle
 
-For any operation with a clear start/end, use `Begin`/`Complete`:
+For any operation with a clear start and end, use `Begin` and `Complete`:
 
 | Function | Emits |
 |----------|-------|
-| `Begin(scope, name, ...tags)` | `{name}.called` counter +1, returns `Op` |
-| `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 |
+| `Begin(scope, name, buckets, ...tags)` | `{name}.start` counter +1 and returns an `Op` |
+| `op.Complete(err)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` |
 
-`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.
+`buckets` is required at `Begin` because operations differ widely in expected latency. The finish histogram records both the duration distribution and the number of completed operations, so `Complete` does not emit a separate counter.
 
 ```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, metrics.StorageLatencyBuckets) }()
-
-    // ... business logic ...
-    return &pb.LandResponse{Sqid: request.ID}, nil
-}
-
-// 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, metrics.LongLatencyBuckets) }()
+    op := metrics.Begin(c.scope, "process", metrics.LongLatencyBuckets)
+    defer func() { op.Complete(retErr) }()
 
     // ... business logic ...
     return nil
 }
 ```
 
-On success, `Complete` emits:
-- `{name}.succeeded` counter +1
-- `{name}.latency` histogram tagged `result=success`
-
-On failure, `Complete` emits:
-- `{name}.failed` counter +1
-- `{name}.latency` histogram tagged `result=error`, `error_origin=user|infra`, `retryable=true|false`, and optionally `dependency=true`
-
 ## Named Helpers
 
-For ad-hoc metrics that don't fit the Begin/Complete lifecycle. All follow the `{name}.{sub}` sub-scope pattern:
+For ad-hoc metrics that do not fit the operation lifecycle:
 
 | Function | Emits | Example |
 |----------|-------|---------|
 | `NamedCounter(scope, name, counter, value, ...tags)` | `{name}.{counter}` counter | `publish.attempts` |
 | `NamedHistogram(scope, name, histogram, buckets, ...tags)` | `{name}.{histogram}` histogram | `process.duration` |
-| `NamedGauge(scope, name, gauge, value, ...tags)` | `{name}.{gauge}` gauge | `consumer.pending_messages` |
 
 ```go
-// Count a specific sub-event
 metrics.NamedCounter(c.scope, "publish", "attempts", 1)
 
-// 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)))
-
-// 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:
-
-| Tag | Values | Source |
-|-----|--------|--------|
-| `error_origin` | `user`, `infra` | `errs.IsUserError` |
-| `retryable` | `true`, `false` | `errs.IsRetryable` |
-| `dependency` | `true` (only when applicable) | `errs.IsDependencyError` |
+Use `tally.Scope` directly for gauges:
 
 ```go
-tags := metrics.ErrorTags(err)
-// Generic error:     [{error_origin, infra}, {retryable, false}]
-// User error:        [{error_origin, user},  {retryable, false}]
-// Retryable error:   [{error_origin, infra}, {retryable, true}]
-// Dependency error:  [{error_origin, infra}, {retryable, false}, {dependency, true}]
+c.scope.SubScope("consumer").Gauge("pending_messages").Update(float64(len(pending)))
 ```
 
+### Why histograms, not timers
+
+Durations are recorded as histograms rather than timers. Timer percentiles cannot be combined accurately across time series, while bucketed histogram counts can be summed to reconstruct a combined distribution for correct aggregate percentiles.
+
 ## Tags
 
-Use `NewTag` to pass additional dimensional tags to any helper:
+Use `NewTag` to pass dimensional tags to a helper:
 
 ```go
-op := metrics.Begin(c.scope, "process", metrics.NewTag("queue", req.Queue))
-defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+op := metrics.Begin(c.scope, "process", metrics.LongLatencyBuckets, metrics.NewTag("queue", req.Queue))
+defer func() { op.Complete(retErr) }()
 
 metrics.NamedCounter(c.scope, "publish", "attempts", 1, metrics.NewTag("topic", c.topic))
 ```
 
 ## Latency Buckets
 
-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:
+There is no default bucket set. The package exports three common sets:
 
 | 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 |
-
-Pass one of these, or your own `tally.DurationBuckets` when none fits:
+| `FastLatencyBuckets` | ~100µs – 5s | Fast in-process work such as scoring, cache lookups, and CPU-bound operations |
+| `StorageLatencyBuckets` | ~1ms – 1m | Storage and message-queue round trips such as database reads, writes, publishing, and consuming |
+| `LongLatencyBuckets` | ~5ms – 4h | Long-running pipeline work and external calls such as builds, merges, pushes, and provider calls |
 
-```go
-defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
-
-h := metrics.NamedHistogram(c.scope, "build", "duration", tally.DurationBuckets{ /* custom */ })
-```
+Pass one of these sets or a custom `tally.DurationBuckets` to `Begin` or `NamedHistogram`.
diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go
index 858ebf1ba..0e7a568b2 100644
--- a/platform/metrics/metrics.go
+++ b/platform/metrics/metrics.go
@@ -15,10 +15,11 @@
 package metrics
 
 import (
+	"context"
+	"errors"
 	"time"
 
 	"github.com/uber-go/tally"
-	"github.com/uber/submitqueue/platform/errs"
 )
 
 // Tag is a key-value pair attached to a metric for dimensional filtering.
@@ -36,7 +37,7 @@ func NewTag(key, value string) Tag {
 
 // 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
+// matches the operation and pass it to Begin or NamedHistogram. Buckets
 // far outside an operation's real latency waste series cardinality and lose
 // resolution where the data actually lands.
 var (
@@ -108,14 +109,14 @@ var (
 )
 
 // 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 histogram with error classification tags) when Complete is called.
+// creation, emits a {name}.start counter, and records the duration and result
+// on a {name}.finish histogram 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, metrics.StorageLatencyBuckets) }()
+//	    op := metrics.Begin(c.scope, "process", metrics.StorageLatencyBuckets)
+//	    defer func() { op.Complete(retErr) }()
 //	    // ... business logic ...
 //	}
 type Op struct {
@@ -123,47 +124,38 @@ type Op struct {
 	scope tally.Scope
 	// start is the time the operation began.
 	start time.Time
+	// buckets defines the finish histogram's duration buckets.
+	buckets tally.Buckets
 }
 
-// Begin starts a new operation. It emits a {name}.called counter and captures
-// the start time. Call Complete on the returned Op to record the outcome.
-func Begin(scope tally.Scope, name string, tags ...Tag) Op {
+// Begin starts a new operation. It emits a {name}.start counter, captures the
+// start time, and retains the buckets used by Complete.
+func Begin(scope tally.Scope, name string, buckets tally.Buckets, tags ...Tag) Op {
 	sub := tagged(scope, tags).SubScope(name)
-	sub.Counter("called").Inc(1)
-	return Op{scope: sub, start: time.Now()}
+	sub.Counter("start").Inc(1)
+	return Op{
+		scope:   sub,
+		start:   time.Now(),
+		buckets: buckets,
+	}
 }
 
-// Complete records the outcome of the operation. It emits a {name}.succeeded or
-// {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.Histogram("latency", buckets).RecordDuration(elapsed)
-		return
+// Complete records elapsed time on the {name}.finish histogram, tagged with
+// result=success|error|cancel. The histogram records both duration and count.
+// Cancellation is detected through the error chain.
+func (o Op) Complete(err error) {
+	result := "success"
+	if err != nil {
+		result = "error"
+		if errors.Is(err, context.Canceled) {
+			result = "cancel"
+		}
 	}
 
-	o.scope.Counter("failed").Inc(1)
-
-	latencyTags := map[string]string{"result": "error"}
-	for _, t := range ErrorTags(err) {
-		latencyTags[t.Key] = t.Value
-	}
-	s := o.scope.Tagged(latencyTags)
-	s.Histogram("latency", buckets).RecordDuration(elapsed)
+	o.scope.
+		Tagged(map[string]string{"result": result}).
+		Histogram("finish", o.buckets).
+		RecordDuration(time.Since(o.start))
 }
 
 // NamedCounter increments the {name}.{counter} counter by value.
@@ -178,43 +170,6 @@ func NamedHistogram(scope tally.Scope, name string, histogram string, buckets ta
 	return tagged(scope, tags).SubScope(name).Histogram(histogram, buckets)
 }
 
-// NamedGauge updates the {name}.{gauge} gauge to value. Gauges represent a
-// current point-in-time measurement that can go up or down, such as queue depth,
-// active connections, or in-flight requests.
-func NamedGauge(scope tally.Scope, name string, gauge string, value float64, tags ...Tag) {
-	tagged(scope, tags).SubScope(name).Gauge(gauge).Update(value)
-}
-
-// ErrorTags returns classification tags for an error using platform/errs.
-// Returns error_origin (user|infra), retryable (true|false), and
-// dependency (true) tags. Returns nil for a nil error.
-func ErrorTags(err error) []Tag {
-	if err == nil {
-		return nil
-	}
-
-	origin := "infra"
-	if errs.IsUserError(err) {
-		origin = "user"
-	}
-
-	retryable := "false"
-	if errs.IsRetryable(err) {
-		retryable = "true"
-	}
-
-	tags := []Tag{
-		{Key: "error_origin", Value: origin},
-		{Key: "retryable", Value: retryable},
-	}
-
-	if errs.IsDependencyError(err) {
-		tags = append(tags, Tag{Key: "dependency", Value: "true"})
-	}
-
-	return tags
-}
-
 // tagsToMap converts a slice of Tag to a map for tally.
 func tagsToMap(tags []Tag) map[string]string {
 	m := make(map[string]string, len(tags))
diff --git a/platform/metrics/metrics_test.go b/platform/metrics/metrics_test.go
index 7318e4918..aa83302bb 100644
--- a/platform/metrics/metrics_test.go
+++ b/platform/metrics/metrics_test.go
@@ -15,137 +15,95 @@
 package metrics
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/uber-go/tally"
-	"github.com/uber/submitqueue/platform/errs"
 )
 
-func TestBegin_EmitsCalled(t *testing.T) {
+func TestBegin_EmitsStart(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
-	_ = Begin(scope, "process")
+	_ = Begin(scope, "process", FastLatencyBuckets)
 
 	snapshot := scope.Snapshot()
 	counters := snapshot.Counters()
-	c, ok := counters["process.called+"]
-	assert.True(t, ok, "expected process.called counter")
+	c, ok := counters["process.start+"]
+	assert.True(t, ok, "expected process.start counter")
 	assert.Equal(t, int64(1), c.Value())
 }
 
 func TestComplete(t *testing.T) {
 	tests := []struct {
-		name             string
-		err              error
-		expectSucceeded  bool
-		expectResultTag  string
-		expectOrigin     string
-		expectRetryable  string
-		expectDependency bool
+		name   string
+		err    error
+		result string
 	}{
 		{
-			name:            "nil error records success",
-			err:             nil,
-			expectSucceeded: true,
-			expectResultTag: "success",
+			name:   "nil error records success",
+			result: "success",
 		},
 		{
-			name:            "generic error records failure with infra origin",
-			err:             fmt.Errorf("something broke"),
-			expectSucceeded: false,
-			expectResultTag: "error",
-			expectOrigin:    "infra",
-			expectRetryable: "false",
+			name:   "generic error records error",
+			err:    fmt.Errorf("something broke"),
+			result: "error",
 		},
 		{
-			name:            "retryable error records retryable=true",
-			err:             errs.NewRetryableError(fmt.Errorf("timeout")),
-			expectSucceeded: false,
-			expectResultTag: "error",
-			expectOrigin:    "infra",
-			expectRetryable: "true",
+			name:   "canceled context records cancel",
+			err:    context.Canceled,
+			result: "cancel",
 		},
 		{
-			name:            "user error records error_origin=user",
-			err:             errs.NewUserError(fmt.Errorf("bad input")),
-			expectSucceeded: false,
-			expectResultTag: "error",
-			expectOrigin:    "user",
-			expectRetryable: "false",
+			name:   "wrapped canceled context records cancel",
+			err:    fmt.Errorf("process: %w", context.Canceled),
+			result: "cancel",
 		},
 		{
-			name:             "dependency error records dependency=true",
-			err:              errs.NewDependencyError(fmt.Errorf("db down")),
-			expectSucceeded:  false,
-			expectResultTag:  "error",
-			expectOrigin:     "infra",
-			expectRetryable:  "false",
-			expectDependency: true,
+			name:   "deadline exceeded records error",
+			err:    context.DeadlineExceeded,
+			result: "error",
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			scope := tally.NewTestScope("", nil)
-			op := Begin(scope, "process")
-			op.Complete(tt.err, FastLatencyBuckets)
+			op := Begin(scope, "process", FastLatencyBuckets)
+			op.Complete(tt.err)
 
 			snapshot := scope.Snapshot()
 			counters := snapshot.Counters()
-
-			// Begin always emits called
-			c, ok := counters["process.called+"]
-			assert.True(t, ok, "expected process.called counter")
+			c, ok := counters["process.start+"]
+			assert.True(t, ok, "expected process.start counter")
 			assert.Equal(t, int64(1), c.Value())
+			assert.Len(t, counters, 1, "Complete should not emit a counter")
 
-			if tt.expectSucceeded {
-				c, ok := counters["process.succeeded+"]
-				assert.True(t, ok, "expected process.succeeded counter")
-				assert.Equal(t, int64(1), c.Value())
-
-				histograms := snapshot.Histograms()
-				_, 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")
-				assert.Equal(t, int64(1), c.Value())
-
-				// Build expected tag suffix (tally sorts tags alphabetically)
-				tagSuffix := ""
-				if tt.expectDependency {
-					tagSuffix += "dependency=true,"
-				}
-				tagSuffix += "error_origin=" + tt.expectOrigin
-				tagSuffix += ",result=" + tt.expectResultTag
-				tagSuffix += ",retryable=" + tt.expectRetryable
-
-				histogramKey := "process.latency+" + tagSuffix
-				histograms := snapshot.Histograms()
-				_, ok = histograms[histogramKey]
-				assert.True(t, ok, "expected histogram key %s, got keys: %v", histogramKey, histogramKeys(histograms))
-			}
+			histograms := snapshot.Histograms()
+			histogramKey := "process.finish+result=" + tt.result
+			_, ok = histograms[histogramKey]
+			assert.True(t, ok, "expected histogram key %s, got keys: %v", histogramKey, histogramKeys(histograms))
+			assert.Len(t, histograms, 1, "finish histogram should only include the result tag")
 		})
 	}
 }
 
 func TestBegin_WithTags(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
-	op := Begin(scope, "process", NewTag("env", "prod"))
-	op.Complete(nil, FastLatencyBuckets)
+	op := Begin(scope, "process", FastLatencyBuckets, NewTag("env", "prod"))
+	op.Complete(nil)
 
 	snapshot := scope.Snapshot()
 	counters := snapshot.Counters()
 
-	c, ok := counters["process.called+env=prod"]
-	assert.True(t, ok, "expected tagged called counter, got keys: %v", counterKeys(counters))
+	c, ok := counters["process.start+env=prod"]
+	assert.True(t, ok, "expected tagged start counter, got keys: %v", counterKeys(counters))
 	assert.Equal(t, int64(1), c.Value())
 
-	c, ok = counters["process.succeeded+env=prod"]
-	assert.True(t, ok, "expected tagged succeeded counter, got keys: %v", counterKeys(counters))
-	assert.Equal(t, int64(1), c.Value())
+	histograms := snapshot.Histograms()
+	_, ok = histograms["process.finish+env=prod,result=success"]
+	assert.True(t, ok, "expected tagged finish histogram, got keys: %v", histogramKeys(histograms))
 }
 
 func TestNamedCounter(t *testing.T) {
@@ -172,17 +130,6 @@ func TestNamedHistogram(t *testing.T) {
 	assert.True(t, ok, "expected process.duration histogram")
 }
 
-func TestNamedGauge(t *testing.T) {
-	scope := tally.NewTestScope("", nil)
-	NamedGauge(scope, "consumer", "pending_messages", 42)
-
-	snapshot := scope.Snapshot()
-	gauges := snapshot.Gauges()
-	g, ok := gauges["consumer.pending_messages+"]
-	assert.True(t, ok, "expected consumer.pending_messages gauge")
-	assert.Equal(t, float64(42), g.Value())
-}
-
 func TestLatencyBuckets_Sorted(t *testing.T) {
 	sets := map[string]tally.DurationBuckets{
 		"FastLatencyBuckets":    FastLatencyBuckets,
@@ -200,60 +147,6 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 	}
 }
 
-func TestErrorTags(t *testing.T) {
-	tests := []struct {
-		name     string
-		err      error
-		expected []Tag
-	}{
-		{
-			name:     "nil error returns nil",
-			err:      nil,
-			expected: nil,
-		},
-		{
-			name: "generic error returns infra non-retryable",
-			err:  fmt.Errorf("fail"),
-			expected: []Tag{
-				{Key: "error_origin", Value: "infra"},
-				{Key: "retryable", Value: "false"},
-			},
-		},
-		{
-			name: "user error returns user origin",
-			err:  errs.NewUserError(fmt.Errorf("bad")),
-			expected: []Tag{
-				{Key: "error_origin", Value: "user"},
-				{Key: "retryable", Value: "false"},
-			},
-		},
-		{
-			name: "retryable error returns retryable=true",
-			err:  errs.NewRetryableError(fmt.Errorf("timeout")),
-			expected: []Tag{
-				{Key: "error_origin", Value: "infra"},
-				{Key: "retryable", Value: "true"},
-			},
-		},
-		{
-			name: "dependency error includes dependency tag",
-			err:  errs.NewDependencyError(fmt.Errorf("db down")),
-			expected: []Tag{
-				{Key: "error_origin", Value: "infra"},
-				{Key: "retryable", Value: "false"},
-				{Key: "dependency", Value: "true"},
-			},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tags := ErrorTags(tt.err)
-			assert.Equal(t, tt.expected, tags)
-		})
-	}
-}
-
 // 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 49712e724..5e065f210 100644
--- a/runway/controller/merge/merge.go
+++ b/runway/controller/merge/merge.go
@@ -79,8 +79,8 @@ func NewController(p Params) *Controller {
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/runway/controller/mergeconflictcheck/mergeconflictcheck.go b/runway/controller/mergeconflictcheck/mergeconflictcheck.go
index d5cb176ec..fb53c62d7 100644
--- a/runway/controller/mergeconflictcheck/mergeconflictcheck.go
+++ b/runway/controller/mergeconflictcheck/mergeconflictcheck.go
@@ -79,8 +79,8 @@ func NewController(p Params) *Controller {
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/runway/controller/ping.go b/runway/controller/ping.go
index c2416edbb..04e9b9915 100644
--- a/runway/controller/ping.go
+++ b/runway/controller/ping.go
@@ -43,8 +43,8 @@ func NewPingController(logger *zap.Logger, scope tally.Scope) *PingController {
 func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) {
 	const opName = "ping"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.FastLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	message := "pong!"
 	isEcho := false
diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go
index 5d8d666f0..557ece2da 100644
--- a/stovepipe/controller/build/build.go
+++ b/stovepipe/controller/build/build.go
@@ -80,8 +80,8 @@ func NewController(
 // its decided scope, and publishes the build id to buildsignal. Returns nil to
 // ack (success) or an error to nack (retry) / reject (DLQ).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
-	op := metrics.Begin(c.metricsScope, _opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go
index de026b6e7..3f8fb0a3c 100644
--- a/stovepipe/controller/buildsignal/buildsignal.go
+++ b/stovepipe/controller/buildsignal/buildsignal.go
@@ -95,8 +95,8 @@ func NewController(
 // poll or, once terminal, publishes the build id to record. Returns nil to
 // ack (success) or an error to nack (retry) / reject (DLQ).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
-	op := metrics.Begin(c.metricsScope, _opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go
index 56261c449..9484bb91f 100644
--- a/stovepipe/controller/dlq/request.go
+++ b/stovepipe/controller/dlq/request.go
@@ -67,8 +67,8 @@ func NewController(
 // wired with errs.AlwaysRetryableProcessor so a transient reconcile failure retries
 // 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, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go
index dd60e493d..729420eb6 100644
--- a/stovepipe/controller/ingest.go
+++ b/stovepipe/controller/ingest.go
@@ -89,8 +89,8 @@ func NewIngestController(
 func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest) (result entity.IngestResult, retErr error) {
 	const opName = "ingest"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	if req.Queue == "" {
 		return entity.IngestResult{}, fmt.Errorf("requires the request to have a queue name specified: %w", ErrInvalidRequest)
diff --git a/stovepipe/controller/ping.go b/stovepipe/controller/ping.go
index 6347f363d..d8a857a2b 100644
--- a/stovepipe/controller/ping.go
+++ b/stovepipe/controller/ping.go
@@ -43,8 +43,8 @@ func NewPingController(logger *zap.Logger, scope tally.Scope) *PingController {
 func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) {
 	const opName = "ping"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.FastLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	message := "pong!"
 	isEcho := false
diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go
index c0b7d72a4..0fbf33ed9 100644
--- a/stovepipe/controller/process/process.go
+++ b/stovepipe/controller/process/process.go
@@ -84,8 +84,8 @@ func NewController(
 // Process reloads the request referenced by the delivery, coalesces older heads,
 // 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, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/stovepipe/extension/storage/mysql/build_store.go b/stovepipe/extension/storage/mysql/build_store.go
index 58eab1cb3..578478132 100644
--- a/stovepipe/extension/storage/mysql/build_store.go
+++ b/stovepipe/extension/storage/mysql/build_store.go
@@ -39,8 +39,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(b.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	_, err := b.db.ExecContext(ctx,
 		`INSERT INTO build (id, request_id, status, version)
@@ -62,8 +62,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(b.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var build entity.Build
 	err := b.db.QueryRowContext(ctx,
@@ -92,8 +92,8 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 // match (including when the build does not exist). This is a pure conditional write; the
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(b.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 63492e247..6c050d522 100644
--- a/stovepipe/extension/storage/mysql/queue_store.go
+++ b/stovepipe/extension/storage/mysql/queue_store.go
@@ -38,8 +38,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(q.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	_, err := q.db.ExecContext(ctx,
 		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
@@ -61,8 +61,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(q.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var queue entity.Queue
 	err := q.db.QueryRowContext(ctx,
@@ -89,8 +89,8 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re
 // Update persists the mutable fields of queue if the stored version matches oldVersion,
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(q.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 38a2b12d7..bfc1627bb 100644
--- a/stovepipe/extension/storage/mysql/request_store.go
+++ b/stovepipe/extension/storage/mysql/request_store.go
@@ -44,8 +44,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	_, err := r.db.ExecContext(ctx,
 		`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
@@ -70,8 +70,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var req entity.Request
 	err := r.db.QueryRowContext(ctx,
@@ -103,8 +103,8 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 // (including when the request does not exist). This is a pure conditional write; the caller owns
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 c63f33bb8..8106596f1 100644
--- a/stovepipe/extension/storage/mysql/request_uri_store.go
+++ b/stovepipe/extension/storage/mysql/request_uri_store.go
@@ -44,8 +44,8 @@ func NewRequestURIStore(db *sql.DB, scope tally.Scope) storage.RequestURIStore {
 // Create records the (queue, uri) -> id reverse index. Returns ErrAlreadyExists if (queue, uri)
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	_, err := r.db.ExecContext(ctx,
 		"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?)",
@@ -63,8 +63,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "get_id_by_uri", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 58beb85ee..2119a7f83 100644
--- a/submitqueue/extension/changeprovider/github/provider.go
+++ b/submitqueue/extension/changeprovider/github/provider.go
@@ -45,8 +45,8 @@ func NewProvider(params Params) changeprovider.ChangeProvider {
 // Get retrieves change information from GitHub for the request's change.
 // 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, coremetrics.LongLatencyBuckets) }()
+	op := coremetrics.Begin(p.metricsScope, "get", coremetrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	change := request.Change
 
diff --git a/submitqueue/extension/changeprovider/phabricator/provider.go b/submitqueue/extension/changeprovider/phabricator/provider.go
index a5553e479..bca5a0e9d 100644
--- a/submitqueue/extension/changeprovider/phabricator/provider.go
+++ b/submitqueue/extension/changeprovider/phabricator/provider.go
@@ -48,8 +48,8 @@ func NewProvider(params Params) changeprovider.ChangeProvider {
 // Get retrieves change information from Phabricator for the request's change.
 // 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, coremetrics.LongLatencyBuckets) }()
+	op := coremetrics.Begin(p.metricsScope, "get", coremetrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	change := request.Change
 
diff --git a/submitqueue/extension/mergechecker/github/checker.go b/submitqueue/extension/mergechecker/github/checker.go
index 4fcb76a56..5c6b394a5 100644
--- a/submitqueue/extension/mergechecker/github/checker.go
+++ b/submitqueue/extension/mergechecker/github/checker.go
@@ -64,8 +64,8 @@ func NewMergeChecker(params Params) mergechecker.MergeChecker {
 func (c *mergeChecker) Check(ctx context.Context, request entity.Request) (result entity.MergeResult, retErr error) {
 	const opName = "check"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	change := request.Change
 
diff --git a/submitqueue/extension/pusher/git/git_pusher.go b/submitqueue/extension/pusher/git/git_pusher.go
index 97dde484d..56e406fc8 100644
--- a/submitqueue/extension/pusher/git/git_pusher.go
+++ b/submitqueue/extension/pusher/git/git_pusher.go
@@ -135,8 +135,8 @@ 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, coremetrics.LongLatencyBuckets) }()
+	op := coremetrics.Begin(p.metricsScope, "push", coremetrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	// 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 c82471dce..4a9302415 100644
--- a/submitqueue/extension/scorer/composite/scorer.go
+++ b/submitqueue/extension/scorer/composite/scorer.go
@@ -91,8 +91,8 @@ func New(scorers map[string]scorer.Scorer, reduce ReduceFunc, scope tally.Scope)
 // Score evaluates all child scorers on the batch and combines their results using the
 // 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, metrics.FastLatencyBuckets) }()
+	op := metrics.Begin(c.scope, "score", metrics.FastLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 3cdd8928d..dc49693f7 100644
--- a/submitqueue/extension/scorer/heuristic/scorer.go
+++ b/submitqueue/extension/scorer/heuristic/scorer.go
@@ -69,8 +69,8 @@ func New(resolver changeset.Resolver, buckets []Bucket, valueFunc ValueFunc, sco
 // score for the first bucket whose range [Min, Max] contains the value. Returns an error
 // 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, metrics.FastLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "score", metrics.FastLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 	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 e7f3cd7b3..be32bc479 100644
--- a/submitqueue/extension/storage/mysql/batch_dependent_store.go
+++ b/submitqueue/extension/storage/mysql/batch_dependent_store.go
@@ -41,8 +41,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var bd entity.BatchDependent
 	var dependentsJSON []byte
@@ -68,8 +68,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	dependentsJSON, err := json.Marshal(batchDependent.Dependents)
 	if err != nil {
@@ -95,8 +95,8 @@ func (s *batchDependentStore) Create(ctx context.Context, batchDependent entity.
 // if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "update_dependents", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 170490150..00df23069 100644
--- a/submitqueue/extension/storage/mysql/batch_store.go
+++ b/submitqueue/extension/storage/mysql/batch_store.go
@@ -42,8 +42,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var batch entity.Batch
 	var containsJSON []byte
@@ -74,8 +74,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	containsJSON, err := json.Marshal(batch.Contains)
 	if err != nil {
@@ -106,8 +106,8 @@ func (s *batchStore) Create(ctx context.Context, batch entity.Batch) (retErr err
 // if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "update_state", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	result, err := s.db.ExecContext(ctx,
 		"UPDATE batch SET state = ?, version = ? WHERE id = ? AND version = ?",
@@ -142,8 +142,8 @@ func (s *batchStore) UpdateState(ctx context.Context, id string, oldVersion, new
 // if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "update_score_and_state", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	result, err := s.db.ExecContext(ctx,
 		"UPDATE batch SET score = ?, state = ?, version = ? WHERE id = ? AND version = ?",
@@ -176,8 +176,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get_by_queue_and_states", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 250d59581..45ef7b14b 100644
--- a/submitqueue/extension/storage/mysql/build_store.go
+++ b/submitqueue/extension/storage/mysql/build_store.go
@@ -40,8 +40,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var build entity.Build
 
@@ -62,8 +62,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	_, err := s.db.ExecContext(ctx,
 		"INSERT INTO build (id, batch_id, status) VALUES (?, ?, ?)",
@@ -82,8 +82,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "update_status", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 dd7755cc2..1167fb00e 100644
--- a/submitqueue/extension/storage/mysql/change_store.go
+++ b/submitqueue/extension/storage/mysql/change_store.go
@@ -41,8 +41,8 @@ func NewChangeStore(db *sql.DB, scope tally.Scope) storage.ChangeStore {
 // (queue, uri, request_id) is silently ignored via INSERT IGNORE, so
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	detailsJSON, err := marshalDetails(record.Details)
 	if err != nil {
@@ -61,8 +61,8 @@ func (s *changeStore) Create(ctx context.Context, record entity.ChangeRecord) (r
 // GetByURI returns every ChangeRecord for (queue, uri). queue leads the WHERE
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get_by_uri", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 afdb7cffe..ad55673d5 100644
--- a/submitqueue/extension/storage/mysql/request_log_store.go
+++ b/submitqueue/extension/storage/mysql/request_log_store.go
@@ -43,8 +43,8 @@ func NewRequestLogStore(db *sql.DB, scope tally.Scope) storage.RequestLogStore {
 // millisecond-precision collisions), so a random salt is generated to guarantee uniqueness
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "insert", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	metadataJSON, err := json.Marshal(log.Metadata)
 	if err != nil {
@@ -71,8 +71,8 @@ func (r *requestLogStore) Insert(ctx context.Context, log entity.RequestLog) (re
 // Salt is used as a secondary sort key to provide stable ordering for entries that share a
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "list", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 97871d152..3e59b9fd7 100644
--- a/submitqueue/extension/storage/mysql/request_queue_summary_store.go
+++ b/submitqueue/extension/storage/mysql/request_queue_summary_store.go
@@ -40,8 +40,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata)
 	if err != nil {
@@ -66,8 +66,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var changeURIsJSON []byte
 	var metadataJSON []byte
@@ -90,8 +90,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	metadataJSON, err := json.Marshal(normalizeMetadata(summary.Metadata))
 	if err != nil {
@@ -118,8 +118,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 12fdaeeb8..2860246d5 100644
--- a/submitqueue/extension/storage/mysql/request_store.go
+++ b/submitqueue/extension/storage/mysql/request_store.go
@@ -41,8 +41,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var req entity.Request
 	var changeURIsJSON []byte
@@ -69,8 +69,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	// Marshal the change URIs to JSON
 	changeURIsJSON, err := json.Marshal(request.Change.URIs)
@@ -97,8 +97,8 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
 // if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
 // 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(r.scope, "update_state", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 f94f1a285..3894564e4 100644
--- a/submitqueue/extension/storage/mysql/request_summary_store.go
+++ b/submitqueue/extension/storage/mysql/request_summary_store.go
@@ -40,8 +40,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata)
 	if err != nil {
@@ -70,8 +70,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	var changeURIsJSON []byte
 	var metadataJSON []byte
@@ -100,8 +100,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 bda76569a..b38a5d69f 100644
--- a/submitqueue/extension/storage/mysql/request_uri_store.go
+++ b/submitqueue/extension/storage/mysql/request_uri_store.go
@@ -39,8 +39,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	_, err := s.db.ExecContext(ctx,
 		"INSERT INTO change_uri_request_mapping (change_uri, received_at_ms, request_id) VALUES (?, ?, ?)",
@@ -57,8 +57,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(s.scope, "list_by_uri", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	rows, err := s.db.QueryContext(ctx, `
 		SELECT change_uri, received_at_ms, request_id
diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go
index 8b753b7a8..1352388c1 100644
--- a/submitqueue/gateway/controller/land.go
+++ b/submitqueue/gateway/controller/land.go
@@ -91,8 +91,8 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter cou
 func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (result entity.LandResult, retErr error) {
 	const opName = "land"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	// 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 f0d28d0ef..ed2d91117 100644
--- a/submitqueue/gateway/controller/list.go
+++ b/submitqueue/gateway/controller/list.go
@@ -64,8 +64,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, "list", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	if err := validateStoredIdentifier("queue", req.Queue); err != nil {
 		return entity.ListResult{}, fmt.Errorf("invalid queue: %w", err)
diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go
index 8b10fefa7..5deb70e74 100644
--- a/submitqueue/gateway/controller/log/log.go
+++ b/submitqueue/gateway/controller/log/log.go
@@ -68,8 +68,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/gateway/controller/request_history.go b/submitqueue/gateway/controller/request_history.go
index 4311fa251..58d0a6b1a 100644
--- a/submitqueue/gateway/controller/request_history.go
+++ b/submitqueue/gateway/controller/request_history.go
@@ -49,8 +49,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, "get_by_id", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	if err := validateStoredIdentifier("sqid", req.ID); err != nil {
 		return nil, fmt.Errorf("GetRequestHistoryByID invalid request: %w", err)
@@ -73,8 +73,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, "get_by_change_uri", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 0cf0eb26c..fc0929d33 100644
--- a/submitqueue/gateway/controller/request_summary.go
+++ b/submitqueue/gateway/controller/request_summary.go
@@ -46,8 +46,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, "get_by_id", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	if err := validateStoredIdentifier("sqid", req.ID); err != nil {
 		return entity.RequestSummary{}, fmt.Errorf("GetRequestSummaryByID invalid request: %w", err)
@@ -73,8 +73,8 @@ 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, metrics.StorageLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, "get_by_change_uri", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	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 931b5da54..8e25435c4 100644
--- a/submitqueue/orchestrator/controller/batch/batch.go
+++ b/submitqueue/orchestrator/controller/batch/batch.go
@@ -78,8 +78,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go
index 1e583035b..e9170ed24 100644
--- a/submitqueue/orchestrator/controller/build/build.go
+++ b/submitqueue/orchestrator/controller/build/build.go
@@ -73,8 +73,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
index e8744d356..a30dfddf7 100644
--- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
+++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
@@ -104,8 +104,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go
index 5f9bad6f7..f5c36e4af 100644
--- a/submitqueue/orchestrator/controller/conclude/conclude.go
+++ b/submitqueue/orchestrator/controller/conclude/conclude.go
@@ -65,8 +65,8 @@ func NewController(
 // Deserializes the batch and completes the pipeline processing.
 // 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, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, "process", metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go
index 09853effd..c91b03bb0 100644
--- a/submitqueue/orchestrator/controller/dlq/batch.go
+++ b/submitqueue/orchestrator/controller/dlq/batch.go
@@ -73,8 +73,8 @@ func NewDLQBatchController(
 func (c *batchController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go
index b77fdce2f..aa8e690eb 100644
--- a/submitqueue/orchestrator/controller/dlq/buildsignal.go
+++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go
@@ -71,8 +71,8 @@ func NewDLQBuildSignalController(
 func (c *buildSignalController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/log.go b/submitqueue/orchestrator/controller/dlq/log.go
index 726ceca5d..7b02b8dbe 100644
--- a/submitqueue/orchestrator/controller/dlq/log.go
+++ b/submitqueue/orchestrator/controller/dlq/log.go
@@ -60,8 +60,8 @@ func NewDLQLogController(
 func (c *logController) Process(_ context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 	dmeta := delivery.Metadata()
diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
index 9e2f878a3..52489d91b 100644
--- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
+++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
@@ -67,8 +67,8 @@ func NewDLQMergeConflictSignalController(
 func (c *mergeConflictSignalController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go
index 0a15951c9..155efa556 100644
--- a/submitqueue/orchestrator/controller/dlq/mergesignal.go
+++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go
@@ -66,8 +66,8 @@ func NewDLQMergeSignalController(
 func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go
index a319d4d24..ee8df0fc9 100644
--- a/submitqueue/orchestrator/controller/dlq/request.go
+++ b/submitqueue/orchestrator/controller/dlq/request.go
@@ -109,8 +109,8 @@ func NewDLQRequestController(
 func (c *requestController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go
index 7e8b2257e..74c52323b 100644
--- a/submitqueue/orchestrator/controller/merge/merge.go
+++ b/submitqueue/orchestrator/controller/merge/merge.go
@@ -93,8 +93,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
index c2f54e63a..5007b7266 100644
--- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
+++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
@@ -79,8 +79,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
index e584f1fde..2d619bdd6 100644
--- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
+++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
@@ -80,8 +80,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/ping.go b/submitqueue/orchestrator/controller/ping.go
index 5370a048d..c788211a2 100644
--- a/submitqueue/orchestrator/controller/ping.go
+++ b/submitqueue/orchestrator/controller/ping.go
@@ -43,8 +43,8 @@ func NewPingController(logger *zap.Logger, scope tally.Scope) *PingController {
 func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) {
 	const opName = "ping"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.FastLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.FastLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	message := "pong!"
 	isEcho := false
diff --git a/submitqueue/orchestrator/controller/score/score.go b/submitqueue/orchestrator/controller/score/score.go
index b0191f6aa..055a26386 100644
--- a/submitqueue/orchestrator/controller/score/score.go
+++ b/submitqueue/orchestrator/controller/score/score.go
@@ -76,8 +76,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go
index cd69dd34e..4a92885c5 100644
--- a/submitqueue/orchestrator/controller/speculate/speculate.go
+++ b/submitqueue/orchestrator/controller/speculate/speculate.go
@@ -90,8 +90,8 @@ func NewController(
 // Process advances a batch one step along the naive happy-path.
 // 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, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go
index fd8d0493c..a67964068 100644
--- a/submitqueue/orchestrator/controller/start/start.go
+++ b/submitqueue/orchestrator/controller/start/start.go
@@ -73,8 +73,8 @@ func NewController(
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	const opName = "process"
 
-	op := metrics.Begin(c.metricsScope, opName)
-	defer func() { op.Complete(retErr, metrics.LongLatencyBuckets) }()
+	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()
 
diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go
index 3c0a65e25..7cafc95cf 100644
--- a/submitqueue/orchestrator/controller/validate/validate.go
+++ b/submitqueue/orchestrator/controller/validate/validate.go
@@ -89,8 +89,8 @@ func NewController(
 // asynchronous merge-conflict check by publishing the full check request to runway.
 // 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, coremetrics.LongLatencyBuckets) }()
+	op := coremetrics.Begin(c.metricsScope, "process", coremetrics.LongLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
 
 	msg := delivery.Message()