WIP

Author: ericallam

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

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

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -234,7 +234,7 @@ export class RunEngineTriggerTaskService {
         lockedQueueId,
       });
 
-      const masterQueue = await this.queueConcern.getMasterQueue(environment);
+      const masterQueue = await this.queueConcern.getWorkerQueue(environment);
 
       try {
         return await this.traceEventConcern.traceRun(triggerRequest, async (event) => {
@@ -271,7 +271,7 @@ export class RunEngineTriggerTaskService {
                   concurrencyKey: body.options?.concurrencyKey,
                   queue: queueName,
                   lockedQueueId,
-                  masterQueue: masterQueue,
+                  workerQueue: masterQueue,
                   isTest: body.options?.test ?? false,
                   delayUntil,
                   queuedAt: delayUntil ? undefined : new Date(),

apps/webapp/app/runEngine/types.ts

@@ -66,7 +66,7 @@ export interface QueueManager {
   ): Promise<QueueProperties>;
   getQueueName(request: TriggerTaskRequest): Promise<string>;
   validateQueueLimits(env: AuthenticatedEnvironment): Promise<QueueValidationResult>;
-  getMasterQueue(env: AuthenticatedEnvironment): Promise<string | undefined>;
+  getWorkerQueue(env: AuthenticatedEnvironment): Promise<string | undefined>;
 }
 
 export interface PayloadProcessor {

internal-packages/database/prisma/schema.prisma

@@ -1764,7 +1764,8 @@ model TaskRun {
   lockedQueueId String?
 
   /// The main queue that this run is part of
-  masterQueue          String  @default("main")
+  workerQueue          String  @default("main") @map("masterQueue")
+  /// @deprecated
   secondaryMasterQueue String?
 
   /// From engine v2+ this will be defined after a run has been dequeued (starting at 1)

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

@@ -7,12 +7,11 @@ import {
   CreateCheckpointResult,
   DequeuedMessage,
   ExecutionResult,
-  MachineResources,
   RunExecutionData,
   StartRunAttemptResult,
   TaskRunExecutionResult,
 } from "@trigger.dev/core/v3";
-import { BatchId, RunId, WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { RunId, WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import {
   Prisma,
   PrismaClient,
@@ -37,20 +36,20 @@ import { DelayedRunSystem } from "./systems/delayedRunSystem.js";
 import { DequeueSystem } from "./systems/dequeueSystem.js";
 import { EnqueueSystem } from "./systems/enqueueSystem.js";
 import {
+  executionDataFromSnapshot,
   ExecutionSnapshotSystem,
-  getLatestExecutionSnapshot,
   getExecutionSnapshotsSince,
-  executionDataFromSnapshot,
+  getLatestExecutionSnapshot,
 } from "./systems/executionSnapshotSystem.js";
 import { PendingVersionSystem } from "./systems/pendingVersionSystem.js";
+import { RaceSimulationSystem } from "./systems/raceSimulationSystem.js";
 import { ReleaseConcurrencySystem } from "./systems/releaseConcurrencySystem.js";
 import { RunAttemptSystem } from "./systems/runAttemptSystem.js";
 import { SystemResources } from "./systems/systems.js";
 import { TtlSystem } from "./systems/ttlSystem.js";
 import { WaitpointSystem } from "./systems/waitpointSystem.js";
 import { EngineWorker, HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types.js";
 import { workerCatalog } from "./workerCatalog.js";
-import { RaceSimulationSystem } from "./systems/raceSimulationSystem.js";
 
 export class RunEngine {
   private runLockRedis: Redis;
@@ -318,7 +317,7 @@ export class RunEngine {
       sdkVersion,
       cliVersion,
       concurrencyKey,
-      masterQueue,
+      workerQueue,
       queue,
       lockedQueueId,
       isTest,
@@ -359,22 +358,6 @@ export class RunEngine {
       async (span) => {
         const status = delayUntil ? "DELAYED" : "PENDING";
 
-        let secondaryMasterQueue: string | undefined = undefined;
-
-        if (environment.type === "DEVELOPMENT") {
-          // In dev we use the environment id as the master queue, or the locked worker id
-          masterQueue = this.#environmentMasterQueueKey(environment.id);
-          if (lockedToVersionId) {
-            masterQueue = this.#backgroundWorkerQueueKey(lockedToVersionId);
-          }
-        } else {
-          // For deployed runs, we add the env/worker id as the secondary master queue
-          secondaryMasterQueue = this.#environmentMasterQueueKey(environment.id);
-          if (lockedToVersionId) {
-            secondaryMasterQueue = this.#backgroundWorkerQueueKey(lockedToVersionId);
-          }
-        }
-
         //create run
         let taskRun: TaskRun;
         try {
@@ -406,8 +389,7 @@ export class RunEngine {
               concurrencyKey,
               queue,
               lockedQueueId,
-              masterQueue,
-              secondaryMasterQueue,
+              workerQueue,
               isTest,
               delayUntil,
               queuedAt,
@@ -557,93 +539,58 @@ export class RunEngine {
   /**
    * Gets a fairly selected run from the specified master queue, returning the information required to run it.
    * @param consumerId: The consumer that is pulling, allows multiple consumers to pull from the same queue
-   * @param masterQueue: The shared queue to pull from, can be an individual environment (for dev)
+   * @param workerQueue: The worker queue to pull from, can be an individual environment (for dev)
    * @returns
    */
-  async dequeueFromMasterQueue({
+  async dequeueFromWorkerQueue({
     consumerId,
-    masterQueue,
-    maxRunCount,
-    maxResources,
+    workerQueue,
     backgroundWorkerId,
     workerId,
     runnerId,
     tx,
   }: {
     consumerId: string;
-    masterQueue: string;
-    maxRunCount: number;
-    maxResources?: MachineResources;
+    workerQueue: string;
     backgroundWorkerId?: string;
     workerId?: string;
     runnerId?: string;
     tx?: PrismaClientOrTransaction;
   }): Promise<DequeuedMessage[]> {
-    return this.dequeueSystem.dequeueFromMasterQueue({
+    const dequeuedMessage = await this.dequeueSystem.dequeueFromWorkerQueue({
       consumerId,
-      masterQueue,
-      maxRunCount,
-      maxResources,
+      workerQueue,
       backgroundWorkerId,
       workerId,
       runnerId,
       tx,
     });
+
+    if (!dequeuedMessage) {
+      return [];
+    }
+
+    return [dequeuedMessage];
   }
 
-  async dequeueFromEnvironmentMasterQueue({
+  async dequeueFromEnvironmentWorkerQueue({
     consumerId,
     environmentId,
-    maxRunCount,
-    maxResources,
     backgroundWorkerId,
     workerId,
     runnerId,
     tx,
   }: {
     consumerId: string;
     environmentId: string;
-    maxRunCount: number;
-    maxResources?: MachineResources;
     backgroundWorkerId?: string;
     workerId?: string;
     runnerId?: string;
     tx?: PrismaClientOrTransaction;
   }): Promise<DequeuedMessage[]> {
-    return this.dequeueFromMasterQueue({
-      consumerId,
-      masterQueue: this.#environmentMasterQueueKey(environmentId),
-      maxRunCount,
-      maxResources,
-      backgroundWorkerId,
-      workerId,
-      runnerId,
-      tx,
-    });
-  }
-
-  async dequeueFromBackgroundWorkerMasterQueue({
-    consumerId,
-    backgroundWorkerId,
-    maxRunCount,
-    maxResources,
-    workerId,
-    runnerId,
-    tx,
-  }: {
-    consumerId: string;
-    backgroundWorkerId: string;
-    maxRunCount: number;
-    maxResources?: MachineResources;
-    workerId?: string;
-    runnerId?: string;
-    tx?: PrismaClientOrTransaction;
-  }): Promise<DequeuedMessage[]> {
-    return this.dequeueFromMasterQueue({
+    return this.dequeueFromWorkerQueue({
       consumerId,
-      masterQueue: this.#backgroundWorkerQueueKey(backgroundWorkerId),
-      maxRunCount,
-      maxResources,
+      workerQueue: environmentId,
       backgroundWorkerId,
       workerId,
       runnerId,
@@ -1318,12 +1265,4 @@ export class RunEngine {
       }
     });
   }
-
-  #environmentMasterQueueKey(environmentId: string) {
-    return `master-env:${environmentId}`;
-  }
-
-  #backgroundWorkerQueueKey(backgroundWorkerId: string) {
-    return `master-background-worker:${backgroundWorkerId}`;
-  }
 }