feat(webapp,supervisor): isolate scheduled runs on a dedicated worker queue (#3839)

## Summary

Scheduled runs and their descendants can now be routed to a dedicated
per-region worker queue, processed by a separate worker fleet, so a
burst of scheduled crons no longer competes with standard and agent runs
for the same queue and inflates their startup latency. It is off by
default and enabled per organization via a feature flag (with a global
default), so nothing changes until it is turned on.

## Design

At trigger time, any run whose lineage originates from a schedule
(`rootTriggerSource === "schedule"`, which already propagates from a
scheduled run down to all of its children) gets its worker queue
suffixed with `:scheduled`. The worker queue name is an opaque string
persisted on the run and used verbatim by enqueue and dequeue, so this
needs no Lua, message-envelope, or concurrency changes. Concurrency
stays keyed by environment and queue, not by worker queue.

On the consumer side, the dequeue endpoint gains an optional
`queueClass` selector. A supervisor sends `queueClass: "scheduled"` and
the server derives the actual queue from the worker's own group, so a
token can only ever reach its own region's queues. A fleet picks its
class with the `TRIGGER_WORKER_QUEUE_CLASS` env var (`default` or
`scheduled`), so a dedicated scheduled fleet can run alongside the
standard one.

Verified end to end against a local managed-worker setup: scheduled runs
route to the dedicated queue, are drained only by the scheduled fleet,
and standard runs are left untouched.

Author: ericallam

.server-changes/scheduled-worker-queue-split.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Scheduled runs and their descendants can be routed to a dedicated worker queue and processed by a separate worker fleet, isolating standard and agent run startup latency from scheduled-cron bursts. Off by default, enabled per organization via a feature flag.

apps/supervisor/src/env.ts

@@ -34,6 +34,10 @@ const Env = z
 
     // Dequeue settings (provider mode)
     TRIGGER_DEQUEUE_ENABLED: BoolEnv.default(true),
+    // Which worker-queue class this supervisor fleet serves. "default" pulls the
+    // region queue (standard/agent runs); "scheduled" pulls the dedicated
+    // scheduled-lineage queue. Run a separate fleet per class for isolation.
+    TRIGGER_WORKER_QUEUE_CLASS: z.enum(["default", "scheduled"]).default("default"),
     TRIGGER_DEQUEUE_INTERVAL_MS: z.coerce.number().int().default(250),
     TRIGGER_DEQUEUE_IDLE_INTERVAL_MS: z.coerce.number().int().default(1000),
     TRIGGER_DEQUEUE_MAX_RUN_COUNT: z.coerce.number().int().default(1),

apps/supervisor/src/index.ts

@@ -190,6 +190,7 @@ class ManagedSupervisor {
       dequeueIdleIntervalMs: env.TRIGGER_DEQUEUE_IDLE_INTERVAL_MS,
       queueConsumerEnabled: env.TRIGGER_DEQUEUE_ENABLED,
       maxRunCount: env.TRIGGER_DEQUEUE_MAX_RUN_COUNT,
+      queueClass: env.TRIGGER_WORKER_QUEUE_CLASS,
       metricsRegistry: register,
       scaling: {
         strategy: env.TRIGGER_DEQUEUE_SCALING_STRATEGY,

apps/webapp/app/env.server.ts

@@ -1059,6 +1059,16 @@ const EnvironmentSchema = z
     COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
     COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
+    // Global default for the scheduled worker-queue split. When "1", runs in a
+    // scheduled lineage (rootTriggerSource === "schedule") are routed to a
+    // dedicated `<region>:scheduled` worker queue so a separate consumer fleet
+    // can dequeue them independently of standard/agent runs. The per-org
+    // `workerQueueScheduledSplitEnabled` feature flag overrides this default in
+    // BOTH directions (an org set to false stays on the single queue even when
+    // this is "1"; an org set to true splits even when this is "0"). Never
+    // applies to DEVELOPMENT environments.
+    TRIGGER_WORKER_QUEUE_SCHEDULED_SPLIT_ENABLED: z.string().default("0"),
+
     TRIGGER_MOLLIFIER_ENABLED: z.string().default("0"),
     // Separate switch for the drainer (consumer side) so it can be split
     // off onto a dedicated worker service. Unset → inherits

apps/webapp/app/routes/engine.v1.worker-actions.dequeue.ts

@@ -7,12 +7,13 @@ import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.
 
 export const action = createActionWorkerApiRoute(
   {
-    body: WorkerApiDequeueRequestBody, // Even though we don't use it, we need to keep it for backwards compatibility
+    body: WorkerApiDequeueRequestBody,
   },
   async ({
     authenticatedWorker,
     runnerId,
+    body,
   }): Promise<TypedResponse<WorkerApiDequeueResponseBody>> => {
-    return json(await authenticatedWorker.dequeue({ runnerId }));
+    return json(await authenticatedWorker.dequeue({ runnerId, queueClass: body.queueClass }));
   }
 );

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

@@ -0,0 +1,90 @@
+import type { WorkerQueueClass } from "@trigger.dev/core/v3/workers";
+import { FEATURE_FLAG, FeatureFlagCatalog } from "~/v3/featureFlags";
+
+/**
+ * Suffix appended to a region's worker queue name to route scheduled-lineage
+ * runs onto their own Redis list (e.g. `us-nyc-3` -> `us-nyc-3:scheduled`). A
+ * dedicated consumer fleet dequeues the suffixed list so the top-of-hour
+ * scheduled-cron herd can't starve standard/agent run startup. The worker queue
+ * name is opaque everywhere downstream (it's only ever `:`-joined into a Redis
+ * key and persisted on the run), so encoding the class in the suffix needs no
+ * Lua, envelope, or resolver changes.
+ */
+export const SCHEDULED_WORKER_QUEUE_SUFFIX = ":scheduled";
+
+/** `TriggerSource` value used for runs originating from a schedule. */
+const SCHEDULE_TRIGGER_SOURCE = "schedule";
+
+/**
+ * Resolve whether the scheduled worker-queue split is enabled for a run, reading
+ * only the in-memory org feature-flags JSON (already loaded on the authenticated
+ * environment) — never a DB query, so it is safe on the trigger hot path.
+ *
+ * Precedence: a per-org override wins in BOTH directions; the global default is
+ * used only when the org has not set the flag.
+ */
+export function resolveScheduledQueueSplitEnabled({
+  orgFeatureFlags,
+  globalDefault,
+}: {
+  orgFeatureFlags: Record<string, unknown> | null | undefined;
+  globalDefault: boolean;
+}): boolean {
+  const override = orgFeatureFlags?.[FEATURE_FLAG.workerQueueScheduledSplitEnabled];
+
+  if (override !== undefined) {
+    const parsed =
+      FeatureFlagCatalog[FEATURE_FLAG.workerQueueScheduledSplitEnabled].safeParse(override);
+
+    if (parsed.success) {
+      return parsed.data;
+    }
+  }
+
+  return globalDefault;
+}
+
+/**
+ * Pick the worker queue a run should be enqueued onto. Runs in a scheduled
+ * lineage (`rootTriggerSource === "schedule"`, which propagates from a scheduled
+ * root down to every descendant) route to the suffixed list when the split is
+ * enabled; everything else is unchanged. Idempotent — never double-suffixes.
+ */
+export function workerQueueForRun({
+  workerQueue,
+  rootTriggerSource,
+  splitEnabled,
+}: {
+  workerQueue: string;
+  rootTriggerSource: string | undefined;
+  splitEnabled: boolean;
+}): string {
+  if (
+    !splitEnabled ||
+    rootTriggerSource !== SCHEDULE_TRIGGER_SOURCE ||
+    workerQueue.endsWith(SCHEDULED_WORKER_QUEUE_SUFFIX)
+  ) {
+    return workerQueue;
+  }
+
+  return `${workerQueue}${SCHEDULED_WORKER_QUEUE_SUFFIX}`;
+}
+
+/**
+ * Consumer-side counterpart to {@link workerQueueForRun}: given a worker's base
+ * (region) queue and the requested queue class, return the worker queue to
+ * dequeue from. `"scheduled"` targets the suffixed list; anything else is the
+ * base queue. The server always derives this from the authenticated worker's
+ * own `masterQueue`, so a token can only ever reach its own region's queues.
+ * Idempotent — never double-suffixes.
+ */
+export function workerQueueForClass(
+  masterQueue: string,
+  queueClass: WorkerQueueClass | undefined
+): string {
+  if (queueClass === "scheduled" && !masterQueue.endsWith(SCHEDULED_WORKER_QUEUE_SUFFIX)) {
+    return `${masterQueue}${SCHEDULED_WORKER_QUEUE_SUFFIX}`;
+  }
+
+  return masterQueue;
+}

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

@@ -34,6 +34,10 @@ import {
   IdempotencyKeyConcern,
   type ClaimedIdempotency,
 } from "../concerns/idempotencyKeys.server";
+import {
+  resolveScheduledQueueSplitEnabled,
+  workerQueueForRun,
+} from "../concerns/workerQueueSplit.server";
 import {
   publishClaim as publishMollifierClaim,
   releaseClaim as releaseMollifierClaim,
@@ -351,7 +355,7 @@ export class RunEngineTriggerTaskService {
             environment,
             body.options?.region
           );
-          const workerQueue = workerQueueResult?.masterQueue;
+          const baseWorkerQueue = workerQueueResult?.masterQueue;
           const enableFastPath = workerQueueResult?.enableFastPath ?? false;
 
           // Build annotations for this run
@@ -366,6 +370,30 @@ export class RunEngineTriggerTaskService {
             taskKind: taskKind ?? "STANDARD",
           };
 
+          // Route runs in a scheduled lineage (the scheduled run itself and every
+          // descendant, via the propagated rootTriggerSource) to a dedicated
+          // `<region>:scheduled` worker queue so a separate consumer fleet can
+          // dequeue them independently of standard/agent runs. Gated per-org with
+          // a global default, never applied to dev. Reads only the in-memory org
+          // flags already on the environment — no DB query on the hot path.
+          const scheduledQueueSplitEnabled =
+            environment.type !== "DEVELOPMENT" &&
+            resolveScheduledQueueSplitEnabled({
+              orgFeatureFlags: environment.organization.featureFlags as Record<
+                string,
+                unknown
+              > | null,
+              globalDefault: env.TRIGGER_WORKER_QUEUE_SCHEDULED_SPLIT_ENABLED === "1",
+            });
+          const workerQueue =
+            baseWorkerQueue !== undefined
+              ? workerQueueForRun({
+                  workerQueue: baseWorkerQueue,
+                  rootTriggerSource: annotations.rootTriggerSource,
+                  splitEnabled: scheduledQueueSplitEnabled,
+                })
+              : baseWorkerQueue;
+
           try {
             return await this.traceEventConcern.traceRun(
               triggerRequest,

apps/webapp/app/v3/featureFlags.ts

@@ -9,6 +9,7 @@ export const FEATURE_FLAG = {
   hasComputeAccess: "hasComputeAccess",
   hasPrivateConnections: "hasPrivateConnections",
   mollifierEnabled: "mollifierEnabled",
+  workerQueueScheduledSplitEnabled: "workerQueueScheduledSplitEnabled",
 } as const;
 
 export const FeatureFlagCatalog = {
@@ -20,6 +21,7 @@ export const FeatureFlagCatalog = {
   [FEATURE_FLAG.hasComputeAccess]: z.coerce.boolean(),
   [FEATURE_FLAG.hasPrivateConnections]: z.coerce.boolean(),
   [FEATURE_FLAG.mollifierEnabled]: z.coerce.boolean(),
+  [FEATURE_FLAG.workerQueueScheduledSplitEnabled]: z.coerce.boolean(),
 };
 
 export type FeatureFlagKey = keyof typeof FeatureFlagCatalog;

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

@@ -10,7 +10,7 @@ import {
   TaskRunExecutionResult,
 } from "@trigger.dev/core/v3";
 import { fromFriendlyId } from "@trigger.dev/core/v3/isomorphic";
-import { WORKER_HEADERS } from "@trigger.dev/core/v3/workers";
+import { WORKER_HEADERS, type WorkerQueueClass } from "@trigger.dev/core/v3/workers";
 import {
   Prisma,
   RuntimeEnvironment,
@@ -27,6 +27,7 @@ import { defaultMachine } from "~/services/platform.v3.server";
 import { singleton } from "~/utils/singleton";
 import { resolveVariablesForEnvironment } from "~/v3/environmentVariables/environmentVariablesRepository.server";
 import { machinePresetFromName } from "~/v3/machinePresets.server";
+import { workerQueueForClass } from "~/runEngine/concerns/workerQueueSplit.server";
 import { WithRunEngine, WithRunEngineOptions } from "../baseService.server";
 
 const authenticatedWorkerInstanceCache = singleton(
@@ -369,10 +370,18 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     });
   }
 
-  async dequeue({ runnerId }: { runnerId?: string }): Promise<DequeuedMessage[]> {
+  async dequeue({
+    runnerId,
+    queueClass,
+  }: {
+    runnerId?: string;
+    queueClass?: WorkerQueueClass;
+  }): Promise<DequeuedMessage[]> {
+    // Derive the actual queue from this worker's own masterQueue + class, so a
+    // token can only ever reach its own region's queues (default or :scheduled).
     return await this._engine.dequeueFromWorkerQueue({
       consumerId: this.workerInstanceId,
-      workerQueue: this.masterQueue,
+      workerQueue: workerQueueForClass(this.masterQueue, queueClass),
       workerId: this.workerInstanceId,
       runnerId,
     });

apps/webapp/test/engine/triggerTask.test.ts

@@ -267,6 +267,124 @@ describe("RunEngineTriggerTaskService", () => {
     await engine.quit();
   });
 
+  containerTest(
+    "routes scheduled-lineage runs to a separate worker queue that dequeues independently",
+    async ({ prisma, redisOptions }) => {
+      const engine = new RunEngine({
+        prisma,
+        worker: {
+          redis: redisOptions,
+          workers: 1,
+          tasksPerWorker: 10,
+          pollIntervalMs: 100,
+        },
+        queue: {
+          redis: redisOptions,
+          // Disable the background master-queue consumers so our manual
+          // processMasterQueueForEnvironment + dequeue calls are deterministic.
+          masterQueueConsumersDisabled: true,
+          processWorkerQueueDebounceMs: 50,
+        },
+        runLock: { redis: redisOptions },
+        machines: {
+          defaultMachine: "small-1x",
+          machines: {
+            "small-1x": {
+              name: "small-1x" as const,
+              cpu: 0.5,
+              memory: 0.5,
+              centsPerMs: 0.0001,
+            },
+          },
+          baseCostInCents: 0.0005,
+        },
+        tracer: trace.getTracer("test", "0.0.0"),
+      });
+
+      try {
+        const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+
+        // Turn the per-org split flag on in-memory — the resolver reads this
+        // object directly (no DB round-trip on the trigger hot path).
+        (authenticatedEnvironment.organization as { featureFlags?: unknown }).featureFlags = {
+          workerQueueScheduledSplitEnabled: true,
+        };
+
+        const taskIdentifier = "test-task";
+        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
+
+        const triggerTaskService = new RunEngineTriggerTaskService({
+          engine,
+          prisma,
+          payloadProcessor: new MockPayloadProcessor(),
+          queueConcern: new DefaultQueueManager(prisma, engine),
+          idempotencyKeyConcern: new IdempotencyKeyConcern(
+            prisma,
+            engine,
+            new MockTraceEventConcern()
+          ),
+          validator: new MockTriggerTaskValidator(),
+          traceEventConcern: new MockTraceEventConcern(),
+          tracer: trace.getTracer("test", "0.0.0"),
+          metadataMaximumSize: 1024 * 1024 * 1,
+        });
+
+        // A standard run (default triggerSource) stays on the region queue.
+        const standardResult = await triggerTaskService.call({
+          taskId: taskIdentifier,
+          environment: authenticatedEnvironment,
+          body: { payload: { kind: "standard" } },
+        });
+        assertNonNullable(standardResult);
+
+        // A scheduled run routes to the `<region>:scheduled` queue. Descendants
+        // would too, via rootTriggerSource propagation.
+        const scheduledResult = await triggerTaskService.call({
+          taskId: taskIdentifier,
+          environment: authenticatedEnvironment,
+          body: { payload: { kind: "scheduled" } },
+          options: { triggerSource: "schedule" },
+        });
+        assertNonNullable(scheduledResult);
+
+        const standardRun = await prisma.taskRun.findUniqueOrThrow({
+          where: { id: standardResult.run.id },
+        });
+        const scheduledRun = await prisma.taskRun.findUniqueOrThrow({
+          where: { id: scheduledResult.run.id },
+        });
+
+        // Producer routing: the persisted worker queue carries the class.
+        const baseWorkerQueue = standardRun.workerQueue;
+        expect(scheduledRun.workerQueue).toBe(`${baseWorkerQueue}:scheduled`);
+
+        // Move both runs from the env queue onto their respective worker queues.
+        await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10);
+        await setTimeout(500);
+
+        // Dequeue isolation: the scheduled queue yields only the scheduled run...
+        const dequeuedScheduled = await engine.dequeueFromWorkerQueue({
+          consumerId: "test-scheduled-consumer",
+          workerQueue: `${baseWorkerQueue}:scheduled`,
+        });
+        expect(dequeuedScheduled.length).toBe(1);
+        assertNonNullable(dequeuedScheduled[0]);
+        expect(dequeuedScheduled[0].run.id).toBe(scheduledResult.run.id);
+
+        // ...and the base queue yields only the standard run.
+        const dequeuedStandard = await engine.dequeueFromWorkerQueue({
+          consumerId: "test-standard-consumer",
+          workerQueue: baseWorkerQueue,
+        });
+        expect(dequeuedStandard.length).toBe(1);
+        assertNonNullable(dequeuedStandard[0]);
+        expect(dequeuedStandard[0].run.id).toBe(standardResult.run.id);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
   // The BatchQueue worker rebuilds body.options from Redis-stored items
   // (Record<string, unknown>), so the Phase-2 schema coercion doesn't apply
   // to in-flight items enqueued before the schema fix. The defensive