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..304819f9f3 100644 --- a/apps/webapp/app/v3/runQueue.server.ts +++ b/apps/webapp/app/v3/runQueue.server.ts @@ -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 */ 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 af9edff856..e10f342743 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. 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); + } + ); +});