Run Engine 2.0 (alpha)

.github/workflows/e2e.yml

@@ -38,6 +38,9 @@ jobs:
       - name: 📥 Download deps
         run: pnpm install --frozen-lockfile --filter trigger.dev...
 
+      - name: 📀 Generate Prisma Client
+        run: pnpm run generate
+
       - name: 🔧 Build v3 cli monorepo dependencies
         run: pnpm run build --filter trigger.dev^...
 

.github/workflows/pr_checks.yml

@@ -54,6 +54,9 @@ jobs:
       - name: 📥 Download deps
         run: pnpm install --frozen-lockfile
 
+      - name: 📀 Generate Prisma Client
+        run: pnpm run generate
+
       - name: 🏗️ Build
         run: pnpm run build --filter "@trigger.dev/*" --filter "trigger.dev"
 

.github/workflows/unit-tests.yml

@@ -24,6 +24,13 @@ jobs:
           node-version: 20.11.1
           cache: "pnpm"
 
+      # ..to avoid rate limits when pulling images
+      - name: 🐳 Login to DockerHub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
       - name: 📥 Download deps
         run: pnpm install --frozen-lockfile
 

.npmrc

@@ -1,3 +1,4 @@
 link-workspace-packages=false
 public-hoist-pattern[]=*prisma*
-prefer-workspace-packages=true
+prefer-workspace-packages=true
+update-notifier=false

.vscode/launch.json

@@ -116,6 +116,14 @@
       "command": "pnpm exec trigger dev",
       "cwd": "${workspaceFolder}/references/hello-world",
       "sourceMaps": true
+    },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug RunEngine tests",
+      "command": "pnpm run test --filter @internal/run-engine",
+      "cwd": "${workspaceFolder}",
+      "sourceMaps": true
     }
   ]
 }

apps/coordinator/package.json

@@ -23,10 +23,8 @@
     "tinyexec": "^0.3.0"
   },
   "devDependencies": {
-    "@types/node": "^18",
     "dotenv": "^16.4.2",
     "esbuild": "^0.19.11",
-    "tsx": "^4.7.0",
-    "typescript": "^5.3.3"
+    "tsx": "^4.7.0"
   }
 }

apps/coordinator/src/checkpointer.ts

@@ -193,7 +193,7 @@ export class Checkpointer {
     const start = performance.now();
     this.#logger.log(`checkpointAndPush() start`, { start, opts });
 
-    let interval: NodeJS.Timer | undefined;
+    let interval: NodeJS.Timeout | undefined;
 
     if (opts.shouldHeartbeat) {
       interval = setInterval(() => {

apps/docker-provider/package.json

@@ -20,10 +20,8 @@
     "execa": "^8.0.1"
   },
   "devDependencies": {
-    "@types/node": "^18.19.8",
     "dotenv": "^16.4.2",
     "esbuild": "^0.19.11",
-    "tsx": "^4.7.0",
-    "typescript": "^5.3.3"
+    "tsx": "^4.7.0"
   }
 }

apps/kubernetes-provider/package.json

@@ -23,7 +23,6 @@
   "devDependencies": {
     "dotenv": "^16.4.2",
     "esbuild": "^0.19.11",
-    "tsx": "^4.7.0",
-    "typescript": "^5.3.3"
+    "tsx": "^4.7.0"
   }
 }

apps/proxy/package.json

@@ -9,7 +9,6 @@
   },
   "devDependencies": {
     "@cloudflare/workers-types": "^4.20240512.0",
-    "typescript": "^5.0.4",
     "wrangler": "^3.57.1"
   },
   "dependencies": {

apps/webapp/app/consts.ts

@@ -1,6 +1,5 @@
 export const LIVE_ENVIRONMENT = "live";
 export const DEV_ENVIRONMENT = "development";
-export const CURRENT_DEPLOYMENT_LABEL = "current";
 export const MAX_LIVE_PROJECTS = 1;
 export const DEFAULT_MAX_CONCURRENT_RUNS = 10;
 export const MAX_CONCURRENT_RUNS_LIMIT = 20;

apps/webapp/app/entry.server.tsx

@@ -205,6 +205,8 @@ process.on("uncaughtException", (error, origin) => {
 
 const sqsEventConsumer = singleton("sqsEventConsumer", getSharedSqsEventConsumer);
 
+singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers);
+
 export { apiRateLimiter } from "./services/apiRateLimit.server";
 export { socketIo } from "./v3/handleSocketIo.server";
 export { wss } from "./v3/handleWebsockets.server";
@@ -214,6 +216,7 @@ import { eventLoopMonitor } from "./eventLoopMonitor.server";
 import { env } from "./env.server";
 import { logger } from "./services/logger.server";
 import { Prisma } from "./db.server";
+import { registerRunEngineEventBusHandlers } from "./v3/runEngineHandlers.server";
 
 if (env.EVENT_LOOP_MONITOR_ENABLED === "1") {
   eventLoopMonitor.enable();

apps/webapp/app/env.server.ts

@@ -84,6 +84,33 @@ const EnvironmentSchema = z.object({
   REDIS_PASSWORD: z.string().optional(),
   REDIS_TLS_DISABLED: z.string().optional(),
 
+  // Valkey options (used in Run Engine 2.0+)
+  VALKEY_HOST: z
+    .string()
+    .nullish()
+    .default(process.env.REDIS_HOST ?? null),
+  VALKEY_READER_HOST: z
+    .string()
+    .nullish()
+    .default(process.env.REDIS_READER_HOST ?? null),
+  VALKEY_READER_PORT: z.coerce
+    .number()
+    .nullish()
+    .default(process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : null),
+  VALKEY_PORT: z.coerce
+    .number()
+    .nullish()
+    .default(process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : null),
+  VALKEY_USERNAME: z
+    .string()
+    .nullish()
+    .default(process.env.REDIS_USERNAME ?? null),
+  VALKEY_PASSWORD: z
+    .string()
+    .nullish()
+    .default(process.env.REDIS_PASSWORD ?? null),
+  VALKEY_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+
   DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),
@@ -155,6 +182,7 @@ const EnvironmentSchema = z.object({
   SHARED_QUEUE_CONSUMER_POOL_SIZE: z.coerce.number().int().default(10),
   SHARED_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(100),
   SHARED_QUEUE_CONSUMER_NEXT_TICK_INTERVAL_MS: z.coerce.number().int().default(100),
+  MANAGED_WORKER_SECRET: z.string().default("managed-secret"),
 
   // Development OTEL environment variables
   DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
@@ -245,6 +273,11 @@ const EnvironmentSchema = z.object({
   MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
 
   REALTIME_STREAM_VERSION: z.enum(["v1", "v2"]).default("v1"),
+
+  // Run Engine 2.0
+  RUN_ENGINE_WORKER_COUNT: z.coerce.number().int().default(4),
+  RUN_ENGINE_TASKS_PER_WORKER: z.coerce.number().int().default(10),
+  RUN_ENGINE_WORKER_POLL_INTERVAL: z.coerce.number().int().default(100),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

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

@@ -1,4 +1,4 @@
-import { WorkerDeploymentStatus } from "@trigger.dev/database";
+import { WorkerDeploymentStatus, WorkerInstanceGroupType } from "@trigger.dev/database";
 import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
@@ -95,29 +95,31 @@ export class DeploymentListPresenter {
         userName: string | null;
         userDisplayName: string | null;
         userAvatarUrl: string | null;
+        type: WorkerInstanceGroupType;
       }[]
     >`
-    SELECT 
-  wd."id", 
-  wd."shortCode", 
-  wd."version", 
-  (SELECT COUNT(*) FROM ${sqlDatabaseSchema}."BackgroundWorkerTask" WHERE "BackgroundWorkerTask"."workerId" = wd."workerId") AS "tasksCount", 
-  wd."environmentId", 
-  wd."status", 
-  u."id" AS "userId", 
-  u."name" AS "userName", 
-  u."displayName" AS "userDisplayName", 
-  u."avatarUrl" AS "userAvatarUrl", 
+    SELECT
+  wd."id",
+  wd."shortCode",
+  wd."version",
+  (SELECT COUNT(*) FROM ${sqlDatabaseSchema}."BackgroundWorkerTask" WHERE "BackgroundWorkerTask"."workerId" = wd."workerId") AS "tasksCount",
+  wd."environmentId",
+  wd."status",
+  u."id" AS "userId",
+  u."name" AS "userName",
+  u."displayName" AS "userDisplayName",
+  u."avatarUrl" AS "userAvatarUrl",
   wd."builtAt",
-  wd."deployedAt"
-FROM 
+  wd."deployedAt",
+  wd."type"
+FROM
   ${sqlDatabaseSchema}."WorkerDeployment" as wd
-INNER JOIN 
-  ${sqlDatabaseSchema}."User" as u ON wd."triggeredById" = u."id" 
-WHERE 
+INNER JOIN
+  ${sqlDatabaseSchema}."User" as u ON wd."triggeredById" = u."id"
+WHERE
   wd."projectId" = ${project.id}
-ORDER BY 
-  string_to_array(wd."version", '.')::int[] DESC 
+ORDER BY
+  string_to_array(wd."version", '.')::int[] DESC
 LIMIT ${pageSize} OFFSET ${pageSize * (page - 1)};`;
 
     return {
@@ -146,6 +148,7 @@ LIMIT ${pageSize} OFFSET ${pageSize * (page - 1)};`;
           isCurrent: label?.label === "current",
           isDeployed: deployment.status === "DEPLOYED",
           isLatest: page === 1 && index === 0,
+          type: deployment.type,
           environment: {
             id: environment.id,
             type: environment.type,

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

@@ -68,6 +68,7 @@ export class DeploymentPresenter {
         imageReference: true,
         externalBuildData: true,
         projectId: true,
+        type: true,
         environment: {
           select: {
             id: true,
@@ -154,6 +155,7 @@ export class DeploymentPresenter {
         organizationId: project.organizationId,
         errorData: DeploymentPresenter.prepareErrorData(deployment.errorData),
         isBuilt: !!deployment.builtAt,
+        type: deployment.type,
       },
     };
   }

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

@@ -39,7 +39,7 @@ export class RunStreamPresenter {
       traceId: run.traceId,
     });
 
-    let pinger: NodeJS.Timer | undefined = undefined;
+    let pinger: NodeJS.Timeout | undefined = undefined;
 
     const { unsubscribe, eventEmitter } = await eventRepository.subscribeToTrace(run.traceId);
 

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

@@ -9,7 +9,7 @@ import { eventRepository } from "~/v3/eventRepository.server";
 import { machinePresetFromName } from "~/v3/machinePresets.server";
 import { FINAL_ATTEMPT_STATUSES, isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
 import { BasePresenter } from "./basePresenter.server";
-import { getMaxDuration } from "~/v3/utils/maxDuration";
+import { getMaxDuration } from "@trigger.dev/core/v3/apps";
 
 type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
 export type Span = NonNullable<NonNullable<Result>["span"]>;
@@ -83,6 +83,12 @@ export class SpanPresenter extends BasePresenter {
             sdkVersion: true,
           },
         },
+        engine: true,
+        masterQueue: true,
+        secondaryMasterQueue: true,
+        error: true,
+        output: true,
+        outputType: true,
         //status + duration
         status: true,
         startedAt: true,
@@ -183,13 +189,33 @@ export class SpanPresenter extends BasePresenter {
         })
       : null;
 
+    const finishedData =
+      run.engine === "V2"
+        ? run
+        : isFinished
+        ? await this._replica.taskRunAttempt.findFirst({
+            select: {
+              output: true,
+              outputType: true,
+              error: true,
+            },
+            where: {
+              status: { in: FINAL_ATTEMPT_STATUSES },
+              taskRunId: run.id,
+            },
+            orderBy: {
+              createdAt: "desc",
+            },
+          })
+        : null;
+
     const output =
-      finishedAttempt === null
+      finishedData === null
         ? undefined
-        : finishedAttempt.outputType === "application/store"
-        ? `/resources/packets/${run.runtimeEnvironment.id}/${finishedAttempt.output}`
-        : typeof finishedAttempt.output !== "undefined" && finishedAttempt.output !== null
-        ? await prettyPrintPacket(finishedAttempt.output, finishedAttempt.outputType ?? undefined)
+        : finishedData.outputType === "application/store"
+        ? `/resources/packets/${run.runtimeEnvironment.id}/${finishedData.output}`
+        : typeof finishedData.output !== "undefined" && finishedData.output !== null
+        ? await prettyPrintPacket(finishedData.output, finishedData.outputType ?? undefined)
         : undefined;
 
     const payload =
@@ -200,14 +226,14 @@ export class SpanPresenter extends BasePresenter {
         : undefined;
 
     let error: TaskRunError | undefined = undefined;
-    if (finishedAttempt?.error) {
-      const result = TaskRunError.safeParse(finishedAttempt.error);
+    if (finishedData?.error) {
+      const result = TaskRunError.safeParse(finishedData.error);
       if (result.success) {
         error = result.data;
       } else {
         error = {
           type: "CUSTOM_ERROR",
-          raw: JSON.stringify(finishedAttempt.error),
+          raw: JSON.stringify(finishedData.error),
         };
       }
     }
@@ -320,6 +346,9 @@ export class SpanPresenter extends BasePresenter {
       metadata,
       maxDurationInSeconds: getMaxDuration(run.maxDurationInSeconds),
       batch: run.batch ? { friendlyId: run.batch.friendlyId } : undefined,
+      engine: run.engine,
+      masterQueue: run.masterQueue,
+      secondaryMasterQueue: run.secondaryMasterQueue,
     };
   }
 

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

@@ -19,8 +19,8 @@ import {
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
 import { TaskRunStatus } from "~/database-types";
-import { CURRENT_DEPLOYMENT_LABEL } from "~/consts";
 import { concurrencyTracker } from "~/v3/services/taskRunConcurrencyTracker.server";
+import { CURRENT_DEPLOYMENT_LABEL } from "@trigger.dev/core/v3/apps";
 
 export type Task = {
   slug: string;

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

@@ -58,7 +58,7 @@ export class TasksStreamPresenter {
       projectSlug,
     });
 
-    let pinger: NodeJS.Timer | undefined = undefined;
+    let pinger: NodeJS.Timeout | undefined = undefined;
 
     const subscriber = await projectPubSub.subscribe(`project:${project.id}:*`);