triggerdotdev · ericallam · Dec 3, 2024 · Dec 3, 2024
diff --git a/.changeset/shaggy-donkeys-hammer.md b/.changeset/shaggy-donkeys-hammer.md
@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Fix an issue that caused errors when using realtime with a run that is cancelled
diff --git a/apps/webapp/app/v3/services/finalizeTaskRun.server.ts b/apps/webapp/app/v3/services/finalizeTaskRun.server.ts
@@ -64,9 +64,13 @@ export class FinalizeTaskRunService extends BaseService {
      completedAt,
    });

+    // I moved the error update here for two reasons:
+    // - A single update is more efficient than two
+    // - If the status updates to a final status, realtime will receive that status and then shut down the stream
+    //   before the error is updated, which would cause the error to be lost
    const run = await this._prisma.taskRun.update({
      where: { id },
-      data: { status, expiredAt, completedAt },
+      data: { status, expiredAt, completedAt, error: error ? sanitizeError(error) : undefined },
      ...(include ? { include } : {}),
    });

@@ -78,10 +82,6 @@ export class FinalizeTaskRunService extends BaseService {
      await this.finalizeAttempt({ attemptStatus, error, run });
    }

-    if (error) {
-      await this.finalizeRunError(run, error);
-    }
-
    try {
      await this.#finalizeBatch(run);
    } catch (finalizeBatchError) {
@@ -211,15 +211,6 @@ export class FinalizeTaskRunService extends BaseService {
    }
  }

-  async finalizeRunError(run: TaskRun, error: TaskRunError) {
-    await this._prisma.taskRun.update({
-      where: { id: run.id },
-      data: {
-        error: sanitizeError(error),
-      },
-    });
-  }
-
  async finalizeAttempt({
    attemptStatus,
    error,

diff --git a/packages/core/src/v3/apiClient/runStream.ts b/packages/core/src/v3/apiClient/runStream.ts
@@ -1,4 +1,5 @@
 import { DeserializedJson } from "../../schemas/json.js";
+import { createJsonErrorObject } from "../errors.js";
 import { RunStatus, SubscribeRunRawShape } from "../schemas/api.js";
 import { SerializedError } from "../schemas/common.js";
 import { AnyRunTypes, AnyTask, InferRunTypes } from "../types/tasks.js";
@@ -347,7 +348,7 @@ export class RunSubscription<TRunTypes extends AnyRunTypes> {
      startedAt: row.startedAt ?? undefined,
      delayedUntil: row.delayUntil ?? undefined,
      queuedAt: row.queuedAt ?? undefined,
-      error: row.error ?? undefined,
+      error: row.error ? createJsonErrorObject(row.error) : undefined,
      isTest: row.isTest,
      metadata,
    } as RunShape<TRunTypes>;

diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts
@@ -1,6 +1,6 @@
 import { z } from "zod";
 import { DeserializedJsonSchema } from "../../schemas/json.js";
-import { SerializedError } from "./common.js";
+import { SerializedError, TaskRunError } from "./common.js";
 import { BackgroundWorkerMetadata } from "./resources.js";
 import { QueueOptions } from "./schemas.js";

@@ -708,7 +708,7 @@ export const SubscribeRunRawShape = z.object({
  output: z.string().nullish(),
  outputType: z.string().nullish(),
  runTags: z.array(z.string()).nullish().default([]),
-  error: SerializedError.nullish(),
+  error: TaskRunError.nullish(),
 });

 export type SubscribeRunRawShape = z.infer<typeof SubscribeRunRawShape>;

diff --git a/references/nextjs-realtime/src/trigger/example.ts b/references/nextjs-realtime/src/trigger/example.ts
@@ -15,7 +15,12 @@ export const exampleTask = schemaTask({

    metadata.set("status", { type: "started", progress: 0.1 });

-    await setTimeout(2000);
+    if (Math.random() < 0.9) {
+      // Simulate a failure
+      throw new Error("Random failure");
+    }
+
+    await setTimeout(20000);

    metadata.set("status", { type: "processing", progress: 0.5 });