1+ import { RunEngine } from "@internal/run-engine" ;
12import { containerTest } from "@internal/testcontainers" ;
3+ import { trace } from "@opentelemetry/api" ;
24import { EnvironmentPauseSource , type PrismaClient } from "@trigger.dev/database" ;
35import type { RedisOptions } from "ioredis" ;
4- import { describe , expect , vi } from "vitest" ;
6+ import { describe , expect , onTestFinished , vi } from "vitest" ;
57import type { AuthenticatedEnvironment } from "~/services/apiAuth.server" ;
68import {
79 createRuntimeEnvironment ,
@@ -11,6 +13,51 @@ import {
1113
1214vi . setConfig ( { testTimeout : 60_000 } ) ;
1315
16+ // test/setup.ts stubs the app's engine singleton to a no-op, which would make any
17+ // assertion about the RunQueue limits vacuous. The tests that care about those limits
18+ // swap in a real RunEngine built on their own Redis container via `useEngine`; the
19+ // others keep the no-op.
20+ const { engineHolder } = vi . hoisted ( ( ) => ( {
21+ engineHolder : {
22+ current : { runQueue : { updateEnvConcurrencyLimits : async ( ) => undefined } } as any ,
23+ } ,
24+ } ) ) ;
25+
26+ vi . mock ( "~/v3/runEngine.server" , ( ) => ( {
27+ engine : new Proxy ( { } as Record < string , any > , {
28+ get : ( _target , prop ) => {
29+ const value = engineHolder . current [ prop as string ] ;
30+ return typeof value === "function" ? value . bind ( engineHolder . current ) : value ;
31+ } ,
32+ } ) ,
33+ } ) ) ;
34+
35+ function useEngine ( prisma : PrismaClient , redisOptions : RedisOptions ) {
36+ const engine = new RunEngine ( {
37+ prisma,
38+ worker : { redis : redisOptions , disabled : true } ,
39+ queue : { redis : redisOptions , masterQueueConsumersDisabled : true } ,
40+ runLock : { redis : redisOptions } ,
41+ machines : {
42+ defaultMachine : "small-1x" ,
43+ machines : {
44+ "small-1x" : { name : "small-1x" as const , cpu : 0.5 , memory : 0.5 , centsPerMs : 0.0001 } ,
45+ } ,
46+ baseCostInCents : 0.0001 ,
47+ } ,
48+ tracer : trace . getTracer ( "test" , "0.0.0" ) ,
49+ } ) ;
50+
51+ const previous = engineHolder . current ;
52+ engineHolder . current = engine ;
53+ onTestFinished ( async ( ) => {
54+ engineHolder . current = previous ;
55+ await engine . quit ( ) ;
56+ } ) ;
57+
58+ return engine ;
59+ }
60+
1461// The service's import chain reaches module-level singletons that throw at load
1562// time when REDIS_HOST/REDIS_PORT are unset (autoIncrementCounter via
1663// triggerTaskV1), so the env must point at the redis container BEFORE the
@@ -20,11 +67,16 @@ async function loadService(redisOptions: RedisOptions) {
2067 process . env . REDIS_HOST = redisOptions . host ;
2168 process . env . REDIS_PORT = String ( redisOptions . port ) ;
2269 process . env . REDIS_TLS_DISABLED = "true" ;
23- const [ { PauseEnvironmentService } , { authIncludeBase, toAuthenticated } ] = await Promise . all ( [
70+ const [
71+ { PauseEnvironmentService } ,
72+ { FinalizeDeploymentService } ,
73+ { authIncludeBase, toAuthenticated } ,
74+ ] = await Promise . all ( [
2475 import ( "~/v3/services/pauseEnvironment.server" ) ,
76+ import ( "~/v3/services/finalizeDeployment.server" ) ,
2577 import ( "~/models/runtimeEnvironment.server" ) ,
2678 ] ) ;
27- return { PauseEnvironmentService, authIncludeBase, toAuthenticated } ;
79+ return { PauseEnvironmentService, FinalizeDeploymentService , authIncludeBase, toAuthenticated } ;
2880}
2981
3082type Loaded = Awaited < ReturnType < typeof loadService > > ;
@@ -41,17 +93,119 @@ async function authEnv(
4193 return loaded . toAuthenticated ( row ) ;
4294}
4395
44- async function seedProductionEnv ( prisma : PrismaClient ) {
96+ async function seedProductionEnv ( prisma : PrismaClient , maximumConcurrencyLimit ?: number ) {
4597 const { organization, project } = await createTestOrgProjectWithMember ( prisma ) ;
4698 const environment = await createRuntimeEnvironment ( prisma , {
4799 projectId : project . id ,
48100 organizationId : organization . id ,
49101 type : "PRODUCTION" ,
50102 slug : uniqueId ( "prod" ) ,
51103 } ) ;
104+
105+ if ( maximumConcurrencyLimit !== undefined ) {
106+ await prisma . runtimeEnvironment . update ( {
107+ where : { id : environment . id } ,
108+ data : { maximumConcurrencyLimit } ,
109+ } ) ;
110+ }
111+
52112 return { organization, project, environment } ;
53113}
54114
115+ /** Runs a deploy through to DEPLOYED, the way the finalize deployment endpoint does. */
116+ async function finalizeADeployment (
117+ loaded : Loaded ,
118+ prisma : PrismaClient ,
119+ environment : AuthenticatedEnvironment
120+ ) {
121+ const version = uniqueId ( "2026.01.01" ) ;
122+ const worker = await prisma . backgroundWorker . create ( {
123+ data : {
124+ friendlyId : uniqueId ( "worker" ) ,
125+ contentHash : uniqueId ( "hash" ) ,
126+ projectId : environment . projectId ,
127+ runtimeEnvironmentId : environment . id ,
128+ version,
129+ metadata : { } ,
130+ engine : "V2" ,
131+ } ,
132+ } ) ;
133+
134+ const deployment = await prisma . workerDeployment . create ( {
135+ data : {
136+ friendlyId : uniqueId ( "deployment" ) ,
137+ contentHash : worker . contentHash ,
138+ shortCode : uniqueId ( "short" ) ,
139+ version,
140+ status : "DEPLOYING" ,
141+ imageReference : "registry.example.com/image:latest" ,
142+ projectId : environment . projectId ,
143+ environmentId : environment . id ,
144+ workerId : worker . id ,
145+ } ,
146+ } ) ;
147+
148+ const service = new loaded . FinalizeDeploymentService ( prisma ) ;
149+ await service . call ( environment , deployment . friendlyId , { skipPromotion : true } ) ;
150+ }
151+
152+ // Kept first in this file: the app's Redis-backed module singletons (the deploy path's
153+ // project pub/sub, for one) bind to the first container this file touches.
154+ describe ( "environment pause and the RunQueue env concurrency limit" , ( ) => {
155+ containerTest (
156+ "a finalized deployment does not resume a paused environment" ,
157+ async ( { prisma, redisOptions } ) => {
158+ const loaded = await loadService ( redisOptions ) ;
159+ const engine = useEngine ( prisma , redisOptions ) ;
160+
161+ const paused = await seedProductionEnv ( prisma , 17 ) ;
162+ const pausedEnv = await authEnv ( loaded , prisma , paused . environment . id ) ;
163+
164+ const pauseResult = await new loaded . PauseEnvironmentService ( prisma ) . call (
165+ pausedEnv ,
166+ "paused"
167+ ) ;
168+ expect ( pauseResult ) . toEqual ( { success : true , state : "paused" } ) ;
169+ expect ( await engine . runQueue . getEnvConcurrencyLimit ( pausedEnv ) ) . toBe ( 0 ) ;
170+
171+ // A deploy request authenticates first, so the deploy sees the env as it is now.
172+ await finalizeADeployment ( loaded , prisma , await authEnv ( loaded , prisma , pausedEnv . id ) ) ;
173+
174+ // The 0 limit is the only thing stopping dequeues, so a deploy must not push the
175+ // environment's real limit back into the queue while the env is still paused.
176+ expect ( await engine . runQueue . getEnvConcurrencyLimit ( pausedEnv ) ) . toBe ( 0 ) ;
177+ const after = await prisma . runtimeEnvironment . findFirstOrThrow ( {
178+ where : { id : paused . environment . id } ,
179+ } ) ;
180+ expect ( after . paused ) . toBe ( true ) ;
181+
182+ // Control for the assertion above: the same deploy path DOES push the real limit for
183+ // a running environment, so a limit of 0 can't just mean "the push never happened".
184+ const running = await seedProductionEnv ( prisma , 17 ) ;
185+ const runningEnv = await authEnv ( loaded , prisma , running . environment . id ) ;
186+ await finalizeADeployment ( loaded , prisma , runningEnv ) ;
187+ expect ( await engine . runQueue . getEnvConcurrencyLimit ( runningEnv ) ) . toBe ( 17 ) ;
188+ }
189+ ) ;
190+
191+ containerTest ( "resuming restores the environment limit" , async ( { prisma, redisOptions } ) => {
192+ const loaded = await loadService ( redisOptions ) ;
193+ const engine = useEngine ( prisma , redisOptions ) ;
194+
195+ const { environment } = await seedProductionEnv ( prisma , 17 ) ;
196+ const service = new loaded . PauseEnvironmentService ( prisma ) ;
197+ const env = await authEnv ( loaded , prisma , environment . id ) ;
198+
199+ expect ( await service . call ( env , "paused" ) ) . toEqual ( { success : true , state : "paused" } ) ;
200+ expect ( await engine . runQueue . getEnvConcurrencyLimit ( env ) ) . toBe ( 0 ) ;
201+
202+ // The service holds an environment read before the resume update, so its `paused` is
203+ // stale by the time the limit is pushed — resuming must still restore the real limit.
204+ expect ( await service . call ( env , "resumed" ) ) . toEqual ( { success : true , state : "resumed" } ) ;
205+ expect ( await engine . runQueue . getEnvConcurrencyLimit ( env ) ) . toBe ( 17 ) ;
206+ } ) ;
207+ } ) ;
208+
55209describe ( "PauseEnvironmentService" , ( ) => {
56210 containerTest (
57211 "resumes a manually paused env (pauseSource stays null through pause and resume)" ,
0 commit comments