feat: Add ability to cancel all runs for job from SDK (#819)

* feat: Add ability to cancel all runs for job from SDK

* Create nine-trainers-vanish.md

* Cancel runs that are executing and waiting

---------

Co-authored-by: Matt Aitken <matt@mattaitken.com>

.changeset/nine-trainers-vanish.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+feat: Add ability to cancel all runs for job from SDK

apps/webapp/app/routes/api.v1.jobs.$jobSlug.cancel-runs.ts

@@ -0,0 +1,48 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { CancelRunsForJobService } from "~/services/jobs/cancelRunsForJob.server";
+
+const ParamsSchema = z.object({
+  jobSlug: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or Missing jobSlug" }, { status: 400 });
+  }
+
+  const { jobSlug } = parsed.data;
+
+  const service = new CancelRunsForJobService();
+  try {
+    const res = await service.call(authenticatedEnv, jobSlug);
+
+    if (!res) {
+      return json({ error: "Job not found" }, { status: 404 });
+    }
+
+    return json(res);
+  } catch (err) {
+    logger.error("CancelRunsForJobService.call() error", { error: err });
+    return json({ error: "Internal Server Error" }, { status: 500 });
+  }
+}

apps/webapp/app/services/jobs/cancelRunsForJob.server.ts

@@ -0,0 +1,73 @@
+import { $transaction, PrismaClient, prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { JobRunStatus } from "@trigger.dev/database";
+import { CancelRunService } from "../runs/cancelRun.server";
+import { logger } from "../logger.server";
+import { CancelRunsForJob } from "@trigger.dev/core";
+
+const CANCELLABLE_JOB_RUN_STATUS: JobRunStatus[] = [
+  JobRunStatus.PENDING,
+  JobRunStatus.QUEUED,
+  JobRunStatus.WAITING_ON_CONNECTIONS,
+  JobRunStatus.PREPROCESSING,
+  JobRunStatus.STARTED,
+  JobRunStatus.EXECUTING,
+  JobRunStatus.WAITING_TO_CONTINUE,
+  JobRunStatus.WAITING_TO_EXECUTE,
+];
+
+export class CancelRunsForJobService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(environment: AuthenticatedEnvironment, jobSlug: string) {
+    return await $transaction<CancelRunsForJob | undefined>(this.#prismaClient, async (tx) => {
+      const job = await tx.job.findUnique({
+        where: {
+          projectId_slug: {
+            projectId: environment.projectId,
+            slug: jobSlug,
+          },
+        },
+      });
+
+      if (!job) {
+        return;
+      }
+
+      const jobRuns = await tx.jobRun.findMany({
+        where: {
+          jobId: job.id,
+          status: {
+            in: CANCELLABLE_JOB_RUN_STATUS,
+          },
+        },
+        select: {
+          id: true,
+        },
+      });
+
+      const cancelRunService = new CancelRunService(this.#prismaClient);
+      const cancelledRunIds: string[] = [];
+      const failedToCancelRunIds: string[] = [];
+
+      for (const jobRun of jobRuns) {
+        try {
+          await cancelRunService.call({ runId: jobRun.id });
+          cancelledRunIds.push(jobRun.id);
+        } catch (err) {
+          logger.debug(`failed to cancel job run with id ${jobRun.id} for job ${jobSlug}`);
+          failedToCancelRunIds.push(jobRun.id);
+        }
+      }
+
+      return {
+        cancelledRunIds: cancelledRunIds,
+        failedToCancelRunIds: failedToCancelRunIds,
+      };
+    });
+  }
+}

packages/core/src/schemas/index.ts

@@ -15,3 +15,4 @@ export * from "./runs";
 export * from "./addMissingVersionField";
 export * from "./statuses";
 export * from "./request";
+export * from "./jobs";

packages/core/src/schemas/jobs.ts

@@ -0,0 +1,8 @@
+import { z } from "zod";
+
+export const CancelRunsForJobSchema = z.object({
+  cancelledRunIds: z.array(z.string()),
+  failedToCancelRunIds: z.array(z.string()),
+});
+
+export type CancelRunsForJob = z.infer<typeof CancelRunsForJobSchema>;

packages/trigger-sdk/src/apiClient.ts

@@ -3,6 +3,7 @@ import {
   ApiEventLog,
   ApiEventLogSchema,
   CancelRunsForEventSchema,
+  CancelRunsForJobSchema,
   CompleteTaskBodyV2Input,
   ConnectionAuthSchema,
   EphemeralEventDispatcherRequestBody,
@@ -553,6 +554,26 @@ export class ApiClient {
     });
   }
 
+  async cancelRunsForJob(jobId: string) {
+    const apiKey = await this.#apiKey();
+
+    this.#logger.debug("Cancelling Runs for Job", {
+      jobId,
+    });
+
+    return await zodfetch(
+      CancelRunsForJobSchema,
+      `${this.#apiUrl}/api/v1/jobs/${jobId}/cancel-runs`,
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          Authorization: `Bearer ${apiKey}`,
+        },
+      }
+    );
+  }
+
   async createEphemeralEventDispatcher(payload: EphemeralEventDispatcherRequestBody) {
     const apiKey = await this.#apiKey();
 

packages/trigger-sdk/src/triggerClient.ts

@@ -1065,6 +1065,10 @@ export class TriggerClient {
     return this.#client.invokeJob(jobId, payload, options);
   }
 
+  async cancelRunsForJob(jobId: string) {
+    return this.#client.cancelRunsForJob(jobId);
+  }
+
   async createEphemeralEventDispatcher(payload: EphemeralEventDispatcherRequestBody) {
     return this.#client.createEphemeralEventDispatcher(payload);
   }
@@ -1684,15 +1688,15 @@ export class TriggerClient {
         auth:
           resolvedAuth.type === "apiKey"
             ? {
-              type: "apiKey",
-              accessToken: resolvedAuth.token,
-              additionalFields: resolvedAuth.additionalFields,
-            }
+                type: "apiKey",
+                accessToken: resolvedAuth.token,
+                additionalFields: resolvedAuth.additionalFields,
+              }
             : {
-              type: "oauth2",
-              accessToken: resolvedAuth.token,
-              additionalFields: resolvedAuth.additionalFields,
-            },
+                type: "oauth2",
+                accessToken: resolvedAuth.token,
+                additionalFields: resolvedAuth.additionalFields,
+              },
       };
     } catch (resolverError) {
       if (resolverError instanceof Error) {
@@ -1709,8 +1713,9 @@ export class TriggerClient {
 
       return {
         ok: false,
-        error: `Auth could not be resolved for ${integration.id
-          }: auth resolver threw an unknown error: ${JSON.stringify(resolverError)}`,
+        error: `Auth could not be resolved for ${
+          integration.id
+        }: auth resolver threw an unknown error: ${JSON.stringify(resolverError)}`,
       };
     }
   }
@@ -1737,8 +1742,8 @@ export class TriggerClient {
         typeof job.options.concurrencyLimit === "number"
           ? job.options.concurrencyLimit
           : typeof job.options.concurrencyLimit === "object"
-            ? { id: job.options.concurrencyLimit.id, limit: job.options.concurrencyLimit.limit }
-            : undefined,
+          ? { id: job.options.concurrencyLimit.id, limit: job.options.concurrencyLimit.limit }
+          : undefined,
     };
   }