Merge branch 'main' into task-page-table-scroll-view-fix

Author: samejr

.server-changes/plan-aware-compute-migration.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Gradually roll out a new run execution backend to a configurable percentage of organizations.

apps/webapp/app/entry.server.tsx

@@ -43,6 +43,10 @@ import { registerRunChangeNotifierHandlers } from "./services/realtime/runChange
 import { sessionsReplicationInstance } from "./services/sessionsReplicationInstance.server";
 (globalThis as Record<string, unknown>).__sessionsReplicationInstance =
   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/env.server.ts

@@ -158,6 +158,9 @@ const EnvironmentSchema = z
     WORKER_SCHEMA: z.string().default("graphile_worker"),
     WORKER_CONCURRENCY: z.coerce.number().int().default(10),
     WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
+    // How often each replica reloads the global flags snapshot from the DB.
+    // Sets kill/ramp propagation latency.
+    GLOBAL_FLAGS_RELOAD_INTERVAL_MS: z.coerce.number().int().min(1000).default(5000),
     WORKER_ENABLED: z.string().default("true"),
     GRACEFUL_SHUTDOWN_TIMEOUT: z.coerce.number().int().default(60000),
     DISABLE_SSE: z.string().optional(),

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

@@ -15,7 +15,7 @@ import { Prisma, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/databa
 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 { regionForDisplay } from "~/runEngine/concerns/workerQueueSplit.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import {
   findRunByIdWithMollifierFallback,
@@ -49,6 +49,7 @@ const commonRunSelect = {
   depth: true,
   scheduleId: true,
   workerQueue: true,
+  region: true,
   lockedToVersion: {
     select: {
       version: true,
@@ -520,7 +521,7 @@ async function createCommonRunStructure(run: CommonRelatedRun, apiVersion: API_V
     triggerFunction: resolveTriggerFunction(run),
     batchId: run.batch?.friendlyId,
     metadata,
-    region: run.workerQueue ? baseWorkerQueue(run.workerQueue) : undefined,
+    region: regionForDisplay(run.region, run.workerQueue),
   };
 }
 
@@ -684,6 +685,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

@@ -11,7 +11,7 @@ import { timeFilters } from "~/components/runs/v3/SharedFilters";
 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 { regionForDisplay } from "~/runEngine/concerns/workerQueueSplit.server";
 import { machinePresetFromRun } from "~/v3/machinePresets.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";
@@ -260,7 +260,7 @@ export class NextRunListPresenter {
             name: run.queue.replace("task/", ""),
             type: run.queue.startsWith("task/") ? "task" : "custom",
           },
-          region: run.workerQueue ? baseWorkerQueue(run.workerQueue) : undefined,
+          region: regionForDisplay(run.region, run.workerQueue),
           taskKind: RunAnnotations.safeParse(run.annotations).data?.taskKind ?? "STANDARD",
         };
       }),

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

@@ -303,11 +303,17 @@ export class SpanPresenter extends BasePresenter {
           location: true,
         },
         where: {
+          // masterQueue is unique and IS the run's backing queue, so this finds
+          // the group the run actually ran on.
           masterQueue: baseWorkerQueue(run.workerQueue),
         },
       });
 
-      region = workerGroup ?? null;
+      // Show the stamped geo region as the name so a migrated run never reveals
+      // its compute backing; fall back to the group name for unstamped runs.
+      region = workerGroup
+        ? { name: run.region ?? workerGroup.name, location: workerGroup.location }
+        : null;
     }
 
     // Only AGENT-tagged runs (chat.agent and friends) can be session-bound,
@@ -513,6 +519,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

@@ -18,7 +18,7 @@ import {
   type SyntheticReplayTaskRun,
 } from "~/v3/mollifier/syntheticReplayTaskRun.server";
 import parseDuration from "parse-duration";
-import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
+import { regionForDisplay } from "~/runEngine/concerns/workerQueueSplit.server";
 import { findCurrentWorkerDeployment } from "~/v3/models/workerDeployment.server";
 import { queueTypeFromType } from "~/presenters/v3/QueueRetrievePresenter.server";
 import { ReplayRunData } from "~/v3/replayTask";
@@ -52,6 +52,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
       maxDurationInSeconds: true,
       machinePreset: true,
       workerQueue: true,
+      region: true,
       ttl: true,
       idempotencyKey: true,
       runTags: true,
@@ -163,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,
@@ -210,7 +212,10 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     maxAttempts: run.maxAttempts,
     maxDurationSeconds: run.maxDurationInSeconds,
     machinePreset: run.machinePreset,
-    region: environment.type === "DEVELOPMENT" ? undefined : baseWorkerQueue(run.workerQueue),
+    region:
+      environment.type === "DEVELOPMENT"
+        ? undefined
+        : regionForDisplay(run.region, 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

@@ -0,0 +1,77 @@
+import { hashBucket } from "~/utils/computeBucket";
+
+/** Subset of the global flags snapshot this resolver reads. */
+export type ComputeMigrationFlags = {
+  computeMigrationEnabled?: boolean;
+  computeMigrationFreePercentage?: number;
+  computeMigrationPaidPercentage?: number;
+};
+
+type MigrationDecisionInput = {
+  planType: string | undefined;
+  orgId: string;
+  orgFeatureFlags: Record<string, unknown> | null | undefined;
+  flags: ComputeMigrationFlags | undefined;
+};
+
+/**
+ * Whether this org should run on the compute backing. Shared by the trigger-time
+ * transform and the deploy-time template decision so a migrated org always gets a
+ * compute template. Precedence: per-org override (both directions) wins; otherwise
+ * global enable + the plan's percentage bucket. Enterprise and unknown plans are
+ * never enrolled by percentage (override only). The sole opt-out is the per-org
+ * `computeMigrationEnabled: false`.
+ */
+export function isOrgMigrated({
+  planType,
+  orgId,
+  orgFeatureFlags,
+  flags,
+}: MigrationDecisionInput): boolean {
+  const override = orgFeatureFlags?.["computeMigrationEnabled"];
+  if (override === false) return false;
+  if (override === true) return true;
+
+  if (!(flags?.computeMigrationEnabled ?? false)) return false;
+
+  const pct =
+    planType === "free"
+      ? flags?.computeMigrationFreePercentage ?? 0
+      : planType === "paid"
+      ? flags?.computeMigrationPaidPercentage ?? 0
+      : 0; // enterprise / undefined
+
+  return hashBucket(orgId) < pct;
+}
+
+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;
+};
+
+/**
+ * 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,
+  baseEnableFastPath,
+  region,
+  backing,
+  envType,
+  ...decision
+}: 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/concerns/workerQueueSplit.server.ts

@@ -33,6 +33,19 @@ export function baseWorkerQueue(workerQueue: string | null | undefined): string
   return colon === -1 ? workerQueue : workerQueue.slice(0, colon);
 }
 
+/**
+ * User-facing region for read surfaces: the explicit geo region if set, else the
+ * region derived from the worker queue, else undefined. Use everywhere a run's
+ * region is displayed so an empty queue never surfaces as `""` and all surfaces
+ * agree. Not for query keys — those want the raw worker queue, not this fallback.
+ */
+export function regionForDisplay(
+  region: string | null | undefined,
+  workerQueue: string | null | undefined
+): string | undefined {
+  return region || (workerQueue ? baseWorkerQueue(workerQueue) : undefined);
+}
+
 /** `TriggerSource` value used for runs originating from a schedule. */
 const SCHEDULE_TRIGGER_SOURCE = "schedule";
 

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

@@ -38,6 +38,9 @@ import {
   resolveScheduledQueueSplitEnabled,
   workerQueueForRun,
 } from "../concerns/workerQueueSplit.server";
+import { resolveComputeMigration } from "../concerns/computeMigration.server";
+import { workerRegionRegistry, backingForQueue, regionForQueue } from "~/v3/workerRegions.server";
+import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server";
 import {
   publishClaim as publishMollifierClaim,
   releaseClaim as releaseMollifierClaim,
@@ -358,6 +361,24 @@ export class RunEngineTriggerTaskService {
           const baseWorkerQueue = workerQueueResult?.masterQueue;
           const enableFastPath = workerQueueResult?.enableFastPath ?? false;
 
+          // Rewrite the region to its compute backing for migration-enrolled orgs,
+          // from the in-memory snapshots (no DB query). A cold read (registry not yet
+          // loaded) returns undefined/[] and the resolver falls back to not-migrated.
+          const workerGroups = workerRegionRegistry.current() ?? [];
+          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,
+          });
+
           // Build annotations for this run
           const triggerSource = options.triggerSource ?? "api";
           const triggerAction = options.triggerAction ?? "trigger";
@@ -386,13 +407,13 @@ export class RunEngineTriggerTaskService {
               globalDefault: env.TRIGGER_WORKER_QUEUE_SCHEDULED_SPLIT_ENABLED === "1",
             });
           const workerQueue =
-            baseWorkerQueue !== undefined
+            migrated.workerQueue !== undefined
               ? workerQueueForRun({
-                  workerQueue: baseWorkerQueue,
+                  workerQueue: migrated.workerQueue,
                   rootTriggerSource: annotations.rootTriggerSource,
                   splitEnabled: scheduledQueueSplitEnabled,
                 })
-              : baseWorkerQueue;
+              : migrated.workerQueue;
 
           try {
             return await this.traceEventConcern.traceRun(
@@ -491,7 +512,8 @@ export class RunEngineTriggerTaskService {
                       queueName,
                       lockedQueueId,
                       workerQueue,
-                      enableFastPath,
+                      region: migrated.region,
+                      enableFastPath: migrated.enableFastPath,
                       lockedToBackgroundWorker: lockedToBackgroundWorker ?? undefined,
                       delayUntil,
                       ttl,
@@ -569,7 +591,8 @@ export class RunEngineTriggerTaskService {
                   queueName,
                   lockedQueueId,
                   workerQueue,
-                  enableFastPath,
+                  region: migrated.region,
+                  enableFastPath: migrated.enableFastPath,
                   lockedToBackgroundWorker: lockedToBackgroundWorker ?? undefined,
                   delayUntil,
                   ttl,
@@ -718,6 +741,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;
@@ -771,6 +795,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,