perf(run-engine): bound group-set reconciliation to one SSCAN batch per pass

Reconciling a saturated queue's group set with SMEMBERS plus one EXISTS per
member runs the whole traversal inside a single Lua call, which blocks Redis
for the duration on a large set. Scan one bounded batch per pass instead,
persisting the SSCAN cursor between passes so successive intervals cover the
whole set. Covered by a test that drains a 1,200-member leaked backlog.

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

@@ -4705,12 +4705,17 @@ if totalConcurrencyEnabled then
     -- 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 instead of one per dequeue attempt.
+    -- 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.
     if groupCurrentConcurrency >= totalConcurrencyLimit then
       local reconcileLockKey = groupConcurrencyKey .. ':reconcileLock'
       if redis.call('SET', reconcileLockKey, '1', 'NX', 'EX', '10') then
-        local groupMembers = redis.call('SMEMBERS', groupConcurrencyKey)
-        for _, groupMemberId in ipairs(groupMembers) do
+        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

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

@@ -364,4 +364,49 @@ describe("RunQueue total concurrency limit", () => {
       }
     }
   );
+
+  redisTest(
+    "reconciles a large leaked backlog across bounded passes",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, true);
+      try {
+        const keys = testOptions.keys;
+        const groupKey = keys.queueGroupConcurrencyKey(authenticatedEnvDev, "task/my-task");
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+        /** 1,200 dead members: more than one SSCAN batch, none with a message key. */
+        const dead = Array.from({ length: 1200 }, (_, i) => `dead-${i}`);
+        await queue.redis.sadd(groupKey, ...dead);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1200);
+
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "r0",
+            concurrencyKey: "ck-a",
+            timestamp: Date.now() - 1000,
+          }),
+          workerQueue: "main",
+        });
+
+        /**
+         * Each dequeue attempt reconciles at most one SSCAN batch behind a 10s
+         * lock; dropping the lock between polls lets the passes run back to
+         * back instead of waiting out the interval.
+         */
+        const r0Admitted = await waitFor(async () => {
+          await queue.redis.del(`${groupKey}: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();
+      }
+    }
+  );
 });