fix(run-engine): address review on total concurrency limit

Strengthen the nack test so it proves the group slot is released (a second
key's run must be admitted after the nack), document the enable-time
convergence window on the flag, and correct the totalConcurrencyOfQueue doc
to describe the drain-on-disable behavior.

Author: matt-aitken

.changeset/queue-total-concurrency-limit.md

@@ -15,4 +15,4 @@ export const perUserQueue = queue({
 });
 ```
 
-Enforcement happens server side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet.
+Enforcement happens server-side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet.

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

@@ -185,6 +185,13 @@ export type RunQueueOptions = {
    * limit at admit time. Default false: admit paths are byte-identical to before, and
    * only the release-side SREM mirror runs (a no-op on an absent set), so the flag can
    * be flipped on a fleet that has fully rolled onto this build without draining queues.
+   *
+   * Runs already in flight when the flag turns on are not in the set, so a queue can
+   * transiently exceed its total limit by the number of those runs. The excess is
+   * one-time and self-corrects as each pre-flag run completes (its release mirror
+   * no-ops), after which the limit is enforced exactly. Every pod must run this build
+   * before enabling: an older pod releases without the mirror and would leak members
+   * the gate then counts forever.
    */
   totalConcurrencyEnabled?: boolean;
   workerOptions?: {
@@ -495,8 +502,10 @@ export class RunQueue {
 
   /**
    * Total in-flight runs across all concurrency-key variants of a queue (the
-   * groupConcurrency SET cardinality). Only maintained while totalConcurrencyEnabled
-   * is on; returns 0 otherwise.
+   * groupConcurrency SET cardinality). Admits only populate the set while
+   * totalConcurrencyEnabled is on. After the flag is turned off the set drains
+   * to zero through the release-side mirrors, so a nonzero read reflects real
+   * runs admitted while it was on, never stale state.
    */
   public async totalConcurrencyOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) {
     return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue));

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

@@ -259,24 +259,38 @@ describe("RunQueue total concurrency limit", () => {
       );
       expect(admitted).toBe(true);
 
+      /** A second run on another key waits behind the total limit of 1. */
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r1", concurrencyKey: "ck-b", timestamp: Date.now() - 500 }),
+        workerQueue: "main",
+      });
+
+      await setTimeout(2000);
+      expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+
       const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main");
       assertNonNullable(dequeued);
+      expect(dequeued.messageId).toBe("r0");
 
+      /**
+       * Nack r0 with a far-future retryAt so it cannot immediately reclaim the
+       * slot. If the nack released r0's group slot, r1 is the only eligible run
+       * and must be admitted; if the slot leaked, the queue stays blocked and r1
+       * never surfaces.
+       */
       await queue.nackMessage({
         orgId: authenticatedEnvDev.organization.id,
-        messageId: dequeued.messageId,
+        messageId: "r0",
+        retryAt: Date.now() + 120_000,
       });
 
-      /**
-       * The nack returns the run to the queue and must release its group slot; the
-       * master consumer then re-admits it, taking the slot again. Assert the
-       * intermediate release by checking the run is never counted twice.
-       */
-      const settled = await waitFor(async () => {
-        const total = await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task");
-        return total <= 1;
+      const r1Admitted = await waitFor(async () => {
+        const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main");
+        return next?.messageId === "r1";
       });
-      expect(settled).toBe(true);
+      expect(r1Admitted).toBe(true);
+      expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
     } finally {
       await queue.quit();
     }