fix(webapp,run-engine): limit mutations survive every race direction

The optimistic guard now carries the row's updatedAt, so a deploy
refreshing declared values (which deliberately keeps the markers)
conflicts a concurrent mutation instead of losing its write. A reset
whose persist fails after the engine already reverted re-syncs the
engine from a fresh read, so a concurrent actor's override or pause is
enforced again. Override bounds are validated against the environment
maximum like queue overrides. Counter reads refresh the TTL so an
observed gate never re-anchors while idle, and the routes drop a
redundant decode that turned malformed names into 500s.

Author: matt-aitken

apps/webapp/app/routes/api.v1.concurrency-limits.$name.override.ts

@@ -5,7 +5,7 @@ import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server
 import { concurrencyLimitsSystem } from "~/v3/services/concurrencyLimitsSystemInstance.server";
 
 const ParamsSchema = z.object({
-  name: z.string().transform((val) => decodeURIComponent(val)),
+  name: z.string(),
 });
 
 const route = createActionApiRoute(

apps/webapp/app/routes/api.v1.concurrency-limits.$name.reset.ts

@@ -4,7 +4,7 @@ import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server
 import { concurrencyLimitsSystem } from "~/v3/services/concurrencyLimitsSystemInstance.server";
 
 const ParamsSchema = z.object({
-  name: z.string().transform((val) => decodeURIComponent(val)),
+  name: z.string(),
 });
 
 const route = createActionApiRoute(

apps/webapp/app/routes/api.v1.concurrency-limits.$name.ts

@@ -4,7 +4,7 @@ import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server
 import { concurrencyLimitsSystem } from "~/v3/services/concurrencyLimitsSystemInstance.server";
 
 const ParamsSchema = z.object({
-  name: z.string().transform((val) => decodeURIComponent(val)),
+  name: z.string(),
 });
 
 export const loader = createLoaderApiRoute(

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

@@ -114,6 +114,12 @@ export class ConcurrencyLimitsSystem {
               message: `\`${field}\` must be an integer between 0 and 100000`,
             });
           }
+          if (value > environment.maximumConcurrencyLimit) {
+            return errAsync({
+              type: "invalid_override" as const,
+              message: `\`${field}\` (${value}) cannot exceed the environment limit (${environment.maximumConcurrencyLimit})`,
+            });
+          }
         }
 
         return findLimitByName(this.db, environment, name)
@@ -129,7 +135,13 @@ export class ConcurrencyLimitsSystem {
       reset: (environment: AuthenticatedEnvironment, name: string) => {
         return findLimitByName(this.db, environment, name)
           .andThen((row) => syncResetToEngine(environment, row))
-          .andThen((row) => resetLimitOverrides(this.db, row))
+          .andThen((row) =>
+            resetLimitOverrides(this.db, row).orElse((error) =>
+              compensateEngineFromFreshRow(this.db, environment, row.id).andThen(() =>
+                errAsync(error)
+              )
+            )
+          )
           .andThen((row) =>
             fromPromise(toLimitItems(environment, [row]), (error) => ({
               type: "other" as const,
@@ -303,9 +315,10 @@ function resetLimitOverrides(db: PrismaClientOrTransaction, row: TaskQueue) {
 }
 
 /**
- * Optimistic update: the where clause carries the override markers as read, so a
- * concurrent override or reset makes this update miss (P2025) and the caller gets
- * a conflict instead of silently clobbering the newer state.
+ * Optimistic update: the where clause carries the row's updatedAt as read, so ANY
+ * concurrent write — another override or reset, or a deploy refreshing the declared
+ * values — makes this update miss (P2025) and the caller gets a conflict instead of
+ * persisting values computed from a stale row.
  */
 function guardedLimitUpdate(
   db: PrismaClientOrTransaction,
@@ -316,8 +329,7 @@ function guardedLimitUpdate(
     db.taskQueue.update({
       where: {
         id: row.id,
-        concurrencyLimitOverriddenAt: row.concurrencyLimitOverriddenAt,
-        totalConcurrencyLimitOverriddenAt: row.totalConcurrencyLimitOverriddenAt,
+        updatedAt: row.updatedAt,
       },
       data,
     }),
@@ -330,6 +342,24 @@ function guardedLimitUpdate(
   );
 }
 
+/**
+ * A reset's engine write precedes its guarded persist (enforce-first, so an engine
+ * failure retries cleanly), which leaves the engine reverted when the persist
+ * conflicts or fails. This re-syncs the engine from a fresh read of the row so a
+ * concurrent actor's state (or the still-standing override) is enforced again;
+ * the original error still reaches the caller.
+ */
+function compensateEngineFromFreshRow(
+  db: PrismaClientOrTransaction,
+  environment: AuthenticatedEnvironment,
+  rowId: string
+) {
+  return fromPromise(db.taskQueue.findFirst({ where: { id: rowId } }), (error) => ({
+    type: "other" as const,
+    cause: error,
+  })).andThen((fresh) => (fresh ? syncLimitToEngine(environment, fresh) : okAsync(null)));
+}
+
 /**
  * Pushes both engine keys from the row: the per-key limit and the total. Limit
  * rows are never paused (pausing a limit is an override to `{ total: 0 }`), so

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

@@ -78,6 +78,10 @@ const SemanticAttributes = {
  * during a rolling upgrade), including runs parked in the DLQ or suspended on
  * checkpoints, which older rules based on the queue zset could never prune.
  */
+/** TTL for gate queued counters: refreshed on every delta and on every read, so an
+ * active or observed gate never re-anchors; the Lua helper inlines the same value. */
+const GATE_QUEUED_COUNTER_TTL_SECONDS = 86400;
+
 const QUEUE_GATES_LUA_HELPERS = `
 local function __gateKeys(gatesKeyPrefix, msg, gate)
   local base = gatesKeyPrefix .. '{org:' .. msg.orgId .. '}:proj:' .. msg.projectId .. ':env:' .. msg.environmentId .. ':queue:' .. gate.queue
@@ -696,29 +700,34 @@ export class RunQueue {
 
   /**
    * Runs that are queued and must clear this gate queue to execute (the per-gate
-   * queued counter). Exact by construction: incremented per gate on enqueue and
-   * decremented on admit and on every queued-removal path, floored at zero.
+   * queued counter): incremented per gate on enqueue and decremented on admit and
+   * on every queued-removal path, floored at zero. Reads refresh the counter's
+   * TTL, so a gate anyone observes (dashboard, API) never expires while idle —
+   * e.g. a paused limit with a stalled backlog keeps its count over a quiet
+   * weekend; only gates nobody touches or reads for a day re-anchor.
    */
   public async gateQueuedCountOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) {
-    const result = await this.redis.get(this.keys.gateQueuedCounterKey(env, queue));
-    return result ? Math.max(Number(result), 0) : 0;
+    const counts = await this.gateQueuedCountOfQueues(env, [queue]);
+    return counts[queue] ?? 0;
   }
 
-  /** Batch variant of gateQueuedCountOfQueue: one pipeline of GETs. */
+  /** Batch variant: one pipeline of GETs, each with a TTL refresh. */
   public async gateQueuedCountOfQueues(
     env: MinimalAuthenticatedEnvironment,
     queues: string[]
   ): Promise<Record<string, number>> {
     const pipeline = this.redis.pipeline();
     queues.forEach((queue) => {
-      pipeline.get(this.keys.gateQueuedCounterKey(env, queue));
+      const key = this.keys.gateQueuedCounterKey(env, queue);
+      pipeline.get(key);
+      pipeline.expire(key, GATE_QUEUED_COUNTER_TTL_SECONDS, "XX");
     });
 
     const results = await pipeline.exec();
 
     return queues.reduce(
       (acc, queue, index) => {
-        const value = results?.[index]?.[1];
+        const value = results?.[index * 2]?.[1];
         const parsed = typeof value === "string" ? Number(value) : 0;
         acc[queue] = Number.isFinite(parsed) ? Math.max(parsed, 0) : 0;
         return acc;