v4: dequeue performance improvements (split concurrency from dequeue) (#2127)

* WIP

* Run queue now works with the worker queue / master queue split

* Acking should also cause the master queue to be processed

* Convert run engine tests and run engine to use runQueue changes

* Include the util files in the test tsconfig

* coordinator target should be es2020 as well

* providers target 2020

* Fix the triggerTask tests in the webapp

* v4 now working with the new worker queues, and added the legacy master queue migration stuff

* report worker queue lengths via opentelemetry metrics

* Adding lock metrics

* Release concurrency bucket metrics

* • Updated RunQueue.removeEnvironmentQueuesFromMasterQueue() method signature to take runtimeEnvironmentId instead of masterQueue parameter
• Added automatic master queue shard calculation using this.keys.masterQueueKeyForEnvironment(runtimeEnvironmentId, this.shardCount) 
• Updated RunEngine wrapper method to use new runtimeEnvironmentId parameter
• Updated DeleteProjectService to call the method once per environment instead of once per master queue
• Simplified API by encapsulating master queue sharding logic within RunQueue class

* metrics now working, configure the run queue settings, additional metrics for run engine and redis-worker

* Fix CodeRabbit suggestions

* return undefined from dequeueFromWorkerQueue, not null

* Remove message from worker queue in certain circumstances when acking

* Update log

* Ensure master queue consumers cannot stop from a processing error, and make the consumer interval configurable via an env var

* Change how the run queue master queue consumers are disabled internally

* Fixed tests

* process the queue on nack

* Fix more tests

* Fix priority tests

* Fixed dequeueing test

Author: ericallam

.vscode/launch.json

@@ -146,7 +146,7 @@
       "type": "node-terminal",
       "request": "launch",
       "name": "Debug RunQueue tests",
-      "command": "pnpm run test ./src/run-queue/index.test.ts",
+      "command": "pnpm run test ./src/run-queue/index.test.ts --run",
       "cwd": "${workspaceFolder}/internal-packages/run-engine",
       "sourceMaps": true
     },

apps/coordinator/tsconfig.json

@@ -1,6 +1,6 @@
 {
   "compilerOptions": {
-    "target": "es2018",
+    "target": "es2020",
     "module": "commonjs",
     "esModuleInterop": true,
     "resolveJsonModule": true,

apps/docker-provider/tsconfig.json

@@ -1,6 +1,6 @@
 {
   "compilerOptions": {
-    "target": "es2018",
+    "target": "es2020",
     "module": "commonjs",
     "esModuleInterop": true,
     "forceConsistentCasingInFileNames": true,

apps/kubernetes-provider/tsconfig.json

@@ -1,6 +1,6 @@
 {
   "compilerOptions": {
-    "target": "es2018",
+    "target": "es2020",
     "module": "commonjs",
     "esModuleInterop": true,
     "forceConsistentCasingInFileNames": true,

apps/webapp/app/env.server.ts

@@ -324,6 +324,10 @@ const EnvironmentSchema = z.object({
   INTERNAL_OTEL_TRACE_DISABLED: z.string().default("0"),
 
   INTERNAL_OTEL_LOG_EXPORTER_URL: z.string().optional(),
+  INTERNAL_OTEL_METRIC_EXPORTER_URL: z.string().optional(),
+  INTERNAL_OTEL_METRIC_EXPORTER_AUTH_HEADERS: z.string().optional(),
+  INTERNAL_OTEL_METRIC_EXPORTER_ENABLED: z.string().default("0"),
+  INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS: z.coerce.number().int().default(30_000),
 
   ORG_SLACK_INTEGRATION_CLIENT_ID: z.string().optional(),
   ORG_SLACK_INTEGRATION_CLIENT_SECRET: z.string().optional(),
@@ -460,8 +464,12 @@ const EnvironmentSchema = z.object({
   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_RUN_QUEUE_SHARD_COUNT: z.coerce.number().int().default(4),
   RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
   RUN_ENGINE_RETRY_WARM_START_THRESHOLD_MS: z.coerce.number().int().default(30_000),
+  RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),
+  RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
+  RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS: z.coerce.number().int().default(500),
 
   RUN_ENGINE_WORKER_REDIS_HOST: z
     .string()
@@ -617,6 +625,7 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_RELEASE_CONCURRENCY_BATCH_SIZE: z.coerce.number().int().default(10),
 
   RUN_ENGINE_WORKER_ENABLED: z.string().default("1"),
+  RUN_ENGINE_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
 
   /** How long should the presence ttl last */
   DEV_PRESENCE_SSE_TIMEOUT: z.coerce.number().int().default(30_000),

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -145,8 +145,7 @@ export class SpanPresenter extends BasePresenter {
           },
         },
         engine: true,
-        masterQueue: true,
-        secondaryMasterQueue: true,
+        workerQueue: true,
         error: true,
         output: true,
         outputType: true,
@@ -364,8 +363,7 @@ export class SpanPresenter extends BasePresenter {
       maxDurationInSeconds: getMaxDuration(run.maxDurationInSeconds),
       batch: run.batch ? { friendlyId: run.batch.friendlyId } : undefined,
       engine: run.engine,
-      masterQueue: run.masterQueue,
-      secondaryMasterQueue: run.secondaryMasterQueue,
+      workerQueue: run.workerQueue,
       spanId: run.spanId,
       isCached: !!span.originalRun,
     };

apps/webapp/app/routes/admin.api.v1.migrate-legacy-master-queues.ts

@@ -0,0 +1,37 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { engine } from "~/v3/runEngine.server";
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  try {
+    await engine.migrateLegacyMasterQueues();
+
+    return json({
+      success: true,
+    });
+  } catch (error) {
+    return json({ error: error instanceof Error ? error.message : error }, { status: 400 });
+  }
+}

apps/webapp/app/routes/engine.v1.dev.dequeue.ts

@@ -1,92 +1,22 @@
 import { json } from "@remix-run/server-runtime";
-import { DequeuedMessage, DevDequeueRequestBody, MachineResources } from "@trigger.dev/core/v3";
-import { BackgroundWorkerId } from "@trigger.dev/core/v3/isomorphic";
-import { env } from "~/env.server";
+import { DevDequeueRequestBody } from "@trigger.dev/core/v3";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { engine } from "~/v3/runEngine.server";
 
 const { action } = createActionApiRoute(
   {
-    body: DevDequeueRequestBody,
+    body: DevDequeueRequestBody, // Even though we don't use it, we need to keep it for backwards compatibility
     maxContentLength: 1024 * 10, // 10KB
     method: "POST",
   },
-  async ({ authentication, body }) => {
-    //we won't return more runs than this in one API call
-    let maxDequeueCount = env.DEV_DEQUEUE_MAX_RUNS_PER_PULL;
-
-    //we can't use more than the max resources
-    const availableResources = body.maxResources ?? {
-      cpu: 8,
-      memory: 16,
-    };
-
-    let dequeuedMessages: DequeuedMessage[] = [];
-
-    //we need to check the current worker, because a run might have been locked to it
-    const workers = body.oldWorkers.concat(body.currentWorker);
-
-    //first we want to clear out old runs
-    for (const worker of workers) {
-      //dequeue
-      const latestResult = await engine.dequeueFromBackgroundWorkerMasterQueue({
-        consumerId: authentication.environment.id,
-        //specific version
-        backgroundWorkerId: BackgroundWorkerId.toId(worker),
-        maxRunCount: maxDequeueCount,
-        maxResources: availableResources,
-      });
-
-      //add runs to the array
-      dequeuedMessages.push(...latestResult);
-
-      //update availableResources
-      const consumedResources = latestResult.reduce(
-        (acc, r) => {
-          return {
-            cpu: acc.cpu + r.run.machine.cpu,
-            memory: acc.memory + r.run.machine.memory,
-          };
-        },
-        { cpu: 0, memory: 0 }
-      );
-      updateAvailableResources(availableResources, consumedResources);
-
-      //update maxDequeueCount
-      maxDequeueCount -= latestResult.length;
-
-      //if we have no resources left, we exit the loop
-      if (!hasAvailableResources(availableResources)) break;
-      //we've already dequeued the max number of runs
-      if (maxDequeueCount <= 0) break;
-    }
-
-    //dequeue from the current version if we still have space
-    if (hasAvailableResources(availableResources) && maxDequeueCount > 0) {
-      const latestResult = await engine.dequeueFromEnvironmentMasterQueue({
-        consumerId: authentication.environment.id,
-        //current dev version (no specific version specified)
-        environmentId: authentication.environment.id,
-        maxRunCount: maxDequeueCount,
-        maxResources: availableResources,
-      });
-      dequeuedMessages.push(...latestResult);
-    }
+  async ({ authentication }) => {
+    const dequeuedMessages = await engine.dequeueFromEnvironmentWorkerQueue({
+      consumerId: authentication.environment.id,
+      environmentId: authentication.environment.id,
+    });
 
     return json({ dequeuedMessages }, { status: 200 });
   }
 );
 
-function updateAvailableResources(
-  availableResources: MachineResources,
-  resources: MachineResources
-) {
-  availableResources.cpu -= resources.cpu;
-  availableResources.memory -= resources.memory;
-}
-
-function hasAvailableResources(availableResources: MachineResources) {
-  return availableResources.cpu > 0 && availableResources.memory > 0;
-}
-
 export { action };

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

@@ -1,10 +1,9 @@
 import { json, TypedResponse } from "@remix-run/server-runtime";
-import { CURRENT_DEPLOYMENT_LABEL } from "@trigger.dev/core/v3/isomorphic";
 import { WorkerApiDequeueResponseBody } from "@trigger.dev/core/v3/workers";
 import { z } from "zod";
-import { $replica, prisma } from "~/db.server";
 import { createLoaderWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
+// Keep this route for backwards compatibility
 export const loader = createLoaderWorkerApiRoute(
   {
     params: z.object({
@@ -14,55 +13,7 @@ export const loader = createLoaderWorkerApiRoute(
       maxRunCount: z.coerce.number().optional(),
     }),
   },
-  async ({
-    authenticatedWorker,
-    params,
-    searchParams,
-  }): Promise<TypedResponse<WorkerApiDequeueResponseBody>> => {
-    const deployment = await $replica.workerDeployment.findUnique({
-      where: {
-        friendlyId: params.deploymentFriendlyId,
-      },
-      include: {
-        worker: true,
-      },
-    });
-
-    if (!deployment) {
-      throw new Error("Deployment not found");
-    }
-
-    if (!deployment.worker) {
-      throw new Error("Worker not found");
-    }
-
-    const dequeuedMessages = (await isCurrentDeployment(deployment.id, deployment.environmentId))
-      ? await authenticatedWorker.dequeueFromEnvironment(
-          deployment.worker.id,
-          deployment.environmentId
-        )
-      : await authenticatedWorker.dequeueFromVersion(
-          deployment.worker.id,
-          searchParams.maxRunCount
-        );
-
-    return json(dequeuedMessages);
+  async (): Promise<TypedResponse<WorkerApiDequeueResponseBody>> => {
+    return json([]);
   }
 );
-
-async function isCurrentDeployment(deploymentId: string, environmentId: string): Promise<boolean> {
-  const promotion = await prisma.workerDeploymentPromotion.findUnique({
-    where: {
-      environmentId_label: {
-        environmentId,
-        label: CURRENT_DEPLOYMENT_LABEL,
-      },
-    },
-  });
-
-  if (!promotion) {
-    return false;
-  }
-
-  return promotion.deploymentId === deploymentId;
-}

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

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