@@ -171,8 +171,9 @@ local __qm_g = false
171171local function __qmret(r) if r == nil then r = false end return {r, __qm_g} end` ;
172172
173173// Fresh-read gauge for splice points with no reusable locals: enqueue slow-path (before
174- // return 0) and the base dequeue top. Gated on the last ARGV so it is inert unless the
175- // caller opts in. CK queues emit per-subqueue depth (queue_name aggregates via the MV).
174+ // return 0) and the base dequeue's sample-at-return wrapper. Gated on the last ARGV so it
175+ // is inert unless the caller opts in. CK queues emit per-subqueue depth (queue_name
176+ // aggregates via the MV).
176177const QUEUE_METRICS_GAUGE_LUA = createMetricsGaugeComputeLua ( {
177178 enabledArg : "ARGV[#ARGV] == '1'" ,
178179 queued : "redis.call('ZCARD', queueKey)" ,
@@ -4694,7 +4695,16 @@ local gatesEnabled = ARGV[7] == '1'
46944695local totalConcurrencyEnabled = ARGV[8] == '1'
46954696${ QUEUE_METRICS_GAUGE_PRELUDE }
46964697${ QUEUE_GATES_LUA_HELPERS }
4698+ -- Sample-at-return: the gauge is computed once, by the return wrapper, so every
4699+ -- exit emits the state as of that exit (post-admission on the success path) and
4700+ -- no path pays for a sample that a later one would overwrite.
4701+ local function __qmsample()
46974702${ QUEUE_METRICS_GAUGE_LUA }
4703+ end
4704+ do
4705+ local __qmret_inner = __qmret
4706+ __qmret = function(r) __qmsample() return __qmret_inner(r) end
4707+ end
46984708
46994709-- Check current env concurrency against the limit
47004710local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
@@ -4806,10 +4816,6 @@ else
48064816 redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
48074817end
48084818
4809- -- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
4810- -- the top-of-script sample only covers the early returns where nothing was admitted.
4811- ${ QUEUE_METRICS_GAUGE_LUA }
4812-
48134819-- Return results as a flat array: [messageId1, messageScore1, messagePayload1, messageId2, messageScore2, messagePayload2, ...]
48144820return __qmret(results)
48154821 ` ,
@@ -5003,7 +5009,16 @@ local totalConcurrencyEnabled = ARGV[7] == '1'
50035009local gatesEnabled = ARGV[8] == '1'
50045010${ QUEUE_METRICS_GAUGE_PRELUDE }
50055011${ QUEUE_GATES_LUA_HELPERS }
5012+ -- Sample-at-return: the gauge is computed once, by the return wrapper, so every
5013+ -- exit emits the state as of that exit (post-admission on the success path) and
5014+ -- no path pays for a sample that a later one would overwrite.
5015+ local function __qmsample()
50065016${ QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA }
5017+ end
5018+ do
5019+ local __qmret_inner = __qmret
5020+ __qmret = function(r) __qmsample() return __qmret_inner(r) end
5021+ end
50075022
50085023local function decrLengthCounter()
50095024 if tonumber(redis.call('GET', lengthCounterKey) or '0') > 0 then
@@ -5192,10 +5207,6 @@ else
51925207 redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
51935208end
51945209
5195- -- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
5196- -- the top-of-script sample only covers the early returns where nothing was admitted.
5197- ${ QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA }
5198-
51995210return __qmret(results)
52005211 ` ,
52015212 } ) ;
0 commit comments