Added SDK and API support to stop runs when they’re canceled

.changeset/fifty-melons-enjoy.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Added support for Runs being canceled

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts

@@ -117,7 +117,8 @@ export class CompleteRunTaskService {
 
       if (
         existingTask.status === "COMPLETED" ||
-        existingTask.status === "ERRORED"
+        existingTask.status === "ERRORED" ||
+        existingTask.status === "CANCELED"
       ) {
         logger.debug("Task already completed", {
           existingTask,

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts

@@ -121,7 +121,8 @@ export class FailRunTaskService {
 
       if (
         existingTask.status === "COMPLETED" ||
-        existingTask.status === "ERRORED"
+        existingTask.status === "ERRORED" ||
+        existingTask.status === "CANCELED"
       ) {
         logger.debug("Task already completed", {
           existingTask,

apps/webapp/app/services/runs/cancelRun.server.ts

@@ -11,21 +11,42 @@ export class CancelRunService {
   public async call({ runId }: { runId: string }) {
     try {
       return await this.#prismaClient.$transaction(async (tx) => {
-        const run = await tx.jobRun.update({
+        const run = await tx.jobRun.findUniqueOrThrow({
-          select: {
+          where: {
-            queueId: true,
+            id: runId,
           },
+        });
+
+        const shouldDecrementQueue =
+          run.status === "STARTED" || run.status === "PREPROCESSING";
+        await tx.jobRun.update({
           where: { id: runId },
           data: {
             status: "CANCELED",
-            queue: {
+            completedAt: new Date(),
-              update: {
+            queue: shouldDecrementQueue
-                jobCount: {
+              ? {
-                  decrement: 1,
+                  update: {
-                },
+                    jobCount: {
-              },
+                      decrement: 1,
+                    },
+                  },
+                }
+              : undefined,
+          },
+        });
+
+        await tx.task.updateMany({
+          where: {
+            runId,
+            status: {
+              in: ["PENDING", "RUNNING", "WAITING"],
             },
           },
+          data: {
+            status: "CANCELED",
+            completedAt: new Date(),
+          },
         });
 
         await workerQueue.enqueue(

apps/webapp/app/services/runs/performRunExecution.server.ts

@@ -2,6 +2,7 @@ import type { Task } from "@trigger.dev/database";
 import {
   ApiEventLogSchema,
   CachedTaskSchema,
+  RunJobCanceledWithTask,
   RunJobError,
   RunJobResumeWithTask,
   RunJobRetryWithTask,
@@ -193,6 +194,11 @@ export class PerformRunExecutionService {
   async #executeJob(execution: FoundRunExecution) {
     const { run } = execution;
 
+    if (run.status === "CANCELED") {
+      await this.#cancelExecution(execution);
+      return;
+    }
+
     const client = new EndpointApi(
       run.environment.apiKey,
       run.endpoint.url,
@@ -339,6 +345,10 @@ export class PerformRunExecutionService {
 
         break;
       }
+      case "CANCELED": {
+        await this.#cancelExecution(execution);
+        break;
+      }
       default: {
         const _exhaustiveCheck: never = status;
         throw new Error(`Non-exhaustive match for value: ${status}`);
@@ -704,6 +714,19 @@ export class PerformRunExecutionService {
       });
     });
   }
+
+  async #cancelExecution(execution: FoundRunExecution) {
+    await this.#prismaClient.jobRunExecution.update({
+      where: {
+        id: execution.id,
+      },
+      data: {
+        status: "FAILURE",
+        completedAt: new Date(),
+        error: "This never ran because it was canceled by the user.",
+      },
+    });
+  }
 }
 
 async function findRunExecution(prisma: PrismaClientOrTransaction, id: string) {

examples/nextjs-example/src/jobs/general.ts

@@ -45,6 +45,12 @@ new Job(client, {
       body: z.any().optional(),
       retry: z.any().optional(),
     }),
+    examples: {
+      successfulRequest: {
+        url: "https://httpbin.org/status/200",
+        method: "GET",
+      },
+    },
   }),
   run: async (payload, io, ctx) => {
     return await io.backgroundFetch<any>(
@@ -635,10 +641,14 @@ new Job(client, {
     repo: "basic-starter-12k",
   }),
   run: async (payload, io, ctx) => {
-    await io.wait("wait", 5); // wait for 5 seconds
+    await io.runTask("slow task", { name: "slow task" }, async () => {
+      await new Promise((resolve) => setTimeout(resolve, 5000));
+    });
 
     await io.logger.info("This is a simple log info message");
 
+    await io.wait("wait", 5); // wait for 5 seconds
+
     const response = await io.slack.postMessage("Slack 📝", {
       text: `New Issue opened: ${payload.issue.html_url}`,
       channel: "C04GWUTDC3W",

packages/internal/src/schemas/api.ts

@@ -356,6 +356,15 @@ export const RunJobRetryWithTaskSchema = z.object({
 
 export type RunJobRetryWithTask = z.infer<typeof RunJobRetryWithTaskSchema>;
 
+export const RunJobCanceledWithTaskSchema = z.object({
+  status: z.literal("CANCELED"),
+  task: TaskSchema,
+});
+
+export type RunJobCanceledWithTask = z.infer<
+  typeof RunJobCanceledWithTaskSchema
+>;
+
 export const RunJobSuccessSchema = z.object({
   status: z.literal("SUCCESS"),
   output: DeserializedJsonSchema.optional(),
@@ -367,6 +376,7 @@ export const RunJobResponseSchema = z.discriminatedUnion("status", [
   RunJobErrorSchema,
   RunJobResumeWithTaskSchema,
   RunJobRetryWithTaskSchema,
+  RunJobCanceledWithTaskSchema,
   RunJobSuccessSchema,
 ]);
 

packages/trigger-sdk/src/errors.ts

@@ -12,6 +12,10 @@ export class RetryWithTaskError {
   ) {}
 }
 
+export class CanceledWithTaskError {
+  constructor(public task: ServerTask) {}
+}
+
 /** Use this function if you're using a `try/catch` block to catch errors.
  * It checks if a thrown error is a special internal error that you should ignore.
  * If this returns `true` then you must rethrow the error: `throw err;`
@@ -20,8 +24,10 @@ export class RetryWithTaskError {
  */
 export function isTriggerError(
   err: unknown
-): err is ResumeWithTaskError | RetryWithTaskError {
+): err is ResumeWithTaskError | RetryWithTaskError | CanceledWithTaskError {
   return (
-    err instanceof ResumeWithTaskError || err instanceof RetryWithTaskError
+    err instanceof ResumeWithTaskError ||
+    err instanceof RetryWithTaskError ||
+    err instanceof CanceledWithTaskError
   );
 }

packages/trigger-sdk/src/io.ts

@@ -19,6 +19,7 @@ import { AsyncLocalStorage } from "node:async_hooks";
 import { webcrypto } from "node:crypto";
 import { ApiClient } from "./apiClient";
 import {
+  CanceledWithTaskError,
   ResumeWithTaskError,
   RetryWithTaskError,
   isTriggerError,
@@ -513,6 +514,15 @@ export class IO {
       parentId,
     });
 
+    if (task.status === "CANCELED") {
+      this._logger.debug("Task canceled", {
+        idempotencyKey,
+        task,
+      });
+
+      throw new CanceledWithTaskError(task);
+    }
+
     if (task.status === "COMPLETED") {
       this._logger.debug("Using task output", {
         idempotencyKey,
@@ -560,10 +570,18 @@ export class IO {
           task,
         });
 
-        await this._apiClient.completeTask(this._id, task.id, {
+        const completedTask = await this._apiClient.completeTask(
-          output: result ?? undefined,
+          this._id,
-          properties: task.outputProperties ?? undefined,
+          task.id,
-        });
+          {
+            output: result ?? undefined,
+            properties: task.outputProperties ?? undefined,
+          }
+        );
+
+        if (completedTask.status === "CANCELED") {
+          throw new CanceledWithTaskError(completedTask);
+        }
 
         return result;
       } catch (error) {

packages/trigger-sdk/src/triggerClient.ts

@@ -23,7 +23,11 @@ import {
   SourceMetadata,
 } from "@trigger.dev/internal";
 import { ApiClient } from "./apiClient";
-import { ResumeWithTaskError, RetryWithTaskError } from "./errors";
+import {
+  CanceledWithTaskError,
+  ResumeWithTaskError,
+  RetryWithTaskError,
+} from "./errors";
 import { IO } from "./io";
 import { createIOWithIntegrations } from "./ioWithIntegrations";
 import { Job } from "./job";
@@ -659,6 +663,13 @@ export class TriggerClient {
         };
       }
 
+      if (error instanceof CanceledWithTaskError) {
+        return {
+          status: "CANCELED",
+          task: error.task,
+        };
+      }
+
       if (error instanceof RetryWithTaskError) {
         const errorWithStack = ErrorWithStackSchema.safeParse(error.cause);