@@ -173,8 +173,9 @@ local __qm_g = false
173173local function __qmret(r) if r == nil then r = false end return {r, __qm_g} end` ;
174174
175175// Fresh-read gauge for splice points with no reusable locals: enqueue slow-path (before
176- // return 0) and the base dequeue top. Gated on the last ARGV so it is inert unless the
177- // caller opts in. CK queues emit per-subqueue depth (queue_name aggregates via the MV).
176+ // return 0) and the base dequeue's sample-at-return wrapper. Gated on the last ARGV so it
177+ // is inert unless the caller opts in. CK queues emit per-subqueue depth (queue_name
178+ // aggregates via the MV).
178179const QUEUE_METRICS_GAUGE_LUA = createMetricsGaugeComputeLua ( {
179180 enabledArg : "ARGV[#ARGV] == '1'" ,
180181 queued : "redis.call('ZCARD', queueKey)" ,
@@ -4696,7 +4697,16 @@ local gatesEnabled = ARGV[7] == '1'
46964697local totalConcurrencyEnabled = ARGV[8] == '1'
46974698${ QUEUE_METRICS_GAUGE_PRELUDE }
46984699${ QUEUE_GATES_LUA_HELPERS }
4700+ -- Sample-at-return: the gauge is computed once, by the return wrapper, so every
4701+ -- exit emits the state as of that exit (post-admission on the success path) and
4702+ -- no path pays for a sample that a later one would overwrite.
4703+ local function __qmsample()
46994704${ QUEUE_METRICS_GAUGE_LUA }
4705+ end
4706+ do
4707+ local __qmret_inner = __qmret
4708+ __qmret = function(r) __qmsample() return __qmret_inner(r) end
4709+ end
47004710
47014711-- Check current env concurrency against the limit
47024712local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
@@ -4808,10 +4818,6 @@ else
48084818 redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
48094819end
48104820
4811- -- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
4812- -- the top-of-script sample only covers the early returns where nothing was admitted.
4813- ${ QUEUE_METRICS_GAUGE_LUA }
4814-
48154821-- Return results as a flat array: [messageId1, messageScore1, messagePayload1, messageId2, messageScore2, messagePayload2, ...]
48164822return __qmret(results)
48174823 ` ,
@@ -5005,7 +5011,16 @@ local totalConcurrencyEnabled = ARGV[7] == '1'
50055011local gatesEnabled = ARGV[8] == '1'
50065012${ QUEUE_METRICS_GAUGE_PRELUDE }
50075013${ QUEUE_GATES_LUA_HELPERS }
5014+ -- Sample-at-return: the gauge is computed once, by the return wrapper, so every
5015+ -- exit emits the state as of that exit (post-admission on the success path) and
5016+ -- no path pays for a sample that a later one would overwrite.
5017+ local function __qmsample()
50085018${ QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA }
5019+ end
5020+ do
5021+ local __qmret_inner = __qmret
5022+ __qmret = function(r) __qmsample() return __qmret_inner(r) end
5023+ end
50095024
50105025local function decrLengthCounter()
50115026 if tonumber(redis.call('GET', lengthCounterKey) or '0') > 0 then
@@ -5194,10 +5209,6 @@ else
51945209 redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
51955210end
51965211
5197- -- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
5198- -- the top-of-script sample only covers the early returns where nothing was admitted.
5199- ${ QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA }
5200-
52015212return __qmret(results)
52025213 ` ,
52035214 } ) ;
0 commit comments