Configure the new queue selection strategy in the webapp and get it all building and typechecks passing

Author: ericallam

apps/webapp/app/env.server.ts

@@ -419,6 +419,12 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_TIMEOUT_EXECUTING: z.coerce.number().int().default(60_000),
   RUN_ENGINE_TIMEOUT_EXECUTING_WITH_WAITPOINTS: z.coerce.number().int().default(60_000),
   RUN_ENGINE_DEBUG_WORKER_NOTIFICATIONS: z.coerce.boolean().default(false),
+  RUN_ENGINE_PARENT_QUEUE_LIMIT: z.coerce.number().int().default(1000),
+  RUN_ENGINE_CONCURRENCY_LIMIT_BIAS: z.coerce.number().default(0.75),
+  RUN_ENGINE_AVAILABLE_CAPACITY_BIAS: z.coerce.number().default(0.3),
+  RUN_ENGINE_QUEUE_AGE_RANDOMIZATION_BIAS: z.coerce.number().default(0.25),
+  RUN_ENGINE_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
+  RUN_ENGINE_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
 
   RUN_ENGINE_WORKER_REDIS_HOST: z
     .string()

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

@@ -43,6 +43,17 @@ function createRunEngine() {
         enableAutoPipelining: true,
         ...(env.RUN_ENGINE_RUN_QUEUE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
       },
+      queueSelectionStrategyOptions: {
+        parentQueueLimit: env.RUN_ENGINE_PARENT_QUEUE_LIMIT,
+        biases: {
+          concurrencyLimitBias: env.RUN_ENGINE_CONCURRENCY_LIMIT_BIAS,
+          availableCapacityBias: env.RUN_ENGINE_AVAILABLE_CAPACITY_BIAS,
+          queueAgeRandomization: env.RUN_ENGINE_QUEUE_AGE_RANDOMIZATION_BIAS,
+        },
+        reuseSnapshotCount: env.RUN_ENGINE_REUSE_SNAPSHOT_COUNT,
+        maximumEnvCount: env.RUN_ENGINE_MAXIMUM_ENV_COUNT,
+        tracer,
+      },
     },
     runLock: {
       redis: {

apps/webapp/remix.config.js

@@ -26,6 +26,9 @@ module.exports = {
     "superjson",
     "prismjs/components/prism-json",
     "prismjs/components/prism-typescript",
+    "@internal/run-engine",
+    "@internal/redis",
+    "@internal/tracing",
   ],
   browserNodeBuiltinsPolyfill: { modules: { path: true, os: true, crypto: true } },
   watchPaths: async () => {

apps/webapp/tsconfig.json

@@ -40,7 +40,9 @@
       "@internal/redis-worker": ["../../internal-packages/redis-worker/src/index"],
       "@internal/redis-worker/*": ["../../internal-packages/redis-worker/src/*"],
       "@internal/redis": ["../../internal-packages/redis/src/index"],
-      "@internal/redis/*": ["../../internal-packages/redis/src/*"]
+      "@internal/redis/*": ["../../internal-packages/redis/src/*"],
+      "@internal/tracing": ["../../internal-packages/tracing/src/index"],
+      "@internal/tracing/*": ["../../internal-packages/tracing/src/*"]
     },
     "noEmit": true
   }

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

@@ -53,7 +53,7 @@ import { nanoid } from "nanoid";
 import { EventEmitter } from "node:events";
 import { z } from "zod";
 import { RunQueue } from "../run-queue/index.js";
-import { FairDequeuingStrategy } from "../run-queue/fairDequeuingStrategy.js";
+import { FairQueueSelectionStrategy } from "../run-queue/fairQueueSelectionStrategy.js";
 import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
 import { MAX_TASK_RUN_ATTEMPTS } from "./consts.js";
 import { getRunWithBackgroundWorkerTasks } from "./db/worker.js";
@@ -159,9 +159,10 @@ export class RunEngine {
       name: "rq",
       tracer: trace.getTracer("rq"),
       keys,
-      queuePriorityStrategy: new FairDequeuingStrategy({
+      queueSelectionStrategy: new FairQueueSelectionStrategy({
         keys,
         redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
+        defaultEnvConcurrencyLimit: options.queue?.defaultEnvConcurrency ?? 10,
       }),
       defaultEnvConcurrency: options.queue?.defaultEnvConcurrency ?? 10,
       logger: new Logger("RunQueue", "debug"),

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

@@ -1,14 +1,16 @@
-import Redlock, { RedlockAbortSignal } from "redlock";
+// import { default: Redlock } from "redlock";
+const { default: Redlock } = require("redlock");
 import { AsyncLocalStorage } from "async_hooks";
 import { Redis } from "@internal/redis";
+import * as redlock from "redlock";
 
 interface LockContext {
   resources: string;
-  signal: RedlockAbortSignal;
+  signal: redlock.RedlockAbortSignal;
 }
 
 export class RunLocker {
-  private redlock: Redlock;
+  private redlock: InstanceType<typeof redlock.default>;
   private asyncLocalStorage: AsyncLocalStorage<LockContext>;
 
   constructor(options: { redis: Redis }) {
@@ -26,7 +28,7 @@ export class RunLocker {
   async lock<T>(
     resources: string[],
     duration: number,
-    routine: (signal: RedlockAbortSignal) => Promise<T>
+    routine: (signal: redlock.RedlockAbortSignal) => Promise<T>
   ): Promise<T> {
     const currentContext = this.asyncLocalStorage.getStore();
     const joinedResources = resources.sort().join(",");

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

@@ -4,6 +4,7 @@ import { MachinePreset, MachinePresetName, QueueOptions, RetryOptions } from "@t
 import { PrismaClient } from "@trigger.dev/database";
 import { type RedisOptions } from "@internal/redis";
 import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
+import { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelectionStrategy.js";
 
 export type RunEngineOptions = {
   prisma: PrismaClient;
@@ -21,6 +22,10 @@ export type RunEngineOptions = {
     redis: RedisOptions;
     retryOptions?: RetryOptions;
     defaultEnvConcurrency?: number;
+    queueSelectionStrategyOptions?: Pick<
+      FairQueueSelectionStrategyOptions,
+      "parentQueueLimit" | "tracer" | "biases" | "reuseSnapshotCount" | "maximumEnvCount"
+    >;
   };
   runLock: {
     redis: RedisOptions;

internal-packages/run-engine/src/run-queue/fairDequeuingStrategy.test.ts renamed to internal-packages/run-engine/src/run-queue/fairQueueSelectionStrategy.test.ts

@@ -1,6 +1,6 @@
 import { redisTest } from "@internal/testcontainers";
 import { describe, expect, vi } from "vitest";
-import { FairDequeuingStrategy } from "./fairDequeuingStrategy.js";
+import { FairQueueSelectionStrategy } from "./fairQueueSelectionStrategy.js";
 import { RunQueueFullKeyProducer } from "./keyProducer.js";
 import { createRedisClient, Redis, RedisOptions } from "@internal/redis";
 import { EnvQueues, RunQueueKeyProducer } from "./types.js";
@@ -13,7 +13,7 @@ describe("FairDequeuingStrategy", () => {
     "should distribute a single queue from a single env",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 5,
@@ -47,7 +47,7 @@ describe("FairDequeuingStrategy", () => {
 
   redisTest("should respect env concurrency limits", async ({ redisOptions: redis }) => {
     const keyProducer = new RunQueueFullKeyProducer();
-    const strategy = new FairDequeuingStrategy({
+    const strategy = new FairQueueSelectionStrategy({
       redis,
       keys: keyProducer,
       defaultEnvConcurrencyLimit: 2,
@@ -80,7 +80,7 @@ describe("FairDequeuingStrategy", () => {
     "should give extra concurrency when the env has reserve concurrency",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 2,
@@ -126,7 +126,7 @@ describe("FairDequeuingStrategy", () => {
 
   redisTest("should respect parentQueueLimit", async ({ redisOptions: redis }) => {
     const keyProducer = new RunQueueFullKeyProducer();
-    const strategy = new FairDequeuingStrategy({
+    const strategy = new FairQueueSelectionStrategy({
       redis,
       keys: keyProducer,
       defaultEnvConcurrencyLimit: 5,
@@ -185,7 +185,7 @@ describe("FairDequeuingStrategy", () => {
     "should reuse snapshots across calls for the same consumer",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 5,
@@ -282,7 +282,7 @@ describe("FairDequeuingStrategy", () => {
     "should fairly distribute queues across environments over time",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 5,
@@ -440,7 +440,7 @@ describe("FairDequeuingStrategy", () => {
     "should shuffle environments while maintaining age order within environments",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 5,
@@ -623,7 +623,7 @@ describe("FairDequeuingStrategy", () => {
       const strategies = Array.from(
         { length: numStrategies },
         (_, i) =>
-          new FairDequeuingStrategy({
+          new FairQueueSelectionStrategy({
             redis,
             keys: keyProducer,
             defaultEnvConcurrencyLimit: 5,
@@ -708,7 +708,7 @@ describe("FairDequeuingStrategy", () => {
 
       // Helper function to run iterations with a specific age influence
       async function runWithQueueAgeRandomization(queueAgeRandomization: number) {
-        const strategy = new FairDequeuingStrategy({
+        const strategy = new FairQueueSelectionStrategy({
           redis,
           keys: keyProducer,
           defaultEnvConcurrencyLimit: 5,
@@ -799,7 +799,7 @@ describe("FairDequeuingStrategy", () => {
     "should respect maximumEnvCount and select envs based on queue ages",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 5,
@@ -942,7 +942,7 @@ describe("FairDequeuingStrategy", () => {
     "should not overly bias picking environments when queue have priority offset ages",
     async ({ redisOptions: redis }) => {
       const keyProducer = new RunQueueFullKeyProducer();
-      const strategy = new FairDequeuingStrategy({
+      const strategy = new FairQueueSelectionStrategy({
         redis,
         keys: keyProducer,
         defaultEnvConcurrencyLimit: 5,

internal-packages/run-engine/src/run-queue/fairDequeuingStrategy.ts renamed to internal-packages/run-engine/src/run-queue/fairQueueSelectionStrategy.ts

@@ -1,17 +1,17 @@
+import { createRedisClient, Redis, type RedisOptions } from "@internal/redis";
+import { startSpan, type Tracer } from "@internal/tracing";
 import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } from "@unkey/cache";
 import { MemoryStore } from "@unkey/cache/stores";
 import { randomUUID } from "crypto";
+import seedrandom from "seedrandom";
 import {
   EnvDescriptor,
   EnvQueues,
-  RunQueueFairDequeueStrategy,
   RunQueueKeyProducer,
+  RunQueueSelectionStrategy,
 } from "./types.js";
-import seedrandom from "seedrandom";
-import { startSpan, type Tracer } from "@internal/tracing";
-import { Redis, type RedisOptions, createRedisClient } from "@internal/redis";
 
-export type FairDequeuingStrategyBiases = {
+export type FairQueueSelectionStrategyBiases = {
   /**
    * How much to bias towards environments with higher concurrency limits
    * 0 = no bias, 1 = full bias based on limit differences
@@ -33,7 +33,7 @@ export type FairDequeuingStrategyBiases = {
   queueAgeRandomization: number;
 };
 
-export type FairDequeuingStrategyOptions = {
+export type FairQueueSelectionStrategyOptions = {
   redis: RedisOptions;
   keys: RunQueueKeyProducer;
   defaultEnvConcurrencyLimit?: number;
@@ -44,7 +44,7 @@ export type FairDequeuingStrategyOptions = {
    * Configure biasing for environment shuffling
    * If not provided, no biasing will be applied (completely random shuffling)
    */
-  biases?: FairDequeuingStrategyBiases;
+  biases?: FairQueueSelectionStrategyBiases;
   reuseSnapshotCount?: number;
   maximumEnvCount?: number;
 };
@@ -79,13 +79,13 @@ const emptyFairQueueSnapshot: FairQueueSnapshot = {
   queues: [],
 };
 
-const defaultBiases: FairDequeuingStrategyBiases = {
+const defaultBiases: FairQueueSelectionStrategyBiases = {
   concurrencyLimitBias: 0,
   availableCapacityBias: 0,
   queueAgeRandomization: 0, // Default to completely age-based ordering
 };
 
-export class FairDequeuingStrategy implements RunQueueFairDequeueStrategy {
+export class FairQueueSelectionStrategy implements RunQueueSelectionStrategy {
   private _cache: UnkeyCache<{
     concurrencyLimit: number;
   }>;
@@ -100,7 +100,7 @@ export class FairDequeuingStrategy implements RunQueueFairDequeueStrategy {
   private _defaultEnvConcurrencyLimit: number;
   private _parentQueueLimit: number;
 
-  constructor(private options: FairDequeuingStrategyOptions) {
+  constructor(private options: FairQueueSelectionStrategyOptions) {
     const ctx = new DefaultStatefulContext();
     const memory = new MemoryStore({ persistentMap: new Map() });
 
@@ -612,7 +612,7 @@ export class FairDequeuingStrategy implements RunQueueFairDequeueStrategy {
   }
 }
 
-export class NoopFairDequeuingStrategy implements RunQueueFairDequeueStrategy {
+export class NoopFairDequeuingStrategy implements RunQueueSelectionStrategy {
   async distributeFairQueuesFromParentQueue(
     parentQueue: string,
     consumerId: string