feat(engine): enqueue fast path; skip the queue under certain conditions

Author: ericallam

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

@@ -254,9 +254,9 @@ export class DefaultQueueManager implements QueueManager {
   async getWorkerQueue(
     environment: AuthenticatedEnvironment,
     regionOverride?: string
-  ): Promise<string | undefined> {
+  ): Promise<{ masterQueue: string; enableFastPath: boolean } | undefined> {
     if (environment.type === "DEVELOPMENT") {
-      return environment.id;
+      return { masterQueue: environment.id, enableFastPath: true };
     }
 
     const workerGroupService = new WorkerGroupService({
@@ -279,7 +279,10 @@ export class DefaultQueueManager implements QueueManager {
       throw new ServiceValidationError("No worker group found");
     }
 
-    return workerGroup.masterQueue;
+    return {
+      masterQueue: workerGroup.masterQueue,
+      enableFastPath: workerGroup.enableFastPath,
+    };
   }
 }
 

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

@@ -290,7 +290,12 @@ export class RunEngineTriggerTaskService {
 
       const depth = parentRun ? parentRun.depth + 1 : 0;
 
-      const workerQueue = await this.queueConcern.getWorkerQueue(environment, body.options?.region);
+      const workerQueueResult = await this.queueConcern.getWorkerQueue(
+        environment,
+        body.options?.region
+      );
+      const workerQueue = workerQueueResult?.masterQueue;
+      const enableFastPath = workerQueueResult?.enableFastPath ?? false;
 
       // Build annotations for this run
       const triggerSource = options.triggerSource ?? "api";
@@ -344,6 +349,7 @@ export class RunEngineTriggerTaskService {
                 queue: queueName,
                 lockedQueueId,
                 workerQueue,
+                enableFastPath,
                 isTest: body.options?.test ?? false,
                 delayUntil,
                 queuedAt: delayUntil ? undefined : new Date(),

apps/webapp/app/runEngine/types.ts

@@ -70,7 +70,7 @@ export interface QueueManager {
   getWorkerQueue(
     env: AuthenticatedEnvironment,
     regionOverride?: string
-  ): Promise<string | undefined>;
+  ): Promise<{ masterQueue: string; enableFastPath: boolean } | undefined>;
 }
 
 export interface PayloadProcessor {

internal-packages/database/prisma/migrations/20260330210247_add_enable_fast_path_to_worker_instance_group/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "public"."WorkerInstanceGroup" ADD COLUMN     "enableFastPath" BOOLEAN NOT NULL DEFAULT false;

internal-packages/database/prisma/schema.prisma

@@ -1331,6 +1331,10 @@ model WorkerInstanceGroup {
 
   workloadType WorkloadType @default(CONTAINER)
 
+  /// When true, runs enqueued to this worker queue may skip the intermediate queue
+  /// and be pushed directly to the worker queue when concurrency is available.
+  enableFastPath Boolean @default(false)
+
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
 }

internal-packages/run-engine/src/engine/index.ts

@@ -462,6 +462,7 @@ export class RunEngine {
       cliVersion,
       concurrencyKey,
       workerQueue,
+      enableFastPath,
       queue,
       lockedQueueId,
       isTest,
@@ -799,6 +800,7 @@ export class RunEngine {
               tx: prisma,
               skipRunLock: true,
               includeTtl: true,
+              enableFastPath,
             });
           } catch (enqueueError) {
             this.logger.error("engine.trigger(): failed to schedule TTL or enqueue run", {

internal-packages/run-engine/src/engine/systems/enqueueSystem.ts

@@ -36,6 +36,7 @@ export class EnqueueSystem {
     runnerId,
     skipRunLock,
     includeTtl = false,
+    enableFastPath = false,
   }: {
     run: TaskRun;
     env: MinimalAuthenticatedEnvironment;
@@ -57,6 +58,8 @@ export class EnqueueSystem {
     skipRunLock?: boolean;
     /** When true, include TTL in the queued message (only for first enqueue from trigger). Default false. */
     includeTtl?: boolean;
+    /** When true, allow the queue to push directly to worker queue if concurrency is available. */
+    enableFastPath?: boolean;
   }) {
     const prisma = tx ?? this.$.prisma;
 
@@ -98,6 +101,7 @@ export class EnqueueSystem {
       await this.$.runQueue.enqueueMessage({
         env,
         workerQueue,
+        enableFastPath,
         message: {
           runId: run.id,
           taskIdentifier: run.taskIdentifier,

internal-packages/run-engine/src/engine/types.ts

@@ -181,6 +181,9 @@ export type TriggerParams = {
   cliVersion?: string;
   concurrencyKey?: string;
   workerQueue?: string;
+  /** When true, the run queue may push directly to the worker queue if concurrency is available.
+   *  Gated per WorkerInstanceGroup (production) or always true (development). */
+  enableFastPath?: boolean;
   queue: string;
   lockedQueueId?: string;
   isTest: boolean;