Identify reacquire in traces and metrics.

Author: samuel-williams-shopify

lib/async/limiter/queued.rb

@@ -54,7 +54,7 @@ def expand(count, value = true)
 			protected
 
 			# Acquire a resource from the queue with optional deadline.
-			def acquire_resource(deadline, **options)
+			def acquire_resource(deadline, reacquire: false, **options)
 				@mutex.unlock
 				return @queue.pop(timeout: deadline&.remaining, **options)
 			ensure

lib/async/limiter/token.rb

@@ -73,7 +73,7 @@ def release
 			def acquire(**options, &block)
 				raise "Token already acquired!" if @resource
 
-				@resource = @limiter.acquire(**options)
+				@resource = @limiter.acquire(reacquire: true, **options)
 
 				return @resource unless block_given?
 

lib/metrics/provider/async/limiter/generic.rb

@@ -18,7 +18,8 @@
 
 	def acquire_synchronized(timeout, cost, **options)
 		# Build base tags and extend with instance tags if present
-		tags = ["limiter_class:#{self.class.name}", "cost:#{cost}"]
+		is_reacquire = options[:reacquire] || false
+		tags = ["limiter:#{self.class.name}", "cost:#{cost}", "reacquire:#{is_reacquire}"]
 		tags = Metrics::Tags.normalize(@tags, tags)
 
 		clock = Async::Clock.start
@@ -31,15 +32,15 @@ def acquire_synchronized(timeout, cost, **options)
 				ACQUIRE_COUNTER.emit(1, tags: success_tags)
 				ACQUIRE_DURATION.emit(clock.total, tags: success_tags)
 			else
-				# Emit failure metrics
-				error_tags = Metrics::Tags.normalize(["result:error", "error:#{error.class.name}"], tags)
-				ACQUIRE_COUNTER.emit(1, tags: error_tags)
-				ACQUIRE_DURATION.emit(clock.total, tags: error_tags)
+				# Emit failure metrics (timeout/contention)
+				failure_tags = Metrics::Tags.normalize(["result:timeout"], tags)
+				ACQUIRE_COUNTER.emit(1, tags: failure_tags)
+				ACQUIRE_DURATION.emit(clock.total, tags: failure_tags)
 			end
 
 			return result
 		rescue => error
-			# Emit failure metrics
+			# Emit error metrics
 			error_tags = Metrics::Tags.normalize(["result:error", "error:#{error.class.name}"], tags)
 			ACQUIRE_COUNTER.emit(1, tags: error_tags)
 			ACQUIRE_DURATION.emit(clock.total, tags: error_tags)
@@ -50,7 +51,7 @@ def acquire_synchronized(timeout, cost, **options)
 
 	def release(resource = true)
 		# Build base tags and extend with instance tags if present
-		tags = ["limiter_class:#{self.class.name}"]
+		tags = ["limiter:#{self.class.name}"]
 		tags = Metrics::Tags.normalize(@tags, tags)
 
 		begin

lib/traces/provider/async/limiter/generic.rb

@@ -16,6 +16,7 @@ def acquire_synchronized(timeout, cost, **options)
 			"cost" => cost,
 			"timeout" => timeout,
 			"priority" => options[:priority],
+			"reacquire" => options[:reacquire] || false,
 		}
 
 		if @tags