feat(run-engine): fair virtual-time scheduling for the concurrency-key dequeue

.server-changes/2026-07-24-ck-fair-scheduling.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+The run queue can now serve concurrency keys fairly, so one key with a large backlog no longer starves runs waiting on other keys. It is opt-in and off by default.

apps/webapp/app/env.server.ts

@@ -978,6 +978,13 @@ const EnvironmentSchema = z
     RUN_ENGINE_TTL_CONSUMERS_DISABLED: BoolEnv.default(false),
     RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS: z.coerce.number().int().default(5_000),
 
+    // Fair (virtual-time) ordering across concurrency-key variants of a base queue.
+    // Off by default; when off the run queue behaves exactly as before.
+    RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED: BoolEnv.default(false),
+    RUN_ENGINE_CK_VTIME_QUANTUM: z.coerce.number().int().positive().default(1),
+    RUN_ENGINE_CK_VTIME_WINDOW_MULTIPLIER: z.coerce.number().int().positive().default(3),
+    RUN_ENGINE_CK_VTIME_STATE_TTL_SECONDS: z.coerce.number().int().positive().default(86400),
+
     /** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
      *  will use this as their TTL, and runs with a TTL larger than this will be clamped. */
     RUN_ENGINE_DEFAULT_MAX_TTL: z.string().optional(),

apps/webapp/app/v3/runEngine.server.ts

@@ -108,6 +108,14 @@ function createRunEngine() {
         batchMaxSize: env.RUN_ENGINE_TTL_WORKER_BATCH_MAX_SIZE,
         batchMaxWaitMs: env.RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS,
       },
+      ckVirtualTimeScheduling: env.RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED
+        ? {
+            enabled: true,
+            quantum: env.RUN_ENGINE_CK_VTIME_QUANTUM,
+            scanWindowMultiplier: env.RUN_ENGINE_CK_VTIME_WINDOW_MULTIPLIER,
+            stateTtlSeconds: env.RUN_ENGINE_CK_VTIME_STATE_TTL_SECONDS,
+          }
+        : undefined,
     },
     runLock: {
       redis: {

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

@@ -235,6 +235,7 @@ export class RunEngine {
             workerItemsSuffix: "ttl-worker:{queue:ttl-expiration:}items",
             visibilityTimeoutMs: options.queue?.ttlSystem?.visibilityTimeoutMs ?? 30_000,
           },
+      ckVirtualTimeScheduling: options.queue?.ckVirtualTimeScheduling,
     });
 
     this.worker = new Worker({

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

@@ -16,7 +16,7 @@ import {
 } from "@trigger.dev/redis-worker";
 import type { ControlPlaneResolver } from "./controlPlaneResolver.js";
 import type { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelectionStrategy.js";
-import type { RunQueueMetricsEmitter } from "../run-queue/index.js";
+import type { RunQueueMetricsEmitter, RunQueueOptions } from "../run-queue/index.js";
 import type { MinimalAuthenticatedEnvironment } from "../shared/index.js";
 import type { LockRetryConfig } from "./locking.js";
 import type { workerCatalog } from "./workerCatalog.js";
@@ -126,6 +126,9 @@ export type RunEngineOptions = {
       /** Max time (ms) to wait for more items before flushing a batch (default: 5000) */
       batchMaxWaitMs?: number;
     };
+    /** Fair (virtual-time) ordering across concurrency-key variants of a base queue.
+     *  Passed through to RunQueue; off by default (undefined = today's behaviour). */
+    ckVirtualTimeScheduling?: RunQueueOptions["ckVirtualTimeScheduling"];
   };
   runLock: {
     redis: RedisOptions;