Fixed dequeueing test

Author: ericallam

internal-packages/run-engine/src/engine/tests/dequeuing.test.ts

@@ -1,13 +1,12 @@
 import { containerTest } from "@internal/testcontainers";
 import { trace } from "@internal/tracing";
+import { DequeuedMessage } from "@trigger.dev/core/v3";
 import { generateFriendlyId } from "@trigger.dev/core/v3/isomorphic";
 import { PrismaClientOrTransaction } from "@trigger.dev/database";
 import { expect } from "vitest";
 import { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
 import { RunEngine } from "../index.js";
 import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
-import { DequeuedMessage } from "@trigger.dev/core/v3";
-import { setTimeout } from "timers/promises";
 
 vi.setConfig({ testTimeout: 60_000 });
 
@@ -62,12 +61,9 @@ describe("RunEngine dequeuing", () => {
       });
       expect(runs.length).toBe(10);
 
-      //check the queue length
-      const queueLength = await engine.runQueue.lengthOfEnvQueue(authenticatedEnvironment);
-      expect(queueLength).toBe(10);
-
       //dequeue
-      await setTimeout(500);
+      await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 5);
+
       const dequeued: DequeuedMessage[] = [];
       for (let i = 0; i < 5; i++) {
         dequeued.push(
@@ -83,93 +79,6 @@ describe("RunEngine dequeuing", () => {
       await engine.quit();
     }
   });
-
-  //This will fail until we support dequeuing multiple runs from a single environment
-  containerTest.fails(
-    "Dequeues runs within machine constraints",
-    async ({ prisma, redisOptions }) => {
-      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
-
-      const engine = new RunEngine({
-        prisma,
-        worker: {
-          redis: redisOptions,
-          workers: 1,
-          tasksPerWorker: 10,
-          pollIntervalMs: 100,
-        },
-        queue: {
-          redis: redisOptions,
-          masterQueueConsumersDisabled: true,
-          processWorkerQueueDebounceMs: 50,
-        },
-        runLock: {
-          redis: redisOptions,
-        },
-        machines: {
-          defaultMachine: "small-1x",
-          machines: {
-            "small-1x": {
-              name: "small-1x" as const,
-              cpu: 0.5,
-              memory: 0.5,
-              centsPerMs: 0.0001,
-            },
-          },
-          baseCostInCents: 0.0005,
-        },
-        tracer: trace.getTracer("test", "0.0.0"),
-      });
-
-      try {
-        const taskIdentifier = "test-task";
-
-        //create background worker
-        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier, {
-          preset: "small-1x",
-        });
-
-        //trigger the runs
-        const runs = await triggerRuns({
-          engine,
-          environment: authenticatedEnvironment,
-          taskIdentifier,
-          prisma,
-          count: 20,
-        });
-        expect(runs.length).toBe(20);
-
-        //check the queue length
-        const queueLength = await engine.runQueue.lengthOfEnvQueue(authenticatedEnvironment);
-        expect(queueLength).toBe(20);
-
-        //dequeue
-        await setTimeout(500);
-        const dequeued = await engine.dequeueFromWorkerQueue({
-          consumerId: "test_12345",
-          workerQueue: "main",
-        });
-        expect(dequeued.length).toBe(2);
-
-        //check the queue length
-        const queueLength2 = await engine.runQueue.lengthOfEnvQueue(authenticatedEnvironment);
-        expect(queueLength2).toBe(18);
-
-        await setTimeout(500);
-        const dequeued2 = await engine.dequeueFromWorkerQueue({
-          consumerId: "test_12345",
-          workerQueue: "main",
-        });
-        expect(dequeued2.length).toBe(6);
-
-        //check the queue length
-        const queueLength3 = await engine.runQueue.lengthOfEnvQueue(authenticatedEnvironment);
-        expect(queueLength3).toBe(12);
-      } finally {
-        await engine.quit();
-      }
-    }
-  );
 });
 
 async function triggerRuns({