diff --git a/apps/webapp/app/v3/services/resumeAttempt.server.ts b/apps/webapp/app/v3/services/resumeAttempt.server.ts index f7719034e..0abf0bd48 100644 --- a/apps/webapp/app/v3/services/resumeAttempt.server.ts +++ b/apps/webapp/app/v3/services/resumeAttempt.server.ts @@ -4,14 +4,13 @@ import { TaskRunExecutionResult, } from "@trigger.dev/core/v3"; import type { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket"; -import { $transaction, PrismaClientOrTransaction } from "~/db.server"; import { logger } from "~/services/logger.server"; import { marqs } from "~/v3/marqs/index.server"; import { socketIo } from "../handleSocketIo.server"; import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server"; import { BaseService } from "./baseService.server"; import { TaskRunAttempt } from "@trigger.dev/database"; -import { isFinalRunStatus } from "../taskStatus"; +import { FINAL_ATTEMPT_STATUSES, FINAL_RUN_STATUSES, isFinalRunStatus } from "../taskStatus"; export class ResumeAttemptService extends BaseService { private _logger = logger; @@ -21,145 +20,139 @@ export class ResumeAttemptService extends BaseService { ): Promise { this._logger.debug(`ResumeAttemptService.call()`, params); - await $transaction(this._prisma, async (tx) => { - const attempt = await tx.taskRunAttempt.findFirst({ - where: { - friendlyId: params.attemptFriendlyId, - }, - include: { - taskRun: true, - dependencies: { - include: { - taskRun: { - include: { - attempts: { - orderBy: { - number: "desc", - }, - take: 1, - select: { - id: true, - }, + const attempt = await this._prisma.taskRunAttempt.findFirst({ + where: { + friendlyId: params.attemptFriendlyId, + }, + include: { + taskRun: true, + dependencies: { + include: { + taskRun: { + include: { + attempts: { + orderBy: { + number: "desc", + }, + take: 1, + select: { + id: true, }, }, }, }, - orderBy: { - createdAt: "desc", - }, - take: 1, }, - batchDependencies: { - include: { - items: { - include: { - taskRun: { - include: { - attempts: { - orderBy: { - number: "desc", - }, - take: 1, - select: { - id: true, - }, + orderBy: { + createdAt: "desc", + }, + take: 1, + }, + batchDependencies: { + include: { + items: { + include: { + taskRun: { + include: { + attempts: { + orderBy: { + number: "desc", + }, + take: 1, + select: { + id: true, }, }, }, }, }, }, - orderBy: { - createdAt: "desc", - }, - take: 1, }, + orderBy: { + createdAt: "desc", + }, + take: 1, }, - }); - - if (!attempt) { - this._logger.error("Could not find attempt", params); - return; - } - - this._logger = logger.child({ - attemptId: attempt.id, - attemptFriendlyId: attempt.friendlyId, - taskRun: attempt.taskRun, - }); - - if (isFinalRunStatus(attempt.taskRun.status)) { - this._logger.error("Run is not resumable"); - return; - } - - let completedAttemptIds: string[] = []; - - switch (params.type) { - case "WAIT_FOR_DURATION": { - this._logger.debug("Sending duration wait resume message"); - - await this.#setPostResumeStatuses(attempt, tx); - - socketIo.coordinatorNamespace.emit("RESUME_AFTER_DURATION", { - version: "v1", - attemptId: attempt.id, - attemptFriendlyId: attempt.friendlyId, - }); - break; - } - case "WAIT_FOR_TASK": { - if (attempt.dependencies.length) { - // We only care about the latest dependency - const dependentAttempt = attempt.dependencies[0].taskRun.attempts[0]; - - if (!dependentAttempt) { - this._logger.error("No dependent attempt"); - return; - } - - completedAttemptIds = [dependentAttempt.id]; - } else { - this._logger.error("No task dependency"); - return; - } - - await this.#handleDependencyResume(attempt, completedAttemptIds, tx); - - break; - } - case "WAIT_FOR_BATCH": { - if (attempt.batchDependencies) { - // We only care about the latest batch dependency - const dependentBatchItems = attempt.batchDependencies[0].items; - - if (!dependentBatchItems) { - this._logger.error("No dependent batch items"); - return; - } - - completedAttemptIds = dependentBatchItems.map((item) => item.taskRun.attempts[0]?.id); - } else { - this._logger.error("No batch dependency"); - return; - } - - await this.#handleDependencyResume(attempt, completedAttemptIds, tx); - - break; - } - default: { - break; - } - } + }, }); + + if (!attempt) { + this._logger.error("Could not find attempt", params); + return; + } + + this._logger = logger.child({ + attemptId: attempt.id, + attemptFriendlyId: attempt.friendlyId, + taskRun: attempt.taskRun, + }); + + if (isFinalRunStatus(attempt.taskRun.status)) { + this._logger.error("Run is not resumable"); + return; + } + + let completedAttemptIds: string[] = []; + + switch (params.type) { + case "WAIT_FOR_DURATION": { + this._logger.debug("Sending duration wait resume message"); + + await this.#setPostResumeStatuses(attempt); + + socketIo.coordinatorNamespace.emit("RESUME_AFTER_DURATION", { + version: "v1", + attemptId: attempt.id, + attemptFriendlyId: attempt.friendlyId, + }); + break; + } + case "WAIT_FOR_TASK": { + if (attempt.dependencies.length) { + // We only care about the latest dependency + const dependentAttempt = attempt.dependencies[0].taskRun.attempts[0]; + + if (!dependentAttempt) { + this._logger.error("No dependent attempt"); + return; + } + + completedAttemptIds = [dependentAttempt.id]; + } else { + this._logger.error("No task dependency"); + return; + } + + await this.#handleDependencyResume(attempt, completedAttemptIds); + + break; + } + case "WAIT_FOR_BATCH": { + if (attempt.batchDependencies) { + // We only care about the latest batch dependency + const dependentBatchItems = attempt.batchDependencies[0].items; + + if (!dependentBatchItems) { + this._logger.error("No dependent batch items"); + return; + } + + completedAttemptIds = dependentBatchItems.map((item) => item.taskRun.attempts[0]?.id); + } else { + this._logger.error("No batch dependency"); + return; + } + + await this.#handleDependencyResume(attempt, completedAttemptIds); + + break; + } + default: { + break; + } + } } - async #handleDependencyResume( - attempt: TaskRunAttempt, - completedAttemptIds: string[], - tx: PrismaClientOrTransaction - ) { + async #handleDependencyResume(attempt: TaskRunAttempt, completedAttemptIds: string[]) { if (completedAttemptIds.length === 0) { this._logger.error("No completed attempt IDs"); return; @@ -169,7 +162,7 @@ export class ResumeAttemptService extends BaseService { const executions: TaskRunExecution[] = []; for (const completedAttemptId of completedAttemptIds) { - const completedAttempt = await tx.taskRunAttempt.findFirst({ + const completedAttempt = await this._prisma.taskRunAttempt.findFirst({ where: { id: completedAttemptId, taskRun: { @@ -221,7 +214,7 @@ export class ResumeAttemptService extends BaseService { executions.push(executionPayload.execution); } - await this.#setPostResumeStatuses(attempt, tx); + await this.#setPostResumeStatuses(attempt); socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", { version: "v1", @@ -233,21 +226,63 @@ export class ResumeAttemptService extends BaseService { }); } - async #setPostResumeStatuses(attempt: TaskRunAttempt, tx: PrismaClientOrTransaction) { - return await tx.taskRunAttempt.update({ - where: { - id: attempt.id, - }, - data: { - status: "EXECUTING", - taskRun: { - update: { - data: { - status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING", + async #setPostResumeStatuses(attempt: TaskRunAttempt) { + try { + const updatedAttempt = await this._prisma.taskRunAttempt.update({ + where: { + id: attempt.id, + status: { + notIn: FINAL_ATTEMPT_STATUSES, + }, + taskRun: { + status: { + notIn: FINAL_RUN_STATUSES, }, }, }, - }, - }); + data: { + status: "EXECUTING", + taskRun: { + update: { + data: { + status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING", + }, + }, + }, + }, + select: { + id: true, + status: true, + taskRun: { + select: { + id: true, + status: true, + }, + }, + }, + }); + + this._logger.debug("Set post resume statuses", { + run: { + id: updatedAttempt.taskRun.id, + status: updatedAttempt.taskRun.status, + }, + attempt: { + id: updatedAttempt.id, + status: updatedAttempt.status, + }, + }); + } catch (error) { + this._logger.error("Failed to set post resume statuses", { + error: + error instanceof Error + ? { + name: error.name, + message: error.message, + stack: error.stack, + } + : error, + }); + } } }