feat(run-engine): enforce per-key limit overrides at admit time

The ck dequeue admit and both enqueue fast paths read the queue's ckLimits
HASH for the variant being admitted and use the env-clamped override in
place of the queue's per-key limit, behind the totalConcurrencyEnabled
flag. Covered by tests for lowered and raised keys, removal, the
cardinality cap, and flag-off behavior.

Author: matt-aitken

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

@@ -2449,6 +2449,7 @@ export class RunQueue {
       const totalConcurrencyLimitKey = this.keys.queueTotalConcurrencyLimitKeyFromQueue(
         message.queue
       );
+      const ckLimitsKey = this.keys.queueCkLimitsKeyFromQueue(message.queue);
       const totalConcurrencyEnabledArg = this.options.totalConcurrencyEnabled ? "1" : "0";
 
       if (ttlInfo) {
@@ -2472,6 +2473,7 @@ export class RunQueue {
           baseQueueKey,
           groupConcurrencyKey,
           totalConcurrencyLimitKey,
+          ckLimitsKey,
           // args
           queueName,
           messageId,
@@ -2511,6 +2513,7 @@ export class RunQueue {
           baseQueueKey,
           groupConcurrencyKey,
           totalConcurrencyLimitKey,
+          ckLimitsKey,
           // args
           queueName,
           messageId,
@@ -2798,6 +2801,7 @@ export class RunQueue {
         runningCounterKey,
         this.keys.queueGroupConcurrencyKeyFromQueue(ckWildcardQueue),
         this.keys.queueTotalConcurrencyLimitKeyFromQueue(ckWildcardQueue),
+        this.keys.queueCkLimitsKeyFromQueue(ckWildcardQueue),
         //args
         ckWildcardQueue,
         String(Date.now()),
@@ -4086,7 +4090,7 @@ return __qmret(0)
     // *Tracked variants of dequeueMessageFromKey and the ack/nack/dlq/release/clear
     // scripts.
     this.redis.defineCommand("enqueueMessageCkTracked", {
-      numberOfKeys: 17,
+      numberOfKeys: 18,
       lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -4108,6 +4112,7 @@ local baseQueueKey = KEYS[15]
 -- Total-cap keys (KEYS 16-17)
 local groupConcurrencyKey = KEYS[16]
 local totalConcurrencyLimitKey = KEYS[17]
+local ckLimitsKey = KEYS[18]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -4145,6 +4150,12 @@ if enableFastPath == '1' then
         tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
         envLimit
       )
+      if totalConcurrencyEnabled then
+        local perKeyOverride = redis.call('HGET', ckLimitsKey, queueName)
+        if perKeyOverride then
+          queueLimit = math.min(tonumber(perKeyOverride), envLimit)
+        end
+      end
 
       if queueCurrent < queueLimit then
         -- Total-cap gate: a fast-path admit consumes a group slot, so it must
@@ -4256,7 +4267,7 @@ return __qmret(0)
     });
 
     this.redis.defineCommand("enqueueMessageWithTtlCkTracked", {
-      numberOfKeys: 18,
+      numberOfKeys: 19,
       lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -4279,6 +4290,7 @@ local baseQueueKey = KEYS[16]
 -- Total-cap keys (KEYS 17-18)
 local groupConcurrencyKey = KEYS[17]
 local totalConcurrencyLimitKey = KEYS[18]
+local ckLimitsKey = KEYS[19]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -4318,6 +4330,12 @@ if enableFastPath == '1' then
         tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
         envLimit
       )
+      if totalConcurrencyEnabled then
+        local perKeyOverride = redis.call('HGET', ckLimitsKey, queueName)
+        if perKeyOverride then
+          queueLimit = math.min(tonumber(perKeyOverride), envLimit)
+        end
+      end
 
       if queueCurrent < queueLimit then
         -- Total-cap gate: see enqueueMessageCkTracked.
@@ -4942,7 +4960,7 @@ return results
     // (normal dequeue, TTL-expired, or stale-orphan path — all of which were
     // counted at enqueue time).
     this.redis.defineCommand("dequeueMessagesFromCkQueueTracked", {
-      numberOfKeys: 13,
+      numberOfKeys: 14,
       lua: `
 local ckIndexKey = KEYS[1]
 local queueConcurrencyLimitKey = KEYS[2]
@@ -4957,6 +4975,7 @@ local lengthCounterKey = KEYS[10]
 local runningCounterKey = KEYS[11]
 local groupConcurrencyKey = KEYS[12]
 local totalConcurrencyLimitKey = KEYS[13]
+local ckLimitsKey = KEYS[14]
 
 local ckWildcardName = ARGV[1]
 local currentTime = tonumber(ARGV[2])
@@ -5046,7 +5065,15 @@ for _, ckQueueName in ipairs(ckQueues) do
   local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency'
   local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0')
 
-  if ckCurrentConcurrency < queueConcurrencyLimit then
+  local perKeyLimit = queueConcurrencyLimit
+  if totalConcurrencyEnabled then
+    local perKeyOverride = redis.call('HGET', ckLimitsKey, ckQueueName)
+    if perKeyOverride then
+      perKeyLimit = math.min(tonumber(perKeyOverride), envConcurrencyLimit)
+    end
+  end
+
+  if ckCurrentConcurrency < perKeyLimit then
     local messages = redis.call('ZRANGEBYSCORE', fullQueueKey, '-inf', tostring(currentTime), 'WITHSCORES', 'LIMIT', 0, 1)
 
     if #messages >= 2 then
@@ -6537,6 +6564,7 @@ declare module "@internal/redis" {
       baseQueueKey: string,
       groupConcurrencyKey: string,
       totalConcurrencyLimitKey: string,
+      ckLimitsKey: string,
       queueName: string,
       messageId: string,
       messageData: string,
@@ -6574,6 +6602,7 @@ declare module "@internal/redis" {
       baseQueueKey: string,
       groupConcurrencyKey: string,
       totalConcurrencyLimitKey: string,
+      ckLimitsKey: string,
       queueName: string,
       messageId: string,
       messageData: string,
@@ -6608,6 +6637,7 @@ declare module "@internal/redis" {
       runningCounterKey: string,
       groupConcurrencyKey: string,
       totalConcurrencyLimitKey: string,
+      ckLimitsKey: string,
       ckWildcardName: string,
       currentTime: string,
       defaultEnvConcurrencyLimit: string,