Skip to content

Commit 01ed8f3

Browse files
jingerpieclaude
andcommitted
feat(sdk,core,webapp): let tasks declare the region(s) they may run in
Adds a task-level `region` option, a single region or a list, so a task definition can restrict where its runs execute instead of relying on every trigger call site passing `{ region }` or on the project default. Placement at trigger time: - a `region` passed when triggering must be in the task's list, otherwise the trigger is rejected with a 400 naming the allowed regions - with no trigger-time region, the project's default region is used when the task allows it, otherwise the first region in the list - tasks without `region` keep today's behaviour; DEV ignores it, matching the existing per-trigger option - the existing access checks (hidden groups, project allowlists, MICROVM access, other projects' UNMANAGED groups) still apply to the picked region Deploys now fail early with a message naming the task and region when a listed region doesn't exist or isn't available to the project. The SDK normalises `region` into a `regions` list in the task manifest, which is persisted as a text-array column on BackgroundWorkerTask and rides on the cached task-metadata entry, so the trigger hot path enforces it with no additional query. Pre-existing cache entries decode as unconstrained. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 8b72e6c commit 01ed8f3

22 files changed

Lines changed: 1157 additions & 53 deletions

File tree

‎.changeset/task-level-region.md‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
---
2+
"@trigger.dev/sdk": patch
3+
"@trigger.dev/core": patch
4+
"trigger.dev": patch
5+
---
6+
7+
Tasks can now declare the region(s) they may run in with `region: "eu-central-1"` or `region: ["eu-central-1", "us-east-1"]` in the task definition. A `region` passed when triggering must be one of them, otherwise the trigger is rejected. When no region is passed, the project's default region is used if the task allows it, otherwise the first region in the list. Deploys fail early if a task names a region that doesn't exist or isn't available to the project. The option is ignored in dev.
8+
9+
```ts
10+
export const euOnlyTask = task({
11+
id: "eu-only-task",
12+
region: "eu-central-1",
13+
run: async (payload) => {
14+
// ...
15+
},
16+
});
17+
```

‎apps/webapp/app/runEngine/concerns/queues.server.ts‎

Lines changed: 45 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,12 @@ import type {
99
QueueProperties,
1010
QueueValidationResult,
1111
TriggerTaskRequest,
12+
WorkerQueueOptions,
1213
} from "../types";
13-
import { WorkerGroupService } from "~/v3/services/worker/workerGroupService.server";
14+
import {
15+
RegionNotAllowedForTaskError,
16+
WorkerGroupService,
17+
} from "~/v3/services/worker/workerGroupService.server";
1418
import type { RunEngine } from "~/v3/runEngine.server";
1519
import { env } from "~/env.server";
1620
import { tryCatch } from "@trigger.dev/core/v3";
@@ -95,6 +99,7 @@ export class DefaultQueueManager implements QueueManager {
9599
let lockedQueueId: string | undefined;
96100
let taskTtl: string | null | undefined;
97101
let taskKind: string | undefined;
102+
let taskRegions: string[] | undefined;
98103

99104
// Determine queue name based on lockToVersion and provided options
100105
if (lockedBackgroundWorker) {
@@ -146,6 +151,7 @@ export class DefaultQueueManager implements QueueManager {
146151
taskTtl = lockedMeta?.ttl ?? undefined;
147152
}
148153
taskKind = lockedMeta?.triggerSource;
154+
taskRegions = lockedMeta?.regions;
149155
} else {
150156
// No queue override - resolve default queue + TTL + triggerSource via cache,
151157
// falling back to a single BackgroundWorkerTask lookup on miss.
@@ -184,6 +190,7 @@ export class DefaultQueueManager implements QueueManager {
184190
queueName = lockedMeta.queueName;
185191
lockedQueueId = lockedMeta.queueId ?? undefined;
186192
taskKind = lockedMeta.triggerSource;
193+
taskRegions = lockedMeta.regions;
187194
}
188195
} else {
189196
// Task is not locked to a specific version, use regular logic
@@ -199,6 +206,7 @@ export class DefaultQueueManager implements QueueManager {
199206
queueName = taskInfo.queueName;
200207
taskTtl = taskInfo.taskTtl;
201208
taskKind = taskInfo.taskKind;
209+
taskRegions = taskInfo.taskRegions;
202210
}
203211

204212
// Sanitize the final determined queue name once
@@ -216,12 +224,16 @@ export class DefaultQueueManager implements QueueManager {
216224
lockedQueueId,
217225
taskTtl,
218226
taskKind,
227+
taskRegions,
219228
};
220229
}
221230

222-
private async getTaskQueueInfo(
223-
request: TriggerTaskRequest
224-
): Promise<{ queueName: string; taskTtl?: string | null; taskKind?: string | undefined }> {
231+
private async getTaskQueueInfo(request: TriggerTaskRequest): Promise<{
232+
queueName: string;
233+
taskTtl?: string | null;
234+
taskKind?: string | undefined;
235+
taskRegions?: string[];
236+
}> {
225237
const { taskId, environment, body } = request;
226238
const { queue } = body.options ?? {};
227239

@@ -243,6 +255,7 @@ export class DefaultQueueManager implements QueueManager {
243255
queueName: overriddenQueueName,
244256
taskTtl: meta?.ttl ?? undefined,
245257
taskKind: meta?.triggerSource,
258+
taskRegions: meta?.regions,
246259
};
247260
}
248261

@@ -259,10 +272,20 @@ export class DefaultQueueManager implements QueueManager {
259272
taskId,
260273
environmentId: environment.id,
261274
});
262-
return { queueName: defaultQueueName, taskTtl: meta.ttl, taskKind: meta.triggerSource };
275+
return {
276+
queueName: defaultQueueName,
277+
taskTtl: meta.ttl,
278+
taskKind: meta.triggerSource,
279+
taskRegions: meta.regions,
280+
};
263281
}
264282

265-
return { queueName: meta.queueName, taskTtl: meta.ttl, taskKind: meta.triggerSource };
283+
return {
284+
queueName: meta.queueName,
285+
taskTtl: meta.ttl,
286+
taskKind: meta.triggerSource,
287+
taskRegions: meta.regions,
288+
};
266289
}
267290

268291
/**
@@ -320,6 +343,7 @@ export class DefaultQueueManager implements QueueManager {
320343
triggerSource: row.triggerSource,
321344
queueId: row.queue?.id ?? null,
322345
queueName: row.queue?.name ?? "",
346+
regions: row.regions,
323347
};
324348

325349
// Fire-and-forget back-fill — `setByWorker` upserts the single field and
@@ -340,6 +364,7 @@ export class DefaultQueueManager implements QueueManager {
340364
select: {
341365
ttl: true,
342366
triggerSource: true,
367+
regions: true,
343368
queue: { select: { id: true, name: true } },
344369
},
345370
});
@@ -378,6 +403,7 @@ export class DefaultQueueManager implements QueueManager {
378403
select: {
379404
ttl: true,
380405
triggerSource: true,
406+
regions: true,
381407
queue: { select: { id: true, name: true } },
382408
},
383409
});
@@ -395,6 +421,7 @@ export class DefaultQueueManager implements QueueManager {
395421
triggerSource: row.triggerSource,
396422
queueId: row.queue?.id ?? null,
397423
queueName: row.queue?.name ?? "",
424+
regions: row.regions,
398425
};
399426

400427
// Fire-and-forget back-fill — atomically upserts the slug into both
@@ -438,8 +465,10 @@ export class DefaultQueueManager implements QueueManager {
438465

439466
async getWorkerQueue(
440467
environment: AuthenticatedEnvironment,
441-
regionOverride?: string
468+
options?: WorkerQueueOptions
442469
): Promise<{ masterQueue: string; enableFastPath: boolean } | undefined> {
470+
// Dev runs always dequeue from the environment's own queue, so both the
471+
// per-trigger region and the task-level region list are ignored here.
443472
if (environment.type === "DEVELOPMENT") {
444473
return { masterQueue: environment.id, enableFastPath: true };
445474
}
@@ -452,11 +481,19 @@ export class DefaultQueueManager implements QueueManager {
452481
const [error, workerGroup] = await tryCatch(
453482
workerGroupService.getDefaultWorkerGroupForProject({
454483
projectId: environment.projectId,
455-
regionOverride,
484+
regionOverride: options?.regionOverride,
485+
allowedRegions: options?.taskRegions,
486+
taskId: options?.taskId,
456487
})
457488
);
458489

459490
if (error) {
491+
// A per-trigger region outside the task definition's list is a caller
492+
// mistake, so surface it as a 400 rather than the generic 422 below.
493+
if (error instanceof RegionNotAllowedForTaskError) {
494+
throw new ServiceValidationError(error.message, 400);
495+
}
496+
460497
// getDefaultWorkerGroupForProject queries the writer DB. A Prisma
461498
// infrastructure error (e.g. P1001 "Can't reach database server", whose
462499
// message carries the DB hostname) must NOT be promoted into a

‎apps/webapp/app/runEngine/services/triggerTask.server.ts‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -445,7 +445,7 @@ export class RunEngineTriggerTaskService {
445445
const parkedOnExternalDeploymentId =
446446
externalDeploymentResolution?.outcome === "park" ? externalDeploymentId : undefined;
447447

448-
const { queueName, lockedQueueId, taskTtl, taskKind } =
448+
const { queueName, lockedQueueId, taskTtl, taskKind, taskRegions } =
449449
await this.queueConcern.resolveQueueProperties(
450450
triggerRequest,
451451
lockedToBackgroundWorker ?? undefined
@@ -497,10 +497,13 @@ export class RunEngineTriggerTaskService {
497497

498498
const depth = parentRun ? parentRun.depth + 1 : 0;
499499

500-
const workerQueueResult = await this.queueConcern.getWorkerQueue(
501-
environment,
502-
body.options?.region
503-
);
500+
// The task definition's region list (if any) rode in on the cached task
501+
// metadata above, so no extra query is needed to enforce it here.
502+
const workerQueueResult = await this.queueConcern.getWorkerQueue(environment, {
503+
regionOverride: body.options?.region,
504+
taskRegions,
505+
taskId,
506+
});
504507
const baseWorkerQueue = workerQueueResult?.masterQueue;
505508
const enableFastPath = workerQueueResult?.enableFastPath ?? false;
506509

‎apps/webapp/app/runEngine/types.ts‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,17 @@ export type QueueProperties = {
4646
lockedQueueId?: string;
4747
taskTtl?: string | null;
4848
taskKind?: string;
49+
/** Regions the task definition allows (worker group master queues). Empty/undefined = unconstrained. */
50+
taskRegions?: string[];
51+
};
52+
53+
export type WorkerQueueOptions = {
54+
/** Per-trigger `region` option. */
55+
regionOverride?: string;
56+
/** Regions the task definition allows; the resolved region must be one of them. */
57+
taskRegions?: string[];
58+
/** Task identifier, used only for error messages. */
59+
taskId?: string;
4960
};
5061

5162
export type LockedBackgroundWorker = Pick<
@@ -66,7 +77,7 @@ export interface QueueManager {
6677
): Promise<QueueValidationResult>;
6778
getWorkerQueue(
6879
env: AuthenticatedEnvironment,
69-
regionOverride?: string
80+
options?: WorkerQueueOptions
7081
): Promise<{ masterQueue: string; enableFastPath: boolean } | undefined>;
7182
}
7283

‎apps/webapp/app/services/taskMetadataCache.server.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ export type TaskMetadataEntry = {
88
triggerSource: TaskTriggerSource;
99
queueId: string | null;
1010
queueName: string;
11+
/** Regions the task may run in (worker group master queues). Empty = no constraint. */
12+
regions: string[];
1113
};
1214

1315
export interface TaskMetadataCache {
@@ -57,6 +59,8 @@ type EncodedEntry = {
5759
k: TaskTriggerSource;
5860
q: string | null;
5961
n: string;
62+
/** Regions; omitted when empty so no-region entries stay byte-identical to pre-regions entries. */
63+
r?: string[];
6064
};
6165

6266
function encode(entry: TaskMetadataEntry): string {
@@ -65,6 +69,7 @@ function encode(entry: TaskMetadataEntry): string {
6569
k: entry.triggerSource,
6670
q: entry.queueId,
6771
n: entry.queueName,
72+
r: entry.regions.length > 0 ? entry.regions : undefined,
6873
};
6974
return JSON.stringify(payload);
7075
}
@@ -78,6 +83,8 @@ function decode(slug: string, raw: string): TaskMetadataEntry | null {
7883
triggerSource: parsed.k,
7984
queueId: parsed.q,
8085
queueName: parsed.n,
86+
// Entries written before regions existed have no `r`; treat as unconstrained.
87+
regions: Array.isArray(parsed.r) ? parsed.r : [],
8188
};
8289
} catch (error) {
8390
logger.error("Failed to decode task metadata cache entry", { slug, error });

‎apps/webapp/app/v3/services/changeCurrentDeployment.server.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ export class ChangeCurrentDeploymentService extends BaseService {
119119
slug: true,
120120
triggerSource: true,
121121
ttl: true,
122+
regions: true,
122123
queue: { select: { id: true, name: true } },
123124
},
124125
})
@@ -157,6 +158,7 @@ export class ChangeCurrentDeploymentService extends BaseService {
157158
triggerSource: t.triggerSource,
158159
queueId: t.queue?.id ?? null,
159160
queueName: t.queue?.name ?? "",
161+
regions: t.regions,
160162
}));
161163

162164
// Cache calls log+swallow internally.

0 commit comments

Comments
 (0)