refactor: simplify operation metrics

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,

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

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

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(`

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

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