Track run events in the run engine

Author: ericallam

apps/webapp/app/services/runsDashboardInstance.server.ts

@@ -1,27 +1,21 @@
-import { singleton } from "~/utils/singleton";
 import { ClickHouse } from "@internal/clickhouse";
+import { EventEmitter } from "node:events";
+import { prisma } from "~/db.server";
+import { singleton } from "~/utils/singleton";
+import { engine } from "~/v3/runEngine.server";
+import { logger } from "./logger.server";
 import {
   RunDashboardEventBus,
   RunDashboardEvents,
   RunsDashboardService,
 } from "./runsDashboardService.server";
-import { EventEmitter } from "node:events";
-import { RuntimeEnvironmentType, TaskRun } from "@trigger.dev/database";
-import { engine } from "~/v3/runEngine.server";
-import { logger } from "./logger.server";
 
 const runDashboardEventBus: RunDashboardEventBus = new EventEmitter<RunDashboardEvents>();
 
-export type TaskRunStatusUpdateEnvironment = {
-  type: RuntimeEnvironmentType;
-  organizationId: string;
-};
-
-export function emitRunStatusUpdate(run: TaskRun, environment: TaskRunStatusUpdateEnvironment) {
+export function emitRunStatusUpdate(runId: string) {
   runDashboardEventBus.emit("runStatusUpdate", {
-    run,
-    environment,
-    organization: { id: environment.organizationId },
+    time: new Date(),
+    runId,
   });
 }
 
@@ -31,16 +25,46 @@ export const runsDashboard = singleton("runsDashboard", () => {
   const service = new RunsDashboardService(clickhouse);
 
   runDashboardEventBus.on("runStatusUpdate", async (event) => {
-    await service.upsertRun(event.run, event.environment.type, event.organization.id);
+    await upsertRun(event.time, event.runId, service);
   });
 
   engine.eventBus.on("runStatusChanged", async (event) => {
-    logger.debug("RunDashboard: runStatusChanged", {
-      event,
-    });
-
-    await service.upsertRun(event.run, event.environment.type, event.environment.organization.id);
+    await upsertRun(event.time, event.runId, service);
   });
 
   return service;
 });
+
+async function upsertRun(time: Date, runId: string, service: RunsDashboardService) {
+  const run = await prisma.taskRun.findFirst({
+    where: {
+      id: runId,
+    },
+  });
+
+  if (!run) {
+    logger.error("RunDashboard: upsertRun: run not found", {
+      runId,
+    });
+
+    return;
+  }
+
+  if (!run.environmentType) {
+    logger.error("RunDashboard: upsertRun: run environment type not found", {
+      runId,
+    });
+
+    return;
+  }
+
+  if (!run.organizationId) {
+    logger.error("RunDashboard: upsertRun: run organization id not found", {
+      runId,
+    });
+
+    return;
+  }
+
+  await service.upsertRun(time, run, run.environmentType, run.organizationId);
+}

apps/webapp/app/services/runsDashboardService.server.ts

@@ -13,6 +13,7 @@ export class RunsDashboardService {
   });
 
   async upsertRun(
+    eventTime: Date,
     taskRun: TaskRun,
     environmentType: RuntimeEnvironmentType,
     organizationId: string
@@ -36,7 +37,7 @@ export class RunsDashboardService {
       queue: taskRun.queue,
       schedule_id: taskRun.scheduleId ?? undefined,
       batch_id: taskRun.batchId ?? undefined,
-      event_time: Date.now(),
+      event_time: eventTime.getTime(),
       created_at: taskRun.createdAt.getTime(),
       updated_at: taskRun.updatedAt.getTime(),
       completed_at: taskRun.completedAt ? taskRun.completedAt.getTime() : undefined,
@@ -120,13 +121,8 @@ export class RunsDashboardService {
 export type RunDashboardEvents = {
   runStatusUpdate: [
     {
-      run: TaskRun;
-      organization: {
-        id: string;
-      };
-      environment: {
-        type: RuntimeEnvironmentType;
-      };
+      time: Date;
+      runId: string;
     }
   ];
 };

apps/webapp/app/v3/services/triggerTaskV1.server.ts

@@ -376,6 +376,8 @@ export class TriggerTaskServiceV1 extends BaseService {
                     number: num,
                     friendlyId: runFriendlyId,
                     runtimeEnvironmentId: environment.id,
+                    environmentType: environment.type,
+                    organizationId: environment.organizationId,
                     projectId: environment.projectId,
                     idempotencyKey,
                     idempotencyKeyExpiresAt: idempotencyKey ? idempotencyKeyExpiresAt : undefined,

internal-packages/database/prisma/migrations/20250428211853_add_environment_type_and_org_id_to_task_run/migration.sql

@@ -0,0 +1,4 @@
+-- AlterTable
+ALTER TABLE "TaskRun" ADD COLUMN     "environmentType" "RuntimeEnvironmentType",
+ADD COLUMN     "organizationId" TEXT;
+

internal-packages/database/prisma/schema.prisma

@@ -1731,9 +1731,13 @@ model TaskRun {
   runtimeEnvironment   RuntimeEnvironment @relation(fields: [runtimeEnvironmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   runtimeEnvironmentId String
 
+  environmentType RuntimeEnvironmentType?
+
   project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   projectId String
 
+  organizationId String?
+
   // The specific queue this run is in
   queue         String
   // The queueId is set when the run is locked to a specific queue

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

@@ -7,8 +7,7 @@ export type EventBusEvents = {
   runStatusChanged: [
     {
       time: Date;
-      run: TaskRun;
-      environment: MinimalAuthenticatedEnvironment;
+      runId: string;
     },
   ];
   runAttemptStarted: [

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

@@ -403,6 +403,8 @@ export class RunEngine {
               number,
               friendlyId,
               runtimeEnvironmentId: environment.id,
+              environmentType: environment.type,
+              organizationId: environment.organization.id,
               projectId: environment.project.id,
               idempotencyKey,
               idempotencyKeyExpiresAt,
@@ -550,8 +552,7 @@ export class RunEngine {
 
         this.eventBus.emit("runStatusChanged", {
           time: new Date(),
-          run: taskRun,
-          environment,
+          runId: taskRun.id,
         });
 
         return taskRun;

internal-packages/run-engine/src/engine/systems/checkpointSystem.ts

@@ -143,6 +143,11 @@ export class CheckpointSystem {
         throw new ServiceValidationError("Run not found", 404);
       }
 
+      this.$.eventBus.emit("runStatusChanged", {
+        time: new Date(),
+        runId,
+      });
+
       // Create the checkpoint
       const taskRunCheckpoint = await prisma.taskRunCheckpoint.create({
         data: {
@@ -272,6 +277,11 @@ export class CheckpointSystem {
         throw new ServiceValidationError("Run not found", 404);
       }
 
+      this.$.eventBus.emit("runStatusChanged", {
+        time: new Date(),
+        runId,
+      });
+
       const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(prisma, {
         run,
         snapshot: {

internal-packages/run-engine/src/engine/systems/delayedRunSystem.ts

@@ -68,6 +68,11 @@ export class DelayedRunSystem {
 
           await this.$.worker.reschedule(`enqueueDelayedRun:${updatedRun.id}`, delayUntil);
 
+          this.$.eventBus.emit("runStatusChanged", {
+            time: new Date(),
+            runId: updatedRun.id,
+          });
+
           return updatedRun;
         });
       },
@@ -109,6 +114,11 @@ export class DelayedRunSystem {
       },
     });
 
+    this.$.eventBus.emit("runStatusChanged", {
+      time: new Date(),
+      runId,
+    });
+
     if (run.ttl) {
       const expireAt = parseNaturalLanguageDuration(run.ttl);
 

internal-packages/run-engine/src/engine/systems/dequeueSystem.ts

@@ -356,6 +356,11 @@ export class DequeueSystem {
                   },
                 });
 
+                this.$.eventBus.emit("runStatusChanged", {
+                  time: new Date(),
+                  runId,
+                });
+
                 if (!lockedTaskRun) {
                   this.$.logger.error(
                     "RunEngine.dequeueFromMasterQueue(): Failed to lock task run",
@@ -573,6 +578,11 @@ export class DequeueSystem {
 
           //we ack because when it's deployed it will be requeued
           await this.$.runQueue.acknowledgeMessage(orgId, runId);
+
+          this.$.eventBus.emit("runStatusChanged", {
+            time: new Date(),
+            runId,
+          });
         });
       },
       {