fix(webapp): limit convergence compares enforced values, zero thresholds never read as at-limit

The freshness re-check after an override sync converges on the values the
engine enforces instead of updatedAt, since Prisma's millisecond updatedAt
makes two same-millisecond writes indistinguishable; identical values mean
identical engine state so skipping them is safe. A zero at-limit threshold
(a paused limit, or a queue overridden to 0) no longer shows the at-limit
warning, matching the health badge. The override tests reset mocks per test
and assert the exact compensating call so they fail without the fix.

Author: matt-aitken

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.concurrency/route.tsx

@@ -782,8 +782,12 @@ function QueuesWithMetricsView() {
                       ? Math.min(queue.concurrency.combined.current, environment.concurrencyLimit)
                       : null
                     : limit;
+                  /** A zero threshold is a pause (nothing may run), not saturation —
+                   * without the guard `running >= 0` holds for every row. */
                   const isAtConcurrencyLimit =
-                    atLimitThreshold !== null && queue.running >= atLimitThreshold;
+                    atLimitThreshold !== null &&
+                    atLimitThreshold > 0 &&
+                    queue.running >= atLimitThreshold;
                   const isAtQueueLimit =
                     environment.queueSizeLimit !== null &&
                     queue.queued >= environment.queueSizeLimit;
@@ -1970,8 +1974,12 @@ function ClassicQueuesView() {
                         ? Math.min(queue.concurrency.combined.current, environment.concurrencyLimit)
                         : null
                       : limit;
+                    /** A zero threshold is a pause (nothing may run), not saturation —
+                     * without the guard `running >= 0` holds for every row. */
                     const isAtConcurrencyLimit =
-                      atLimitThreshold !== null && queue.running >= atLimitThreshold;
+                      atLimitThreshold !== null &&
+                      atLimitThreshold > 0 &&
+                      queue.running >= atLimitThreshold;
                     const isAtQueueLimit =
                       environment.queueSizeLimit !== null &&
                       queue.queued >= environment.queueSizeLimit;

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

@@ -128,7 +128,10 @@ export class ConcurrencyLimitsSystem {
             syncLimitToEngine(environment, row)
               .andThen(() =>
                 compensateEngineFromFreshRow(this.db, environment, row.id, {
-                  alreadySyncedAt: row.updatedAt.getTime(),
+                  alreadySynced: {
+                    perKey: row.concurrencyLimit,
+                    total: row.totalConcurrencyLimit,
+                  },
                 })
                   .orElse(() => okAsync(undefined))
                   .map(() => row)
@@ -387,28 +390,39 @@ function guardedLimitUpdate(
   );
 }
 
+type SyncedLimitValues = { perKey: number | null; total: number | null };
+
 /**
- * Re-syncs the engine from fresh reads of the row until updatedAt stops moving
- * (bounded), the same convergence the deploy sync uses: every actor writes
+ * Re-syncs the engine from fresh reads of the row until the enforced values stop
+ * moving (bounded), the same convergence the deploy sync uses: every actor writes
  * Postgres before its own engine sync, so re-syncing whatever is freshest
- * converges. Callers use it two ways: after a failure (a reset's enforce-first
- * engine write preceding a persist that then conflicts, or an override's sync
- * failing after its persist), where the original error still reaches the caller;
- * and after a successful sync with `alreadySyncedAt` set to the synced row's
- * updatedAt, where an unchanged row costs one read and a moved row is re-synced.
+ * converges. The fixpoint compares the values the engine enforces rather than
+ * updatedAt, because Prisma's @updatedAt has millisecond precision and two writes
+ * in the same millisecond are indistinguishable by timestamp; identical values
+ * mean identical engine state, so skipping those is always safe. Callers use it
+ * two ways: after a failure (a reset's enforce-first engine write preceding a
+ * persist that then conflicts, or an override's sync failing after its persist),
+ * where the original error still reaches the caller; and after a successful sync
+ * with `alreadySynced` set to the values just synced, where an unchanged row
+ * costs one read and a moved row is re-synced.
  */
 function compensateEngineFromFreshRow(
   db: PrismaClientOrTransaction,
   environment: AuthenticatedEnvironment,
   rowId: string,
-  options?: { alreadySyncedAt?: number }
+  options?: { alreadySynced?: SyncedLimitValues }
 ) {
   return fromPromise(
     (async () => {
-      let lastSyncedAt: number | null = options?.alreadySyncedAt ?? null;
+      let lastSynced: SyncedLimitValues | null = options?.alreadySynced ?? null;
       for (let i = 0; i < 3; i++) {
         const fresh = await db.taskQueue.findFirst({ where: { id: rowId } });
-        if (!fresh || fresh.updatedAt.getTime() === lastSyncedAt) {
+        if (
+          !fresh ||
+          (lastSynced !== null &&
+            fresh.concurrencyLimit === lastSynced.perKey &&
+            fresh.totalConcurrencyLimit === lastSynced.total)
+        ) {
           return;
         }
         await settleBothEngineWrites(
@@ -423,7 +437,7 @@ function compensateEngineFromFreshRow(
               )
             : removeQueueTotalConcurrencyLimits(environment, fresh.name)
         );
-        lastSyncedAt = fresh.updatedAt.getTime();
+        lastSynced = { perKey: fresh.concurrencyLimit, total: fresh.totalConcurrencyLimit };
       }
     })(),
     (error) => ({ type: "other" as const, cause: error })

apps/webapp/test/concurrencyLimitsSystem.test.ts

@@ -1,6 +1,6 @@
 import { postgresTest } from "@internal/testcontainers";
 import type { PrismaClient } from "@trigger.dev/database";
-import { describe, expect, vi } from "vitest";
+import { beforeEach, describe, expect, vi } from "vitest";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { ConcurrencyLimitsSystem } from "~/v3/services/concurrencyLimitsSystem.server";
 
@@ -82,6 +82,16 @@ async function seedEnvAndLimit(
 }
 
 describe("ConcurrencyLimitsSystem", () => {
+  /** Call counts must start at zero per test and leaked one-off implementations
+   * must not outlive the test that set them; mockReset also restores the default
+   * implementations given to vi.fn above. */
+  beforeEach(() => {
+    perKeySyncMock.mockReset();
+    perKeyRemoveMock.mockReset();
+    totalSyncMock.mockReset();
+    totalRemoveMock.mockReset();
+  });
+
   postgresTest(
     "override changes only the given bound and keeps the declared base",
     async ({ prisma }) => {
@@ -215,11 +225,13 @@ describe("ConcurrencyLimitsSystem", () => {
       }
 
       /** The persist already happened; compensation re-syncs it so the engine
-       * doesn't keep enforcing the old bound while the API reports the new one. */
+       * doesn't keep enforcing the old bound while the API reports the new one.
+       * Exactly two calls: the rejected primary sync, then the compensating
+       * re-sync from the fresh row — without compensation there is only one. */
       const updated = await prisma.taskQueue.findFirstOrThrow({ where: { id: row.id } });
       expect(updated.totalConcurrencyLimit).toBe(50);
+      expect(totalSyncMock).toHaveBeenCalledTimes(2);
       expect(totalSyncMock).toHaveBeenLastCalledWith(authEnv, "limit/openai", 50);
-      expect(totalSyncMock.mock.calls.length).toBeGreaterThanOrEqual(2);
     }
   );
 
@@ -248,8 +260,6 @@ describe("ConcurrencyLimitsSystem", () => {
       const final = await prisma.taskQueue.findFirstOrThrow({ where: { id: row.id } });
       expect(final.totalConcurrencyLimit).toBe(75);
       expect(engineTotal).toBe(75);
-
-      totalSyncMock.mockImplementation(async () => undefined);
     }
   );