Skip to content

Commit da7dcb8

Browse files
committed
fix(run-engine): base-queue admit paths enforce the total limit for keyless runs
Keyless admits join the per-base-queue group set and are gated on the env-clamped total limit across the base enqueue fast path, the base dequeue and the queue mover, with the bounded reconcile at saturation and the release-side group mirror, matching the keyed tracked paths. Base-queue gauges carry the total running/limit fields.
1 parent e6b60cf commit da7dcb8

1 file changed

Lines changed: 135 additions & 12 deletions

File tree

  • internal-packages/run-engine/src/run-queue

‎internal-packages/run-engine/src/run-queue/index.ts‎

Lines changed: 135 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,22 @@ const QUEUE_METRICS_GAUGE_PRELUDE = `
170170
local __qm_g = false
171171
local function __qmret(r) if r == nil then r = false end return {r, __qm_g} end`;
172172

173+
/** Total-concurrency tail (gauge[10]/gauge[11]): live group cardinality + raw stored cap.
174+
* Requires the groupConcurrencyKey local and the __totalLimitRaw memo (one GET shared with
175+
* the total-cap gate); every script that runs a gauge with this tail declares both. The
176+
* group SCARD stays a fresh read: it must be post-admission. */
177+
const QUEUE_METRICS_TOTAL_GAUGE_EXTRAS = {
178+
totalRunning: "redis.call('SCARD', groupConcurrencyKey)",
179+
totalLimit: "__totalLimitRaw() or '0'",
180+
};
181+
182+
/** The gauge layout is positional (totals ride behind the CK slots), so the plain
183+
* scripts zero-fill the CK health fields: a base queue has no CK variants to backlog. */
184+
const QUEUE_METRICS_PLAIN_CK_ZERO_EXTRAS = {
185+
ckBacklogged: "0",
186+
ckMaxWaitMs: "0",
187+
};
188+
173189
// Fresh-read gauge for splice points with no reusable locals: enqueue slow-path (before
174190
// return 0) and the base dequeue top. Gated on the last ARGV so it is inert unless the
175191
// caller opts in. CK queues emit per-subqueue depth (queue_name aggregates via the MV).
@@ -181,6 +197,8 @@ const QUEUE_METRICS_GAUGE_LUA = createMetricsGaugeComputeLua({
181197
envQueued: "redis.call('ZCARD', envQueueKey)",
182198
envRunning: "redis.call('SCARD', envCurrentConcurrencyKey)",
183199
envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit",
200+
...QUEUE_METRICS_PLAIN_CK_ZERO_EXTRAS,
201+
...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS,
184202
});
185203

186204
// Enqueue fast-path gauge: the admission check already computed queueCurrent/envCurrent/
@@ -194,6 +212,8 @@ const QUEUE_METRICS_ENQUEUE_FASTPATH_GAUGE_LUA = createMetricsGaugeComputeLua({
194212
envQueued: "redis.call('ZCARD', envQueueKey)",
195213
envRunning: "envCurrent",
196214
envLimit: "envLimit",
215+
...QUEUE_METRICS_PLAIN_CK_ZERO_EXTRAS,
216+
...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS,
197217
});
198218

199219
// CK-health extras: distinct backlogged keys + most-starved head-of-line wait (ckIndex scores
@@ -284,8 +304,8 @@ export type RunQueueOptions = {
284304
*/
285305
counterTtlSeconds?: number;
286306
/**
287-
* When true, concurrency-keyed queues maintain a per-base-queue groupConcurrency SET
288-
* (total in-flight across all key variants) and enforce the queue's total concurrency
307+
* When true, queues maintain a per-base-queue groupConcurrency SET (total in-flight
308+
* across all key variants AND keyless runs) and enforce the queue's total concurrency
289309
* limit at admit time. Default false: admit paths are byte-identical to before, and
290310
* only the release-side SREM mirror runs (a no-op on an absent set), so the flag can
291311
* be flipped on a fleet that has fully rolled onto this build without draining queues.
@@ -621,7 +641,7 @@ export class RunQueue {
621641
}
622642

623643
/**
624-
* Total in-flight runs across all concurrency-key variants of a queue (the
644+
* Total in-flight runs on a queue, keyed and keyless together (the
625645
* groupConcurrency SET cardinality). Admits only populate the set while
626646
* totalConcurrencyEnabled is on. After the flag is turned off the set drains
627647
* to zero through the release-side mirrors, so a nonzero read reflects real
@@ -2477,6 +2497,8 @@ export class RunQueue {
24772497
queueConcurrencyLimitKey,
24782498
envConcurrencyLimitKey,
24792499
envConcurrencyLimitBurstFactorKey,
2500+
this.keys.queueGroupConcurrencyKeyFromQueue(message.queue),
2501+
this.keys.queueTotalConcurrencyLimitKeyFromQueue(message.queue),
24802502
// args
24812503
queueName,
24822504
messageId,
@@ -2508,6 +2530,8 @@ export class RunQueue {
25082530
queueConcurrencyLimitKey,
25092531
envConcurrencyLimitKey,
25102532
envConcurrencyLimitBurstFactorKey,
2533+
this.keys.queueGroupConcurrencyKeyFromQueue(message.queue),
2534+
this.keys.queueTotalConcurrencyLimitKeyFromQueue(message.queue),
25112535
// args
25122536
queueName,
25132537
messageId,
@@ -2593,6 +2617,8 @@ export class RunQueue {
25932617
envQueueKey,
25942618
masterQueueKey,
25952619
ttlQueueKey,
2620+
this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue),
2621+
this.keys.queueTotalConcurrencyLimitKeyFromQueue(messageQueue),
25962622
//args
25972623
messageQueue,
25982624
String(Date.now()),
@@ -3569,7 +3595,7 @@ end
35693595
// When enableFastPath == '0', the script skips the fast-path check entirely and behaves
35703596
// identically to the pre-fast-path version (with the addition of returning 0).
35713597
this.redis.defineCommand("enqueueMessage", {
3572-
numberOfKeys: 12,
3598+
numberOfKeys: 14,
35733599
lua: `
35743600
local masterQueueKey = KEYS[1]
35753601
local queueKey = KEYS[2]
@@ -3584,6 +3610,16 @@ local workerQueueKey = KEYS[9]
35843610
local queueConcurrencyLimitKey = KEYS[10]
35853611
local envConcurrencyLimitKey = KEYS[11]
35863612
local envConcurrencyLimitBurstFactorKey = KEYS[12]
3613+
-- Total-cap keys (KEYS 13-14)
3614+
local groupConcurrencyKey = KEYS[13]
3615+
local totalConcurrencyLimitKey = KEYS[14]
3616+
local __rawTotalLimit = nil
3617+
local function __totalLimitRaw()
3618+
if __rawTotalLimit == nil then
3619+
__rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey) or false
3620+
end
3621+
return __rawTotalLimit
3622+
end
35873623
35883624
local queueName = ARGV[1]
35893625
local messageId = ARGV[2]
@@ -3618,6 +3654,20 @@ if enableFastPath == '1' then
36183654
)
36193655
36203656
if queueCurrent < queueLimit then
3657+
-- Total-cap gate: a fast-path admit consumes a group slot, so it must
3658+
-- respect the env-clamped total limit. At the cap we fall through to the
3659+
-- slow path (the message queues; the dequeue gate holds it).
3660+
local totalAllowsFastPath = true
3661+
if totalConcurrencyEnabled then
3662+
local rawTotalLimit = __totalLimitRaw()
3663+
if rawTotalLimit then
3664+
local totalLimit = math.min(tonumber(rawTotalLimit), envLimit)
3665+
if tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') >= totalLimit then
3666+
totalAllowsFastPath = false
3667+
end
3668+
end
3669+
end
3670+
36213671
local gateMsg = nil
36223672
local gatesAllowFastPath = true
36233673
if gatesEnabled and string.find(messageData, '"gates"', 1, true) then
@@ -3628,10 +3678,13 @@ if enableFastPath == '1' then
36283678
end
36293679
end
36303680
3631-
if gatesAllowFastPath then
3681+
if totalAllowsFastPath and gatesAllowFastPath then
36323682
redis.call('SET', messageKey, messageData)
36333683
redis.call('SADD', queueCurrentConcurrencyKey, messageId)
36343684
redis.call('SADD', envCurrentConcurrencyKey, messageId)
3685+
if totalConcurrencyEnabled then
3686+
redis.call('SADD', groupConcurrencyKey, messageId)
3687+
end
36353688
if gateMsg then
36363689
__gatesAcquire(keyPrefix, gateMsg, messageId)
36373690
end
@@ -3663,8 +3716,12 @@ else
36633716
redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
36643717
end
36653718
3666-
-- Update the concurrency keys
3667-
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
3719+
-- Update the concurrency keys. The groupConcurrency SREM mirrors the base SREM
3720+
-- unconditionally (no flag check) so a disabled flag still drains the group set.
3721+
local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
3722+
if removedFromCurrentConcurrency == 1 then
3723+
redis.call('SREM', groupConcurrencyKey, messageId)
3724+
end
36683725
redis.call('SREM', envCurrentConcurrencyKey, messageId)
36693726
redis.call('SREM', queueCurrentDequeuedKey, messageId)
36703727
redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -3680,7 +3737,7 @@ return __qmret(0)
36803737
// (scheduled independently before enqueue) handles TTL expiry. This mirrors what
36813738
// dequeueMessagesFromQueue does: it removes from the TTL set when dequeuing.
36823739
this.redis.defineCommand("enqueueMessageWithTtl", {
3683-
numberOfKeys: 13,
3740+
numberOfKeys: 15,
36843741
lua: `
36853742
local masterQueueKey = KEYS[1]
36863743
local queueKey = KEYS[2]
@@ -3696,6 +3753,16 @@ local workerQueueKey = KEYS[10]
36963753
local queueConcurrencyLimitKey = KEYS[11]
36973754
local envConcurrencyLimitKey = KEYS[12]
36983755
local envConcurrencyLimitBurstFactorKey = KEYS[13]
3756+
-- Total-cap keys (KEYS 14-15)
3757+
local groupConcurrencyKey = KEYS[14]
3758+
local totalConcurrencyLimitKey = KEYS[15]
3759+
local __rawTotalLimit = nil
3760+
local function __totalLimitRaw()
3761+
if __rawTotalLimit == nil then
3762+
__rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey) or false
3763+
end
3764+
return __rawTotalLimit
3765+
end
36993766
37003767
local queueName = ARGV[1]
37013768
local messageId = ARGV[2]
@@ -3732,6 +3799,20 @@ if enableFastPath == '1' then
37323799
)
37333800
37343801
if queueCurrent < queueLimit then
3802+
-- Total-cap gate: a fast-path admit consumes a group slot, so it must
3803+
-- respect the env-clamped total limit. At the cap we fall through to the
3804+
-- slow path (the message queues; the dequeue gate holds it).
3805+
local totalAllowsFastPath = true
3806+
if totalConcurrencyEnabled then
3807+
local rawTotalLimit = __totalLimitRaw()
3808+
if rawTotalLimit then
3809+
local totalLimit = math.min(tonumber(rawTotalLimit), envLimit)
3810+
if tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') >= totalLimit then
3811+
totalAllowsFastPath = false
3812+
end
3813+
end
3814+
end
3815+
37353816
local gateMsg = nil
37363817
local gatesAllowFastPath = true
37373818
if gatesEnabled and string.find(messageData, '"gates"', 1, true) then
@@ -3742,10 +3823,13 @@ if enableFastPath == '1' then
37423823
end
37433824
end
37443825
3745-
if gatesAllowFastPath then
3826+
if totalAllowsFastPath and gatesAllowFastPath then
37463827
redis.call('SET', messageKey, messageData)
37473828
redis.call('SADD', queueCurrentConcurrencyKey, messageId)
37483829
redis.call('SADD', envCurrentConcurrencyKey, messageId)
3830+
if totalConcurrencyEnabled then
3831+
redis.call('SADD', groupConcurrencyKey, messageId)
3832+
end
37493833
if gateMsg then
37503834
__gatesAcquire(keyPrefix, gateMsg, messageId)
37513835
end
@@ -3781,8 +3865,12 @@ else
37813865
redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
37823866
end
37833867
3784-
-- Update the concurrency keys
3785-
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
3868+
-- Update the concurrency keys. The groupConcurrency SREM mirrors the base SREM
3869+
-- unconditionally (no flag check) so a disabled flag still drains the group set.
3870+
local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
3871+
if removedFromCurrentConcurrency == 1 then
3872+
redis.call('SREM', groupConcurrencyKey, messageId)
3873+
end
37863874
redis.call('SREM', envCurrentConcurrencyKey, messageId)
37873875
redis.call('SREM', queueCurrentDequeuedKey, messageId)
37883876
redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -4579,7 +4667,7 @@ return results
45794667
});
45804668

45814669
this.redis.defineCommand("dequeueMessagesFromQueue", {
4582-
numberOfKeys: 10,
4670+
numberOfKeys: 12,
45834671
lua: `
45844672
local queueKey = KEYS[1]
45854673
local queueConcurrencyLimitKey = KEYS[2]
@@ -4591,6 +4679,16 @@ local messageKeyPrefix = KEYS[7]
45914679
local envQueueKey = KEYS[8]
45924680
local masterQueueKey = KEYS[9]
45934681
local ttlQueueKey = KEYS[10] -- Optional: TTL sorted set key (empty string if not used)
4682+
-- Total-cap keys (KEYS 11-12)
4683+
local groupConcurrencyKey = KEYS[11]
4684+
local totalConcurrencyLimitKey = KEYS[12]
4685+
local __rawTotalLimit = nil
4686+
local function __totalLimitRaw()
4687+
if __rawTotalLimit == nil then
4688+
__rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey) or false
4689+
end
4690+
return __rawTotalLimit
4691+
end
45944692
45954693
local queueName = ARGV[1]
45964694
local currentTime = tonumber(ARGV[2])
@@ -4628,6 +4726,22 @@ local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConc
46284726
local queueAvailableCapacity = totalQueueConcurrencyLimit - queueCurrentConcurrency
46294727
local actualMaxCount = math.min(maxCount, envAvailableCapacity, queueAvailableCapacity)
46304728
4729+
-- Total-cap gate: every admit joins the group set, so the batch is bounded by the
4730+
-- env-clamped total limit's remaining capacity. At saturation, run the bounded
4731+
-- reconcile once (heals leaked members) and re-check before giving up.
4732+
if totalConcurrencyEnabled then
4733+
local rawTotalLimit = __totalLimitRaw()
4734+
if rawTotalLimit then
4735+
local totalLimit = math.min(tonumber(rawTotalLimit), envConcurrencyLimit)
4736+
local groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
4737+
if groupCurrentConcurrency >= totalLimit then
4738+
__gateReconcile(groupConcurrencyKey, messageKeyPrefix, keyPrefix)
4739+
groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
4740+
end
4741+
actualMaxCount = math.min(actualMaxCount, totalLimit - groupCurrentConcurrency)
4742+
end
4743+
end
4744+
46314745
if actualMaxCount <= 0 then
46324746
return __qmret(nil)
46334747
end
@@ -4679,6 +4793,9 @@ for i = 1, #messages, 2 do
46794793
redis.call('ZREM', envQueueKey, messageId)
46804794
redis.call('SADD', queueCurrentConcurrencyKey, messageId)
46814795
redis.call('SADD', envCurrentConcurrencyKey, messageId)
4796+
if totalConcurrencyEnabled then
4797+
redis.call('SADD', groupConcurrencyKey, messageId)
4798+
end
46824799
if gatesEnabled then
46834800
__gatesAcquire(keyPrefix, messageData, messageId)
46844801
end
@@ -6071,6 +6188,8 @@ declare module "@internal/redis" {
60716188
queueConcurrencyLimitKey: string,
60726189
envConcurrencyLimitKey: string,
60736190
envConcurrencyLimitBurstFactorKey: string,
6191+
groupConcurrencyKey: string,
6192+
totalConcurrencyLimitKey: string,
60746193
//args
60756194
queueName: string,
60766195
messageId: string,
@@ -6102,6 +6221,8 @@ declare module "@internal/redis" {
61026221
queueConcurrencyLimitKey: string,
61036222
envConcurrencyLimitKey: string,
61046223
envConcurrencyLimitBurstFactorKey: string,
6224+
groupConcurrencyKey: string,
6225+
totalConcurrencyLimitKey: string,
61056226
//args
61066227
queueName: string,
61076228
messageId: string,
@@ -6146,6 +6267,8 @@ declare module "@internal/redis" {
61466267
envQueueKey: string,
61476268
masterQueueKey: string,
61486269
ttlQueueKey: string,
6270+
groupConcurrencyKey: string,
6271+
totalConcurrencyLimitKey: string,
61496272
//args
61506273
childQueueName: string,
61516274
currentTime: string,

0 commit comments

Comments
 (0)