fix(webapp,run-engine): recoverable resets, conflict-guarded limit mutations, activity-refreshed counter TTL

Limit resets sync the engine to the declared base before clearing the
markers, so an engine failure leaves the override intact and a retry
converges. Both limit mutations carry the read markers in their where
clause and surface a conflict instead of clobbering a concurrent
change. The gate queued counter's TTL now refreshes on every delta, so
an active gate's count never resets while drift from delta-less paths
still clears once the gate goes quiet; the flag comment documents that
total bounds enforce solely through the total-concurrency flag.

Author: matt-aitken

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

@@ -1,6 +1,6 @@
 import type { TaskQueue, User } from "@trigger.dev/database";
 import { errAsync, fromPromise, okAsync } from "neverthrow";
-import type { PrismaClientOrTransaction } from "~/db.server";
+import { Prisma, type PrismaClientOrTransaction } from "~/db.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import {
   removeQueueConcurrencyLimits,
@@ -128,8 +128,8 @@ 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) => syncLimitToEngine(environment, row))
           .andThen((row) =>
             fromPromise(toLimitItems(environment, [row]), (error) => ({
               type: "other" as const,
@@ -246,17 +246,43 @@ function applyLimitOverride(
     data.totalConcurrencyLimitOverriddenBy = overriddenBy?.id ?? null;
   }
 
-  return fromPromise(db.taskQueue.update({ where: { id: row.id }, data }), (error) => ({
-    type: "limit_update_failed" as const,
-    cause: error,
-  }));
+  return guardedLimitUpdate(db, row, data);
 }
 
-function resetLimitOverrides(db: PrismaClientOrTransaction, row: TaskQueue) {
+/**
+ * Enforce first, then persist: the engine syncs to the declared base BEFORE the
+ * override markers clear, so an engine failure leaves the markers set and a retry
+ * converges instead of being rejected while the overridden limit stays enforced.
+ */
+function syncResetToEngine(environment: AuthenticatedEnvironment, row: TaskQueue) {
   if (row.concurrencyLimitOverriddenAt === null && row.totalConcurrencyLimitOverriddenAt === null) {
     return errAsync({ type: "limit_not_overridden" as const });
   }
 
+  const perKeyTarget = row.concurrencyLimitOverriddenAt
+    ? row.concurrencyLimitBase
+    : row.concurrencyLimit;
+  const totalTarget = row.totalConcurrencyLimitOverriddenAt
+    ? row.totalConcurrencyLimitBase
+    : row.totalConcurrencyLimit;
+
+  const perKeySync =
+    typeof perKeyTarget === "number"
+      ? updateQueueConcurrencyLimits(environment, row.name, perKeyTarget)
+      : removeQueueConcurrencyLimits(environment, row.name);
+
+  const totalSync =
+    typeof totalTarget === "number"
+      ? updateQueueTotalConcurrencyLimits(environment, row.name, totalTarget)
+      : removeQueueTotalConcurrencyLimits(environment, row.name);
+
+  return fromPromise(Promise.all([perKeySync, totalSync]), (error) => ({
+    type: "sync_limit_to_engine_failed" as const,
+    cause: error,
+  })).map(() => row);
+}
+
+function resetLimitOverrides(db: PrismaClientOrTransaction, row: TaskQueue) {
   const data: Record<string, unknown> = {};
 
   if (row.concurrencyLimitOverriddenAt !== null) {
@@ -273,10 +299,35 @@ function resetLimitOverrides(db: PrismaClientOrTransaction, row: TaskQueue) {
     data.totalConcurrencyLimitOverriddenBy = null;
   }
 
-  return fromPromise(db.taskQueue.update({ where: { id: row.id }, data }), (error) => ({
-    type: "limit_update_failed" as const,
-    cause: error,
-  }));
+  return guardedLimitUpdate(db, row, data);
+}
+
+/**
+ * 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.
+ */
+function guardedLimitUpdate(
+  db: PrismaClientOrTransaction,
+  row: TaskQueue,
+  data: Record<string, unknown>
+) {
+  return fromPromise(
+    db.taskQueue.update({
+      where: {
+        id: row.id,
+        concurrencyLimitOverriddenAt: row.concurrencyLimitOverriddenAt,
+        totalConcurrencyLimitOverriddenAt: row.totalConcurrencyLimitOverriddenAt,
+      },
+      data,
+    }),
+    (error) => {
+      if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2025") {
+        return { type: "conflict" as const };
+      }
+      return { type: "limit_update_failed" as const, cause: error };
+    }
+  );
 }
 
 /**

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

@@ -166,10 +166,11 @@ end
 -- Callers gate the delta on the actual queue-zset transition (ZADD added == 1 /
 -- ZREM removed == 1) so re-enqueues and already-removed members never double count;
 -- gates sharing a base (duplicate entries, key variants) count once per run. The
--- 24h absolute TTL (set at creation, never extended) re-anchors drift from paths
--- without the delta (rolling deploys, stale-entry cleanup): the counter resets,
--- floored decrements absorb the pre-reset backlog as it drains, and counts converge
--- to exact for every run enqueued after the reset. Payload-driven and
+-- 24h TTL refreshes on every delta, so an ACTIVE gate's count never resets while
+-- drift from delta-less paths (a mixed-version rollout, a stale-entry cleanup)
+-- clears once the gate has been quiet for a day. The residual gap is a gate idle
+-- for 24h with runs still queued (e.g. paused with no new enqueues): its counter
+-- expires and under-counts until the backlog fully drains. Payload-driven and
 -- flag-independent, like release, so counts stay exact across flag flips.
 local function __gateQueuedDelta(gatesKeyPrefix, msg, delta)
   if type(msg) ~= 'table' or not msg.gates then return end
@@ -180,12 +181,11 @@ local function __gateQueuedDelta(gatesKeyPrefix, msg, delta)
       seenBases[base] = true
       local counterKey = base .. ':gateQueuedCounter'
       if delta > 0 then
-        if redis.call('EXISTS', counterKey) == 0 then
-          redis.call('SET', counterKey, '0', 'EX', '86400')
-        end
         redis.call('INCRBY', counterKey, delta)
+        redis.call('EXPIRE', counterKey, '86400')
       elseif tonumber(redis.call('GET', counterKey) or '0') > 0 then
         redis.call('DECRBY', counterKey, -delta)
+        redis.call('EXPIRE', counterKey, '86400')
       end
     end
   end
@@ -347,7 +347,9 @@ export type RunQueueOptions = {
   /**
    * When true, queues maintain a per-base-queue groupConcurrency SET (total in-flight
    * across all key variants AND keyless runs) and enforce the queue's total concurrency
-   * limit at admit time. Default false: admit paths are byte-identical to before, and
+   * limit at admit time. V2 concurrency semantics (a limit's `total` bound, including
+   * total-only declarations) are enforced solely through this flag: with it off, a
+   * total-only limit caps nothing. Default false: admit paths are byte-identical to before, and
    * only the release-side SREM mirror runs (a no-op on an absent set), so the flag can
    * be flipped on a fleet that has fully rolled onto this build without draining queues.
    *