Fix cancelled runs breaking realtime subscriptions (#1533)

.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

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,

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>;

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>;

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 });