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

Author: ericallam

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());
   }
 );

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -700,12 +700,8 @@ function RunBody({
                 {isAdmin && (
                   <>
                     <Property.Item>
-                      <Property.Label>Primary master queue</Property.Label>
-                      <Property.Value>{run.masterQueue}</Property.Value>
-                    </Property.Item>
-                    <Property.Item>
-                      <Property.Label>Secondary master queue</Property.Label>
-                      <Property.Value>{run.secondaryMasterQueue ?? "–"}</Property.Value>
+                      <Property.Label>Worker queue</Property.Label>
+                      <Property.Value>{run.workerQueue}</Property.Value>
                     </Property.Item>
                   </>
                 )}

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

@@ -198,7 +198,7 @@ export class DefaultQueueManager implements QueueManager {
 
   async getWorkerQueue(environment: AuthenticatedEnvironment): Promise<string | undefined> {
     if (environment.type === "DEVELOPMENT") {
-      return;
+      return environment.id;
     }
 
     const workerGroupService = new WorkerGroupService({

apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts

@@ -1,5 +1,5 @@
 import { customAlphabet } from "nanoid";
-import { WithRunEngine, WithRunEngineOptions } from "../baseService.server";
+import { ServiceValidationError, WithRunEngine, WithRunEngineOptions } from "../baseService.server";
 import { createHash, timingSafeEqual } from "crypto";
 import { logger } from "~/services/logger.server";
 import {
@@ -537,19 +537,11 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     });
   }
 
-  async dequeue({
-    maxRunCount = 10,
-    maxResources,
-  }: {
-    maxRunCount?: number;
-    maxResources?: MachineResources;
-  } = {}): Promise<DequeuedMessage[]> {
+  async dequeue(): Promise<DequeuedMessage[]> {
     if (this.type === WorkerInstanceGroupType.MANAGED) {
-      return await this._engine.dequeueFromMasterQueue({
+      return await this._engine.dequeueFromWorkerQueue({
         consumerId: this.workerInstanceId,
-        masterQueue: this.masterQueue,
-        maxRunCount,
-        maxResources,
+        workerQueue: this.masterQueue,
         workerId: this.workerInstanceId,
         runnerId: this.runnerId,
       });
@@ -572,51 +564,21 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     });
 
     if (this.isLatestDeployment) {
-      return await this._engine.dequeueFromEnvironmentMasterQueue({
+      return await this._engine.dequeueFromEnvironmentWorkerQueue({
         consumerId: this.workerInstanceId,
         environmentId: this.environment.id,
-        maxRunCount,
-        backgroundWorkerId: this.backgroundWorkerId,
         workerId: this.workerInstanceId,
         runnerId: this.runnerId,
       });
     }
 
-    return await this._engine.dequeueFromBackgroundWorkerMasterQueue({
-      consumerId: this.workerInstanceId,
-      backgroundWorkerId: this.backgroundWorkerId,
-      maxRunCount,
-      workerId: this.workerInstanceId,
-      runnerId: this.runnerId,
-    });
-  }
-
-  /** Allows managed workers to dequeue from a specific version */
-  async dequeueFromVersion(
-    backgroundWorkerId: string,
-    maxRunCount = 1
-  ): Promise<DequeuedMessage[]> {
-    if (this.type !== WorkerInstanceGroupType.MANAGED) {
-      logger.error("[AuthenticatedWorkerInstance] Worker instance is not managed", {
-        ...this.toJSON(),
-      });
-      return [];
-    }
-
-    return await this._engine.dequeueFromBackgroundWorkerMasterQueue({
-      consumerId: this.workerInstanceId,
-      backgroundWorkerId,
-      maxRunCount,
-      workerId: this.workerInstanceId,
-      runnerId: this.runnerId,
-    });
+    throw new ServiceValidationError("Unmanaged workers cannot dequeue from a specific version");
   }
 
   /** Allows managed workers to dequeue from a specific environment */
   async dequeueFromEnvironment(
     backgroundWorkerId: string,
-    environmentId: string,
-    maxRunCount = 1
+    environmentId: string
   ): Promise<DequeuedMessage[]> {
     if (this.type !== WorkerInstanceGroupType.MANAGED) {
       logger.error("[AuthenticatedWorkerInstance] Worker instance is not managed", {
@@ -625,11 +587,10 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
       return [];
     }
 
-    return await this._engine.dequeueFromEnvironmentMasterQueue({
+    return await this._engine.dequeueFromEnvironmentWorkerQueue({
       consumerId: this.workerInstanceId,
       backgroundWorkerId,
       environmentId,
-      maxRunCount,
       workerId: this.workerInstanceId,
       runnerId: this.runnerId,
     });