fix(run-engine,webapp,clickhouse): review fixes for the metrics tier

The plain dequeue gauge re-samples after admissions like the keyed one, the
repair path clears a keyed run's variant and group slots by concurrency key,
pause responses include the combined limit, stale wording and a leftover
changeset from before the rename are cleaned up, and two empty flag blocks
are removed from the fast-path scripts.

Author: matt-aitken

apps/webapp/app/v3/querySchemas.ts

@@ -1426,7 +1426,7 @@ const queueMetricsByKeySchema: TableSchema = {
       name: "max_limit",
       ...column("UInt32", {
         description:
-          "The queue concurrency limit that applied to this key in the bucket. Aggregate with max().",
+          "The queue concurrency limit that applied to this key in the bucket (1000000 = no explicit limit). Aggregate with max().",
         fillMode: "carry",
       }),
     },

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

@@ -92,6 +92,9 @@ export class PauseQueueService extends BaseService {
           concurrencyLimitOverriddenAt: updatedQueue.concurrencyLimitOverriddenAt ?? null,
           concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy ?? null,
           paused: updatedQueue.paused,
+          totalConcurrencyLimit: updatedQueue.totalConcurrencyLimit ?? null,
+          totalConcurrencyLimitBase: updatedQueue.totalConcurrencyLimitBase ?? null,
+          totalConcurrencyLimitOverriddenAt: updatedQueue.totalConcurrencyLimitOverriddenAt ?? null,
         }),
       };
     } catch (error) {

internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql

@@ -3,8 +3,9 @@
 -- Total-concurrency gauges: combined_running is the in-flight count across ALL
 -- concurrency-key variants of a queue (the groupConcurrency set), combined_limit the
 -- RAW stored total cap (0 = none, readers clamp against max_env_limit). Emitted on
--- base-queue gauge rows only. Per-key gauge rows now carry the EFFECTIVE per-key
--- limit in queue_limit, surfaced in the ck tier as max_limit.
+-- base-queue gauge rows only. Per-key gauge rows carry the queue concurrency
+-- limit that applied in queue_limit, surfaced in the ck tier as max_limit
+-- (1000000 = no explicit limit).
 
 ALTER TABLE trigger_dev.queue_metrics_raw_v1
   ADD COLUMN IF NOT EXISTS combined_running UInt32 DEFAULT 0,

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

@@ -3001,6 +3001,7 @@ export class RunEngine {
             {
               select: {
                 queue: true,
+                concurrencyKey: true,
               },
             },
             this.prisma
@@ -3022,6 +3023,7 @@ export class RunEngine {
             runId,
             orgId: latestSnapshot.organizationId,
             queue: taskRun.queue,
+            concurrencyKey: taskRun.concurrencyKey ?? undefined,
             env: {
               id: latestSnapshot.environmentId,
               type: latestSnapshot.environmentType,

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

@@ -349,7 +349,6 @@ export type RunQueueOptions = {
    * the total cap covering releases from builds without the mirror.
    */
   gatesEnabled?: boolean;
-  /** Cap on per-concurrency-key limit overrides stored per queue. Default 1000. */
   workerOptions?: {
     pollIntervalMs?: number;
     immediatePollIntervalMs?: number;
@@ -1567,6 +1566,7 @@ export class RunQueue {
     runId: string;
     orgId: string;
     queue: string;
+    concurrencyKey?: string;
     env: RunQueueKeyProducerEnvironment;
   }) {
     return this.#callClearMessageFromConcurrencySets(params);
@@ -3108,18 +3108,30 @@ export class RunQueue {
     runId,
     orgId,
     queue,
+    concurrencyKey,
     env,
   }: {
     runId: string;
     orgId: string;
     queue: string;
+    concurrencyKey?: string;
     env: RunQueueKeyProducerEnvironment;
   }) {
     const messageId = runId;
     const messageKey = this.keys.messageKey(orgId, messageId);
-    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKey(env, queue);
+    /**
+     * Callers pass the bare TaskRun queue name plus its concurrencyKey; the run's
+     * slots live on the ck variant, and the tracked clear additionally mirrors the
+     * group set and counters that only keyed queues maintain.
+     */
+    const fullQueue = concurrencyKey ? this.keys.queueKey(env, queue, concurrencyKey) : queue;
+    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKey(
+      env,
+      queue,
+      concurrencyKey
+    );
     const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKey(env);
-    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKey(env, queue);
+    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKey(env, queue, concurrencyKey);
     const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKey(env);
 
     this.logger.debug("Calling clearMessageFromConcurrencySets", {
@@ -3134,15 +3146,15 @@ export class RunQueue {
       service: this.name,
     });
 
-    if (queue.includes(":ck:")) {
+    if (fullQueue.includes(":ck:")) {
       return this.redis.clearMessageFromConcurrencySetsTracked(
         queueCurrentConcurrencyKey,
         envCurrentConcurrencyKey,
         queueCurrentDequeuedKey,
         envCurrentDequeuedKey,
-        this.keys.queueRunningCounterKeyFromQueue(queue),
-        this.keys.ckIndexKeyFromQueue(queue),
-        this.keys.queueGroupConcurrencyKeyFromQueue(queue),
+        this.keys.queueRunningCounterKeyFromQueue(fullQueue),
+        this.keys.ckIndexKeyFromQueue(fullQueue),
+        this.keys.queueGroupConcurrencyKeyFromQueue(fullQueue),
         messageKey,
         messageId,
         this.options.redis.keyPrefix ?? "",
@@ -4229,8 +4241,6 @@ if enableFastPath == '1' then
         tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
         envLimit
       )
-      if totalConcurrencyEnabled then
-      end
 
       if queueCurrent < queueLimit then
         -- Total-cap gate: a fast-path admit consumes a group slot, so it must
@@ -4404,8 +4414,6 @@ if enableFastPath == '1' then
         tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
         envLimit
       )
-      if totalConcurrencyEnabled then
-      end
 
       if queueCurrent < queueLimit then
         -- Total-cap gate: see enqueueMessageCkTracked.
@@ -4900,6 +4908,10 @@ else
   redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
 end
 
+-- Re-sample the gauge so the emitted snapshot includes this batch's admissions;
+-- the top-of-script sample only covers the early returns where nothing was admitted.
+${QUEUE_METRICS_GAUGE_LUA}
+
 -- Return results as a flat array: [messageId1, messageScore1, messagePayload1, messageId2, messageScore2, messagePayload2, ...]
 return __qmret(results)
       `,