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

## Summary

Currently, every triggered run follows a two-step path through Redis:

1. **Enqueue** — A Lua script atomically adds the message to a queue
sorted set (ordered by priority-adjusted timestamp)
2. **Dequeue** — A debounced `processQueueForWorkerQueue` job fires
~500ms later, checks concurrency limits, removes the message from the
sorted set, and pushes it to a worker queue (Redis list) where workers
pick it up via `BLPOP`

This means every run pays at least ~500ms of latency between being
triggered and being available for a worker to execute, even when the
queue is empty and concurrency is wide open.

### What changed

The enqueue Lua scripts now atomically decide whether to **skip the
queue sorted set entirely** and push directly to the worker queue. This
happens inside the same Lua script that handles normal enqueue, so the
decision is atomic with respect to concurrency bookkeeping.

A run takes the **fast path** when all of these are true:
- **Fast path is enabled** for this worker queue (gated per
`WorkerInstanceGroup`)
- **No available messages** in the queue (`ZRANGEBYSCORE` finds nothing
with score ≤ now) — this respects priority ordering and allows fast path
even when the queue has future-scored messages (e.g. nacked retries with
delay)
- **Environment concurrency** has capacity
- **Queue concurrency** has capacity (including per-concurrency-key
limits for CK queues)

When the fast path is taken:
- The message is stored and pushed directly to the worker queue
(`RPUSH`)
- Concurrency slots are claimed (`SADD` to the same sets used by the
normal dequeue path)
- The `processQueueForWorkerQueue` job is **not scheduled** (no work to
do)
- TTL sorted set is skipped (the `expireRun` worker job handles TTL
independently)

When any condition fails, the existing slow path runs unchanged.

### Rollout gating

- **Development environments**: Fast path is always enabled
- **Production environments**: Gated by a new `enableFastPath` boolean
on `WorkerInstanceGroup` (defaults to `false`), allowing
region-by-region rollout

### Rolling deploy safety

Each process registers its own Lua scripts via `defineCommand`
(identified by SHA hash). Old and new processes never share scripts. The
Redis data structures are fully compatible in both directions — ack,
nack, and release operations work identically regardless of which path a
message took.

## Test plan

- [x] Fast path taken when queue is empty and concurrency available
- [x] Slow path when `enableFastPath` is false
- [x] Slow path when queue has available messages (respects priority
ordering)
- [x] Fast path when queue only has future-scored messages
- [x] Slow path when env concurrency is full
- [x] Fast-path message can be acknowledged correctly
- [x] Fast-path message can be nacked and re-enqueued to the queue
sorted set
- [x] Run all existing run-queue tests (ack, nack, CK, concurrency
sweeper, dequeue) to verify no regressions
- [x] Typecheck passes for run-engine and webapp

Author: ericallam

.server-changes/enqueue-fast-path.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Reduce run start latency by skipping the intermediate queue when concurrency is available. This optimization is rolled out per-region and enabled automatically for development environments.

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

@@ -295,9 +295,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({
@@ -320,7 +320,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

@@ -298,7 +298,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";
@@ -352,6 +357,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

@@ -1333,6 +1333,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;