fix(run-engine,webapp): harden the pause sweep retry and report it honestly

Guard the pass count against non-finite input. Math.max(1, NaN) is NaN, so a
NaN maxPasses skipped the loop entirely and reported a successful no-op sweep,
and Infinity spun at one pass per 50ms inside an awaited request. The count is
now clamped to a ceiling and falls back to the default when it is not finite.

Retry a pass that errored, not just one that skipped. Rejected script calls
never increment skipped, so a transient Redis failure part-way through a batch
ended the sweep with those runs still admitted and still on the worker queue —
the exact outcome the sweep exists to prevent — while the result read as a
success with a non-zero error count nobody was gated on.

Rename the residual count to skippedLastPass. It is the final pass's value
rather than a total, and it conflates a run a worker claimed first (which
escaped the pause and is executing) with a leaked concurrency slot that has no
worker queue entry behind it. The name and its doc comment now say so instead
of implying it partitions the candidate set alongside returned.

Tests: pin the non-finite pass-count guard, cover that an acknowledged run is
never resurrected by a later sweep, and stop the worker queue override test
leaking its environment variable if the queue constructor throws.

Author: matt-aitken

apps/webapp/test/pauseSweepWiring.test.ts

@@ -0,0 +1,94 @@
+import { type PrismaClient } from "@trigger.dev/database";
+import { describe, expect, vi } from "vitest";
+import { postgresTest } from "@internal/testcontainers";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import {
+  createRuntimeEnvironment,
+  createTestOrgProjectWithMember,
+  uniqueId,
+} from "./fixtures/environmentVariablesFixtures";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+const calls: string[] = [];
+
+vi.mock("~/v3/runQueue.server", () => ({
+  updateEnvConcurrencyLimits: vi.fn(async (_env: unknown, limit?: number) => {
+    calls.push(`updateEnvConcurrencyLimits:${limit}`);
+  }),
+  updateQueueConcurrencyLimits: vi.fn(async (_env: unknown, name: string, limit: number) => {
+    calls.push(`updateQueueConcurrencyLimits:${name}:${limit}`);
+  }),
+  removeQueueConcurrencyLimits: vi.fn(async () => {
+    calls.push("removeQueueConcurrencyLimits");
+  }),
+  returnUnclaimedMessagesToQueue: vi.fn(async () => ({
+    returned: 0,
+    skipped: 0,
+    errors: 0,
+    passes: 1,
+  })),
+  sweepUnclaimedRuns: vi.fn(async (_env: unknown, queue?: string) => {
+    calls.push(`sweepUnclaimedRuns:${queue ?? "*"}`);
+  }),
+}));
+
+async function loadServices() {
+  const [{ PauseEnvironmentService }, { authIncludeBase, toAuthenticated }] = await Promise.all([
+    import("~/v3/services/pauseEnvironment.server"),
+    import("~/models/runtimeEnvironment.server"),
+  ]);
+  return { PauseEnvironmentService, authIncludeBase, toAuthenticated };
+}
+
+type Loaded = Awaited<ReturnType<typeof loadServices>>;
+
+async function seedEnv(
+  loaded: Loaded,
+  prisma: PrismaClient
+): Promise<{ environment: AuthenticatedEnvironment; environmentId: string }> {
+  const { organization, project } = await createTestOrgProjectWithMember(prisma);
+  const created = await createRuntimeEnvironment(prisma, {
+    projectId: project.id,
+    organizationId: organization.id,
+    type: "PRODUCTION",
+    slug: uniqueId("prod"),
+  });
+
+  const row = await prisma.runtimeEnvironment.findFirstOrThrow({
+    where: { id: created.id },
+    include: loaded.authIncludeBase,
+  });
+
+  return { environment: loaded.toAuthenticated(row), environmentId: created.id };
+}
+
+describe("pause sweep wiring", () => {
+  postgresTest("pausing an environment sweeps after the limit is zeroed", async ({ prisma }) => {
+    calls.length = 0;
+    const loaded = await loadServices();
+    const { environment } = await seedEnv(loaded, prisma);
+
+    const result = await new loaded.PauseEnvironmentService(prisma).call(environment, "paused");
+
+    expect(result).toEqual({ success: true, state: "paused" });
+    expect(calls).toEqual(["updateEnvConcurrencyLimits:0", "sweepUnclaimedRuns:*"]);
+  });
+
+  postgresTest("resuming an environment does not sweep", async ({ prisma }) => {
+    const loaded = await loadServices();
+    const { environment, environmentId } = await seedEnv(loaded, prisma);
+
+    await prisma.runtimeEnvironment.update({
+      where: { id: environmentId },
+      data: { paused: true },
+    });
+
+    calls.length = 0;
+
+    const result = await new loaded.PauseEnvironmentService(prisma).call(environment, "resumed");
+
+    expect(result).toEqual({ success: true, state: "resumed" });
+    expect(calls).toEqual(["updateEnvConcurrencyLimits:undefined"]);
+  });
+});

apps/webapp/test/sweepUnclaimedRuns.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, it, vi } from "vitest";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+
+const returnUnclaimedMessagesToQueue = vi.fn();
+
+vi.mock("~/v3/runEngine.server", () => ({
+  engine: { returnUnclaimedMessagesToQueue },
+}));
+
+const environment = { id: "env_1234" } as AuthenticatedEnvironment;
+
+describe("sweepUnclaimedRuns", () => {
+  it("swallows a failing sweep so a pause that is already in force is not reported as failed", async () => {
+    const { sweepUnclaimedRuns } = await import("~/v3/runQueue.server");
+
+    returnUnclaimedMessagesToQueue.mockRejectedValueOnce(new Error("run queue unavailable"));
+
+    await expect(sweepUnclaimedRuns(environment)).resolves.toBeUndefined();
+    expect(returnUnclaimedMessagesToQueue).toHaveBeenCalledWith({ environment, queue: undefined });
+  });
+
+  it("passes the queue through when one is targeted", async () => {
+    const { sweepUnclaimedRuns } = await import("~/v3/runQueue.server");
+
+    returnUnclaimedMessagesToQueue.mockResolvedValueOnce({
+      returned: 2,
+      skipped: 0,
+      errors: 0,
+      passes: 1,
+    });
+
+    await sweepUnclaimedRuns(environment, "task/my-task");
+
+    expect(returnUnclaimedMessagesToQueue).toHaveBeenCalledWith({
+      environment,
+      queue: "task/my-task",
+    });
+  });
+});

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

@@ -49,7 +49,7 @@ import type {
   BatchCompletionCallback,
 } from "../batch-queue/types.js";
 import { FairQueueSelectionStrategy } from "../run-queue/fairQueueSelectionStrategy.js";
-import { RunQueue } from "../run-queue/index.js";
+import { RunQueue, type ReturnUnclaimedMessagesResult } from "../run-queue/index.js";
 import { RunQueueFullKeyProducer } from "../run-queue/keyProducer.js";
 import type { AuthenticatedEnvironment, MinimalAuthenticatedEnvironment } from "../shared/index.js";
 import { BillingCache } from "./billingCache.js";
@@ -1648,7 +1648,7 @@ export class RunEngine {
   }: {
     environment: MinimalAuthenticatedEnvironment;
     queue?: string;
-  }): Promise<{ returned: number; skipped: number }> {
+  }): Promise<ReturnUnclaimedMessagesResult> {
     return this.runQueue.returnUnclaimedMessagesToQueue({ env: environment, queue });
   }
 

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

@@ -140,9 +140,25 @@ type MarkedRun = {
 const RETURN_UNCLAIMED_BATCH_SIZE = 50;
 const READ_MESSAGES_BATCH_SIZE = 200;
 
+/**
+ * One retry covers the admission-to-RPUSH gap, which is a single round trip. `skipped` cannot
+ * distinguish that gap from a leaked concurrency slot, which never becomes claimable, so the
+ * pass count is kept low enough that a leak costs one short delay rather than a stall — these
+ * sweeps run per environment inside bulk pause loops.
+ */
+const RETURN_UNCLAIMED_MAX_PASSES = 2;
+const RETURN_UNCLAIMED_PASS_DELAY_MS = 50;
+const RETURN_UNCLAIMED_MAX_PASS_CEILING = 10;
+
 export type ReturnUnclaimedMessagesResult = {
+  /** Distinct runs put back on their queue. */
   returned: number;
-  skipped: number;
+  /**
+   * Candidates the final pass could not claim. Each is either a run a worker took first —
+   * which means it escaped the pause and is executing — or a leaked concurrency slot with no
+   * worker queue entry behind it. The two are indistinguishable from here.
+   */
+  skippedLastPass: number;
   errors: number;
   passes: number;
 };
@@ -1058,8 +1074,8 @@ export class RunQueue {
    *
    * 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
-   * is a candidate whose claim fails, so a pass reporting skipped runs is retried — by then
-   * the run has either landed on the worker queue (and is claimable) or been picked up by a
+   * is a candidate whose claim fails, so a pass that skips or errors is retried — by then the
+   * run has either landed on the worker queue (and is claimable) or been picked up by a
    * worker (and has left the candidate set).
    *
    * @param queue - Restrict to a single queue (including its concurrency key variants).
@@ -1068,8 +1084,8 @@ export class RunQueue {
   public async returnUnclaimedMessagesToQueue({
     env,
     queue,
-    maxPasses = 3,
-    passDelayMs = 250,
+    maxPasses = RETURN_UNCLAIMED_MAX_PASSES,
+    passDelayMs = RETURN_UNCLAIMED_PASS_DELAY_MS,
   }: {
     env: MinimalAuthenticatedEnvironment;
     queue?: string;
@@ -1080,30 +1096,39 @@ export class RunQueue {
       "returnUnclaimedMessagesToQueue",
       async (span) => {
         const targetBaseQueueKey = queue ? this.keys.queueKey(env, queue) : undefined;
+        const totalPasses = Number.isFinite(maxPasses)
+          ? Math.min(RETURN_UNCLAIMED_MAX_PASS_CEILING, Math.max(1, Math.floor(maxPasses)))
+          : RETURN_UNCLAIMED_MAX_PASSES;
 
-        let returned = 0;
-        let skipped = 0;
+        const returnedRunIds = new Set<string>();
+        let skippedLastPass = 0;
         let errors = 0;
         let passes = 0;
 
-        for (let pass = 0; pass < Math.max(1, maxPasses); pass++) {
+        for (let pass = 0; pass < totalPasses; pass++) {
           passes++;
 
           const passResult = await this.#returnUnclaimedMessagesPass(env, targetBaseQueueKey);
 
-          returned += passResult.returned;
+          for (const runId of passResult.returnedRunIds) {
+            returnedRunIds.add(runId);
+          }
           errors += passResult.errors;
-          skipped = passResult.skipped;
+          skippedLastPass = passResult.skipped;
 
-          if (passResult.skipped === 0 || pass === Math.max(1, maxPasses) - 1) {
+          const settled = passResult.skipped === 0 && passResult.errors === 0;
+
+          if (settled || pass === totalPasses - 1) {
             break;
           }
 
           await setTimeout(passDelayMs);
         }
 
+        const returned = returnedRunIds.size;
+
         span.setAttribute("returned_count", returned);
-        span.setAttribute("skipped_count", skipped);
+        span.setAttribute("skipped_last_pass", skippedLastPass);
         span.setAttribute("error_count", errors);
         span.setAttribute("passes", passes);
 
@@ -1112,12 +1137,12 @@ export class RunQueue {
           environmentId: env.id,
           queue,
           returned,
-          skipped,
+          skippedLastPass,
           errors,
           passes,
         });
 
-        return { returned, skipped, errors, passes };
+        return { returned, skippedLastPass, errors, passes };
       },
       {
         kind: SpanKind.INTERNAL,
@@ -1129,25 +1154,31 @@ export class RunQueue {
     );
   }
 
+  /**
+   * Candidates come from the environment-level sets even when a single queue is targeted.
+   * The per-queue sets would be cheaper but `ckIndex` only tracks concurrency key variants
+   * that still have pending messages, so a variant whose only runs are already in flight is
+   * not discoverable from it.
+   */
   async #returnUnclaimedMessagesPass(
     env: MinimalAuthenticatedEnvironment,
     targetBaseQueueKey: string | undefined
-  ): Promise<{ returned: number; skipped: number; errors: number }> {
+  ): Promise<{ returnedRunIds: string[]; skipped: number; errors: number }> {
     const unclaimedRunIds = await this.redis.sdiff(
       this.keys.envCurrentConcurrencyKey(env),
       this.keys.envCurrentDequeuedKey(env)
     );
 
     if (unclaimedRunIds.length === 0) {
-      return { returned: 0, skipped: 0, errors: 0 };
+      return { returnedRunIds: [], skipped: 0, errors: 0 };
     }
 
     const messages = (await this.#readMessages(env.organization.id, unclaimedRunIds)).filter(
       (message) =>
         !targetBaseQueueKey || this.keys.baseQueueKeyFromQueue(message.queue) === targetBaseQueueKey
     );
 
-    let returned = 0;
+    const returnedRunIds: string[] = [];
     let skipped = 0;
     let errors = 0;
 
@@ -1171,14 +1202,17 @@ export class RunQueue {
         }
 
         if (result.value) {
-          returned++;
+          const runId = batch[index]?.runId;
+          if (runId) {
+            returnedRunIds.push(runId);
+          }
         } else {
           skipped++;
         }
       }
     }
 
-    return { returned, skipped, errors };
+    return { returnedRunIds, skipped, errors };
   }
 
   async #readMessages(orgId: string, runIds: string[]): Promise<OutputPayload[]> {