fix(run-engine,webapp): flag-consistent gate overrides; generation-safe cap cleanup

Gate admission reads a per-key override only when concurrency limit
enforcement is enabled, matching the primary admit paths (the flag now
threads through the unkeyed enqueue and dequeue scripts too). The
cap-rejection cleanup deletes only the exact row generation the rejected
request wrote, so a concurrent request that succeeded after capacity freed
keeps its durable record.

Author: matt-aitken

apps/webapp/app/v3/services/concurrencySystem.server.ts

@@ -537,7 +537,7 @@ function overrideQueueConcurrencyKeyLimit(
     }),
     (error) => ({ type: "other" as const, cause: error })
   )
-    .andThen((existing) =>
+    .andThen(() =>
       fromPromise(
         db.taskQueueConcurrencyKeyOverride.upsert({
           where: { taskQueueId_concurrencyKey: { taskQueueId: queue.id, concurrencyKey } },
@@ -557,9 +557,9 @@ function overrideQueueConcurrencyKeyLimit(
           type: "queue_update_failed" as const,
           cause: error,
         })
-      ).map(() => existing)
+      )
     )
-    .andThen((existing) =>
+    .andThen((written) =>
       fromPromise(
         engine.runQueue.updateQueueConcurrencyKeyLimit(
           environment,
@@ -576,8 +576,13 @@ function overrideQueueConcurrencyKeyLimit(
       ).orElse((error) => {
         if (error.type === "too_many_key_overrides") {
           return fromPromise(
+            /**
+             * Deletes only the exact row generation this request wrote, so a
+             * concurrent request that succeeded after capacity freed keeps its
+             * durable record.
+             */
             db.taskQueueConcurrencyKeyOverride.deleteMany({
-              where: { taskQueueId: queue.id, concurrencyKey },
+              where: { id: written.id, overriddenAt: written.overriddenAt },
             }),
             () => error
           ).andThen(() => errAsync(error));

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

@@ -116,13 +116,13 @@ local function __gateReconcile(setKey, msgKeyPrefix, reconcileKeyPrefix)
   end
 end
 
-local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msgKeyPrefix)
+local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msgKeyPrefix, ckOverridesEnabled)
   if not msg.gates then return true end
   for _, gate in ipairs(msg.gates) do
     local base, variant, gateKey = __gateKeys(gatesKeyPrefix, msg, gate)
     local occupancy = tonumber(redis.call('SCARD', variant .. ':currentConcurrency') or '0')
     local perKeyLimit = math.min(tonumber(redis.call('GET', base .. ':concurrency') or '1000000'), envLimit)
-    if gateKey and gateKey ~= '' then
+    if ckOverridesEnabled and gateKey and gateKey ~= '' then
       local gateOverride = redis.call('HGET', base .. ':ckLimits', string.sub(variant, #gatesKeyPrefix + 1))
       if gateOverride then
         perKeyLimit = math.min(tonumber(gateOverride), envLimit)
@@ -2571,6 +2571,7 @@ export class RunQueue {
         enableFastPathArg,
         this.options.redis.keyPrefix ?? "",
         this.options.gatesEnabled ? "1" : "0",
+        this.options.totalConcurrencyEnabled ? "1" : "0",
         metricsGaugeArg
       );
     } else {
@@ -2600,6 +2601,7 @@ export class RunQueue {
         enableFastPathArg,
         this.options.redis.keyPrefix ?? "",
         this.options.gatesEnabled ? "1" : "0",
+        this.options.totalConcurrencyEnabled ? "1" : "0",
         metricsGaugeArg
       );
     }
@@ -2681,6 +2683,7 @@ export class RunQueue {
         this.options.redis.keyPrefix ?? "",
         String(maxCount),
         this.options.gatesEnabled ? "1" : "0",
+        this.options.totalConcurrencyEnabled ? "1" : "0",
         metricsGaugeArg
       );
 
@@ -3678,6 +3681,7 @@ local currentTime = ARGV[8]
 local enableFastPath = ARGV[9]
 local keyPrefix = ARGV[10]
 local gatesEnabled = ARGV[11] == '1'
+local totalConcurrencyEnabled = ARGV[12] == '1'
 
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 ${QUEUE_GATES_LUA_HELPERS}
@@ -3705,7 +3709,7 @@ if enableFastPath == '1' then
           local okDecode, decoded = pcall(cjson.decode, messageData)
           if okDecode and type(decoded) == 'table' and decoded.gates then
             gateMsg = decoded
-            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil)
+            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled)
           end
         end
 
@@ -3792,6 +3796,7 @@ local currentTime = ARGV[10]
 local enableFastPath = ARGV[11]
 local keyPrefix = ARGV[12]
 local gatesEnabled = ARGV[13] == '1'
+local totalConcurrencyEnabled = ARGV[14] == '1'
 
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 ${QUEUE_GATES_LUA_HELPERS}
@@ -3819,7 +3824,7 @@ if enableFastPath == '1' then
           local okDecode, decoded = pcall(cjson.decode, messageData)
           if okDecode and type(decoded) == 'table' and decoded.gates then
             gateMsg = decoded
-            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil)
+            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled)
           end
         end
 
@@ -4186,7 +4191,7 @@ if enableFastPath == '1' then
           local okDecode, decoded = pcall(cjson.decode, messageData)
           if okDecode and type(decoded) == 'table' and decoded.gates then
             gateMsg = decoded
-            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil)
+            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled)
           end
         end
 
@@ -4364,7 +4369,7 @@ if enableFastPath == '1' then
           local okDecode, decoded = pcall(cjson.decode, messageData)
           if okDecode and type(decoded) == 'table' and decoded.gates then
             gateMsg = decoded
-            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil)
+            gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled)
           end
         end
 
@@ -4694,6 +4699,7 @@ local defaultEnvConcurrencyBurstFactor = ARGV[4]
 local keyPrefix = ARGV[5]
 local maxCount = tonumber(ARGV[6] or '1')
 local gatesEnabled = ARGV[7] == '1'
+local totalConcurrencyEnabled = ARGV[8] == '1'
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 ${QUEUE_GATES_LUA_HELPERS}
 ${QUEUE_METRICS_GAUGE_LUA}
@@ -4766,7 +4772,7 @@ for i = 1, #messages, 2 do
         else
             local gatesAllow = true
             if gatesEnabled then
-              gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix)
+              gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix, totalConcurrencyEnabled)
             end
 
             if gatesAllow then
@@ -5115,7 +5121,7 @@ for _, ckQueueName in ipairs(ckQueues) do
         else
           local gatesAllow = true
           if gatesEnabled then
-            gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix)
+            gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix, totalConcurrencyEnabled)
           end
           if not gatesAllow then
             blockedByGates = true
@@ -6223,6 +6229,7 @@ declare module "@internal/redis" {
       enableFastPath: string,
       keyPrefix: string,
       gatesEnabled: string,
+      totalConcurrencyEnabled: string,
       metricsEnabled: string,
       callback?: Callback<[number, number[] | null]>
     ): Result<[number, number[] | null], Context>;
@@ -6256,6 +6263,7 @@ declare module "@internal/redis" {
       enableFastPath: string,
       keyPrefix: string,
       gatesEnabled: string,
+      totalConcurrencyEnabled: string,
       metricsEnabled: string,
       callback?: Callback<[number, number[] | null]>
     ): Result<[number, number[] | null], Context>;
@@ -6294,6 +6302,7 @@ declare module "@internal/redis" {
       keyPrefix: string,
       maxCount: string,
       gatesEnabled: string,
+      totalConcurrencyEnabled: string,
       metricsEnabled: string,
       callback?: Callback<[string[] | null, number[] | null]>
     ): Result<[string[] | null, number[] | null], Context>;