refactor(run-engine,webapp,sdk,database): remove per-key concurrency limit overrides

Per-key runtime overrides are deferred: composing a per-key hash with pause
semantics and keeping it convergent with durable rows across deploys and
resets needs its own design. Combined-limit overrides stay. Declared per-key
behavior is unchanged (concurrencyLimit applies per key as before).

Author: matt-aitken

.changeset/queue-concurrency-overrides.md

@@ -3,16 +3,13 @@
 "@trigger.dev/core": patch
 ---
 
-Adjust queue concurrency at runtime, per key and combined. `queues.overrideConcurrencyLimit` accepts a `concurrencyKey` to raise or lower one key's limit without touching the rest of the queue, and the new `queues.overrideCombinedConcurrencyLimit` and `queues.resetCombinedConcurrencyLimit` adjust the cap across all keys.
+Adjust a queue's combined concurrency limit at runtime. `queues.overrideCombinedConcurrencyLimit` raises or lowers the cap on concurrent runs across all of a queue's `concurrencyKey` values, and `queues.resetCombinedConcurrencyLimit` reverts to the declared configuration.
 
 ```ts
 import { queues } from "@trigger.dev/sdk";
 
-await queues.overrideConcurrencyLimit("my-queue", 20, { concurrencyKey: "tenant-123" });
-await queues.resetConcurrencyLimit("my-queue", { concurrencyKey: "tenant-123" });
-
 await queues.overrideCombinedConcurrencyLimit("my-queue", 100);
 await queues.resetCombinedConcurrencyLimit("my-queue");
 ```
 
-Overrides survive deploys and reset back to the declared configuration. Enforcement happens server-side on servers with combined concurrency limits enabled.
+Overrides survive deploys. Enforcement happens server-side on servers with combined concurrency limits enabled.

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

@@ -9,7 +9,6 @@ import {
   updateQueueConcurrencyLimits,
   updateQueueTotalConcurrencyLimits,
 } from "../runQueue.server";
-import { RunQueueConcurrencyKeyLimitExceededError } from "@internal/run-engine";
 import { engine } from "../runEngine.server";
 
 export type ConcurrencySystemOptions = {
@@ -109,37 +108,6 @@ export class ConcurrencySystem {
           .andThen((queue) => syncQueueTotalConcurrencyToEngine(environment, queue))
           .andThen((queue) => getQueueStats(environment, queue));
       },
-      overrideConcurrencyKeyLimit: (
-        environment: AuthenticatedEnvironment,
-        queue: QueueInput,
-        concurrencyKey: string,
-        concurrencyLimit: number,
-        overriddenBy?: User
-      ) => {
-        return findQueueFromInput(this.db, environment, queue)
-          .andThen((queue) =>
-            overrideQueueConcurrencyKeyLimit(
-              this.db,
-              environment,
-              queue,
-              concurrencyKey,
-              concurrencyLimit,
-              overriddenBy
-            )
-          )
-          .andThen((queue) => getQueueStats(environment, queue));
-      },
-      resetConcurrencyKeyLimit: (
-        environment: AuthenticatedEnvironment,
-        queue: QueueInput,
-        concurrencyKey: string
-      ) => {
-        return findQueueFromInput(this.db, environment, queue)
-          .andThen((queue) =>
-            resetQueueConcurrencyKeyLimit(this.db, environment, queue, concurrencyKey)
-          )
-          .andThen((queue) => getQueueStats(environment, queue));
-      },
       /**
        * Recalculates the materialized limit of every percent-based override in the environment
        * against its CURRENT maximumConcurrencyLimit and syncs changed queues to the run engine.
@@ -498,153 +466,6 @@ function syncQueueTotalConcurrencyToEngine(
   })).andThen(() => okAsync(queue));
 }
 
-/**
- * Persist first, then enforce: a database failure leaves nothing enforced and the
- * request errors cleanly, while an engine failure after persistence leaves a durable
- * record and a retry converges (the upsert is idempotent). A cap rejection deletes
- * the row unconditionally: the cap only rejects keys absent from the engine hash
- * (updates to present keys always succeed), so a cap-rejected row is never backing
- * an enforced limit and must not survive as an authoritative override.
- */
-function overrideQueueConcurrencyKeyLimit(
-  db: PrismaClientOrTransaction,
-  environment: AuthenticatedEnvironment,
-  queue: TaskQueue,
-  concurrencyKey: string,
-  concurrencyLimit: number,
-  overriddenBy?: User
-) {
-  const maximum = environment.maximumConcurrencyLimit;
-
-  if (!Number.isFinite(concurrencyLimit) || concurrencyLimit < 0) {
-    return errAsync({
-      type: "invalid_override" as const,
-      message: "Concurrency limit must be a non-negative number",
-    });
-  }
-
-  if (concurrencyLimit > maximum) {
-    return errAsync({
-      type: "concurrency_limit_exceeds_maximum" as const,
-      message: `Concurrency limit (${concurrencyLimit}) cannot exceed the environment limit (${maximum})`,
-    });
-  }
-
-  return fromPromise(
-    db.taskQueueConcurrencyKeyOverride.findFirst({
-      where: { taskQueueId: queue.id, concurrencyKey },
-      select: { id: true },
-    }),
-    (error) => ({ type: "other" as const, cause: error })
-  )
-    .andThen(() =>
-      fromPromise(
-        db.taskQueueConcurrencyKeyOverride.upsert({
-          where: { taskQueueId_concurrencyKey: { taskQueueId: queue.id, concurrencyKey } },
-          create: {
-            taskQueueId: queue.id,
-            concurrencyKey,
-            concurrencyLimit,
-            overriddenBy: overriddenBy?.id ?? null,
-          },
-          update: {
-            concurrencyLimit,
-            overriddenAt: new Date(),
-            overriddenBy: overriddenBy?.id ?? null,
-          },
-        }),
-        (error) => ({
-          type: "queue_update_failed" as const,
-          cause: error,
-        })
-      )
-    )
-    .andThen((written) =>
-      fromPromise(
-        engine.runQueue.updateQueueConcurrencyKeyLimit(
-          environment,
-          queue.name,
-          concurrencyKey,
-          concurrencyLimit
-        ),
-        (error) => {
-          if (error instanceof RunQueueConcurrencyKeyLimitExceededError) {
-            return { type: "too_many_key_overrides" as const, message: error.message };
-          }
-          return { type: "sync_queue_concurrency_to_engine_failed" as const, cause: error };
-        }
-      ).orElse((error) => {
-        if (error.type === "too_many_key_overrides") {
-          return fromPromise(
-            /**
-             * Deletes only the exact row generation this request wrote, so a
-             * concurrent request that succeeded after capacity freed keeps its
-             * durable record.
-             */
-            db.taskQueueConcurrencyKeyOverride.deleteMany({
-              where: { id: written.id, overriddenAt: written.overriddenAt },
-            }),
-            () => error
-          ).andThen(() => errAsync(error));
-        }
-        return errAsync(error);
-      })
-    )
-    .andThen(() => okAsync(queue));
-}
-
-/**
- * Enforce first, then persist: removing the engine limit is idempotent, so an engine
- * failure leaves the override row in place and a retry converges instead of being
- * rejected as not overridden while the old limit is still enforced.
- */
-function resetQueueConcurrencyKeyLimit(
-  db: PrismaClientOrTransaction,
-  environment: AuthenticatedEnvironment,
-  queue: TaskQueue,
-  concurrencyKey: string
-) {
-  return fromPromise(
-    db.taskQueueConcurrencyKeyOverride.findFirst({
-      where: { taskQueueId: queue.id, concurrencyKey },
-      select: { id: true, overriddenAt: true },
-    }),
-    (error) => ({ type: "other" as const, cause: error })
-  ).andThen((existing) => {
-    if (!existing) {
-      return errAsync({ type: "queue_not_overridden" as const });
-    }
-
-    return fromPromise(
-      engine.runQueue.removeQueueConcurrencyKeyLimit(environment, queue.name, concurrencyKey),
-      (error) => ({
-        type: "sync_queue_concurrency_to_engine_failed" as const,
-        cause: error,
-      })
-    )
-      .andThen(() =>
-        fromPromise(
-          /**
-           * Deletes only the exact row generation this reset read, so a concurrent
-           * override that re-wrote the row after the reset began keeps its record
-           * (its next write, or the deploy-time restore, re-syncs the engine).
-           */
-          db.taskQueueConcurrencyKeyOverride.deleteMany({
-            where: {
-              id: existing.id,
-              overriddenAt: existing.overriddenAt,
-            },
-          }),
-          (error) => ({
-            type: "queue_update_failed" as const,
-            cause: error,
-          })
-        )
-      )
-      .andThen(() => okAsync(queue));
-  });
-}
-
 function getQueueStats(environment: AuthenticatedEnvironment, queue: TaskQueue) {
   return fromPromise(
     Promise.all([

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

@@ -582,31 +582,6 @@ async function createWorkerQueue(
     await removeQueueTotalConcurrencyLimits(environment, taskQueue.name);
   }
 
-  /**
-   * Restore per-key limit overrides into the engine so a fresh or flushed Redis
-   * converges back to the durable records on the next deploy. Row-level failures
-   * are logged rather than failing the deploy; the next deploy retries them.
-   */
-  const keyOverrides = await prisma.taskQueueConcurrencyKeyOverride.findMany({
-    where: { taskQueueId: taskQueue.id },
-  });
-  for (const keyOverride of keyOverrides) {
-    try {
-      await engine.runQueue.updateQueueConcurrencyKeyLimit(
-        environment,
-        taskQueue.name,
-        keyOverride.concurrencyKey,
-        keyOverride.concurrencyLimit
-      );
-    } catch (error) {
-      logger.error("createWorkerQueue: failed to restore concurrency key override", {
-        taskQueueId: taskQueue.id,
-        concurrencyKey: keyOverride.concurrencyKey,
-        error,
-      });
-    }
-  }
-
   if (!taskQueue.paused) {
     if (typeof newConcurrencyLimit === "number") {
       logger.debug("createWorkerQueue: updating concurrency limit", {

internal-packages/database/prisma/migrations/20260829150000_add_concurrency_overrides/migration.sql

@@ -2,23 +2,3 @@
 ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimitOverriddenAt" TIMESTAMP(3);
 ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimitOverriddenBy" TEXT;
 ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimitBase" INTEGER;
-
--- CreateTable
-CREATE TABLE "TaskQueueConcurrencyKeyOverride" (
-    "id" TEXT NOT NULL,
-    "taskQueueId" TEXT NOT NULL,
-    "concurrencyKey" TEXT NOT NULL,
-    "concurrencyLimit" INTEGER NOT NULL,
-    "overriddenAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "overriddenBy" TEXT,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "updatedAt" TIMESTAMP(3) NOT NULL,
-
-    CONSTRAINT "TaskQueueConcurrencyKeyOverride_pkey" PRIMARY KEY ("id")
-);
-
--- CreateIndex
-CREATE UNIQUE INDEX "TaskQueueConcurrencyKeyOverride_taskQueueId_concurrencyKey_key" ON "TaskQueueConcurrencyKeyOverride"("taskQueueId", "concurrencyKey");
-
--- AddForeignKey
-ALTER TABLE "TaskQueueConcurrencyKeyOverride" ADD CONSTRAINT "TaskQueueConcurrencyKeyOverride_taskQueueId_fkey" FOREIGN KEY ("taskQueueId") REFERENCES "TaskQueue"("id") ON DELETE CASCADE ON UPDATE CASCADE;

internal-packages/database/prisma/schema.prisma

@@ -2001,30 +2001,10 @@ model TaskQueue {
   tasks    BackgroundWorkerTask[]
   workers  BackgroundWorker[]
 
-  concurrencyKeyOverrides TaskQueueConcurrencyKeyOverride[]
 
   @@unique([runtimeEnvironmentId, name])
 }
 
-/// A per-concurrency-key limit override for a queue: the named key value gets this
-/// limit instead of the queue's concurrencyLimit. Deleting the row resets the key.
-model TaskQueueConcurrencyKeyOverride {
-  id String @id @default(cuid())
-
-  taskQueue   TaskQueue @relation(fields: [taskQueueId], references: [id], onDelete: Cascade, onUpdate: Cascade)
-  taskQueueId String
-
-  concurrencyKey   String
-  concurrencyLimit Int
-
-  overriddenAt DateTime  @default(now())
-  overriddenBy String?
-
-  createdAt DateTime @default(now())
-  updatedAt DateTime @updatedAt
-
-  @@unique([taskQueueId, concurrencyKey])
-}
 
 enum TaskQueueType {
   VIRTUAL

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

@@ -1,5 +1,4 @@
 export { RunEngine } from "./engine/index.js";
-export { RunQueueConcurrencyKeyLimitExceededError } from "./run-queue/index.js";
 export {
   RunDuplicateIdempotencyKeyError,
   RunOneTimeUseTokenError,

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

@@ -26,7 +26,6 @@ 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 {
@@ -367,20 +366,6 @@ 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:*");
   }