fix(run-engine): reconcile prunes members whose runs are queued

A member whose message still exists but is sitting in a queue zset holds no
legitimate slot: queued and in-flight are mutually exclusive on every path.
The bounded reconcile now prunes those members too (the stored payload's
queue field is the full variant key, so one ZSCORE answers it), which frees
a saturated cap even when the self-holding run's variant lies beyond the
dequeue candidate window and can never exempt itself. Covered by a test
where the leaked member belongs to a different queue entirely.

Author: matt-aitken

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

@@ -85,16 +85,24 @@ local function __gateKeys(gatesKeyPrefix, msg, gate)
   return base, variant, gateKey
 end
 
-local function __gateReconcile(setKey, msgKeyPrefix)
+local function __gateReconcile(setKey, msgKeyPrefix, reconcileKeyPrefix)
   if not msgKeyPrefix then return end
   if redis.call('SET', setKey .. ':reconcileLock', '1', 'NX', 'EX', '10') then
     local cursorKey = setKey .. ':reconcileCursor'
     local cursor = redis.call('GET', cursorKey) or '0'
-    local scanResult = redis.call('SSCAN', setKey, cursor, 'COUNT', '500')
+    local scanResult = redis.call('SSCAN', setKey, cursor, 'COUNT', '100')
     redis.call('SET', cursorKey, scanResult[1], 'EX', '3600')
     for _, memberId in ipairs(scanResult[2]) do
-      if redis.call('EXISTS', msgKeyPrefix .. memberId) == 0 then
+      local rawMemberPayload = redis.call('GET', msgKeyPrefix .. memberId)
+      if not rawMemberPayload then
         redis.call('SREM', setKey, memberId)
+      elseif reconcileKeyPrefix then
+        local okMember, member = pcall(cjson.decode, rawMemberPayload)
+        if okMember and type(member) == 'table' and type(member.queue) == 'string' then
+          if redis.call('ZSCORE', reconcileKeyPrefix .. member.queue, memberId) then
+            redis.call('SREM', setKey, memberId)
+          end
+        end
       end
     end
   end
@@ -107,7 +115,7 @@ local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msg
     local occupancy = tonumber(redis.call('SCARD', variant .. ':currentConcurrency') or '0')
     local perKeyLimit = math.min(tonumber(redis.call('GET', base .. ':concurrency') or '1000000'), envLimit)
     if occupancy >= perKeyLimit and redis.call('SISMEMBER', variant .. ':currentConcurrency', messageId) == 0 then
-      __gateReconcile(variant .. ':currentConcurrency', msgKeyPrefix)
+      __gateReconcile(variant .. ':currentConcurrency', msgKeyPrefix, gatesKeyPrefix)
       return false
     end
     if gateKey and gateKey ~= '' then
@@ -116,7 +124,7 @@ local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msg
         local totalLimit = math.min(tonumber(rawTotal), envLimit)
         local groupKey = base .. ':groupConcurrency'
         if tonumber(redis.call('SCARD', groupKey) or '0') >= totalLimit and redis.call('SISMEMBER', groupKey, messageId) == 0 then
-          __gateReconcile(groupKey, msgKeyPrefix)
+          __gateReconcile(groupKey, msgKeyPrefix, gatesKeyPrefix)
           return false
         end
       end
@@ -4921,29 +4929,14 @@ if totalConcurrencyEnabled then
     local totalConcurrencyLimit = math.min(tonumber(rawTotalLimit), envConcurrencyLimit)
     local groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
 
-    -- Self-heal before holding the queue at its limit. A terminal release path
-    -- that misses the group mirror (an older build, or a future script) leaves
-    -- a member behind, but every terminal path deletes the run's message key,
-    -- so a member with no message key is provably dead. Members of re-queued
-    -- runs keep their message key and clear through the mirrored ack when the
-    -- run completes. The short lock bounds a saturated queue to one pass per
-    -- interval, and SSCAN with a persisted cursor bounds each pass to one
-    -- batch so a large set never blocks Redis for a full traversal; successive
-    -- passes cover the whole set.
+    -- Self-heal before holding the queue at its limit: a member with no message
+    -- key was terminally released without the mirror and is dead, and a member
+    -- whose message is QUEUED (in its variant zset) holds no legitimate slot,
+    -- since queued and in-flight are mutually exclusive on every path. Both are
+    -- pruned by the shared bounded reconcile.
     if groupCurrentConcurrency >= totalConcurrencyLimit then
-      local reconcileLockKey = groupConcurrencyKey .. ':reconcileLock'
-      if redis.call('SET', reconcileLockKey, '1', 'NX', 'EX', '10') then
-        local reconcileCursorKey = groupConcurrencyKey .. ':reconcileCursor'
-        local reconcileCursor = redis.call('GET', reconcileCursorKey) or '0'
-        local scanResult = redis.call('SSCAN', groupConcurrencyKey, reconcileCursor, 'COUNT', '500')
-        redis.call('SET', reconcileCursorKey, scanResult[1], 'EX', '3600')
-        for _, groupMemberId in ipairs(scanResult[2]) do
-          if redis.call('EXISTS', messageKeyPrefix .. groupMemberId) == 0 then
-            redis.call('SREM', groupConcurrencyKey, groupMemberId)
-          end
-        end
-        groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
-      end
+      __gateReconcile(groupConcurrencyKey, messageKeyPrefix, keyPrefix)
+      groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
     end
 
     totalHeadroom = totalConcurrencyLimit - groupCurrentConcurrency

internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts

@@ -365,6 +365,64 @@ describe("RunQueue total concurrency limit", () => {
     }
   );
 
+  redisTest(
+    "reconciles a member whose run went back to waiting in a queue",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, true);
+      try {
+        const keys = testOptions.keys;
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+        /**
+         * A run on ANOTHER queue whose message exists and is queued there, leaked
+         * into this queue's group set by an unmirrored release. It can never be a
+         * dequeue candidate here, so only the reconcile can free the slot: queued
+         * and in-flight are mutually exclusive, so a queued member holds nothing.
+         */
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "x0",
+            queue: "other-queue",
+            concurrencyKey: "ck-x",
+            timestamp: Date.now() + 3_600_000,
+          }),
+          workerQueue: "main",
+        });
+        await queue.redis.sadd(
+          keys.queueGroupConcurrencyKey(authenticatedEnvDev, "task/my-task"),
+          "x0"
+        );
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "r0",
+            concurrencyKey: "ck-a",
+            timestamp: Date.now() - 1000,
+          }),
+          workerQueue: "main",
+        });
+
+        const r0Admitted = await waitFor(async () => {
+          await queue.redis.del(
+            `${keys.queueGroupConcurrencyKey(authenticatedEnvDev, "task/my-task")}:reconcileLock`
+          );
+          const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", {
+            blockingPop: false,
+          });
+          return next?.messageId === "r0";
+        }, 30_000);
+        expect(r0Admitted).toBe(true);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
   redisTest(
     "reconciles a large leaked backlog across bounded passes",
     async ({ redisContainer }) => {