fix(webapp): re-sync engine limits when an override lands during a deploy

The optimistic markers only guard the Postgres write; an override or
reset landing between that write and the engine sync would leave the
engine holding the deploy's stale values. The deploy now re-reads the
markers after syncing and re-syncs once from the fresh row when they
moved, so the last engine write always carries the freshest limits.

Author: matt-aitken

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

@@ -771,53 +771,77 @@ async function createWorkerQueue(
     concurrencyVersion
   );
 
-  const newConcurrencyLimit = taskQueue.concurrencyLimit;
-
-  /**
-   * The total limit key is separate from the per-queue limit key that pause zeroes,
-   * so it is safe to sync it regardless of the paused state. The engine clamps it
-   * to the environment limit at read time, so the raw declared value is stored.
-   */
-  if (typeof taskQueue.totalConcurrencyLimit === "number") {
-    await updateQueueTotalConcurrencyLimits(
-      environment,
-      taskQueue.name,
-      taskQueue.totalConcurrencyLimit
-    );
-  } else {
-    await removeQueueTotalConcurrencyLimits(environment, taskQueue.name);
-  }
+  const syncQueueLimitsToEngine = async (row: {
+    name: string;
+    paused: boolean;
+    concurrencyLimit: number | null;
+    totalConcurrencyLimit: number | null;
+  }) => {
+    /**
+     * The total limit key is separate from the per-queue limit key that pause zeroes,
+     * so it is safe to sync it regardless of the paused state. The engine clamps it
+     * to the environment limit at read time, so the raw declared value is stored.
+     */
+    if (typeof row.totalConcurrencyLimit === "number") {
+      await updateQueueTotalConcurrencyLimits(environment, row.name, row.totalConcurrencyLimit);
+    } else {
+      await removeQueueTotalConcurrencyLimits(environment, row.name);
+    }
 
-  if (!taskQueue.paused) {
-    if (typeof newConcurrencyLimit === "number") {
-      logger.debug("createWorkerQueue: updating concurrency limit", {
+    if (!row.paused) {
+      logger.debug("createWorkerQueue: syncing concurrency limit", {
         workerId: worker.id,
-        taskQueue,
+        taskQueue: row,
         orgId: environment.organizationId,
         projectId: environment.projectId,
         environmentId: environment.id,
-        concurrencyLimit: newConcurrencyLimit,
+        concurrencyLimit: row.concurrencyLimit,
       });
-      await updateQueueConcurrencyLimits(environment, taskQueue.name, newConcurrencyLimit);
+      if (typeof row.concurrencyLimit === "number") {
+        await updateQueueConcurrencyLimits(environment, row.name, row.concurrencyLimit);
+      } else {
+        await removeQueueConcurrencyLimits(environment, row.name);
+      }
     } else {
-      logger.debug("createWorkerQueue: removing concurrency limit", {
+      logger.debug("createWorkerQueue: queue is paused, not updating concurrency limit", {
         workerId: worker.id,
-        taskQueue,
+        taskQueue: row,
         orgId: environment.organizationId,
         projectId: environment.projectId,
         environmentId: environment.id,
-        concurrencyLimit: newConcurrencyLimit,
       });
-      await removeQueueConcurrencyLimits(environment, taskQueue.name);
     }
-  } else {
-    logger.debug("createWorkerQueue: queue is paused, not updating concurrency limit", {
-      workerId: worker.id,
-      taskQueue,
-      orgId: environment.organizationId,
-      projectId: environment.projectId,
-      environmentId: environment.id,
-    });
+  };
+
+  await syncQueueLimitsToEngine(taskQueue);
+
+  /**
+   * The optimistic markers only guard the Postgres write; an override or reset can
+   * still land between that write and the engine sync above, which would leave the
+   * engine holding this deploy's stale values. Re-read the markers and re-sync once
+   * from the fresh row when they moved: every actor writes Postgres before its own
+   * engine sync, so whoever syncs last is syncing the freshest row.
+   */
+  const freshQueue = await prisma.taskQueue.findFirst({
+    where: { id: taskQueue.id },
+    select: {
+      name: true,
+      paused: true,
+      concurrencyLimit: true,
+      totalConcurrencyLimit: true,
+      concurrencyLimitOverriddenAt: true,
+      totalConcurrencyLimitOverriddenAt: true,
+    },
+  });
+
+  if (
+    freshQueue &&
+    (freshQueue.concurrencyLimitOverriddenAt?.getTime() !==
+      taskQueue.concurrencyLimitOverriddenAt?.getTime() ||
+      freshQueue.totalConcurrencyLimitOverriddenAt?.getTime() !==
+        taskQueue.totalConcurrencyLimitOverriddenAt?.getTime())
+  ) {
+    await syncQueueLimitsToEngine(freshQueue);
   }
 
   return taskQueue;