feat(run-engine): per-concurrency-key limit override storage and methods

Sparse ckLimits HASH at the base queue whose fields are the exact ck-variant
queue names, plus engine methods to set (atomic cardinality cap, default
1000 per queue), remove, and read the overrides. The admit-path gate wiring
follows.

Author: matt-aitken

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

@@ -257,6 +257,13 @@ export interface RunQueueMetricsEmitter {
   emitGauge(shardKey: string, fields: Record<string, string | number>): void;
 }
 
+export class RunQueueConcurrencyKeyLimitExceededError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "RunQueueConcurrencyKeyLimitExceededError";
+  }
+}
+
 export type RunQueueOptions = {
   name: string;
   tracer: Tracer;
@@ -316,6 +323,8 @@ 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. */
+  maxConcurrencyKeyOverridesPerQueue?: number;
   workerOptions?: {
     pollIntervalMs?: number;
     immediatePollIntervalMs?: number;
@@ -429,6 +438,7 @@ export class RunQueue {
   private queueSelectionStrategy: RunQueueSelectionStrategy;
   private shardCount: number;
   private counterTtlSeconds: number;
+  private maxConcurrencyKeyOverridesPerQueue: number;
   private abortController: AbortController;
   private worker: Worker<typeof workerCatalog>;
   private workerQueueResolver: WorkerQueueResolver;
@@ -439,6 +449,7 @@ export class RunQueue {
   constructor(public readonly options: RunQueueOptions) {
     this.shardCount = options.shardCount ?? 2;
     this.counterTtlSeconds = options.counterTtlSeconds ?? 86400;
+    this.maxConcurrencyKeyOverridesPerQueue = options.maxConcurrencyKeyOverridesPerQueue ?? 1000;
     this.retryOptions = options.retryOptions ?? defaultRetrySettings;
     this.redis = createRedisClient(options.redis, {
       onError: (error) => {
@@ -633,6 +644,62 @@ export class RunQueue {
     return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue));
   }
 
+  /**
+   * Sets a per-concurrency-key limit override for a queue. The stored value is the
+   * raw requested limit; admit paths clamp to the environment limit at read time.
+   * Throws RunQueueConcurrencyKeyLimitExceededError when a NEW key would push the
+   * queue past maxConcurrencyKeyOverridesPerQueue (updates to existing keys always
+   * succeed).
+   */
+  public async updateQueueConcurrencyKeyLimit(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string,
+    concurrencyKey: string,
+    limit: number
+  ) {
+    const result = await this.redis.setQueueConcurrencyKeyLimit(
+      this.keys.queueCkLimitsKey(env, queue),
+      this.keys.queueKey(env, queue, concurrencyKey),
+      String(limit),
+      String(this.maxConcurrencyKeyOverridesPerQueue)
+    );
+
+    if (result === 0) {
+      throw new RunQueueConcurrencyKeyLimitExceededError(
+        `Cannot add a concurrency key override to queue ${queue}: the queue already has ${this.maxConcurrencyKeyOverridesPerQueue} overrides`
+      );
+    }
+  }
+
+  public async removeQueueConcurrencyKeyLimit(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string,
+    concurrencyKey: string
+  ) {
+    return this.redis.hdel(
+      this.keys.queueCkLimitsKey(env, queue),
+      this.keys.queueKey(env, queue, concurrencyKey)
+    );
+  }
+
+  /** Returns the raw per-concurrency-key limit overrides for a queue, keyed by concurrency key value. */
+  public async getQueueConcurrencyKeyLimits(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string
+  ): Promise<Record<string, number>> {
+    const raw = await this.redis.hgetall(this.keys.queueCkLimitsKey(env, queue));
+
+    const limits: Record<string, number> = {};
+    for (const [variantName, value] of Object.entries(raw)) {
+      const ckIndex = variantName.indexOf(":ck:");
+      if (ckIndex === -1) {
+        continue;
+      }
+      limits[variantName.slice(ckIndex + 4)] = Number(value);
+    }
+    return limits;
+  }
+
   public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) {
     await this.#callUpdateEnvironmentConcurrencyLimits({
       envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env),
@@ -5895,6 +5962,26 @@ __gatesRelease(keyPrefix, redis.call('GET', messageKey), messageId)
 `,
     });
 
+    this.redis.defineCommand("setQueueConcurrencyKeyLimit", {
+      numberOfKeys: 1,
+      lua: `
+local ckLimitsKey = KEYS[1]
+
+local fieldName = ARGV[1]
+local limit = ARGV[2]
+local maxFields = tonumber(ARGV[3])
+
+if redis.call('HEXISTS', ckLimitsKey, fieldName) == 0 then
+  if redis.call('HLEN', ckLimitsKey) >= maxFields then
+    return 0
+  end
+end
+
+redis.call('HSET', ckLimitsKey, fieldName, limit)
+return 1
+`,
+    });
+
     this.redis.defineCommand("updateEnvironmentConcurrencyLimits", {
       numberOfKeys: 2,
       lua: `
@@ -6256,6 +6343,14 @@ declare module "@internal/redis" {
       callback?: Callback<void>
     ): Result<void, Context>;
 
+    setQueueConcurrencyKeyLimit(
+      ckLimitsKey: string,
+      fieldName: string,
+      limit: string,
+      maxFields: string,
+      callback?: Callback<number>
+    ): Result<number, Context>;
+
     updateEnvironmentConcurrencyLimits(
       // keys
       envConcurrencyLimitKey: string,

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

@@ -26,6 +26,7 @@ const constants = {
   RUNNING_COUNTER_PART: "runningCounter",
   GROUP_CONCURRENCY_PART: "groupConcurrency",
   TOTAL_CONCURRENCY_LIMIT_PART: "totalConcurrency",
+  CK_LIMITS_PART: "ckLimits",
 } as const;
 
 export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
@@ -366,6 +367,20 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
     return `${this.baseQueueKeyFromQueue(queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`;
   }
 
+  /**
+   * HASH of per-concurrency-key limit overrides for a queue. Lives at the base
+   * queue; each field is the EXACT full ck-variant queue name (the ckIndex ZSET
+   * member), so reads need no parsing, and values are the raw requested limits
+   * (readers clamp to the environment limit).
+   */
+  queueCkLimitsKey(env: RunQueueKeyProducerEnvironment, queue: string): string {
+    return `${this.queueKey(env, queue)}:${constants.CK_LIMITS_PART}`;
+  }
+
+  queueCkLimitsKeyFromQueue(queue: string): string {
+    return `${this.baseQueueKeyFromQueue(queue)}:${constants.CK_LIMITS_PART}`;
+  }
+
   isCkWildcard(queue: string): boolean {
     return queue.endsWith(":ck:*");
   }

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

@@ -110,6 +110,8 @@ export interface RunQueueKeyProducer {
   queueGroupConcurrencyKeyFromQueue(queue: string): string;
   queueTotalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string;
   queueTotalConcurrencyLimitKeyFromQueue(queue: string): string;
+  queueCkLimitsKey(env: RunQueueKeyProducerEnvironment, queue: string): string;
+  queueCkLimitsKeyFromQueue(queue: string): string;
 
   //env oncurrency
   envCurrentConcurrencyKey(env: EnvDescriptor): string;