converted expired complete span event to new method

Author: ericallam

apps/webapp/app/v3/eventRepository.server.ts

@@ -514,6 +514,60 @@ export class EventRepository {
     });
   }
 
+  async completeExpiredRunEvent({
+    run,
+    endTime,
+    ttl,
+  }: {
+    run: CompleteableTaskRun;
+    endTime?: Date;
+    ttl: string;
+  }) {
+    const startTime = convertDateToNanoseconds(run.createdAt);
+
+    await this.insertImmediate({
+      message: run.taskIdentifier,
+      serviceName: "api server",
+      serviceNamespace: "trigger.dev",
+      level: "TRACE",
+      kind: "SERVER",
+      traceId: run.traceId,
+      spanId: run.spanId,
+      parentId: run.parentSpanId,
+      runId: run.friendlyId,
+      taskSlug: run.taskIdentifier,
+      projectRef: "",
+      projectId: run.projectId,
+      environmentId: run.runtimeEnvironmentId,
+      environmentType: run.environmentType ?? "DEVELOPMENT",
+      organizationId: run.organizationId ?? "",
+      isPartial: false,
+      isError: true,
+      isCancelled: false,
+      status: "ERROR",
+      runIsTest: run.isTest,
+      startTime,
+      events: [
+        {
+          name: "exception",
+          time: endTime ?? new Date(),
+          properties: {
+            exception: {
+              message: `Run expired because the TTL (${ttl}) was reached`,
+            },
+          },
+        },
+      ],
+      properties: {},
+      metadata: undefined,
+      style: undefined,
+      duration: calculateDurationFromStart(startTime, endTime ?? new Date()),
+      output: undefined,
+      payload: undefined,
+      payloadType: undefined,
+    });
+  }
+
   async createAttemptFailedRunEvent({
     run,
     endTime,

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -16,12 +16,11 @@ import {
 import { roomFromFriendlyRunId, socketIo } from "./handleSocketIo.server";
 import { engine } from "./runEngine.server";
 import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server";
-import { getTaskEventStoreTableForRun } from "./taskEventStore.server";
 
 export function registerRunEngineEventBusHandlers() {
   engine.eventBus.on("runSucceeded", async ({ time, run }) => {
     const [taskRunError, taskRun] = await tryCatch(
-      $replica.taskRun.findFirst({
+      $replica.taskRun.findFirstOrThrow({
         where: {
           id: run.id,
         },
@@ -51,13 +50,6 @@ export function registerRunEngineEventBusHandlers() {
       return;
     }
 
-    if (!taskRun) {
-      logger.error("[runSucceeded] Task run not found", {
-        runId: run.id,
-      });
-      return;
-    }
-
     const [completeSuccessfulRunEventError] = await tryCatch(
       eventRepository.completeSuccessfulRunEvent({
         run: taskRun,
@@ -92,7 +84,7 @@ export function registerRunEngineEventBusHandlers() {
     const exception = createExceptionPropertiesFromError(sanitizedError);
 
     const [taskRunError, taskRun] = await tryCatch(
-      $replica.taskRun.findFirst({
+      $replica.taskRun.findFirstOrThrow({
         where: {
           id: run.id,
         },
@@ -122,13 +114,6 @@ export function registerRunEngineEventBusHandlers() {
       return;
     }
 
-    if (!taskRun) {
-      logger.error("[runFailed] Task run not found", {
-        runId: run.id,
-      });
-      return;
-    }
-
     const [completeFailedRunEventError] = await tryCatch(
       eventRepository.completeFailedRunEvent({
         run: taskRun,
@@ -150,7 +135,7 @@ export function registerRunEngineEventBusHandlers() {
     const exception = createExceptionPropertiesFromError(sanitizedError);
 
     const [taskRunError, taskRun] = await tryCatch(
-      $replica.taskRun.findFirst({
+      $replica.taskRun.findFirstOrThrow({
         where: {
           id: run.id,
         },
@@ -180,13 +165,6 @@ export function registerRunEngineEventBusHandlers() {
       return;
     }
 
-    if (!taskRun) {
-      logger.error("[runAttemptFailed] Task run not found", {
-        runId: run.id,
-      });
-      return;
-    }
-
     const [createAttemptFailedRunEventError] = await tryCatch(
       eventRepository.createAttemptFailedRunEvent({
         run: taskRun,
@@ -218,7 +196,7 @@ export function registerRunEngineEventBusHandlers() {
       }
 
       const [cachedRunError, cachedRun] = await tryCatch(
-        $replica.taskRun.findFirst({
+        $replica.taskRun.findFirstOrThrow({
           where: {
             id: cachedRunId,
           },
@@ -248,13 +226,6 @@ export function registerRunEngineEventBusHandlers() {
         return;
       }
 
-      if (!cachedRun) {
-        logger.error("[cachedRunCompleted] Cached run not found", {
-          cachedRunId,
-        });
-        return;
-      }
-
       const [blockedRunError, blockedRun] = await tryCatch(
         $replica.taskRun.findFirst({
           where: {
@@ -314,52 +285,60 @@ export function registerRunEngineEventBusHandlers() {
   );
 
   engine.eventBus.on("runExpired", async ({ time, run }) => {
-    try {
-      const eventStore = getTaskEventStoreTableForRun(run);
+    if (!run.ttl) {
+      return;
+    }
 
-      const completedEvent = await eventRepository.completeEvent(
-        eventStore,
-        run.spanId,
-        run.createdAt,
-        run.completedAt ?? undefined,
-        {
-          endTime: time,
-          attributes: {
-            isError: true,
-          },
-          events: [
-            {
-              name: "exception",
-              time,
-              properties: {
-                exception: {
-                  message: `Run expired because the TTL (${run.ttl}) was reached`,
-                },
-              },
-            },
-          ],
-        }
-      );
+    const [taskRunError, taskRun] = await tryCatch(
+      $replica.taskRun.findFirstOrThrow({
+        where: {
+          id: run.id,
+        },
+        select: {
+          id: true,
+          friendlyId: true,
+          traceId: true,
+          spanId: true,
+          parentSpanId: true,
+          createdAt: true,
+          completedAt: true,
+          taskIdentifier: true,
+          projectId: true,
+          runtimeEnvironmentId: true,
+          environmentType: true,
+          isTest: true,
+          organizationId: true,
+        },
+      })
+    );
 
-      if (!completedEvent) {
-        logger.error("[runFailed] Failed to complete event for unknown reason", {
-          runId: run.id,
-          spanId: run.spanId,
-        });
-        return;
-      }
-    } catch (error) {
-      logger.error("[runExpired] Failed to complete event", {
-        error: error instanceof Error ? error.message : error,
+    if (taskRunError) {
+      logger.error("[runExpired] Failed to find task run", {
+        error: taskRunError,
+        runId: run.id,
+      });
+      return;
+    }
+
+    const [completeExpiredRunEventError] = await tryCatch(
+      eventRepository.completeExpiredRunEvent({
+        run: taskRun,
+        endTime: time,
+        ttl: run.ttl,
+      })
+    );
+
+    if (completeExpiredRunEventError) {
+      logger.error("[runExpired] Failed to complete expired run event", {
+        error: completeExpiredRunEventError,
         runId: run.id,
-        spanId: run.spanId,
       });
     }
   });
 
   engine.eventBus.on("runCancelled", async ({ time, run }) => {
-    try {
-      const taskRun = await $replica.taskRun.findFirst({
+    const [taskRunError, taskRun] = await tryCatch(
+      $replica.taskRun.findFirstOrThrow({
         where: {
           id: run.id,
         },
@@ -378,27 +357,31 @@ export function registerRunEngineEventBusHandlers() {
           isTest: true,
           organizationId: true,
         },
-      });
+      })
+    );
 
-      if (!taskRun) {
-        logger.error("[runCancelled] Task run not found", {
-          runId: run.id,
-        });
-        return;
-      }
+    if (taskRunError) {
+      logger.error("[runCancelled] Task run not found", {
+        error: taskRunError,
+        runId: run.id,
+      });
+      return;
+    }
 
-      const error = createJsonErrorObject(run.error);
+    const error = createJsonErrorObject(run.error);
 
-      await eventRepository.cancelRunEvent({
+    const [cancelRunEventError] = await tryCatch(
+      eventRepository.cancelRunEvent({
         reason: error.message,
         run: taskRun,
         cancelledAt: time,
-      });
-    } catch (error) {
-      logger.error("[runCancelled] Failed to cancel event", {
-        error: error instanceof Error ? error.message : error,
+      })
+    );
+
+    if (cancelRunEventError) {
+      logger.error("[runCancelled] Failed to cancel run event", {
+        error: cancelRunEventError,
         runId: run.id,
-        spanId: run.spanId,
       });
     }
   });