feat(webapp): stamp geo region on runs, keep worker_queue raw, test-safe registries

Author: nicktrn

apps/webapp/app/entry.server.tsx

@@ -45,6 +45,8 @@ import { sessionsReplicationInstance } from "./services/sessionsReplicationInsta
   sessionsReplicationInstance;
 import { globalFlagsRegistry } from "./v3/globalFlagsRegistry.server";
 (globalThis as Record<string, unknown>).__globalFlagsRegistry = globalFlagsRegistry;
+import { workerRegionRegistry } from "./v3/workerRegions.server";
+(globalThis as Record<string, unknown>).__workerRegionRegistry = workerRegionRegistry;
 
 const ABORT_DELAY = 30000;
 

apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts

@@ -16,7 +16,6 @@ import assertNever from "assert-never";
 import { API_VERSIONS, CURRENT_API_VERSION, RunStatusUnspecifiedApiVersion } from "~/api/versions";
 import { $replica, prisma } from "~/db.server";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
-import { regionForQueue, workerRegionRegistry } from "~/v3/workerRegions.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import {
   findRunByIdWithMollifierFallback,
@@ -50,6 +49,7 @@ const commonRunSelect = {
   depth: true,
   scheduleId: true,
   workerQueue: true,
+  region: true,
   lockedToVersion: {
     select: {
       version: true,
@@ -521,9 +521,8 @@ async function createCommonRunStructure(run: CommonRelatedRun, apiVersion: API_V
     triggerFunction: resolveTriggerFunction(run),
     batchId: run.batch?.friendlyId,
     metadata,
-    region: run.workerQueue
-      ? regionForQueue(baseWorkerQueue(run.workerQueue), workerRegionRegistry.current() ?? [])
-      : undefined,
+    region:
+      run.region || run.workerQueue ? run.region ?? baseWorkerQueue(run.workerQueue) : undefined,
   };
 }
 
@@ -687,6 +686,7 @@ export function synthesiseFoundRunFromBuffer(buffered: SyntheticRun): FoundRun {
     // API response's `region` to undefined instead of advertising a
     // misleading "main" region for a not-yet-assigned buffered run).
     workerQueue: buffered.workerQueue ?? "",
+    region: buffered.region ?? "",
     parentTaskRun: null,
     rootTaskRun: null,
     childRuns: [],

apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts

@@ -12,7 +12,6 @@ import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
 import { getTaskIdentifiers } from "~/models/task.server";
 import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
-import { regionForQueue, workerRegionRegistry } from "~/v3/workerRegions.server";
 import { machinePresetFromRun } from "~/v3/machinePresets.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";
@@ -261,9 +260,7 @@ export class NextRunListPresenter {
             name: run.queue.replace("task/", ""),
             type: run.queue.startsWith("task/") ? "task" : "custom",
           },
-          region: run.workerQueue
-            ? regionForQueue(baseWorkerQueue(run.workerQueue), workerRegionRegistry.current() ?? [])
-            : undefined,
+          region: run.region || baseWorkerQueue(run.workerQueue),
           taskKind: RunAnnotations.safeParse(run.annotations).data?.taskKind ?? "STANDARD",
         };
       }),

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -16,7 +16,6 @@ import {
 } from "@trigger.dev/core/v3/serverOnly";
 import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
-import { regionForQueue, workerRegionRegistry } from "~/v3/workerRegions.server";
 import { logger } from "~/services/logger.server";
 import { rehydrateAttribute } from "~/v3/eventRepository/eventRepository.server";
 import { machinePresetFromRun } from "~/v3/machinePresets.server";
@@ -304,7 +303,7 @@ export class SpanPresenter extends BasePresenter {
           location: true,
         },
         where: {
-          masterQueue: regionForQueue(baseWorkerQueue(run.workerQueue), workerRegionRegistry.current() ?? []),
+          masterQueue: run.region ?? baseWorkerQueue(run.workerQueue),
         },
       });
 
@@ -514,6 +513,7 @@ export class SpanPresenter extends BasePresenter {
         },
         engine: true,
         workerQueue: true,
+        region: true,
         error: true,
         output: true,
         outputType: true,

apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts

@@ -19,7 +19,6 @@ import {
 } from "~/v3/mollifier/syntheticReplayTaskRun.server";
 import parseDuration from "parse-duration";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
-import { regionForQueue, workerRegionRegistry } from "~/v3/workerRegions.server";
 import { findCurrentWorkerDeployment } from "~/v3/models/workerDeployment.server";
 import { queueTypeFromType } from "~/presenters/v3/QueueRetrievePresenter.server";
 import { ReplayRunData } from "~/v3/replayTask";
@@ -53,6 +52,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
       maxDurationInSeconds: true,
       machinePreset: true,
       workerQueue: true,
+      region: true,
       ttl: true,
       idempotencyKey: true,
       runTags: true,
@@ -164,6 +164,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
       maxDurationInSeconds: buffered.maxDurationInSeconds ?? null,
       machinePreset: buffered.machinePreset ?? null,
       workerQueue: buffered.workerQueue ?? null,
+      region: buffered.region ?? null,
       ttl: buffered.ttl ?? null,
       idempotencyKey: buffered.idempotencyKey ?? null,
       runTags: buffered.runTags,
@@ -214,7 +215,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     region:
       environment.type === "DEVELOPMENT"
         ? undefined
-        : regionForQueue(baseWorkerQueue(run.workerQueue), workerRegionRegistry.current() ?? []),
+        : run.region ?? baseWorkerQueue(run.workerQueue),
     regions: regionsResult.regions,
     ttlSeconds: run.ttl ? parseDuration(run.ttl, "s") ?? undefined : undefined,
     idempotencyKey: run.idempotencyKey,

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

@@ -46,24 +46,32 @@ export function isOrgMigrated({
 
 type ResolveInput = MigrationDecisionInput & {
   baseWorkerQueue: string | undefined;
+  baseEnableFastPath: boolean;
+  region: string | undefined; // geo of the base queue (same whether migrated or not)
+  backing: { workerQueue: string; enableFastPath: boolean } | undefined;
   envType: string;
-  backing: string | undefined; // the compute backing for this queue's region, or undefined
 };
 
 /**
- * Rewrite the resolved worker queue to its compute backing when the org is
- * migrated and the region has a backing. Same-geo swap (us-east-1 -> us-east-1-next):
- * any explicit placement is a geography preference, honored by staying in-region.
- * Applied after region resolution, mirroring the scheduled-split.
+ * Produce the target descriptor `{ workerQueue, region, enableFastPath }` for a
+ * run. When the org is migrated and the region has a compute backing, the queue
+ * and fast-path setting come from the MICROVM backing group; `region` is the geo
+ * either way. Same-geo swap (us-east-1 -> us-east-1-next): any explicit placement
+ * is a geography preference, honored by staying in-region. Applied after region
+ * resolution, mirroring the scheduled-split.
  */
 export function resolveComputeMigration({
   baseWorkerQueue,
-  envType,
+  baseEnableFastPath,
+  region,
   backing,
+  envType,
   ...decision
-}: ResolveInput): string | undefined {
-  if (baseWorkerQueue === undefined) return baseWorkerQueue;
-  if (envType === "DEVELOPMENT") return baseWorkerQueue;
-  if (!isOrgMigrated(decision)) return baseWorkerQueue;
-  return backing ?? baseWorkerQueue;
+}: ResolveInput): { workerQueue: string | undefined; region: string | undefined; enableFastPath: boolean } {
+  const passthrough = { workerQueue: baseWorkerQueue, region, enableFastPath: baseEnableFastPath };
+  if (baseWorkerQueue === undefined) return passthrough;
+  if (envType === "DEVELOPMENT") return passthrough;
+  if (!isOrgMigrated(decision)) return passthrough;
+  if (!backing) return passthrough;
+  return { workerQueue: backing.workerQueue, region, enableFastPath: backing.enableFastPath };
 }

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

@@ -39,7 +39,7 @@ import {
   workerQueueForRun,
 } from "../concerns/workerQueueSplit.server";
 import { resolveComputeMigration } from "../concerns/computeMigration.server";
-import { workerRegionRegistry, backingForQueue } from "~/v3/workerRegions.server";
+import { workerRegionRegistry, backingForQueue, regionForQueue } from "~/v3/workerRegions.server";
 import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server";
 import {
   publishClaim as publishMollifierClaim,
@@ -372,14 +372,18 @@ export class RunEngineTriggerTaskService {
             await workerRegionRegistry.waitUntilReady(env.GLOBAL_FLAGS_READY_TIMEOUT_MS);
           }
           const workerGroups = workerRegionRegistry.current() ?? [];
-          const migratedWorkerQueue = resolveComputeMigration({
+          const region = baseWorkerQueue ? regionForQueue(baseWorkerQueue, workerGroups) : undefined;
+          const backing = baseWorkerQueue ? backingForQueue(baseWorkerQueue, workerGroups) : undefined;
+          const migrated = resolveComputeMigration({
             baseWorkerQueue,
+            baseEnableFastPath: enableFastPath,
+            region,
+            backing,
             planType,
             orgId: environment.organization.id,
             orgFeatureFlags: environment.organization.featureFlags as Record<string, unknown> | null,
             flags: globalFlagsRegistry.current(),
             envType: environment.type,
-            backing: baseWorkerQueue ? backingForQueue(baseWorkerQueue, workerGroups) : undefined,
           });
 
           // Build annotations for this run
@@ -410,13 +414,13 @@ export class RunEngineTriggerTaskService {
               globalDefault: env.TRIGGER_WORKER_QUEUE_SCHEDULED_SPLIT_ENABLED === "1",
             });
           const workerQueue =
-            migratedWorkerQueue !== undefined
+            migrated.workerQueue !== undefined
               ? workerQueueForRun({
-                  workerQueue: migratedWorkerQueue,
+                  workerQueue: migrated.workerQueue,
                   rootTriggerSource: annotations.rootTriggerSource,
                   splitEnabled: scheduledQueueSplitEnabled,
                 })
-              : migratedWorkerQueue;
+              : migrated.workerQueue;
 
           try {
             return await this.traceEventConcern.traceRun(
@@ -515,7 +519,8 @@ export class RunEngineTriggerTaskService {
                       queueName,
                       lockedQueueId,
                       workerQueue,
-                      enableFastPath,
+                      region: migrated.region,
+                      enableFastPath: migrated.enableFastPath,
                       lockedToBackgroundWorker: lockedToBackgroundWorker ?? undefined,
                       delayUntil,
                       ttl,
@@ -593,7 +598,8 @@ export class RunEngineTriggerTaskService {
                   queueName,
                   lockedQueueId,
                   workerQueue,
-                  enableFastPath,
+                  region: migrated.region,
+                  enableFastPath: migrated.enableFastPath,
                   lockedToBackgroundWorker: lockedToBackgroundWorker ?? undefined,
                   delayUntil,
                   ttl,
@@ -742,6 +748,7 @@ export class RunEngineTriggerTaskService {
     queueName: string;
     lockedQueueId?: string;
     workerQueue?: string;
+    region?: string;
     enableFastPath: boolean;
     lockedToBackgroundWorker?: { id: string; version: string; sdkVersion: string; cliVersion: string };
     delayUntil?: Date;
@@ -795,6 +802,7 @@ export class RunEngineTriggerTaskService {
       queue: args.queueName,
       lockedQueueId: args.lockedQueueId,
       workerQueue: args.workerQueue,
+      region: args.region,
       enableFastPath: args.enableFastPath,
       isTest: args.body.options?.test ?? false,
       delayUntil: args.delayUntil,

apps/webapp/app/services/runsReplicationService.server.ts

@@ -39,7 +39,6 @@ import pLimit from "p-limit";
 import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
 import { calculateErrorFingerprint } from "~/utils/errorFingerprinting";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
-import { regionForQueue, workerRegionRegistry } from "~/v3/workerRegions.server";
 import {
   isClickHouseJsonParseError,
   parseRowNumberFromError,
@@ -1123,7 +1122,8 @@ export class RunsReplicationService {
       event === "delete" ? 1 : 0, // _is_deleted
       run.concurrencyKey ?? "", // concurrency_key
       run.bulkActionGroupIds ?? [], // bulk_action_group_ids
-      regionForQueue(baseWorkerQueue(run.masterQueue ?? ""), workerRegionRegistry.current() ?? []), // worker_queue (geo region; strip `:scheduled` and hide the compute backing - customers query this column)
+      baseWorkerQueue(run.masterQueue ?? ""), // worker_queue (raw - operators slice by this)
+      run.region ?? "", // region (geo for customers)
       run.maxDurationInSeconds ?? null, // max_duration_in_seconds
       annotations?.triggerSource ?? "", // trigger_source
       annotations?.rootTriggerSource ?? "", // root_trigger_source

apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts

@@ -204,6 +204,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
         machinePreset: true,
         queue: true,
         workerQueue: true,
+        region: true,
         annotations: true,
       },
     });
@@ -377,7 +378,9 @@ function applyRunFiltersToQueryBuilder<T>(
   }
 
   if (options.regions && options.regions.length > 0) {
-    queryBuilder.where("worker_queue IN {regions: Array(String)}", { regions: options.regions });
+    queryBuilder.where("if(region != '', region, worker_queue) IN {regions: Array(String)}", {
+      regions: options.regions,
+    });
   }
 
   if (options.machines && options.machines.length > 0) {

apps/webapp/app/services/runsRepository/runsRepository.server.ts

@@ -108,6 +108,7 @@ export type ListedRun = Prisma.TaskRunGetPayload<{
     machinePreset: true;
     queue: true;
     workerQueue: true;
+    region: true;
     annotations: true;
   };
 }>;