diff --git a/.changeset/tidy-pets-smell.md b/.changeset/tidy-pets-smell.md new file mode 100644 index 000000000..c4e04e4f9 --- /dev/null +++ b/.changeset/tidy-pets-smell.md @@ -0,0 +1,6 @@ +--- +"trigger.dev": patch +"@trigger.dev/core": patch +--- + +Fix for when a log flush times out and the process is checkpointed diff --git a/apps/webapp/app/platform/zodWorker.server.ts b/apps/webapp/app/platform/zodWorker.server.ts index 7d67ce1c0..b231a4aca 100644 --- a/apps/webapp/app/platform/zodWorker.server.ts +++ b/apps/webapp/app/platform/zodWorker.server.ts @@ -318,7 +318,7 @@ export class ZodWorker { identifier: K, payload: z.infer, options?: ZodWorkerEnqueueOptions - ): Promise { + ): Promise { const task = this.#tasks[identifier]; const optionsWithoutTx = removeUndefinedKeys(omit(options ?? {}, ["tx"])); @@ -439,11 +439,9 @@ export class ZodWorker { identifier, payload, spec, + error: JSON.stringify(rows.error), }); - - throw new Error( - `Failed to add job to queue, zod parsing error: ${JSON.stringify(rows.error)}` - ); + return { job: undefined, durationInMs: Math.floor(durationInMs) }; } const job = rows.data[0]; diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx index ca0fcd799..7c9ad1de3 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx @@ -796,6 +796,12 @@ function RunTimelineLine({ title, state }: RunTimelineLineProps) { function RunError({ error }: { error: TaskRunError }) { switch (error.type) { case "STRING_ERROR": + return ( +
+ Error + {error.raw} +
+ ); case "CUSTOM_ERROR": { return (
diff --git a/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts b/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts index e9e6fa11d..e36c808ae 100644 --- a/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts +++ b/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts @@ -134,7 +134,7 @@ export class DeliverScheduledEventService { id, }, data: { - workerJobId: workerJob.id, + workerJobId: workerJob?.id, nextEventTimestamp: runAt, }, }); diff --git a/apps/webapp/app/v3/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository.server.ts index 989ba2b27..98872a50b 100644 --- a/apps/webapp/app/v3/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository.server.ts @@ -226,6 +226,7 @@ export class EventRepository { const events = await this.queryIncompleteEvents({ spanId }); if (events.length === 0) { + logger.warn("No incomplete events found for spanId", { spanId, options }); return; } diff --git a/apps/webapp/app/v3/failedTaskRun.server.ts b/apps/webapp/app/v3/failedTaskRun.server.ts index 31ef876e2..c322241df 100644 --- a/apps/webapp/app/v3/failedTaskRun.server.ts +++ b/apps/webapp/app/v3/failedTaskRun.server.ts @@ -42,26 +42,10 @@ export class FailedTaskRunService extends BaseService { id: taskRun.id, status: "SYSTEM_FAILURE", completedAt: new Date(), + attemptStatus: "FAILED", + error: sanitizeError(completion.error), }); - // Get the final attempt and add the error to it, if it's not already set - const finalAttempt = await this._prisma.taskRunAttempt.findFirst({ - where: { - taskRunId: taskRun.id, - }, - orderBy: { id: "desc" }, - }); - - if (finalAttempt && !finalAttempt.error) { - // Haven't set the status because the attempt might still be running - await this._prisma.taskRunAttempt.update({ - where: { id: finalAttempt.id }, - data: { - error: sanitizeError(completion.error), - }, - }); - } - // Now we need to "complete" the task run event/span await eventRepository.completeEvent(taskRun.spanId, { endTime: new Date(), diff --git a/apps/webapp/app/v3/services/cancelAttempt.server.ts b/apps/webapp/app/v3/services/cancelAttempt.server.ts index bdbaa2f3f..77e99c489 100644 --- a/apps/webapp/app/v3/services/cancelAttempt.server.ts +++ b/apps/webapp/app/v3/services/cancelAttempt.server.ts @@ -5,7 +5,6 @@ import { eventRepository } from "../eventRepository.server"; import { isCancellableRunStatus } from "../taskStatus"; import { BaseService } from "./baseService.server"; import { FinalizeTaskRunService } from "./finalizeTaskRun.server"; -import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server"; export class CancelAttemptService extends BaseService { public async call( @@ -61,13 +60,15 @@ export class CancelAttemptService extends BaseService { }, }); + const isCancellable = isCancellableRunStatus(taskRunAttempt.taskRun.status); + const finalizeService = new FinalizeTaskRunService(tx); await finalizeService.call({ id: taskRunId, - status: isCancellableRunStatus(taskRunAttempt.taskRun.status) ? "INTERRUPTED" : undefined, - completedAt: isCancellableRunStatus(taskRunAttempt.taskRun.status) - ? cancelledAt - : undefined, + status: isCancellable ? "INTERRUPTED" : undefined, + completedAt: isCancellable ? cancelledAt : undefined, + attemptStatus: isCancellable ? "CANCELED" : undefined, + error: isCancellable ? { type: "STRING_ERROR", raw: reason } : undefined, }); }); @@ -84,10 +85,6 @@ export class CancelAttemptService extends BaseService { return eventRepository.cancelEvent(event, cancelledAt, reason); }) ); - - if (environment?.type !== "DEVELOPMENT") { - await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma); - } }); } } diff --git a/apps/webapp/app/v3/services/cancelTaskRun.server.ts b/apps/webapp/app/v3/services/cancelTaskRun.server.ts index f6810cfc5..a0f37ab23 100644 --- a/apps/webapp/app/v3/services/cancelTaskRun.server.ts +++ b/apps/webapp/app/v3/services/cancelTaskRun.server.ts @@ -76,6 +76,11 @@ export class CancelTaskRunService extends BaseService { runtimeEnvironment: true, lockedToVersion: true, }, + attemptStatus: "CANCELED", + error: { + type: "STRING_ERROR", + raw: opts.reason, + }, }); const inProgressEvents = await eventRepository.queryIncompleteEvents({ diff --git a/apps/webapp/app/v3/services/completeAttempt.server.ts b/apps/webapp/app/v3/services/completeAttempt.server.ts index 1c5e3a455..a7fb61671 100644 --- a/apps/webapp/app/v3/services/completeAttempt.server.ts +++ b/apps/webapp/app/v3/services/completeAttempt.server.ts @@ -17,7 +17,6 @@ import { createExceptionPropertiesFromError, eventRepository } from "../eventRep import { marqs } from "~/v3/marqs/index.server"; import { BaseService } from "./baseService.server"; import { CancelAttemptService } from "./cancelAttempt.server"; -import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server"; import { MAX_TASK_RUN_ATTEMPTS } from "~/consts"; import { CreateCheckpointService } from "./createCheckpoint.server"; import { TaskRun } from "@trigger.dev/database"; @@ -76,6 +75,12 @@ export class CompleteAttemptService extends BaseService { id: run.id, status: "SYSTEM_FAILURE", completedAt: new Date(), + attemptStatus: "FAILED", + error: { + type: "INTERNAL_ERROR", + code: "TASK_EXECUTION_FAILED", + message: "Tried to complete attempt but it doesn't exist", + }, }); // No attempt, so there's no message to ACK @@ -149,10 +154,6 @@ export class CompleteAttemptService extends BaseService { }, }); - if (!env || env.type !== "DEVELOPMENT") { - await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma); - } - return "COMPLETED"; } @@ -355,10 +356,6 @@ export class CompleteAttemptService extends BaseService { }); } - if (!env || env.type !== "DEVELOPMENT") { - await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma); - } - return "COMPLETED"; } } diff --git a/apps/webapp/app/v3/services/crashTaskRun.server.ts b/apps/webapp/app/v3/services/crashTaskRun.server.ts index 7820fc2b0..9556699a5 100644 --- a/apps/webapp/app/v3/services/crashTaskRun.server.ts +++ b/apps/webapp/app/v3/services/crashTaskRun.server.ts @@ -4,7 +4,6 @@ import { marqs } from "~/v3/marqs/index.server"; import { BaseService } from "./baseService.server"; import { logger } from "~/services/logger.server"; import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; -import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server"; import { CRASHABLE_ATTEMPT_STATUSES, isCrashableRunStatus } from "../taskStatus"; import { sanitizeError } from "@trigger.dev/core/v3"; import { FinalizeTaskRunService } from "./finalizeTaskRun.server"; @@ -69,6 +68,13 @@ export class CrashTaskRunService extends BaseService { }, }, }, + attemptStatus: "FAILED", + error: { + type: "INTERNAL_ERROR", + code: "TASK_RUN_CRASHED", + message: opts.reason, + stackTrace: opts.logs, + }, }); const inProgressEvents = await eventRepository.queryIncompleteEvents( @@ -146,12 +152,6 @@ export class CrashTaskRunService extends BaseService { }), }, }); - - if (environment.type === "DEVELOPMENT") { - return; - } - - await ResumeTaskRunDependenciesService.enqueue(attempt.id, this._prisma); }); } } diff --git a/apps/webapp/app/v3/services/createCheckpoint.server.ts b/apps/webapp/app/v3/services/createCheckpoint.server.ts index 08797096c..b7189716b 100644 --- a/apps/webapp/app/v3/services/createCheckpoint.server.ts +++ b/apps/webapp/app/v3/services/createCheckpoint.server.ts @@ -4,16 +4,11 @@ import type { Checkpoint, CheckpointRestoreEvent } from "@trigger.dev/database"; import { logger } from "~/services/logger.server"; import { marqs } from "~/v3/marqs/index.server"; import { generateFriendlyId } from "../friendlyIdentifiers"; -import { - isFinalAttemptStatus, - isFinalRunStatus, - isFreezableAttemptStatus, - isFreezableRunStatus, -} from "../taskStatus"; +import { isFreezableAttemptStatus, isFreezableRunStatus } from "../taskStatus"; import { BaseService } from "./baseService.server"; import { CreateCheckpointRestoreEventService } from "./createCheckpointRestoreEvent.server"; import { ResumeBatchRunService } from "./resumeBatchRun.server"; -import { ResumeTaskDependencyService } from "./resumeTaskDependency.server"; +import { ResumeDependentParentsService } from "./resumeDependentParents.server"; export class CreateCheckpointService extends BaseService { public async call( @@ -177,127 +172,15 @@ export class CreateCheckpointService extends BaseService { }); await marqs?.cancelHeartbeat(attempt.taskRunId); - const dependency = await this._prisma.taskRunDependency.findFirst({ - select: { - id: true, - taskRunId: true, - }, - where: { - taskRun: { - friendlyId: reason.friendlyId, - }, - }, - }); + const resumeService = new ResumeDependentParentsService(this._prisma); + const result = await resumeService.call({ id: attempt.taskRunId }); - logger.log("CreateCheckpointService: Created checkpoint WAIT_FOR_TASK", { - checkpointId: checkpoint.id, - runFriendlyId: reason.friendlyId, - dependencyId: dependency?.id, - }); - - if (!dependency) { - logger.error("CreateCheckpointService: Dependency not found", { - friendlyId: reason.friendlyId, - }); - - return { - success: true, - checkpoint, - event: checkpointEvent, - keepRunAlive: false, - }; + if (result.success) { + logger.log("CreateCheckpointService: Resumed dependent parents", result); + } else { + logger.error("CreateCheckpointService: Failed to resume dependent parents", result); } - const childRun = await this._prisma.taskRun.findFirst({ - select: { - id: true, - status: true, - }, - where: { - id: dependency.taskRunId, - }, - }); - - if (!childRun) { - logger.error("CreateCheckpointService: Dependency child run not found", { - taskRunId: dependency.taskRunId, - runFriendlyId: reason.friendlyId, - dependencyId: dependency.id, - }); - - return { - success: true, - checkpoint, - event: checkpointEvent, - keepRunAlive: false, - }; - } - - const isFinished = isFinalRunStatus(childRun.status); - if (!isFinished) { - logger.debug("CreateCheckpointService: Dependency child run not finished", { - taskRunId: dependency.taskRunId, - runFriendlyId: reason.friendlyId, - dependencyId: dependency.id, - childRunStatus: childRun.status, - childRunId: childRun.id, - }); - - return { - success: true, - checkpoint, - event: checkpointEvent, - keepRunAlive: false, - }; - } - - const lastAttempt = await this._prisma.taskRunAttempt.findFirst({ - select: { - id: true, - status: true, - }, - where: { - taskRunId: dependency.taskRunId, - }, - orderBy: { - createdAt: "desc", - }, - }); - - if (!lastAttempt) { - logger.debug("CreateCheckpointService: Dependency child attempt not found", { - taskRunId: dependency.taskRunId, - runFriendlyId: reason.friendlyId, - dependencyId: dependency?.id, - }); - return { - success: true, - checkpoint, - event: checkpointEvent, - keepRunAlive: false, - }; - } - - if (!isFinalAttemptStatus(lastAttempt.status)) { - logger.debug("CreateCheckpointService: Dependency child attempt not final", { - taskRunId: dependency.taskRunId, - runFriendlyId: reason.friendlyId, - dependencyId: dependency.id, - lastAttemptId: lastAttempt.id, - lastAttemptStatus: lastAttempt.status, - }); - - return { - success: true, - checkpoint, - event: checkpointEvent, - keepRunAlive: false, - }; - } - - //resume the dependent task - await ResumeTaskDependencyService.enqueue(dependency.id, lastAttempt.id, this._prisma); - return { success: true, checkpoint, diff --git a/apps/webapp/app/v3/services/expireEnqueuedRun.server.ts b/apps/webapp/app/v3/services/expireEnqueuedRun.server.ts index d1a3abb99..0f3cbfb4a 100644 --- a/apps/webapp/app/v3/services/expireEnqueuedRun.server.ts +++ b/apps/webapp/app/v3/services/expireEnqueuedRun.server.ts @@ -45,6 +45,11 @@ export class ExpireEnqueuedRunService extends BaseService { status: "EXPIRED", expiredAt: new Date(), completedAt: new Date(), + attemptStatus: "FAILED", + error: { + type: "STRING_ERROR", + raw: `Run expired because the TTL (${run.ttl}) was reached`, + }, }); await eventRepository.completeEvent(run.spanId, { diff --git a/apps/webapp/app/v3/services/finalizeTaskRun.server.ts b/apps/webapp/app/v3/services/finalizeTaskRun.server.ts index eaf788d5a..6a42549a8 100644 --- a/apps/webapp/app/v3/services/finalizeTaskRun.server.ts +++ b/apps/webapp/app/v3/services/finalizeTaskRun.server.ts @@ -1,16 +1,24 @@ +import { sanitizeError, TaskRunError } from "@trigger.dev/core/v3"; import { type Prisma, type TaskRun } from "@trigger.dev/database"; import { logger } from "~/services/logger.server"; -import { marqs } from "~/v3/marqs/index.server"; -import { BaseService } from "./baseService.server"; -import { isFailedRunStatus, type FINAL_RUN_STATUSES } from "../taskStatus"; -import { PerformTaskAttemptAlertsService } from "./alerts/performTaskAttemptAlerts.server"; +import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server"; +import { + isFailedRunStatus, + type FINAL_ATTEMPT_STATUSES, + type FINAL_RUN_STATUSES, +} from "../taskStatus"; import { PerformTaskRunAlertsService } from "./alerts/performTaskRunAlerts.server"; +import { BaseService } from "./baseService.server"; +import { ResumeDependentParentsService } from "./resumeDependentParents.server"; +import { generateFriendlyId } from "../friendlyIdentifiers"; type BaseInput = { id: string; status?: FINAL_RUN_STATUSES; expiredAt?: Date; completedAt?: Date; + attemptStatus?: FINAL_ATTEMPT_STATUSES; + error?: TaskRunError; }; type InputWithInclude = BaseInput & { @@ -32,6 +40,8 @@ export class FinalizeTaskRunService extends BaseService { expiredAt, completedAt, include, + attemptStatus, + error, }: T extends Prisma.TaskRunInclude ? InputWithInclude : InputWithoutInclude): Promise< Output > { @@ -56,6 +66,20 @@ export class FinalizeTaskRunService extends BaseService { ...(include ? { include } : {}), }); + if (attemptStatus || error) { + await this.finalizeAttempt({ attemptStatus, error, run }); + } + + //resume any dependencies + const resumeService = new ResumeDependentParentsService(this._prisma); + const result = await resumeService.call({ id: run.id }); + + if (result.success) { + logger.log("FinalizeTaskRunService: Resumed dependent parents", { result }); + } else { + logger.error("FinalizeTaskRunService: Failed to resume dependent parents", { result }); + } + //enqueue alert if (isFailedRunStatus(run.status)) { await PerformTaskRunAlertsService.enqueue(run.id, this._prisma); @@ -63,4 +87,85 @@ export class FinalizeTaskRunService extends BaseService { return run as Output; } + + async finalizeAttempt({ + attemptStatus, + error, + run, + }: { + attemptStatus?: FINAL_ATTEMPT_STATUSES; + error?: TaskRunError; + run: TaskRun; + }) { + if (attemptStatus || error) { + const latestAttempt = await this._prisma.taskRunAttempt.findFirst({ + where: { taskRunId: run.id }, + orderBy: { id: "desc" }, + take: 1, + }); + + if (latestAttempt) { + logger.debug("Finalizing run attempt", { + id: latestAttempt.id, + status: attemptStatus, + error, + }); + + await this._prisma.taskRunAttempt.update({ + where: { id: latestAttempt.id }, + data: { status: attemptStatus, error: error ? sanitizeError(error) : undefined }, + }); + } else { + logger.debug("Finalizing run no attempt found", { + runId: run.id, + attemptStatus, + error, + }); + + const workerTask = await this._prisma.backgroundWorkerTask.findFirst({ + select: { + id: true, + workerId: true, + runtimeEnvironmentId: true, + }, + where: { + id: run.lockedById!, + }, + }); + + if (!workerTask) { + logger.error("FinalizeTaskRunService: No worker task found", { runId: run.id }); + return; + } + + const queue = await this._prisma.taskQueue.findUnique({ + where: { + runtimeEnvironmentId_name: { + runtimeEnvironmentId: workerTask.runtimeEnvironmentId, + name: sanitizeQueueName(run.queue), + }, + }, + }); + + if (!queue) { + logger.error("FinalizeTaskRunService: No queue found", { runId: run.id }); + return; + } + + await this._prisma.taskRunAttempt.create({ + data: { + number: 1, + friendlyId: generateFriendlyId("attempt"), + taskRunId: run.id, + backgroundWorkerId: workerTask?.workerId, + backgroundWorkerTaskId: workerTask?.id, + queueId: queue.id, + runtimeEnvironmentId: workerTask.runtimeEnvironmentId, + status: attemptStatus, + error: error ? sanitizeError(error) : undefined, + }, + }); + } + } + } } diff --git a/apps/webapp/app/v3/services/resumeBatchRun.server.ts b/apps/webapp/app/v3/services/resumeBatchRun.server.ts index 0930dddc8..f0a9b57f9 100644 --- a/apps/webapp/app/v3/services/resumeBatchRun.server.ts +++ b/apps/webapp/app/v3/services/resumeBatchRun.server.ts @@ -4,22 +4,13 @@ import { marqs } from "~/v3/marqs/index.server"; import { BaseService } from "./baseService.server"; import { logger } from "~/services/logger.server"; +const finishedBatchRunStatuses = ["COMPLETED", "FAILED", "CANCELED"]; + export class ResumeBatchRunService extends BaseService { public async call(batchRunId: string) { const batchRun = await this._prisma.batchTaskRun.findFirst({ where: { id: batchRunId, - dependentTaskAttemptId: { - not: null, - }, - status: "PENDING", - items: { - every: { - taskRunAttemptId: { - not: null, - }, - }, - }, }, include: { dependentTaskAttempt: { @@ -38,6 +29,26 @@ export class ResumeBatchRunService extends BaseService { }); if (!batchRun || !batchRun.dependentTaskAttempt) { + logger.error( + "ResumeBatchRunService: Batch run doesn't exist or doesn't have a dependent attempt", + { + batchRun, + } + ); + return; + } + + if (batchRun.status === "COMPLETED") { + logger.debug("ResumeBatchRunService: Batch run is already completed", { + batchRun: batchRun, + }); + return; + } + + if (batchRun.items.some((item) => !finishedBatchRunStatuses.includes(item.status))) { + logger.debug("ResumeBatchRunService: All items aren't yet completed", { + batchRun: batchRun, + }); return; } @@ -61,34 +72,45 @@ export class ResumeBatchRunService extends BaseService { const dependentRun = batchRun.dependentTaskAttempt.taskRun; if (batchRun.dependentTaskAttempt.status === "PAUSED" && batchRun.checkpointEventId) { - // We need to update the batchRun status so we don't resume it again - await this._prisma.batchTaskRun.update({ - where: { - id: batchRun.id, - }, - data: { - status: "COMPLETED", - }, + logger.debug("ResumeBatchRunService: Attempt is paused and has a checkpoint event", { + batchRunId: batchRun.id, + dependentTaskAttempt: batchRun.dependentTaskAttempt, + checkpointEventId: batchRun.checkpointEventId, }); - await marqs?.enqueueMessage( - environment, - dependentRun.queue, - dependentRun.id, - { - type: "RESUME", - completedAttemptIds: [], - resumableAttemptId: batchRun.dependentTaskAttempt.id, + // We need to update the batchRun status so we don't resume it again + const wasUpdated = await this.#setBatchToCompletedOnce(batchRun.id); + if (wasUpdated) { + logger.debug("ResumeBatchRunService: Resuming dependent run with checkpoint", { + batchRunId: batchRun.id, + dependentTaskAttemptId: batchRun.dependentTaskAttempt.id, + }); + await marqs?.enqueueMessage( + environment, + dependentRun.queue, + dependentRun.id, + { + type: "RESUME", + completedAttemptIds: [], + resumableAttemptId: batchRun.dependentTaskAttempt.id, + checkpointEventId: batchRun.checkpointEventId, + taskIdentifier: batchRun.dependentTaskAttempt.taskRun.taskIdentifier, + projectId: batchRun.dependentTaskAttempt.runtimeEnvironment.projectId, + environmentId: batchRun.dependentTaskAttempt.runtimeEnvironment.id, + environmentType: batchRun.dependentTaskAttempt.runtimeEnvironment.type, + }, + dependentRun.concurrencyKey ?? undefined + ); + } else { + logger.debug("ResumeBatchRunService: with checkpoint was already completed", { + batchRunId: batchRun.id, + dependentTaskAttempt: batchRun.dependentTaskAttempt, checkpointEventId: batchRun.checkpointEventId, - taskIdentifier: batchRun.dependentTaskAttempt.taskRun.taskIdentifier, - projectId: batchRun.dependentTaskAttempt.runtimeEnvironment.projectId, - environmentId: batchRun.dependentTaskAttempt.runtimeEnvironment.id, - environmentType: batchRun.dependentTaskAttempt.runtimeEnvironment.type, - }, - dependentRun.concurrencyKey ?? undefined - ); + hasCheckpointEvent: !!batchRun.checkpointEventId, + }); + } } else { - logger.debug("Batch run resume: Attempt is not paused or there's no checkpoint event", { + logger.debug("ResumeBatchRunService: attempt is not paused or there's no checkpoint event", { batchRunId: batchRun.id, dependentTaskAttempt: batchRun.dependentTaskAttempt, checkpointEventId: batchRun.checkpointEventId, @@ -98,23 +120,60 @@ export class ResumeBatchRunService extends BaseService { if (batchRun.dependentTaskAttempt.status === "PAUSED" && !batchRun.checkpointEventId) { // In case of race conditions the status can be PAUSED without a checkpoint event // When the checkpoint is created, it will continue the run - logger.error("Batch run resume: Attempt is paused but there's no checkpoint event", { + logger.error("ResumeBatchRunService: attempt is paused but there's no checkpoint event", { batchRunId: batchRun.id, dependentTaskAttemptId: batchRun.dependentTaskAttempt.id, }); return; } - await marqs?.replaceMessage(dependentRun.id, { - type: "RESUME", - completedAttemptIds: batchRun.items.map((item) => item.taskRunAttemptId).filter(Boolean), - resumableAttemptId: batchRun.dependentTaskAttempt.id, - checkpointEventId: batchRun.checkpointEventId ?? undefined, - taskIdentifier: batchRun.dependentTaskAttempt.taskRun.taskIdentifier, - projectId: batchRun.dependentTaskAttempt.runtimeEnvironment.projectId, - environmentId: batchRun.dependentTaskAttempt.runtimeEnvironment.id, - environmentType: batchRun.dependentTaskAttempt.runtimeEnvironment.type, - }); + // We need to update the batchRun status so we don't resume it again + const wasUpdated = await this.#setBatchToCompletedOnce(batchRun.id); + if (wasUpdated) { + logger.debug("ResumeBatchRunService: Resuming dependent run without checkpoint", { + batchRunId: batchRun.id, + dependentTaskAttemptId: batchRun.dependentTaskAttempt.id, + }); + await marqs?.replaceMessage(dependentRun.id, { + type: "RESUME", + completedAttemptIds: batchRun.items.map((item) => item.taskRunAttemptId).filter(Boolean), + resumableAttemptId: batchRun.dependentTaskAttempt.id, + checkpointEventId: batchRun.checkpointEventId ?? undefined, + taskIdentifier: batchRun.dependentTaskAttempt.taskRun.taskIdentifier, + projectId: batchRun.dependentTaskAttempt.runtimeEnvironment.projectId, + environmentId: batchRun.dependentTaskAttempt.runtimeEnvironment.id, + environmentType: batchRun.dependentTaskAttempt.runtimeEnvironment.type, + }); + } else { + logger.debug("ResumeBatchRunService: without checkpoint was already completed", { + batchRunId: batchRun.id, + dependentTaskAttempt: batchRun.dependentTaskAttempt, + checkpointEventId: batchRun.checkpointEventId, + hasCheckpointEvent: !!batchRun.checkpointEventId, + }); + } + } + } + + async #setBatchToCompletedOnce(batchRunId: string) { + const result = await this._prisma.batchTaskRun.updateMany({ + where: { + id: batchRunId, + status: { + not: "COMPLETED", // Ensure the status is not already "COMPLETED" + }, + }, + data: { + status: "COMPLETED", + }, + }); + + // Check if any records were updated + if (result.count > 0) { + // The status was changed, so we return true + return true; + } else { + return false; } } diff --git a/apps/webapp/app/v3/services/resumeDependentParents.server.ts b/apps/webapp/app/v3/services/resumeDependentParents.server.ts new file mode 100644 index 000000000..97a58a7bf --- /dev/null +++ b/apps/webapp/app/v3/services/resumeDependentParents.server.ts @@ -0,0 +1,257 @@ +import { Prisma } from "@trigger.dev/database"; +import { logger } from "~/services/logger.server"; +import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus"; +import { BaseService } from "./baseService.server"; +import { ResumeBatchRunService } from "./resumeBatchRun.server"; +import { ResumeTaskDependencyService } from "./resumeTaskDependency.server"; +import { $transaction } from "~/db.server"; + +type Output = + | { + success: true; + action: + | "resume-scheduled" + | "batch-resume-scheduled" + | "no-dependencies" + | "not-finished" + | "dev"; + } + | { + success: false; + error: string; + }; + +type Dependency = Prisma.TaskRunDependencyGetPayload<{ + include: { + taskRun: true; + dependentAttempt: true; + dependentBatchRun: true; + }; +}>; + +/** This will resume a dependent (parent) run if there is one and it makes sense. */ +export class ResumeDependentParentsService extends BaseService { + public async call({ id }: { id: string }): Promise { + try { + const dependency = await this._prisma.taskRunDependency.findFirst({ + include: { + taskRun: { + include: { + runtimeEnvironment: true, + }, + }, + dependentAttempt: true, + dependentBatchRun: true, + }, + where: { + taskRunId: id, + }, + }); + + logger.log("ResumeDependentParentsService: tried to find dependency", { + runId: id, + dependency: dependency, + }); + + if (!dependency) { + logger.log("ResumeDependentParentsService: dependency not found", { + runId: id, + }); + + //no dependency, that's fine most runs won't have one. + return { + success: true, + action: "no-dependencies", + }; + } + + if (!isFinalRunStatus(dependency.taskRun.status)) { + logger.debug( + "ResumeDependentParentsService: run not finished yet, can't resume parent yet", + { + runId: id, + dependency, + } + ); + + // the child run isn't finished yet, so we can't resume the parent yet. + return { + success: true, + action: "not-finished", + }; + } + + if (dependency.taskRun.runtimeEnvironment.type === "DEVELOPMENT") { + logger.debug("ResumeDependentParentsService: runs are resumed on device for DEV runs.", { + runId: id, + dependency, + }); + + return { + success: true, + action: "dev", + }; + } + + if (dependency.dependentAttempt) { + return this.#singleRunDependency(dependency); + } else if (dependency.dependentBatchRun) { + return this.#batchRunDependency(dependency); + } else { + logger.error("ResumeDependentParentsService: dependency has no dependencies", { + runId: id, + dependency, + }); + + return { + success: false, + error: `Dependency has no dependencies (single or batch)`, + }; + } + } catch (error) { + return { + success: false, + error: error instanceof Error ? error.message : JSON.stringify(error), + }; + } + } + + async #singleRunDependency(dependency: Dependency): Promise { + logger.debug( + `ResumeDependentParentsService.singleRunDependency(): Resuming dependent parent for run`, + { + dependency, + } + ); + + const lastAttempt = await this._prisma.taskRunAttempt.findFirst({ + select: { + id: true, + status: true, + }, + where: { + taskRunId: dependency.taskRunId, + }, + orderBy: { + id: "desc", + }, + }); + + if (!lastAttempt) { + logger.error( + "ResumeDependentParentsService.singleRunDependency(): dependency child attempt not found", + { + dependency, + } + ); + + return { + success: false, + error: `Dependency child attempt not found for run ${dependency.taskRunId}`, + }; + } + + if (!isFinalAttemptStatus(lastAttempt.status)) { + //We still want to continue if this happens because the run is final but log it + logger.error( + "ResumeDependentParentsService.singleRunDependency(): dependency child attempt not final, but the run is.", + { + dependency, + lastAttempt, + } + ); + + return { + success: false, + error: `Dependency child attempt not final, but the run is`, + }; + } + + //resume the dependent task + await ResumeTaskDependencyService.enqueue(dependency.id, lastAttempt.id, this._prisma); + return { + success: true, + action: "resume-scheduled", + }; + } + + async #batchRunDependency(dependency: Dependency): Promise { + logger.debug( + `ResumeDependentParentsService.batchRunDependency(): Resuming dependent batch for run`, + { + dependency, + } + ); + + if (!dependency.dependentBatchRun) { + logger.error( + "ResumeDependentParentsService.batchRunDependency(): dependency has no dependent batch", + { + dependency, + } + ); + + return { + success: false, + error: `Dependency has no dependent batch`, + }; + } + + const lastAttempt = await this._prisma.taskRunAttempt.findFirst({ + select: { + id: true, + status: true, + }, + where: { + taskRunId: dependency.taskRunId, + }, + orderBy: { + id: "desc", + }, + }); + + if (!lastAttempt) { + logger.error( + "ResumeDependentParentsService.singleRunDependency(): dependency child attempt not found", + { + dependency, + } + ); + + return { + success: false, + error: `Dependency child attempt not found for run ${dependency.taskRunId}`, + }; + } + + logger.log( + "ResumeDependentParentsService.batchRunDependency(): Setting the batchTaskRunItem to COMPLETED", + { + dependency, + lastAttempt, + } + ); + + await $transaction(this._prisma, async (tx) => { + await tx.batchTaskRunItem.update({ + where: { + batchTaskRunId_taskRunId: { + batchTaskRunId: dependency.dependentBatchRun!.id, + taskRunId: dependency.taskRunId, + }, + }, + data: { + status: "COMPLETED", + taskRunAttemptId: lastAttempt.id, + }, + }); + + await ResumeBatchRunService.enqueue(dependency.dependentBatchRun!.id, tx); + }); + + return { + success: true, + action: "batch-resume-scheduled", + }; + } +} diff --git a/apps/webapp/app/v3/taskStatus.ts b/apps/webapp/app/v3/taskStatus.ts index a464d7aa4..2ac65fbbf 100644 --- a/apps/webapp/app/v3/taskStatus.ts +++ b/apps/webapp/app/v3/taskStatus.ts @@ -45,7 +45,12 @@ export const FINAL_RUN_STATUSES = [ export type FINAL_RUN_STATUSES = (typeof FINAL_RUN_STATUSES)[number]; -export const FINAL_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["CANCELED", "COMPLETED", "FAILED"]; +export const FINAL_ATTEMPT_STATUSES = [ + "CANCELED", + "COMPLETED", + "FAILED", +] satisfies TaskRunAttemptStatus[]; +export type FINAL_ATTEMPT_STATUSES = (typeof FINAL_ATTEMPT_STATUSES)[number]; export const FREEZABLE_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "RETRYING_AFTER_FAILURE"]; export const FREEZABLE_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["EXECUTING", "FAILED"]; diff --git a/packages/cli-v3/src/executions/taskRunProcess.ts b/packages/cli-v3/src/executions/taskRunProcess.ts index 1ba453257..35bbda4b8 100644 --- a/packages/cli-v3/src/executions/taskRunProcess.ts +++ b/packages/cli-v3/src/executions/taskRunProcess.ts @@ -399,7 +399,7 @@ class FlushingProcess { private _flushPromise: Promise; constructor(private readonly doFlush: () => Promise) { - this._flushPromise = this.doFlush(); + this._flushPromise = this.doFlush().catch(() => {}); } waitForCompletion() { diff --git a/packages/database/prisma/migrations/20240909141925_task_run_attempt_index_on_task_run_id/migration.sql b/packages/database/prisma/migrations/20240909141925_task_run_attempt_index_on_task_run_id/migration.sql new file mode 100644 index 000000000..c2ab37a94 --- /dev/null +++ b/packages/database/prisma/migrations/20240909141925_task_run_attempt_index_on_task_run_id/migration.sql @@ -0,0 +1,2 @@ +-- CreateIndex +CREATE INDEX CONCURRENTLY IF NOT EXISTS "TaskRunAttempt_taskRunId_idx" ON "TaskRunAttempt" ("taskRunId"); \ No newline at end of file diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index ef7c7c25c..8740d22db 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -1796,9 +1796,11 @@ model TaskRunTag { @@index([name, id]) } +/// This is used for triggerAndWait and batchTriggerAndWait. The taskRun is the child task, it points at a parent attempt or a batch model TaskRunDependency { id String @id @default(cuid()) + /// The child run taskRun TaskRun @relation(fields: [taskRunId], references: [id], onDelete: Cascade, onUpdate: Cascade) taskRunId String @unique @@ -1880,6 +1882,7 @@ model TaskRunAttempt { alerts ProjectAlert[] @@unique([taskRunId, number]) + @@index([taskRunId]) } enum TaskRunAttemptStatus { diff --git a/references/v3-catalog/src/trigger/batch.ts b/references/v3-catalog/src/trigger/batch.ts index e4a4490cf..5e25a2eeb 100644 --- a/references/v3-catalog/src/trigger/batch.ts +++ b/references/v3-catalog/src/trigger/batch.ts @@ -26,8 +26,35 @@ export const batchParentTask = task({ }, }); +export const batchParentWitFailsTask = task({ + id: "batch-parent-with-fails-task", + retry: { + maxAttempts: 1, + }, + run: async () => { + const response = await taskThatFails.batchTriggerAndWait([ + { payload: false }, + { payload: true }, + { payload: false }, + ]); + + logger.info("Batch response", { response }); + + const respone2 = await taskThatFails.batchTriggerAndWait([ + { payload: true }, + { payload: false }, + { payload: true }, + ]); + + logger.info("Batch response2", { respone2 }); + }, +}); + export const batchChildTask = task({ id: "batch-child-task", + retry: { + maxAttempts: 2, + }, run: async (payload: string, { ctx }) => { logger.info("Processing child task", { payload }); @@ -36,3 +63,21 @@ export const batchChildTask = task({ return `${payload} - processed`; }, }); + +export const taskThatFails = task({ + id: "task-that-fails", + retry: { + maxAttempts: 2, + }, + run: async (fail: boolean) => { + logger.info(`Will fail ${fail}`); + + if (fail) { + throw new Error("Task failed"); + } + + return { + foo: "bar", + }; + }, +}); diff --git a/references/v3-catalog/src/trigger/crash.ts b/references/v3-catalog/src/trigger/crash.ts new file mode 100644 index 000000000..8a0f9ead3 --- /dev/null +++ b/references/v3-catalog/src/trigger/crash.ts @@ -0,0 +1,35 @@ +import { logger, task } from "@trigger.dev/sdk/v3"; + +type Payload = {}; + +export const crashparent = task({ + id: "crashparent", + run: async (payload: Payload, { ctx }) => { + logger.log("crashparent started"); + + const result = await crash.triggerAndWait({}); + logger.log("crashparent done", { result }); + + const results = await crash.batchTriggerAndWait([ + { payload: {} }, + { payload: {} }, + { payload: {} }, + { payload: {} }, + { payload: {} }, + ]); + logger.log("crashparent batch done", { results }); + }, +}); + +export const crash = task({ + id: "crash", + run: async (payload: Payload, { ctx }) => { + logger.log(`${ctx.run.version}`); + + process.exit(1); + + return { + foo: "bar", + }; + }, +}); diff --git a/references/v3-catalog/src/trigger/scheduled.ts b/references/v3-catalog/src/trigger/scheduled.ts index 8081e8757..4bf2bd280 100644 --- a/references/v3-catalog/src/trigger/scheduled.ts +++ b/references/v3-catalog/src/trigger/scheduled.ts @@ -3,7 +3,7 @@ import { logger, schedules, task } from "@trigger.dev/sdk/v3"; export const firstScheduledTask = schedules.task({ id: "first-scheduled-task", //every other minute - cron: "0 */2 * * *", + // cron: "0 */2 * * *", run: async (payload, { ctx }) => { const distanceInMs = payload.timestamp.getTime() - (payload.lastTimestamp ?? new Date()).getTime(); @@ -22,10 +22,10 @@ export const firstScheduledTask = schedules.task({ export const secondScheduledTask = schedules.task({ id: "second-scheduled-task", - cron: { - pattern: "0 5 * * *", - timezone: "Asia/Tokyo", - }, + // cron: { + // pattern: "0 5 * * *", + // timezone: "Asia/Tokyo", + // }, run: async (payload) => {}, });