@@ -187,8 +187,9 @@ const QUEUE_METRICS_PLAIN_CK_ZERO_EXTRAS = {
187187} ;
188188
189189// Fresh-read gauge for splice points with no reusable locals: enqueue slow-path (before
190- // return 0) and the base dequeue top. Gated on the last ARGV so it is inert unless the
191- // caller opts in. CK queues emit per-subqueue depth (queue_name aggregates via the MV).
190+ // return 0) and the base dequeue's sample-at-return wrapper. Gated on the last ARGV so it
191+ // is inert unless the caller opts in. CK queues emit per-subqueue depth (queue_name
192+ // aggregates via the MV).
192193const QUEUE_METRICS_GAUGE_LUA = createMetricsGaugeComputeLua ( {
193194 enabledArg : "ARGV[#ARGV] == '1'" ,
194195 queued : "redis.call('ZCARD', queueKey)" ,
@@ -4792,7 +4793,16 @@ local gatesEnabled = ARGV[7] == '1'
47924793local totalConcurrencyEnabled = ARGV[8] == '1'
47934794${ QUEUE_METRICS_GAUGE_PRELUDE }
47944795${ QUEUE_GATES_LUA_HELPERS }
4796+ -- Sample-at-return: the gauge is computed once, by the return wrapper, so every
4797+ -- exit emits the state as of that exit (post-admission on the success path) and
4798+ -- no path pays for a sample that a later one would overwrite.
4799+ local function __qmsample()
47954800${ QUEUE_METRICS_GAUGE_LUA }
4801+ end
4802+ do
4803+ local __qmret_inner = __qmret
4804+ __qmret = function(r) __qmsample() return __qmret_inner(r) end
4805+ end
47964806
47974807-- Check current env concurrency against the limit
47984808local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
@@ -4923,10 +4933,6 @@ else
49234933 redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
49244934end
49254935
4926- -- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
4927- -- the top-of-script sample only covers the early returns where nothing was admitted.
4928- ${ QUEUE_METRICS_GAUGE_LUA }
4929-
49304936-- Return results as a flat array: [messageId1, messageScore1, messagePayload1, messageId2, messageScore2, messagePayload2, ...]
49314937return __qmret(results)
49324938 ` ,
@@ -5120,7 +5126,16 @@ local totalConcurrencyEnabled = ARGV[7] == '1'
51205126local gatesEnabled = ARGV[8] == '1'
51215127${ QUEUE_METRICS_GAUGE_PRELUDE }
51225128${ QUEUE_GATES_LUA_HELPERS }
5129+ -- Sample-at-return: the gauge is computed once, by the return wrapper, so every
5130+ -- exit emits the state as of that exit (post-admission on the success path) and
5131+ -- no path pays for a sample that a later one would overwrite.
5132+ local function __qmsample()
51235133${ QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA }
5134+ end
5135+ do
5136+ local __qmret_inner = __qmret
5137+ __qmret = function(r) __qmsample() return __qmret_inner(r) end
5138+ end
51245139
51255140local function decrLengthCounter()
51265141 if tonumber(redis.call('GET', lengthCounterKey) or '0') > 0 then
@@ -5309,10 +5324,6 @@ else
53095324 redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
53105325end
53115326
5312- -- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
5313- -- the top-of-script sample only covers the early returns where nothing was admitted.
5314- ${ QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA }
5315-
53165327return __qmret(results)
53175328 ` ,
53185329 } ) ;
0 commit comments