fix(run-engine,webapp): close gaps found in adversarial review of the pause sweep

Restore the TTL registration when returning a run. Admission removes the run's
member from the TTL sorted set, so returning it to the pending queue without
re-adding left a run that could never expire: on the next admission the dequeue
script takes its already-expired branch, which deliberately leaves cleanup to
the TTL consumer, and that consumer no longer has an entry to find. The per-run
expireRun job is not a backstop because it is only armed for development
environments.

Take the pause sweep out of the rollback window. It ran inside the try that
compensates a failed pause by clearing the DB flag, but by then the Redis limit
was already 0, so a sweep failure left the environment with no concurrency and
no paused flag — dark, and unrecoverable without a manual pause/resume cycle.
For the billing-limit path it also aborted the remaining environments in the
org. The pause is durable and the sweep is idempotent, so it now runs after the
pause has been committed and a failure is logged rather than propagated. That
also stops the queue pause reporting failure for a pause that took effect.

Retry passes while runs remain unclaimed. Admission is not atomic with the push
onto the worker queue: the dequeue script claims the concurrency slot and a
separate round trip does the RPUSH. A run caught in that gap fails the claim and
would previously have been abandoned, then executed during the pause.

Claim from both the resolver-mapped and the raw worker queue key. The trigger-
time fast path pushes to the raw workerQueue on the message while the master
queue consumer pushes to the resolver-mapped one, so with a region override
active the two producers disagree and a single-key LREM missed fast-pathed runs.

Batch the sweep. It read every payload in one MGET and then made one sequential
round trip per run, on a request path, with no error isolation — one failure
abandoned the remaining runs.

Tests: cover the LREM claim directly by emptying the worker queue without
marking the run dequeued (the previous test was excluded by the candidate set
before ever reaching the script), the TTL round trip, and the concurrency key
running counter across a return.

Author: matt-aitken

apps/webapp/app/v3/runQueue.server.ts

@@ -1,4 +1,5 @@
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
 import { engine } from "./runEngine.server";
 
 /** Updates the RunQueue env concurrency limits */
@@ -46,3 +47,31 @@ export async function returnUnclaimedMessagesToQueue({
 }) {
   return engine.returnUnclaimedMessagesToQueue({ environment, queue });
 }
+
+/**
+ * Best-effort sweep for the pause paths.
+ *
+ * By the time this runs the pause is already durable and in force, so a failure here must
+ * not roll it back or surface as a failed pause — the worst case is that some already-admitted
+ * runs still execute, which is the behaviour that existed before the sweep.
+ */
+export async function sweepUnclaimedRuns(
+  environment: AuthenticatedEnvironment,
+  queue?: string
+): Promise<void> {
+  try {
+    const result = await returnUnclaimedMessagesToQueue({ environment, queue });
+
+    logger.debug("sweepUnclaimedRuns", {
+      environmentId: environment.id,
+      queue,
+      ...result,
+    });
+  } catch (error) {
+    logger.error("sweepUnclaimedRuns failed", {
+      environmentId: environment.id,
+      queue,
+      error,
+    });
+  }
+}

apps/webapp/app/v3/services/billingLimit/billingLimitConvergeEnvironments.server.ts

@@ -190,7 +190,6 @@ async function pauseEnvironmentForBillingLimit(
 
   try {
     await updateConcurrency(updated, 0);
-    await returnUnclaimed(updated);
   } catch (error) {
     await db.runtimeEnvironment.update({
       where: { id: environment.id },
@@ -201,6 +200,15 @@ async function pauseEnvironmentForBillingLimit(
     // The env's paused state changed (or was rolled back); drop any cached copy either way.
     controlPlaneResolver.invalidateEnvironment(environment.id);
   }
+
+  try {
+    await returnUnclaimed(updated);
+  } catch (error) {
+    logger.error("Billing limit converge failed to return unclaimed runs", {
+      environmentId: environment.id,
+      error,
+    });
+  }
 }
 
 async function resumeEnvironmentFromBillingLimit(

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

@@ -2,7 +2,7 @@ import { EnvironmentPauseSource, type PrismaClientOrTransaction } from "@trigger
 import { prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { getManualPauseEnvironmentResult } from "~/v3/services/billingLimit/manualPauseEnvironmentGuard.server";
-import { returnUnclaimedMessagesToQueue, updateEnvConcurrencyLimits } from "../runQueue.server";
+import { sweepUnclaimedRuns, updateEnvConcurrencyLimits } from "../runQueue.server";
 import { WithRunEngine } from "./baseService.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
@@ -114,7 +114,6 @@ export class PauseEnvironmentService extends WithRunEngine {
             environmentId: environment.id,
           });
           await updateEnvConcurrencyLimits(environment, 0);
-          await returnUnclaimedMessagesToQueue({ environment });
         } else {
           logger.debug("PauseEnvironmentService: resuming environment", {
             environmentId: environment.id,
@@ -137,6 +136,10 @@ export class PauseEnvironmentService extends WithRunEngine {
       // The env's `paused` state changed in the control-plane; drop any cached copy.
       controlPlaneResolver.invalidateEnvironment(environment.id);
 
+      if (action === "paused") {
+        await sweepUnclaimedRuns(environment);
+      }
+
       return {
         success: true,
         state: action,

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

@@ -6,7 +6,7 @@ import { BaseService } from "./baseService.server";
 import { determineEngineVersion } from "../engineVersion.server";
 import {
   removeQueueConcurrencyLimits,
-  returnUnclaimedMessagesToQueue,
+  sweepUnclaimedRuns,
   updateQueueConcurrencyLimits,
 } from "../runQueue.server";
 import { engine } from "../runEngine.server";
@@ -63,7 +63,6 @@ export class PauseQueueService extends BaseService {
 
       if (action === "paused") {
         await updateQueueConcurrencyLimits(environment, queue.name, 0);
-        await returnUnclaimedMessagesToQueue({ environment, queue: queue.name });
       } else {
         if (queue.concurrencyLimit) {
           await updateQueueConcurrencyLimits(environment, queue.name, queue.concurrencyLimit);
@@ -78,6 +77,10 @@ export class PauseQueueService extends BaseService {
         environmentId: environment.id,
       });
 
+      if (action === "paused") {
+        await sweepUnclaimedRuns(environment, queue.name);
+      }
+
       const results = await Promise.all([
         engine.lengthOfQueues(environment, [queue.name]),
         engine.currentConcurrencyOfQueues(environment, [queue.name]),

apps/webapp/test/billingLimitConvergeEnvironments.test.ts

@@ -95,30 +95,38 @@ describe("convergeBillingLimitEnvironmentsForOrg", () => {
     expect(returnUnclaimed).not.toHaveBeenCalled();
   });
 
-  postgresTest("rolls back pause when returning unclaimed runs fails", async ({ prisma }) => {
+  postgresTest("keeps the pause when returning unclaimed runs fails", async ({ prisma }) => {
     const { organization, project } = await createTestOrgProjectWithMember(prisma);
     const environment = await createRuntimeEnvironment(prisma, {
       projectId: project.id,
       organizationId: organization.id,
       type: "PRODUCTION",
       slug: uniqueId("prod"),
     });
+    const second = await createRuntimeEnvironment(prisma, {
+      projectId: project.id,
+      organizationId: organization.id,
+      type: "PRODUCTION",
+      slug: uniqueId("prod"),
+    });
 
-    await expect(
-      convergeBillingLimitEnvironmentsForOrg(organization.id, "grace", {
-        prismaClient: prisma,
-        updateConcurrency: async () => undefined,
-        returnUnclaimed: async () => {
-          throw new Error("run queue unavailable");
-        },
-      })
-    ).rejects.toThrow("run queue unavailable");
-
-    const envAfter = await prisma.runtimeEnvironment.findUniqueOrThrow({
-      where: { id: environment.id },
+    const result = await convergeBillingLimitEnvironmentsForOrg(organization.id, "grace", {
+      prismaClient: prisma,
+      updateConcurrency: async () => undefined,
+      returnUnclaimed: async () => {
+        throw new Error("run queue unavailable");
+      },
     });
-    expect(envAfter.paused).toBe(false);
-    expect(envAfter.pauseSource).toBeNull();
+
+    expect(result).toEqual({ paused: 2, unpaused: 0 });
+
+    for (const env of [environment, second]) {
+      const envAfter = await prisma.runtimeEnvironment.findUniqueOrThrow({
+        where: { id: env.id },
+      });
+      expect(envAfter.paused).toBe(true);
+      expect(envAfter.pauseSource).toBe(EnvironmentPauseSource.BILLING_LIMIT);
+    }
   });
 
   postgresTest("rolls back pause when concurrency update fails", async ({ prisma }) => {