From ee3c3584ec2367a27997299f2d098e80bbc02a51 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 19:01:34 +0000 Subject: [PATCH 1/2] fix(webapp): keep paused environments paused when concurrency limits are pushed `updateEnvConcurrencyLimits` is the only enforcement of an environment pause: pausing writes a 0 env concurrency limit to the run queue, which is what stops dequeueing. Callers that push the limit without an explicit value (finalizing a deployment, creating a background worker, the admin concurrency/burst-factor routes) rewrote the real limit, silently resuming an environment the dashboard still showed as paused. Clamp the pushed limit to 0 when the environment is paused and no explicit limit is given, so every caller is covered. An explicit limit still wins. Resuming now passes the post-update state, and the helper no longer mutates the caller's environment object (which made a pause + resume on the same object write 0 twice). Co-Authored-By: Claude --- ...d-environment-stays-paused-after-deploy.md | 6 + apps/webapp/app/v3/runQueue.server.ts | 13 +- .../v3/services/pauseEnvironment.server.ts | 4 +- .../test/pauseEnvironment.server.test.ts | 162 +++++++++++++++++- 4 files changed, 175 insertions(+), 10 deletions(-) create mode 100644 .server-changes/paused-environment-stays-paused-after-deploy.md diff --git a/.server-changes/paused-environment-stays-paused-after-deploy.md b/.server-changes/paused-environment-stays-paused-after-deploy.md new file mode 100644 index 0000000000..4c35b6e215 --- /dev/null +++ b/.server-changes/paused-environment-stays-paused-after-deploy.md @@ -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. diff --git a/apps/webapp/app/v3/runQueue.server.ts b/apps/webapp/app/v3/runQueue.server.ts index 0ff28fa088..6574c0ddf0 100644 --- a/apps/webapp/app/v3/runQueue.server.ts +++ b/apps/webapp/app/v3/runQueue.server.ts @@ -6,12 +6,15 @@ export async function updateEnvConcurrencyLimits( environment: AuthenticatedEnvironment, maximumConcurrencyLimit?: number ) { - let updatedEnvironment = environment; - if (maximumConcurrencyLimit !== undefined) { - updatedEnvironment.maximumConcurrencyLimit = maximumConcurrencyLimit; - } + // A paused env is only enforced by a 0 limit in the RunQueue, so a push without an explicit + // limit has to stay 0 — otherwise it silently resumes an env the dashboard still shows as paused. + const limit = + maximumConcurrencyLimit ?? (environment.paused ? 0 : environment.maximumConcurrencyLimit); - await engine.runQueue.updateEnvConcurrencyLimits(updatedEnvironment); + await engine.runQueue.updateEnvConcurrencyLimits({ + ...environment, + maximumConcurrencyLimit: limit, + }); } /** Updates the RunQueue limits for a queue */ diff --git a/apps/webapp/app/v3/services/pauseEnvironment.server.ts b/apps/webapp/app/v3/services/pauseEnvironment.server.ts index af9edff856..7efa650969 100644 --- a/apps/webapp/app/v3/services/pauseEnvironment.server.ts +++ b/apps/webapp/app/v3/services/pauseEnvironment.server.ts @@ -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: pass the + // resumed state or the helper would clamp the limit back to 0. + await updateEnvConcurrencyLimits({ ...environment, paused: false }); } } catch (error) { await this._prisma.runtimeEnvironment.update({ diff --git a/apps/webapp/test/pauseEnvironment.server.test.ts b/apps/webapp/test/pauseEnvironment.server.test.ts index ea31560264..fecfacfbe6 100644 --- a/apps/webapp/test/pauseEnvironment.server.test.ts +++ b/apps/webapp/test/pauseEnvironment.server.test.ts @@ -1,7 +1,9 @@ +import { RunEngine } from "@internal/run-engine"; import { containerTest } from "@internal/testcontainers"; +import { trace } from "@opentelemetry/api"; import { EnvironmentPauseSource, type PrismaClient } from "@trigger.dev/database"; import type { RedisOptions } from "ioredis"; -import { describe, expect, vi } from "vitest"; +import { describe, expect, onTestFinished, vi } from "vitest"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { createRuntimeEnvironment, @@ -11,6 +13,51 @@ import { vi.setConfig({ testTimeout: 60_000 }); +// test/setup.ts stubs the app's engine singleton to a no-op, which would make any +// assertion about the RunQueue limits vacuous. The tests that care about those limits +// swap in a real RunEngine built on their own Redis container via `useEngine`; the +// others keep the no-op. +const { engineHolder } = vi.hoisted(() => ({ + engineHolder: { + current: { runQueue: { updateEnvConcurrencyLimits: async () => undefined } } as any, + }, +})); + +vi.mock("~/v3/runEngine.server", () => ({ + engine: new Proxy({} as Record, { + get: (_target, prop) => { + const value = engineHolder.current[prop as string]; + return typeof value === "function" ? value.bind(engineHolder.current) : value; + }, + }), +})); + +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"), + }); + + const previous = engineHolder.current; + engineHolder.current = engine; + onTestFinished(async () => { + engineHolder.current = previous; + await engine.quit(); + }); + + return engine; +} + // The service's 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 @@ -20,11 +67,16 @@ async function loadService(redisOptions: RedisOptions) { process.env.REDIS_HOST = redisOptions.host; process.env.REDIS_PORT = String(redisOptions.port); process.env.REDIS_TLS_DISABLED = "true"; - const [{ PauseEnvironmentService }, { authIncludeBase, toAuthenticated }] = await Promise.all([ + const [ + { PauseEnvironmentService }, + { FinalizeDeploymentService }, + { authIncludeBase, toAuthenticated }, + ] = await Promise.all([ import("~/v3/services/pauseEnvironment.server"), + import("~/v3/services/finalizeDeployment.server"), import("~/models/runtimeEnvironment.server"), ]); - return { PauseEnvironmentService, authIncludeBase, toAuthenticated }; + return { PauseEnvironmentService, FinalizeDeploymentService, authIncludeBase, toAuthenticated }; } type Loaded = Awaited>; @@ -41,7 +93,7 @@ async function authEnv( return loaded.toAuthenticated(row); } -async function seedProductionEnv(prisma: PrismaClient) { +async function seedProductionEnv(prisma: PrismaClient, maximumConcurrencyLimit?: number) { const { organization, project } = await createTestOrgProjectWithMember(prisma); const environment = await createRuntimeEnvironment(prisma, { projectId: project.id, @@ -49,9 +101,111 @@ async function seedProductionEnv(prisma: PrismaClient) { type: "PRODUCTION", slug: uniqueId("prod"), }); + + if (maximumConcurrencyLimit !== undefined) { + await prisma.runtimeEnvironment.update({ + where: { id: environment.id }, + data: { maximumConcurrencyLimit }, + }); + } + return { organization, project, environment }; } +/** Runs a deploy through to DEPLOYED, the way the finalize deployment endpoint does. */ +async function finalizeADeployment( + loaded: Loaded, + prisma: PrismaClient, + environment: AuthenticatedEnvironment +) { + const version = uniqueId("2026.01.01"); + const worker = await prisma.backgroundWorker.create({ + data: { + friendlyId: uniqueId("worker"), + contentHash: uniqueId("hash"), + projectId: environment.projectId, + runtimeEnvironmentId: environment.id, + version, + metadata: {}, + engine: "V2", + }, + }); + + const deployment = await prisma.workerDeployment.create({ + data: { + friendlyId: uniqueId("deployment"), + contentHash: worker.contentHash, + shortCode: uniqueId("short"), + version, + status: "DEPLOYING", + imageReference: "registry.example.com/image:latest", + projectId: environment.projectId, + environmentId: environment.id, + workerId: worker.id, + }, + }); + + const service = new loaded.FinalizeDeploymentService(prisma); + await service.call(environment, deployment.friendlyId, { skipPromotion: true }); +} + +// Kept first in this file: the app's Redis-backed module singletons (the deploy path's +// project pub/sub, for one) bind to the first container this file touches. +describe("environment pause and the RunQueue env concurrency limit", () => { + containerTest( + "a finalized deployment does not resume a paused environment", + async ({ prisma, redisOptions }) => { + const loaded = await loadService(redisOptions); + const engine = useEngine(prisma, redisOptions); + + const paused = await seedProductionEnv(prisma, 17); + const pausedEnv = await authEnv(loaded, prisma, paused.environment.id); + + const pauseResult = await new loaded.PauseEnvironmentService(prisma).call( + pausedEnv, + "paused" + ); + expect(pauseResult).toEqual({ success: true, state: "paused" }); + expect(await engine.runQueue.getEnvConcurrencyLimit(pausedEnv)).toBe(0); + + // A deploy request authenticates first, so the deploy sees the env as it is now. + await finalizeADeployment(loaded, prisma, await authEnv(loaded, prisma, pausedEnv.id)); + + // The 0 limit is the only thing stopping dequeues, so a deploy must not push the + // environment's real limit back into the queue while the env is still paused. + expect(await engine.runQueue.getEnvConcurrencyLimit(pausedEnv)).toBe(0); + const after = await prisma.runtimeEnvironment.findFirstOrThrow({ + where: { id: paused.environment.id }, + }); + expect(after.paused).toBe(true); + + // Control for the assertion above: the same deploy path DOES push the real limit for + // a running environment, so a limit of 0 can't just mean "the push never happened". + const running = await seedProductionEnv(prisma, 17); + const runningEnv = await authEnv(loaded, prisma, running.environment.id); + await finalizeADeployment(loaded, prisma, runningEnv); + expect(await engine.runQueue.getEnvConcurrencyLimit(runningEnv)).toBe(17); + } + ); + + containerTest("resuming restores the environment limit", async ({ prisma, redisOptions }) => { + const loaded = await loadService(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 the resume update, so its `paused` is + // stale by the time the limit is pushed — resuming must still restore the real limit. + expect(await service.call(env, "resumed")).toEqual({ success: true, state: "resumed" }); + expect(await engine.runQueue.getEnvConcurrencyLimit(env)).toBe(17); + }); +}); + describe("PauseEnvironmentService", () => { containerTest( "resumes a manually paused env (pauseSource stays null through pause and resume)", From 65b215208eac0b8acd8fa160a845a836a1ebc0a1 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 19:31:29 +0000 Subject: [PATCH 2/2] fix(webapp): resolve the pause state from the DB when pushing env concurrency limits The clamp added in the previous commit read `paused` off the environment object the caller passed, which is captured when the request authenticates and can be read from a replica. A stale `paused: false` still silently resumes a paused environment, and a stale `paused: true` writes 0 over a limit that was just restored - leaving the environment stalled with `paused: false` and nothing to put the limit back. Resolve `paused` and `maximumConcurrencyLimit` for the environment id inside `updateEnvConcurrencyLimits` when no explicit limit is given, and let callers hand in the client they wrote with so the read is primary- and transaction-consistent. The explicit-limit path is unchanged: pausing and billing-limit converge still decide the value with no read. Move the RunQueue limit cases into their own test file so the pre-existing suite keeps the shared engine stub instead of a file-level mock, and cover the resumed-while-in-flight ordering. Co-Authored-By: Claude --- apps/webapp/app/v3/runQueue.server.ts | 24 +- .../v3/services/allocateConcurrency.server.ts | 2 +- .../services/createBackgroundWorker.server.ts | 2 +- .../v3/services/finalizeDeployment.server.ts | 2 +- .../v3/services/pauseEnvironment.server.ts | 6 +- .../envConcurrencyLimitPause.server.test.ts | 217 ++++++++++++++++++ .../test/pauseEnvironment.server.test.ts | 162 +------------ 7 files changed, 246 insertions(+), 169 deletions(-) create mode 100644 apps/webapp/test/envConcurrencyLimitPause.server.test.ts diff --git a/apps/webapp/app/v3/runQueue.server.ts b/apps/webapp/app/v3/runQueue.server.ts index 6574c0ddf0..304819f9f3 100644 --- a/apps/webapp/app/v3/runQueue.server.ts +++ b/apps/webapp/app/v3/runQueue.server.ts @@ -1,15 +1,29 @@ +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 ) { - // A paused env is only enforced by a 0 limit in the RunQueue, so a push without an explicit - // limit has to stay 0 — otherwise it silently resumes an env the dashboard still shows as paused. - const limit = - maximumConcurrencyLimit ?? (environment.paused ? 0 : environment.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({ ...environment, diff --git a/apps/webapp/app/v3/services/allocateConcurrency.server.ts b/apps/webapp/app/v3/services/allocateConcurrency.server.ts index 78c8b82915..24aebfefcf 100644 --- a/apps/webapp/app/v3/services/allocateConcurrency.server.ts +++ b/apps/webapp/app/v3/services/allocateConcurrency.server.ts @@ -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 diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index f73f177270..d84247b634 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -238,7 +238,7 @@ export class CreateBackgroundWorkerService extends BaseService { } const [updateConcurrencyLimitsError] = await tryCatch( - updateEnvConcurrencyLimits(environment) + updateEnvConcurrencyLimits(environment, undefined, this._prisma) ); if (updateConcurrencyLimitsError) { diff --git a/apps/webapp/app/v3/services/finalizeDeployment.server.ts b/apps/webapp/app/v3/services/finalizeDeployment.server.ts index 98ef7a73bd..fd559aff30 100644 --- a/apps/webapp/app/v3/services/finalizeDeployment.server.ts +++ b/apps/webapp/app/v3/services/finalizeDeployment.server.ts @@ -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 }); } diff --git a/apps/webapp/app/v3/services/pauseEnvironment.server.ts b/apps/webapp/app/v3/services/pauseEnvironment.server.ts index 7efa650969..e10f342743 100644 --- a/apps/webapp/app/v3/services/pauseEnvironment.server.ts +++ b/apps/webapp/app/v3/services/pauseEnvironment.server.ts @@ -118,9 +118,9 @@ export class PauseEnvironmentService extends WithRunEngine { logger.debug("PauseEnvironmentService: resuming environment", { environmentId: environment.id, }); - // `environment` was read before the update above, so its `paused` is stale: pass the - // resumed state or the helper would clamp the limit back to 0. - await updateEnvConcurrencyLimits({ ...environment, paused: false }); + // `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({ diff --git a/apps/webapp/test/envConcurrencyLimitPause.server.test.ts b/apps/webapp/test/envConcurrencyLimitPause.server.test.ts new file mode 100644 index 0000000000..7bb67c8134 --- /dev/null +++ b/apps/webapp/test/envConcurrencyLimitPause.server.test.ts @@ -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, { + 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>; + +async function authEnv( + loaded: Loaded, + prisma: PrismaClient, + environmentId: string +): Promise { + 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); + } + ); +}); diff --git a/apps/webapp/test/pauseEnvironment.server.test.ts b/apps/webapp/test/pauseEnvironment.server.test.ts index fecfacfbe6..ea31560264 100644 --- a/apps/webapp/test/pauseEnvironment.server.test.ts +++ b/apps/webapp/test/pauseEnvironment.server.test.ts @@ -1,9 +1,7 @@ -import { RunEngine } from "@internal/run-engine"; import { containerTest } from "@internal/testcontainers"; -import { trace } from "@opentelemetry/api"; import { EnvironmentPauseSource, type PrismaClient } from "@trigger.dev/database"; import type { RedisOptions } from "ioredis"; -import { describe, expect, onTestFinished, vi } from "vitest"; +import { describe, expect, vi } from "vitest"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { createRuntimeEnvironment, @@ -13,51 +11,6 @@ import { vi.setConfig({ testTimeout: 60_000 }); -// test/setup.ts stubs the app's engine singleton to a no-op, which would make any -// assertion about the RunQueue limits vacuous. The tests that care about those limits -// swap in a real RunEngine built on their own Redis container via `useEngine`; the -// others keep the no-op. -const { engineHolder } = vi.hoisted(() => ({ - engineHolder: { - current: { runQueue: { updateEnvConcurrencyLimits: async () => undefined } } as any, - }, -})); - -vi.mock("~/v3/runEngine.server", () => ({ - engine: new Proxy({} as Record, { - get: (_target, prop) => { - const value = engineHolder.current[prop as string]; - return typeof value === "function" ? value.bind(engineHolder.current) : value; - }, - }), -})); - -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"), - }); - - const previous = engineHolder.current; - engineHolder.current = engine; - onTestFinished(async () => { - engineHolder.current = previous; - await engine.quit(); - }); - - return engine; -} - // The service's 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 @@ -67,16 +20,11 @@ async function loadService(redisOptions: RedisOptions) { process.env.REDIS_HOST = redisOptions.host; process.env.REDIS_PORT = String(redisOptions.port); process.env.REDIS_TLS_DISABLED = "true"; - const [ - { PauseEnvironmentService }, - { FinalizeDeploymentService }, - { authIncludeBase, toAuthenticated }, - ] = await Promise.all([ + const [{ PauseEnvironmentService }, { authIncludeBase, toAuthenticated }] = await Promise.all([ import("~/v3/services/pauseEnvironment.server"), - import("~/v3/services/finalizeDeployment.server"), import("~/models/runtimeEnvironment.server"), ]); - return { PauseEnvironmentService, FinalizeDeploymentService, authIncludeBase, toAuthenticated }; + return { PauseEnvironmentService, authIncludeBase, toAuthenticated }; } type Loaded = Awaited>; @@ -93,7 +41,7 @@ async function authEnv( return loaded.toAuthenticated(row); } -async function seedProductionEnv(prisma: PrismaClient, maximumConcurrencyLimit?: number) { +async function seedProductionEnv(prisma: PrismaClient) { const { organization, project } = await createTestOrgProjectWithMember(prisma); const environment = await createRuntimeEnvironment(prisma, { projectId: project.id, @@ -101,111 +49,9 @@ async function seedProductionEnv(prisma: PrismaClient, maximumConcurrencyLimit?: type: "PRODUCTION", slug: uniqueId("prod"), }); - - if (maximumConcurrencyLimit !== undefined) { - await prisma.runtimeEnvironment.update({ - where: { id: environment.id }, - data: { maximumConcurrencyLimit }, - }); - } - return { organization, project, environment }; } -/** Runs a deploy through to DEPLOYED, the way the finalize deployment endpoint does. */ -async function finalizeADeployment( - loaded: Loaded, - prisma: PrismaClient, - environment: AuthenticatedEnvironment -) { - const version = uniqueId("2026.01.01"); - const worker = await prisma.backgroundWorker.create({ - data: { - friendlyId: uniqueId("worker"), - contentHash: uniqueId("hash"), - projectId: environment.projectId, - runtimeEnvironmentId: environment.id, - version, - metadata: {}, - engine: "V2", - }, - }); - - const deployment = await prisma.workerDeployment.create({ - data: { - friendlyId: uniqueId("deployment"), - contentHash: worker.contentHash, - shortCode: uniqueId("short"), - version, - status: "DEPLOYING", - imageReference: "registry.example.com/image:latest", - projectId: environment.projectId, - environmentId: environment.id, - workerId: worker.id, - }, - }); - - const service = new loaded.FinalizeDeploymentService(prisma); - await service.call(environment, deployment.friendlyId, { skipPromotion: true }); -} - -// Kept first in this file: the app's Redis-backed module singletons (the deploy path's -// project pub/sub, for one) bind to the first container this file touches. -describe("environment pause and the RunQueue env concurrency limit", () => { - containerTest( - "a finalized deployment does not resume a paused environment", - async ({ prisma, redisOptions }) => { - const loaded = await loadService(redisOptions); - const engine = useEngine(prisma, redisOptions); - - const paused = await seedProductionEnv(prisma, 17); - const pausedEnv = await authEnv(loaded, prisma, paused.environment.id); - - const pauseResult = await new loaded.PauseEnvironmentService(prisma).call( - pausedEnv, - "paused" - ); - expect(pauseResult).toEqual({ success: true, state: "paused" }); - expect(await engine.runQueue.getEnvConcurrencyLimit(pausedEnv)).toBe(0); - - // A deploy request authenticates first, so the deploy sees the env as it is now. - await finalizeADeployment(loaded, prisma, await authEnv(loaded, prisma, pausedEnv.id)); - - // The 0 limit is the only thing stopping dequeues, so a deploy must not push the - // environment's real limit back into the queue while the env is still paused. - expect(await engine.runQueue.getEnvConcurrencyLimit(pausedEnv)).toBe(0); - const after = await prisma.runtimeEnvironment.findFirstOrThrow({ - where: { id: paused.environment.id }, - }); - expect(after.paused).toBe(true); - - // Control for the assertion above: the same deploy path DOES push the real limit for - // a running environment, so a limit of 0 can't just mean "the push never happened". - const running = await seedProductionEnv(prisma, 17); - const runningEnv = await authEnv(loaded, prisma, running.environment.id); - await finalizeADeployment(loaded, prisma, runningEnv); - expect(await engine.runQueue.getEnvConcurrencyLimit(runningEnv)).toBe(17); - } - ); - - containerTest("resuming restores the environment limit", async ({ prisma, redisOptions }) => { - const loaded = await loadService(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 the resume update, so its `paused` is - // stale by the time the limit is pushed — resuming must still restore the real limit. - expect(await service.call(env, "resumed")).toEqual({ success: true, state: "resumed" }); - expect(await engine.runQueue.getEnvConcurrencyLimit(env)).toBe(17); - }); -}); - describe("PauseEnvironmentService", () => { containerTest( "resumes a manually paused env (pauseSource stays null through pause and resume)",