From 1dc42daed4c298735fcb7fff4abab3de1db1aa2a Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Wed, 12 Jul 2023 13:14:23 +0100 Subject: [PATCH] =?UTF-8?q?Added=20SDK=20and=20API=20support=20to=20stop?= =?UTF-8?q?=20runs=20when=20they=E2=80=99re=20canceled?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .changeset/fifty-melons-enjoy.md | 5 +++ .../api.v1.runs.$runId.tasks.$id.complete.ts | 3 +- .../api.v1.runs.$runId.tasks.$id.fail.ts | 3 +- .../app/services/runs/cancelRun.server.ts | 39 ++++++++++++++----- .../runs/performRunExecution.server.ts | 23 +++++++++++ examples/nextjs-example/src/jobs/general.ts | 12 +++++- packages/internal/src/schemas/api.ts | 10 +++++ packages/trigger-sdk/src/errors.ts | 10 ++++- packages/trigger-sdk/src/io.ts | 26 +++++++++++-- packages/trigger-sdk/src/triggerClient.ts | 13 ++++++- 10 files changed, 125 insertions(+), 19 deletions(-) create mode 100644 .changeset/fifty-melons-enjoy.md diff --git a/.changeset/fifty-melons-enjoy.md b/.changeset/fifty-melons-enjoy.md new file mode 100644 index 000000000..cf87a5ec7 --- /dev/null +++ b/.changeset/fifty-melons-enjoy.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Added support for Runs being canceled diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts index 1457c4000..85a6e5ca4 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts +++ b/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, diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts index b4d7a10e5..06b6277f1 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts +++ b/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, diff --git a/apps/webapp/app/services/runs/cancelRun.server.ts b/apps/webapp/app/services/runs/cancelRun.server.ts index 0a1e8841a..546b14df4 100644 --- a/apps/webapp/app/services/runs/cancelRun.server.ts +++ b/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({ - select: { - queueId: true, + const run = await tx.jobRun.findUniqueOrThrow({ + where: { + id: runId, }, + }); + + const shouldDecrementQueue = + run.status === "STARTED" || run.status === "PREPROCESSING"; + await tx.jobRun.update({ where: { id: runId }, data: { status: "CANCELED", - queue: { - update: { - jobCount: { - decrement: 1, - }, - }, + completedAt: new Date(), + queue: shouldDecrementQueue + ? { + 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( diff --git a/apps/webapp/app/services/runs/performRunExecution.server.ts b/apps/webapp/app/services/runs/performRunExecution.server.ts index aa775a89b..5527f3e5f 100644 --- a/apps/webapp/app/services/runs/performRunExecution.server.ts +++ b/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) { diff --git a/examples/nextjs-example/src/jobs/general.ts b/examples/nextjs-example/src/jobs/general.ts index 9dac3b08c..3f50c1883 100644 --- a/examples/nextjs-example/src/jobs/general.ts +++ b/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( @@ -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", diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index 357c793fc..c1025c5d7 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -356,6 +356,15 @@ export const RunJobRetryWithTaskSchema = z.object({ export type RunJobRetryWithTask = z.infer; +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, ]); diff --git a/packages/trigger-sdk/src/errors.ts b/packages/trigger-sdk/src/errors.ts index e2329b960..b4155bc65 100644 --- a/packages/trigger-sdk/src/errors.ts +++ b/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 ); } diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts index 02e39355e..2563072d4 100644 --- a/packages/trigger-sdk/src/io.ts +++ b/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, { - output: result ?? undefined, - properties: task.outputProperties ?? undefined, - }); + const completedTask = await this._apiClient.completeTask( + this._id, + task.id, + { + output: result ?? undefined, + properties: task.outputProperties ?? undefined, + } + ); + + if (completedTask.status === "CANCELED") { + throw new CanceledWithTaskError(completedTask); + } return result; } catch (error) { diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 53e5d0856..1499e6ce8 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/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);