make sure engine v1 runs get synced to CH

Author: ericallam

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -4,6 +4,7 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { createTag, getTagsForRunId, MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
 
 const ParamsSchema = z.object({
@@ -95,6 +96,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
       },
     });
 
+    emitRunStatusUpdate(taskRun.id);
+
     return json({ message: `Successfully set ${newTags.length} new tags.` }, { status: 200 });
   } catch (error) {
     return json(

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

@@ -305,6 +305,8 @@ export class RunEngineTriggerTaskService {
                       ? parentRun.queueTimestamp ?? undefined
                       : undefined,
                   runChainState,
+                  scheduleId: options.scheduleId,
+                  scheduleInstanceId: options.scheduleInstanceId,
                 },
                 this.prisma
               );

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

@@ -9,6 +9,7 @@ import {
   RunDashboardEvents,
   RunsDashboardService,
 } from "./runsDashboardService.server";
+import { tryCatch } from "@trigger.dev/core/utils";
 
 const runDashboardEventBus: RunDashboardEventBus = new EventEmitter<RunDashboardEvents>();
 
@@ -25,11 +26,25 @@ export const runsDashboard = singleton("runsDashboard", () => {
   const service = new RunsDashboardService(clickhouse);
 
   runDashboardEventBus.on("runStatusUpdate", async (event) => {
-    await upsertRun(event.time, event.runId, service);
+    const [upsertError] = await tryCatch(upsertRun(event.time, event.runId, service));
+
+    if (upsertError) {
+      logger.error("RunDashboard: runStatusUpdate: upsertRun error", {
+        runId: event.runId,
+        error: upsertError,
+      });
+    }
   });
 
   engine.eventBus.on("runStatusChanged", async (event) => {
-    await upsertRun(event.time, event.runId, service);
+    const [upsertError] = await tryCatch(upsertRun(event.time, event.runId, service));
+
+    if (upsertError) {
+      logger.error("RunDashboard: runStatusChanged: upsertRun error", {
+        runId: event.runId,
+        error: upsertError,
+      });
+    }
   });
 
   return service;

apps/webapp/app/v3/marqs/devQueueConsumer.server.ts

@@ -23,6 +23,7 @@ import { getMaxDuration } from "@trigger.dev/core/v3/isomorphic";
 import { DevSubscriber, devPubSub } from "./devPubSub.server";
 import { findQueueInEnvironment, sanitizeQueueName } from "~/models/taskQueue.server";
 import { createRedisClient, RedisClient } from "~/redis.server";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 const MessageBody = z.discriminatedUnion("type", [
   z.object({
@@ -539,6 +540,8 @@ export class DevQueueConsumer {
           messageId: message.messageId,
         });
 
+        emitRunStatusUpdate(lockedTaskRun.id);
+
         this._inProgressRuns.set(lockedTaskRun.friendlyId, message.messageId);
       } catch (e) {
         if (e instanceof Error) {

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -66,6 +66,7 @@ import {
 import { tracer } from "../tracer.server";
 import { getMaxDuration } from "../utils/maxDuration";
 import { MessagePayload } from "./types";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 const WithTraceContext = z.object({
   traceparent: z.string().optional(),
@@ -841,6 +842,8 @@ export class SharedQueueConsumer {
           };
         }
 
+        emitRunStatusUpdate(lockedTaskRun.id);
+
         return {
           action: "noop",
           reason: "restored_checkpoint",
@@ -922,6 +925,8 @@ export class SharedQueueConsumer {
             },
           });
 
+          emitRunStatusUpdate(lockedTaskRun.id);
+
           return {
             action: "noop",
             reason: "scheduled_attempt",
@@ -1430,14 +1435,16 @@ export class SharedQueueConsumer {
   async #markRunAsWaitingForDeploy(runId: string) {
     logger.debug("Marking run as waiting for deploy", { runId });
 
-    return await prisma.taskRun.update({
+    await prisma.taskRun.update({
       where: {
         id: runId,
       },
       data: {
         status: "WAITING_FOR_DEPLOY",
       },
     });
+
+    emitRunStatusUpdate(runId);
   }
 
   async #resolveCompletedAttemptsForResumeMessage(

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

@@ -34,6 +34,7 @@ import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
 import { RetryAttemptService } from "./retryAttempt.server";
 import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
 import { socketIo } from "../handleSocketIo.server";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 type FoundAttempt = Awaited<ReturnType<typeof findAttempt>>;
 
@@ -614,6 +615,8 @@ export class CompleteAttemptService extends BaseService {
       },
     });
 
+    emitRunStatusUpdate(taskRunAttempt.taskRunId);
+
     if (environment.type === "DEVELOPMENT") {
       await marqs.requeueMessage(taskRunAttempt.taskRunId, {}, executionRetry.timestamp, "retry");
 

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

@@ -12,6 +12,7 @@ import { CrashTaskRunService } from "./crashTaskRun.server";
 import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
 import { findQueueInEnvironment } from "~/models/taskQueue.server";
 import { FINAL_RUN_STATUSES } from "../taskStatus";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 export class CreateTaskRunAttemptService extends BaseService {
   public async call({
@@ -180,6 +181,8 @@ export class CreateTaskRunAttemptService extends BaseService {
         });
       }
 
+      emitRunStatusUpdate(taskRun.id);
+
       const machinePreset =
         machinePresetFromRun(taskRun) ?? machinePresetFromConfig(lockedBy.machineConfig ?? {});
 

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

@@ -7,6 +7,7 @@ import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
 import { commonWorker } from "../commonWorker.server";
 import { workerQueue } from "~/services/worker.server";
 import { enqueueRun } from "./enqueueRun.server";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 export class EnqueueDelayedRunService extends BaseService {
   public static async enqueue(runId: string, runAt?: Date) {
@@ -102,6 +103,8 @@ export class EnqueueDelayedRunService extends BaseService {
       }
     });
 
+    emitRunStatusUpdate(run.id);
+
     await enqueueRun({
       env: run.runtimeEnvironment,
       run: run,

apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts

@@ -4,6 +4,7 @@ import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 import { env } from "~/env.server";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 export class ExecuteTasksWaitingForDeployService extends BaseService {
   public async call(backgroundWorkerId: string) {
@@ -78,6 +79,10 @@ export class ExecuteTasksWaitingForDeployService extends BaseService {
       });
     }
 
+    for (const run of runsWaitingForDeploy) {
+      emitRunStatusUpdate(run.id);
+    }
+
     for (const run of runsWaitingForDeploy) {
       await marqs?.enqueueMessage(
         backgroundWorker.runtimeEnvironment,

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

@@ -19,6 +19,7 @@ import { completeBatchTaskRunItemV3 } from "./batchTriggerV3.server";
 import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
 import { ResumeBatchRunService } from "./resumeBatchRun.server";
 import { ResumeDependentParentsService } from "./resumeDependentParents.server";
+import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
 
 type BaseInput = {
   id: string;
@@ -100,6 +101,8 @@ export class FinalizeTaskRunService extends BaseService {
       ...(include ? { include } : {}),
     });
 
+    emitRunStatusUpdate(run.id);
+
     if (run.ttl) {
       await ExpireEnqueuedRunService.ack(run.id);
     }