Mirror task run attempt errors and output

ericallam · ericallam · commit 493a30cc9851 · 2024-09-25T22:42:16.000+01:00
diff --git a/apps/webapp/app/v3/services/completeAttempt.server.ts b/apps/webapp/app/v3/services/completeAttempt.server.ts
@@ -128,6 +128,12 @@ export class CompleteAttemptService extends BaseService {
           output: completion.output,
           outputType: completion.outputType,
           usageDurationMs: completion.usage?.durationMs,
+          taskRun: {
+            update: {
+              output: completion.output,
+              outputType: completion.outputType,
+            },
+          },
         },
       });
 
@@ -348,6 +354,15 @@ export class CompleteAttemptService extends BaseService {
           })
         );
       } else {
+        await this._prisma.taskRun.update({
+          where: {
+            id: taskRunAttempt.taskRunId,
+          },
+          data: {
+            error: sanitizedError,
+          },
+        });
+
         const finalizeService = new FinalizeTaskRunService();
         await finalizeService.call({
           id: taskRunAttempt.taskRunId,
diff --git a/apps/webapp/app/v3/services/finalizeTaskRun.server.ts b/apps/webapp/app/v3/services/finalizeTaskRun.server.ts
@@ -2,15 +2,11 @@ import { sanitizeError, TaskRunError } from "@trigger.dev/core/v3";
 import { type Prisma, type TaskRun } from "@trigger.dev/database";
 import { logger } from "~/services/logger.server";
 import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
-import {
-  isFailedRunStatus,
-  type FINAL_ATTEMPT_STATUSES,
-  type FINAL_RUN_STATUSES,
-} from "../taskStatus";
+import { generateFriendlyId } from "../friendlyIdentifiers";
+import { FINAL_ATTEMPT_STATUSES, isFailedRunStatus, type FINAL_RUN_STATUSES } from "../taskStatus";
 import { PerformTaskRunAlertsService } from "./alerts/performTaskRunAlerts.server";
 import { BaseService } from "./baseService.server";
 import { ResumeDependentParentsService } from "./resumeDependentParents.server";
-import { generateFriendlyId } from "../friendlyIdentifiers";
 
 type BaseInput = {
   id: string;
@@ -60,36 +56,20 @@ export class FinalizeTaskRunService extends BaseService {
       completedAt,
     });
 
-    let output: string | undefined = undefined;
-    let outputType: string | undefined = undefined;
-
-    if (status === "COMPLETED_SUCCESSFULLY") {
-      // We need to get the output from the attempt
-      const attempt = await this._prisma.taskRunAttempt.findFirst({
-        where: { taskRunId: id, status: "COMPLETED", output: { not: null } },
-        orderBy: { id: "desc" },
-        select: {
-          output: true,
-          outputType: true,
-        },
-      });
-
-      if (attempt) {
-        output = attempt.output ?? undefined;
-        outputType = attempt.outputType;
-      }
-    }
-
     const run = await this._prisma.taskRun.update({
       where: { id },
-      data: { status, expiredAt, completedAt, output, outputType },
+      data: { status, expiredAt, completedAt },
       ...(include ? { include } : {}),
     });
 
     if (attemptStatus || error) {
       await this.finalizeAttempt({ attemptStatus, error, run });
     }
 
+    if (isFailedRunStatus(run.status)) {
+      await this.finalizeRunError(run, error);
+    }
+
     //resume any dependencies
     const resumeService = new ResumeDependentParentsService(this._prisma);
     const result = await resumeService.call({ id: run.id });
@@ -108,6 +88,19 @@ export class FinalizeTaskRunService extends BaseService {
     return run as Output<T>;
   }
 
+  async finalizeRunError(run: TaskRun, error?: TaskRunError) {
+    if (!error) {
+      return;
+    }
+
+    await this._prisma.taskRun.update({
+      where: { id: run.id },
+      data: {
+        error: sanitizeError(error),
+      },
+    });
+  }
+
   async finalizeAttempt({
     attemptStatus,
     error,
diff --git a/apps/webapp/app/v3/taskStatus.ts b/apps/webapp/app/v3/taskStatus.ts
@@ -50,8 +50,13 @@ export const FINAL_ATTEMPT_STATUSES = [
   "COMPLETED",
   "FAILED",
 ] satisfies TaskRunAttemptStatus[];
+
 export type FINAL_ATTEMPT_STATUSES = (typeof FINAL_ATTEMPT_STATUSES)[number];
 
+export const FAILED_ATTEMPT_STATUSES = ["FAILED", "CANCELED"] satisfies TaskRunAttemptStatus[];
+
+export type FAILED_ATTEMPT_STATUSES = (typeof FAILED_ATTEMPT_STATUSES)[number];
+
 export const FREEZABLE_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "RETRYING_AFTER_FAILURE"];
 export const FREEZABLE_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["EXECUTING", "FAILED"];
 
diff --git a/packages/database/prisma/migrations/20240925205409_add_error_to_task_run/migration.sql b/packages/database/prisma/migrations/20240925205409_add_error_to_task_run/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "TaskRun" ADD COLUMN     "error" JSONB;
diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma
@@ -1759,6 +1759,9 @@ model TaskRun {
   output     String?
   outputType String  @default("application/json")
 
+  /// Run error
+  error Json?
+
   @@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
   // Finding child runs
   @@index([parentTaskRunId])
diff --git a/references/v3-catalog/src/trigger/runMetadata.ts b/references/v3-catalog/src/trigger/runMetadata.ts
@@ -1,4 +1,4 @@
-import { logger, task, metadata } from "@trigger.dev/sdk/v3";
+import { logger, task, metadata, AbortTaskRunError } from "@trigger.dev/sdk/v3";
 
 export const runMetadataTask = task({
   id: "run-metadata-task",
@@ -55,7 +55,9 @@ export const runMetadataChildTask = task({
 
 export const runMetadataChildTask2 = task({
   id: "run-metadata-child-task-2",
-  run: async (payload: any, { ctx }) => {},
+  run: async (payload: any, { ctx }) => {
+    throw new AbortTaskRunError("aborting");
+  },
 });
 
 export const myTask = task({

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+-- AlterTable`
	`2`	`+ALTER TABLE "TaskRun" ADD COLUMN "error" JSONB;`