Pass a region in when triggering

Author: matt-aitken

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

@@ -14,6 +14,7 @@ import { WorkerGroupService } from "~/v3/services/worker/workerGroupService.serv
 import type { RunEngine } from "~/v3/runEngine.server";
 import { env } from "~/env.server";
 import { EngineServiceValidationError } from "./errors";
+import { tryCatch } from "@trigger.dev/core/v3";
 
 export class DefaultQueueManager implements QueueManager {
   constructor(
@@ -196,7 +197,10 @@ export class DefaultQueueManager implements QueueManager {
     };
   }
 
-  async getWorkerQueue(environment: AuthenticatedEnvironment): Promise<string | undefined> {
+  async getWorkerQueue(
+    environment: AuthenticatedEnvironment,
+    regionOverride?: string
+  ): Promise<string | undefined> {
     if (environment.type === "DEVELOPMENT") {
       return environment.id;
     }
@@ -206,9 +210,16 @@ export class DefaultQueueManager implements QueueManager {
       engine: this.engine,
     });
 
-    const workerGroup = await workerGroupService.getDefaultWorkerGroupForProject({
-      projectId: environment.projectId,
-    });
+    const [error, workerGroup] = await tryCatch(
+      workerGroupService.getDefaultWorkerGroupForProject({
+        projectId: environment.projectId,
+        regionOverride,
+      })
+    );
+
+    if (error) {
+      throw new EngineServiceValidationError(error.message);
+    }
 
     if (!workerGroup) {
       throw new EngineServiceValidationError("No worker group found");

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

@@ -234,7 +234,7 @@ export class RunEngineTriggerTaskService {
 
       const depth = parentRun ? parentRun.depth + 1 : 0;
 
-      const workerQueue = await this.queueConcern.getWorkerQueue(environment);
+      const workerQueue = await this.queueConcern.getWorkerQueue(environment, body.options?.region);
 
       try {
         return await this.traceEventConcern.traceRun(triggerRequest, async (event) => {

apps/webapp/app/runEngine/types.ts

@@ -67,7 +67,10 @@ export interface QueueManager {
   ): Promise<QueueProperties>;
   getQueueName(request: TriggerTaskRequest): Promise<string>;
   validateQueueLimits(env: AuthenticatedEnvironment): Promise<QueueValidationResult>;
-  getWorkerQueue(env: AuthenticatedEnvironment): Promise<string | undefined>;
+  getWorkerQueue(
+    env: AuthenticatedEnvironment,
+    regionOverride?: string
+  ): Promise<string | undefined>;
 }
 
 export interface PayloadProcessor {

apps/webapp/app/v3/services/worker/workerGroupService.server.ts

@@ -195,10 +195,12 @@ export class WorkerGroupService extends WithRunEngine {
 
   async getDefaultWorkerGroupForProject({
     projectId,
+    regionOverride,
   }: {
     projectId: string;
+    regionOverride?: string;
   }): Promise<WorkerInstanceGroup | undefined> {
-    const project = await this._prisma.project.findUnique({
+    const project = await this._prisma.project.findFirst({
       where: {
         id: projectId,
       },
@@ -208,8 +210,39 @@ export class WorkerGroupService extends WithRunEngine {
     });
 
     if (!project) {
-      logger.error("[WorkerGroupService] Project not found", { projectId });
-      return;
+      throw new Error("Project not found.");
+    }
+
+    // If they've specified a region, we need to check they have access to it
+    if (regionOverride) {
+      const workerGroup = await this._prisma.workerInstanceGroup.findFirst({
+        where: {
+          masterQueue: regionOverride,
+        },
+      });
+
+      if (!workerGroup) {
+        throw new Error(`The region you specified doesn't exist ("${regionOverride}").`);
+      }
+
+      // If they're restricted, check they have access
+      if (project.allowedWorkerQueues.length > 0) {
+        if (project.allowedWorkerQueues.includes(workerGroup.masterQueue)) {
+          return workerGroup;
+        }
+
+        throw new Error(
+          `You don't have access to this region ("${regionOverride}"). You can use the following regions: ${project.allowedWorkerQueues.join(
+            ", "
+          )}.`
+        );
+      }
+
+      if (workerGroup.hidden) {
+        throw new Error(`The region you specified isn't available to you ("${regionOverride}").`);
+      }
+
+      return workerGroup;
     }
 
     if (project.defaultWorkerGroup) {

packages/core/src/v3/schemas/api.ts

@@ -134,6 +134,7 @@ export const TriggerTaskRequestBody = z.object({
       ttl: z.string().or(z.number().nonnegative().int()).optional(),
       priority: z.number().optional(),
       bulkActionId: z.string().optional(),
+      region: z.string().optional(),
     })
     .optional(),
 });
@@ -181,6 +182,7 @@ export const BatchTriggerTaskItem = z.object({
       test: z.boolean().optional(),
       ttl: z.string().or(z.number().nonnegative().int()).optional(),
       priority: z.number().optional(),
+      region: z.string().optional(),
     })
     .optional(),
 });

packages/core/src/v3/types/tasks.ts

@@ -855,6 +855,21 @@ export type TriggerOptions = {
    * to the same version as the parent task that is triggering the child tasks.
    */
   version?: string;
+
+  /**
+   * Specify the region to run the task in. This overrides the default region set for your project in the dashboard.
+   *
+   * Check the Regions page in the dashboard for regions that are available to you.
+   *
+   * In DEV this won't do anything, so it's fine to set it in your code.
+   *
+   * @example
+   *
+   * ```ts
+   * await myTask.trigger({ foo: "bar" }, { region: "us-east-1" });
+   * ```
+   */
+  region?: string;
 };
 
 export type TriggerAndWaitOptions = Omit<TriggerOptions, "version">;

packages/trigger-sdk/src/v3/shared.ts

@@ -627,6 +627,7 @@ export async function batchTriggerById<TTask extends AnyTask>(
               idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
               machine: item.options?.machine,
               priority: item.options?.priority,
+              region: item.options?.region,
               lockToVersion: item.options?.version ?? getEnvVar("TRIGGER_VERSION"),
             },
           } satisfies BatchTriggerTaskV2RequestBody["items"][0];
@@ -796,6 +797,7 @@ export async function batchTriggerByIdAndWait<TTask extends AnyTask>(
                   idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
                   machine: item.options?.machine,
                   priority: item.options?.priority,
+                  region: item.options?.region,
                 },
               } satisfies BatchTriggerTaskV2RequestBody["items"][0];
             })
@@ -955,6 +957,7 @@ export async function batchTriggerTasks<TTasks extends readonly AnyTask[]>(
               idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
               machine: item.options?.machine,
               priority: item.options?.priority,
+              region: item.options?.region,
               lockToVersion: item.options?.version ?? getEnvVar("TRIGGER_VERSION"),
             },
           } satisfies BatchTriggerTaskV2RequestBody["items"][0];
@@ -1126,6 +1129,7 @@ export async function batchTriggerAndWaitTasks<TTasks extends readonly AnyTask[]
                   idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
                   machine: item.options?.machine,
                   priority: item.options?.priority,
+                  region: item.options?.region,
                 },
               } satisfies BatchTriggerTaskV2RequestBody["items"][0];
             })
@@ -1198,6 +1202,7 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
         parentRunId: taskContext.ctx?.run.id,
         machine: options?.machine,
         priority: options?.priority,
+        region: options?.region,
         lockToVersion: options?.version ?? getEnvVar("TRIGGER_VERSION"),
       },
     },
@@ -1270,6 +1275,7 @@ async function batchTrigger_internal<TRunTypes extends AnyRunTypes>(
               idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
               machine: item.options?.machine,
               priority: item.options?.priority,
+              region: item.options?.region,
               lockToVersion: item.options?.version ?? getEnvVar("TRIGGER_VERSION"),
             },
           } satisfies BatchTriggerTaskV2RequestBody["items"][0];
@@ -1354,6 +1360,7 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
             idempotencyKeyTTL: options?.idempotencyKeyTTL,
             machine: options?.machine,
             priority: options?.priority,
+            region: options?.region,
           },
         },
         {},
@@ -1444,6 +1451,7 @@ async function batchTriggerAndWait_internal<TIdentifier extends string, TPayload
                   idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
                   machine: item.options?.machine,
                   priority: item.options?.priority,
+                  region: item.options?.region,
                 },
               } satisfies BatchTriggerTaskV2RequestBody["items"][0];
             })

references/hello-world/src/trigger/regions.ts

@@ -0,0 +1,9 @@
+import { task } from "@trigger.dev/sdk";
+import { fixedLengthTask } from "./batches.js";
+
+export const regionsTask = task({
+  id: "regions",
+  run: async ({ region }: { region?: string }, { ctx }) => {
+    await fixedLengthTask.triggerAndWait({ waitSeconds: 1 }, { region });
+  },
+});