fix(webapp): keep paused environments paused when concurrency limits are pushed (#4625)
<!-- ccr-slack-attribution --> _Requested by **Matt Aitken** · [Slack thread](https://triggerdotdev.slack.com/archives/C045W9WM3E1/p1786732623292829?thread_ts=1786732623.292829&cid=C045W9WM3E1)_ **Before:** you pause an environment, then a deploy lands (or a background worker is created, or an admin changes the concurrency/burst-factor). The environment starts picking up runs again even though the dashboard still shows it as paused. **After:** a paused environment stays paused until it is resumed, no matter what else pushes its concurrency limit. Pausing an environment sets `paused` in the database and writes a `0` env concurrency limit into the run queue — the `0` is the only thing that actually stops dequeueing. Any caller that pushed the limit without an explicit value (`finalizeDeployment`, `createBackgroundWorker`, the two admin environment routes) rewrote the real limit and silently un-paused the environment. ## ✅ Checklist - [x] I have followed every step in the [contributing guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md) - [x] The PR title follows the convention. - [x] I ran and tested the code works --- ## Testing `apps/webapp/test/pauseEnvironment.server.test.ts` gains two `containerTest` cases that wire a real `RunEngine` (real Redis) in place of the stubbed app singleton and assert the actual run-queue env limit: - pause a PRODUCTION env → limit is `0` → run the real `FinalizeDeploymentService` → limit is still `0`, plus a control on a running env in the same test proving that deploy path really does push the limit (so the `0` can't just mean "nothing happened"). - pause → resume → the real limit is restored, so the clamp can't regress resuming. Both cases fail on `main` (`expected 17 to be +0` and `expected +0 to be 17`) and pass with this change. `pnpm run typecheck --filter webapp` is clean. --- ## Changelog Fix paused environments starting to run work again after a deploy. --- ## How The clamp lives in the shared `updateEnvConcurrencyLimits` helper in `apps/webapp/app/v3/runQueue.server.ts`, so every present and future caller is covered: when no explicit limit is passed and the environment is paused, `0` is written instead of the stored maximum. An explicitly-passed limit still wins, which is what pausing itself relies on. The resume path now passes the post-update environment state (its in-memory copy was read before the un-pause and would otherwise be clamped back to `0`), and the helper no longer mutates the caller's environment object — that aliasing made a pause followed by a resume on the same object write `0` twice. The existing `!paused` guards in `allocateConcurrency` and the queue-level guard in `createBackgroundWorker` are left in place as defence in depth, and queue-level `TaskQueue.paused` behaviour is untouched. --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
---
|
||||
area: webapp
|
||||
type: fix
|
||||
---
|
||||
|
||||
Fix paused environments starting to run work again after a deploy: a paused environment now stays paused until you resume it.
|
||||
@@ -1,17 +1,34 @@
|
||||
import { type PrismaClientOrTransaction } from "@trigger.dev/database";
|
||||
import { prisma } from "~/db.server";
|
||||
import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import { engine } from "./runEngine.server";
|
||||
|
||||
/** Updates the RunQueue env concurrency limits */
|
||||
export async function updateEnvConcurrencyLimits(
|
||||
environment: AuthenticatedEnvironment,
|
||||
maximumConcurrencyLimit?: number
|
||||
maximumConcurrencyLimit?: number,
|
||||
db: PrismaClientOrTransaction = prisma
|
||||
) {
|
||||
let updatedEnvironment = environment;
|
||||
if (maximumConcurrencyLimit !== undefined) {
|
||||
updatedEnvironment.maximumConcurrencyLimit = maximumConcurrencyLimit;
|
||||
let limit = maximumConcurrencyLimit;
|
||||
|
||||
if (limit === undefined) {
|
||||
// A paused env is only enforced by a 0 limit in the RunQueue, so a push without an explicit
|
||||
// limit must not resurrect the real limit. Callers hold an environment read at auth time, so
|
||||
// resolve both values here instead of trusting it: a stale `paused: false` silently resumes a
|
||||
// paused env, and a stale `paused: true` strands a resumed one at 0 with nothing to restore it.
|
||||
const current = await db.runtimeEnvironment.findFirst({
|
||||
where: { id: environment.id },
|
||||
select: { paused: true, maximumConcurrencyLimit: true },
|
||||
});
|
||||
|
||||
const resolved = current ?? environment;
|
||||
limit = resolved.paused ? 0 : resolved.maximumConcurrencyLimit;
|
||||
}
|
||||
|
||||
await engine.runQueue.updateEnvConcurrencyLimits(updatedEnvironment);
|
||||
await engine.runQueue.updateEnvConcurrencyLimits({
|
||||
...environment,
|
||||
maximumConcurrencyLimit: limit,
|
||||
});
|
||||
}
|
||||
|
||||
/** Updates the RunQueue limits for a queue */
|
||||
|
||||
@@ -88,7 +88,7 @@ export class AllocateConcurrencyService extends BaseService {
|
||||
});
|
||||
|
||||
if (!updatedEnvironment.paused) {
|
||||
await updateEnvConcurrencyLimits(updatedEnvironment);
|
||||
await updateEnvConcurrencyLimits(updatedEnvironment, undefined, this._prisma);
|
||||
}
|
||||
|
||||
// Percent-based queue overrides follow the environment limit automatically. Note the
|
||||
|
||||
@@ -238,7 +238,7 @@ export class CreateBackgroundWorkerService extends BaseService {
|
||||
}
|
||||
|
||||
const [updateConcurrencyLimitsError] = await tryCatch(
|
||||
updateEnvConcurrencyLimits(environment)
|
||||
updateEnvConcurrencyLimits(environment, undefined, this._prisma)
|
||||
);
|
||||
|
||||
if (updateConcurrencyLimitsError) {
|
||||
|
||||
@@ -123,7 +123,7 @@ export class FinalizeDeploymentService extends BaseService {
|
||||
}
|
||||
);
|
||||
|
||||
await updateEnvConcurrencyLimits(authenticatedEnv);
|
||||
await updateEnvConcurrencyLimits(authenticatedEnv, undefined, this._prisma);
|
||||
} catch (err) {
|
||||
logger.error("Failed to publish WORKER_CREATED event", { err });
|
||||
}
|
||||
|
||||
@@ -118,7 +118,9 @@ export class PauseEnvironmentService extends WithRunEngine {
|
||||
logger.debug("PauseEnvironmentService: resuming environment", {
|
||||
environmentId: environment.id,
|
||||
});
|
||||
await updateEnvConcurrencyLimits(environment);
|
||||
// `environment` was read before the update above, so its `paused` is stale. The helper
|
||||
// resolves the current state itself - hand it the client that wrote the resume.
|
||||
await updateEnvConcurrencyLimits(environment, undefined, this._prisma);
|
||||
}
|
||||
} catch (error) {
|
||||
await this._prisma.runtimeEnvironment.update({
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
import { RunEngine } from "@internal/run-engine";
|
||||
import { containerTest } from "@internal/testcontainers";
|
||||
import { trace } from "@opentelemetry/api";
|
||||
import type { PrismaClient } from "@trigger.dev/database";
|
||||
import type { RedisOptions } from "ioredis";
|
||||
import { describe, expect, onTestFinished, vi } from "vitest";
|
||||
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import {
|
||||
createRuntimeEnvironment,
|
||||
createTestOrgProjectWithMember,
|
||||
uniqueId,
|
||||
} from "./fixtures/environmentVariablesFixtures";
|
||||
|
||||
vi.setConfig({ testTimeout: 60_000, hookTimeout: 60_000 });
|
||||
|
||||
// test/setup.ts replaces the app's engine singleton with a no-op for every webapp suite, which
|
||||
// would make any assertion about the RunQueue limits vacuous. Every test in this file asserts on
|
||||
// real RunQueue state, so put a real RunEngine - built on the test's own Redis container - back
|
||||
// behind the singleton. No test here uses the no-op default.
|
||||
const { engineHolder } = vi.hoisted(() => ({
|
||||
engineHolder: { current: undefined as any },
|
||||
}));
|
||||
|
||||
vi.mock("~/v3/runEngine.server", () => ({
|
||||
engine: new Proxy({} as Record<string, any>, {
|
||||
get: (_target, prop) => engineHolder.current?.[prop as string],
|
||||
}),
|
||||
}));
|
||||
|
||||
function useEngine(prisma: PrismaClient, redisOptions: RedisOptions) {
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
worker: { redis: redisOptions, disabled: true },
|
||||
queue: { redis: redisOptions, masterQueueConsumersDisabled: true },
|
||||
runLock: { redis: redisOptions },
|
||||
machines: {
|
||||
defaultMachine: "small-1x",
|
||||
machines: {
|
||||
"small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
|
||||
},
|
||||
baseCostInCents: 0.0001,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
|
||||
engineHolder.current = engine;
|
||||
onTestFinished(async () => {
|
||||
engineHolder.current = undefined;
|
||||
await engine.quit();
|
||||
});
|
||||
|
||||
return engine;
|
||||
}
|
||||
|
||||
// The import chain reaches module-level singletons that throw at load time when
|
||||
// REDIS_HOST/REDIS_PORT are unset (autoIncrementCounter via triggerTaskV1), so the env must point
|
||||
// at the redis container BEFORE the modules are imported. Hence dynamic imports; vitest runs each
|
||||
// file in its own fork, so the env mutation cannot leak into other suites.
|
||||
async function loadServices(redisOptions: RedisOptions) {
|
||||
process.env.REDIS_HOST = redisOptions.host;
|
||||
process.env.REDIS_PORT = String(redisOptions.port);
|
||||
process.env.REDIS_TLS_DISABLED = "true";
|
||||
const [{ updateEnvConcurrencyLimits }, { PauseEnvironmentService }, runtimeEnvironment] =
|
||||
await Promise.all([
|
||||
import("~/v3/runQueue.server"),
|
||||
import("~/v3/services/pauseEnvironment.server"),
|
||||
import("~/models/runtimeEnvironment.server"),
|
||||
]);
|
||||
return {
|
||||
updateEnvConcurrencyLimits,
|
||||
PauseEnvironmentService,
|
||||
authIncludeBase: runtimeEnvironment.authIncludeBase,
|
||||
toAuthenticated: runtimeEnvironment.toAuthenticated,
|
||||
};
|
||||
}
|
||||
|
||||
type Loaded = Awaited<ReturnType<typeof loadServices>>;
|
||||
|
||||
async function authEnv(
|
||||
loaded: Loaded,
|
||||
prisma: PrismaClient,
|
||||
environmentId: string
|
||||
): Promise<AuthenticatedEnvironment> {
|
||||
const row = await prisma.runtimeEnvironment.findFirstOrThrow({
|
||||
where: { id: environmentId },
|
||||
include: loaded.authIncludeBase,
|
||||
});
|
||||
return loaded.toAuthenticated(row);
|
||||
}
|
||||
|
||||
async function seedProductionEnv(prisma: PrismaClient, maximumConcurrencyLimit: number) {
|
||||
const { organization, project } = await createTestOrgProjectWithMember(prisma);
|
||||
const environment = await createRuntimeEnvironment(prisma, {
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
type: "PRODUCTION",
|
||||
slug: uniqueId("prod"),
|
||||
});
|
||||
|
||||
await prisma.runtimeEnvironment.update({
|
||||
where: { id: environment.id },
|
||||
data: { maximumConcurrencyLimit },
|
||||
});
|
||||
|
||||
return { organization, project, environment };
|
||||
}
|
||||
|
||||
// An unset RunQueue limit reads back as the engine default (10), so neither the 0 nor the 17
|
||||
// assertions below can pass just because a push never happened.
|
||||
describe("updateEnvConcurrencyLimits", () => {
|
||||
containerTest(
|
||||
"clamps to 0 when the environment is paused, even though the caller's copy says otherwise",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const loaded = await loadServices(redisOptions);
|
||||
const engine = useEngine(prisma, redisOptions);
|
||||
|
||||
const { environment } = await seedProductionEnv(prisma, 17);
|
||||
// What an argument-less caller holds: an environment read when the request authenticated,
|
||||
// before the pause landed (finalizing a deployment, registering a background worker).
|
||||
const atAuthTime = await authEnv(loaded, prisma, environment.id);
|
||||
expect(atAuthTime.paused).toBe(false);
|
||||
|
||||
await prisma.runtimeEnvironment.update({
|
||||
where: { id: environment.id },
|
||||
data: { paused: true },
|
||||
});
|
||||
|
||||
await loaded.updateEnvConcurrencyLimits(atAuthTime, undefined, prisma);
|
||||
|
||||
// The 0 limit is the only thing stopping dequeues, so the real limit must not go back in.
|
||||
expect(await engine.runQueue.getEnvConcurrencyLimit(atAuthTime)).toBe(0);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"pushes the real limit for a running environment",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const loaded = await loadServices(redisOptions);
|
||||
const engine = useEngine(prisma, redisOptions);
|
||||
|
||||
const { environment } = await seedProductionEnv(prisma, 17);
|
||||
const env = await authEnv(loaded, prisma, environment.id);
|
||||
|
||||
await loaded.updateEnvConcurrencyLimits(env, undefined, prisma);
|
||||
|
||||
expect(await engine.runQueue.getEnvConcurrencyLimit(env)).toBe(17);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"restores the real limit when the environment was resumed while the request was in flight",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const loaded = await loadServices(redisOptions);
|
||||
const engine = useEngine(prisma, redisOptions);
|
||||
|
||||
const { environment } = await seedProductionEnv(prisma, 17);
|
||||
await prisma.runtimeEnvironment.update({
|
||||
where: { id: environment.id },
|
||||
data: { paused: true },
|
||||
});
|
||||
|
||||
// Captured while paused, then resumed before the push. Trusting this copy would write 0 over
|
||||
// the restored limit and leave the env stalled with `paused: false` and nothing to fix it.
|
||||
const whilePaused = await authEnv(loaded, prisma, environment.id);
|
||||
expect(whilePaused.paused).toBe(true);
|
||||
|
||||
await prisma.runtimeEnvironment.update({
|
||||
where: { id: environment.id },
|
||||
data: { paused: false },
|
||||
});
|
||||
|
||||
await loaded.updateEnvConcurrencyLimits(whilePaused, undefined, prisma);
|
||||
|
||||
expect(await engine.runQueue.getEnvConcurrencyLimit(whilePaused)).toBe(17);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"an explicit limit wins over the stored pause state",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const loaded = await loadServices(redisOptions);
|
||||
const engine = useEngine(prisma, redisOptions);
|
||||
|
||||
const { environment } = await seedProductionEnv(prisma, 17);
|
||||
await prisma.runtimeEnvironment.update({
|
||||
where: { id: environment.id },
|
||||
data: { paused: true },
|
||||
});
|
||||
const env = await authEnv(loaded, prisma, environment.id);
|
||||
|
||||
// How billing-limit converge restores a limit as it unpauses: the caller decides, no read.
|
||||
await loaded.updateEnvConcurrencyLimits(env, 9, prisma);
|
||||
|
||||
expect(await engine.runQueue.getEnvConcurrencyLimit(env)).toBe(9);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"a pause writes 0 and a resume restores the limit",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const loaded = await loadServices(redisOptions);
|
||||
const engine = useEngine(prisma, redisOptions);
|
||||
|
||||
const { environment } = await seedProductionEnv(prisma, 17);
|
||||
const service = new loaded.PauseEnvironmentService(prisma);
|
||||
const env = await authEnv(loaded, prisma, environment.id);
|
||||
|
||||
expect(await service.call(env, "paused")).toEqual({ success: true, state: "paused" });
|
||||
expect(await engine.runQueue.getEnvConcurrencyLimit(env)).toBe(0);
|
||||
|
||||
// The service holds an environment read before its own resume update, so `env.paused` is
|
||||
// stale here too.
|
||||
expect(await service.call(env, "resumed")).toEqual({ success: true, state: "resumed" });
|
||||
expect(await engine.runQueue.getEnvConcurrencyLimit(env)).toBe(17);
|
||||
}
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user