@@ -104,6 +104,7 @@ export class ConcurrencySystem {
       },
       resetTotalConcurrencyLimit: (environment: AuthenticatedEnvironment, queue: QueueInput) => {
         return findQueueFromInput(this.db, environment, queue)
+          .andThen((queue) => syncQueueTotalConcurrencyResetToEngine(environment, queue))
           .andThen((queue) => resetQueueTotalConcurrencyLimit(this.db, queue))
           .andThen((queue) => syncQueueTotalConcurrencyToEngine(environment, queue))
           .andThen((queue) => getQueueStats(environment, queue));
@@ -134,7 +135,9 @@ export class ConcurrencySystem {
         concurrencyKey: string
       ) => {
         return findQueueFromInput(this.db, environment, queue)
-          .andThen((queue) => resetQueueConcurrencyKeyLimit(this.db, environment, queue, concurrencyKey))
+          .andThen((queue) =>
+            resetQueueConcurrencyKeyLimit(this.db, environment, queue, concurrencyKey)
+          )
           .andThen((queue) => getQueueStats(environment, queue));
       },
       /**
@@ -420,6 +423,35 @@ function overrideQueueTotalConcurrencyLimit(
   );
 }
 
+/**
+ * Enforce first, then persist: syncs the engine to the declared base BEFORE clearing
+ * the override marker, so an engine failure leaves the marker set and a retry
+ * converges instead of being rejected while the overridden limit stays enforced.
+ */
+function syncQueueTotalConcurrencyResetToEngine(
+  environment: AuthenticatedEnvironment,
+  queue: TaskQueue
+) {
+  if (queue.totalConcurrencyLimitOverriddenAt === null) {
+    return errAsync({ type: "queue_not_overridden" as const });
+  }
+
+  if (typeof queue.totalConcurrencyLimitBase === "number") {
+    return fromPromise(
+      updateQueueTotalConcurrencyLimits(environment, queue.name, queue.totalConcurrencyLimitBase),
+      (error) => ({
+        type: "sync_queue_concurrency_to_engine_failed" as const,
+        cause: error,
+      })
+    ).andThen(() => okAsync(queue));
+  }
+
+  return fromPromise(removeQueueTotalConcurrencyLimits(environment, queue.name), (error) => ({
+    type: "sync_queue_concurrency_to_engine_failed" as const,
+    cause: error,
+  })).andThen(() => okAsync(queue));
+}
+
 function resetQueueTotalConcurrencyLimit(db: PrismaClientOrTransaction, queue: TaskQueue) {
   if (queue.totalConcurrencyLimitOverriddenAt === null) {
     return errAsync({ type: "queue_not_overridden" as const });
@@ -466,6 +498,13 @@ 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). When the engine rejects a
+ * NEW key for exceeding the per-queue override cap, the just-created row is removed
+ * again so the record never claims an override the engine refused.
+ */
 function overrideQueueConcurrencyKeyLimit(
   db: PrismaClientOrTransaction,
   environment: AuthenticatedEnvironment,
@@ -491,20 +530,13 @@ function overrideQueueConcurrencyKeyLimit(
   }
 
   return 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 };
-    }
+    db.taskQueueConcurrencyKeyOverride.findFirst({
+      where: { taskQueueId: queue.id, concurrencyKey },
+      select: { id: true },
+    }),
+    (error) => ({ type: "other" as const, cause: error })
   )
-    .andThen(() =>
+    .andThen((existing) =>
       fromPromise(
         db.taskQueueConcurrencyKeyOverride.upsert({
           where: { taskQueueId_concurrencyKey: { taskQueueId: queue.id, concurrencyKey } },
@@ -524,27 +556,56 @@ function overrideQueueConcurrencyKeyLimit(
           type: "queue_update_failed" as const,
           cause: error,
         })
-      )
+      ).map(() => existing)
+    )
+    .andThen((existing) =>
+      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 (!existing && error.type === "too_many_key_overrides") {
+          return fromPromise(
+            db.taskQueueConcurrencyKeyOverride.deleteMany({
+              where: { taskQueueId: queue.id, concurrencyKey },
+            }),
+            () => 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.deleteMany({
+    db.taskQueueConcurrencyKeyOverride.findFirst({
       where: { taskQueueId: queue.id, concurrencyKey },
+      select: { id: true },
     }),
-    (error) => ({
-      type: "queue_update_failed" as const,
-      cause: error,
-    })
-  ).andThen((deleted) => {
-    if (deleted.count === 0) {
+    (error) => ({ type: "other" as const, cause: error })
+  ).andThen((existing) => {
+    if (!existing) {
       return errAsync({ type: "queue_not_overridden" as const });
     }
 
@@ -554,7 +615,19 @@ function resetQueueConcurrencyKeyLimit(
         type: "sync_queue_concurrency_to_engine_failed" as const,
         cause: error,
       })
-    ).andThen(() => okAsync(queue));
+    )
+      .andThen(() =>
+        fromPromise(
+          db.taskQueueConcurrencyKeyOverride.deleteMany({
+            where: { taskQueueId: queue.id, concurrencyKey },
+          }),
+          (error) => ({
+            type: "queue_update_failed" as const,
+            cause: error,
+          })
+        )
+      )
+      .andThen(() => okAsync(queue));
   });
 }
 
 
@@ -582,6 +582,31 @@ 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", {
 
@@ -304,6 +304,10 @@ export type RunQueueOptions = {
    * exists are pruned, so such leaks clear within seconds instead of blocking the
    * queue. Enabling only after every instance runs this build avoids the noise but is
    * no longer load-bearing for correctness.
+   *
+   * Per-concurrency-key limit overrides are part of the same concurrency-limits
+   * feature and are deliberately enforced behind this flag too: writes are always
+   * accepted and durable, and enforcement of both arrives together.
    */
   totalConcurrencyEnabled?: boolean;
   /**
@@ -5038,6 +5042,7 @@ for _, ckQueueName in ipairs(ckQueues) do
   end
 
   local fullQueueKey = keyPrefix .. ckQueueName
+  local blockedByGates = false
 
   local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency'
   local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0')
@@ -5050,6 +5055,14 @@ for _, ckQueueName in ipairs(ckQueues) do
     end
   end
 
+  if ckCurrentConcurrency >= perKeyLimit then
+    -- Back a blocked variant off so it cannot pin the bounded candidate window
+    -- and starve later keys (acute with a zero per-key override, which never
+    -- self-clears). Acks and nacks rebalance the score back to the oldest
+    -- message, so the key is eligible again the moment capacity frees.
+    redis.call('ZADD', ckIndexKey, currentTime + 1000, ckQueueName)
+  end
+
   if ckCurrentConcurrency < perKeyLimit then
     local messages = redis.call('ZRANGEBYSCORE', fullQueueKey, '-inf', tostring(currentTime), 'WITHSCORES', 'LIMIT', 0, 1)
 
@@ -5077,6 +5090,9 @@ for _, ckQueueName in ipairs(ckQueues) do
           if gatesEnabled then
             gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix)
           end
+          if not gatesAllow then
+            blockedByGates = true
+          end
 
           local alreadyInGroup = false
           local totalAllows = true
@@ -5119,11 +5135,15 @@ for _, ckQueueName in ipairs(ckQueues) do
         decrLengthCounter()
       end
 
-      local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
-      if #earliest == 0 then
-        redis.call('ZREM', ckIndexKey, ckQueueName)
+      if blockedByGates then
+        redis.call('ZADD', ckIndexKey, currentTime + 1000, ckQueueName)
       else
-        redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName)
+        local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
+        if #earliest == 0 then
+          redis.call('ZREM', ckIndexKey, ckQueueName)
+        else
+          redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName)
+        end
       end
     else
       local any = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
 
@@ -32,11 +32,17 @@ const authenticatedEnvDev = {
   organization: { id: "o1234" },
 };
 
-function createQueue(redisContainer: any, totalConcurrencyEnabled: boolean, maxOverrides?: number) {
+function createQueue(
+  redisContainer: any,
+  totalConcurrencyEnabled: boolean,
+  maxOverrides?: number,
+  dequeueCount?: number
+) {
   return new RunQueue({
     ...testOptions,
     totalConcurrencyEnabled,
     maxConcurrencyKeyOverridesPerQueue: maxOverrides,
+    masterQueueConsumerDequeueCount: dequeueCount,
     queueSelectionStrategy: new FairQueueSelectionStrategy({
       redis: {
         keyPrefix: "runqueue:test:",
@@ -239,6 +245,61 @@ describe("RunQueue per-concurrency-key limit overrides", () => {
     }
   });
 
+  redisTest(
+    "blocked keys cannot pin the candidate window and starve later keys",
+    async ({ redisContainer }) => {
+      /** dequeueCount 2 makes the candidate window 6 variants wide. */
+      const queue = createQueue(redisContainer, true, undefined, 2);
+      try {
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+
+        /**
+         * Ten zero-limit keys with OLDER messages fill the window many times over;
+         * without the blocked-key backoff the runnable key behind them would never
+         * be examined.
+         */
+        const now = Date.now();
+        for (let i = 0; i < 10; i++) {
+          const ck = `blocked-${i}`;
+          await queue.updateQueueConcurrencyKeyLimit(authenticatedEnvDev, "task/my-task", ck, 0);
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({
+              runId: `b${i}`,
+              concurrencyKey: ck,
+              timestamp: now - 10_000 + i,
+            }),
+            workerQueue: "main",
+          });
+        }
+
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "good-0",
+            concurrencyKey: "ck-good",
+            timestamp: now - 500,
+          }),
+          workerQueue: "main",
+        });
+
+        const goodAdmitted = await waitFor(
+          async () =>
+            (await queue.currentConcurrencyOfQueue(
+              authenticatedEnvDev,
+              "task/my-task",
+              "ck-good"
+            )) === 1,
+          30_000
+        );
+        expect(goodAdmitted).toBe(true);
+        expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(10);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
   redisTest("overrides are ignored when disabled", async ({ redisContainer }) => {
     const queue = createQueue(redisContainer, false);
     try {
 
@@ -170,7 +170,7 @@ export function overrideConcurrencyLimit(
     requestOptions
   );
 
-  if (concurrencyKey) {
+  if (concurrencyKey !== undefined) {
     return apiClient.overrideQueueConcurrencyKeyLimit(
       queue,
       concurrencyKey,
@@ -291,7 +291,7 @@ export function resetConcurrencyLimit(
     requestOptions
   );
 
-  if (concurrencyKey) {
+  if (concurrencyKey !== undefined) {
     return apiClient.resetQueueConcurrencyKeyLimit(queue, concurrencyKey, $requestOptions);
   }