github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Frozen run fixes (#1286)

Browse Source 
* When resuming a batch, only do marqs operations once

* Made TaskRunDependency clearer in the Prisma schema

* New ResumeDependentParentsService service, use it from checkpoints

* WIP on making resuming more robust

* Turn the declarative schedules off because they make debugging other runs painful

* Resuming batches when there’s an attempt is working

* If there’s no attempt then create one

* Added a log if there are no span events to complete

* If Graphile addJob doesn’t return a row, log and return undefined. No throw

* Pass prisma into the ResumeDependentParentsService

* Removed the todos

* Pass Prisma through to the checkpoint service

* Fix for not checking the batch item correctly

* Fix for when a log flush times out and the process is checkpointed

* Fix for when a log flush times out and the process is checkpointed

* Another test run that does batches with failed subtasks

* Don’t call ResumeTaskRunDependenciesService anymore (we have a new service)

* Only resume if the run is in a final state

* If an attempt doesn’t exist, fix for creating queue with sanitized name

* If DEV then don’t resume using marqs/batches. The CLI manages it

* We don’t need to check the run status again, it’s in the main function now

* Added TaskRunAttempt taskRunId index

* Only allow calling ResumeDependentParentsService with a run ID

* Put the flushing back to what it was
This commit is contained in:
Matt Aitken
2024-09-09 16:03:36 +01:00
committed by GitHub
parent e5f0aafb14
commit 392453e8a6
22 changed files with 624 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/tidy-pets-smell.md
+6
View File
@@ -0,0 +1,6 @@
---
"trigger.dev": patch
"@trigger.dev/core": patch
---
Fix for when a log flush times out and the process is checkpointed
apps/webapp/app/platform/zodWorker.server.ts
+3 -5
View File
@@ -318,7 +318,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
identifier: K,
payload: z.infer<TMessageCatalog[K]>,
options?: ZodWorkerEnqueueOptions
): Promise<GraphileJob> {
): Promise<GraphileJob | undefined> {
const task = this.#tasks[identifier];
const optionsWithoutTx = removeUndefinedKeys(omit(options ?? {}, ["tx"]));
@@ -439,11 +439,9 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
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];
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
+6
View File
@@ -796,6 +796,12 @@ function RunTimelineLine({ title, state }: RunTimelineLineProps) {
function RunError({ error }: { error: TaskRunError }) {
switch (error.type) {
case "STRING_ERROR":
return (
<div className="flex flex-col gap-2 rounded-sm border border-rose-500/50 px-3 pb-3 pt-2">
<Header3 className="text-rose-500">Error</Header3>
<Callout variant="error">{error.raw}</Callout>
</div>
);
case "CUSTOM_ERROR": {
return (
<div className="flex flex-col gap-2 rounded-sm border border-rose-500/50 px-3 pb-3 pt-2">
apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts
+1 -1
View File
@@ -134,7 +134,7 @@ export class DeliverScheduledEventService {
id,
},
data: {
workerJobId: workerJob.id,
workerJobId: workerJob?.id,
nextEventTimestamp: runAt,
},
});
apps/webapp/app/v3/eventRepository.server.ts
+1
View File
@@ -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;
}
apps/webapp/app/v3/failedTaskRun.server.ts
+2 -18
View File
@@ -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(),
apps/webapp/app/v3/services/cancelAttempt.server.ts
+6 -9
View File
@@ -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);
}
});
}
}
apps/webapp/app/v3/services/cancelTaskRun.server.ts
+5
View File
@@ -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({
apps/webapp/app/v3/services/completeAttempt.server.ts
+6 -9
View File
@@ -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";
}
}
apps/webapp/app/v3/services/crashTaskRun.server.ts
+7 -7
View File
@@ -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);
});
}
}
apps/webapp/app/v3/services/createCheckpoint.server.ts
+8 -125
View File
@@ -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,
apps/webapp/app/v3/services/expireEnqueuedRun.server.ts
+5
View File
@@ -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, {
apps/webapp/app/v3/services/finalizeTaskRun.server.ts
+109 -4
View File
@@ -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<T extends Prisma.TaskRunInclude> = BaseInput & {
@@ -32,6 +40,8 @@ export class FinalizeTaskRunService extends BaseService {
expiredAt,
completedAt,
include,
attemptStatus,
error,
}: T extends Prisma.TaskRunInclude ? InputWithInclude<T> : InputWithoutInclude): Promise<
Output<T>
> {
@@ -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<T>;
}
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,
},
});
}
}
}
}
apps/webapp/app/v3/services/resumeBatchRun.server.ts
+105 -46
View File
@@ -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;
}
}
apps/webapp/app/v3/services/resumeDependentParents.server.ts
+257
View File
@@ -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<Output> {
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<Output> {
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<Output> {
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",
};
}
}
apps/webapp/app/v3/taskStatus.ts
+6 -1
View File
@@ -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"];
packages/cli-v3/src/executions/taskRunProcess.ts
+1 -1
View File
@@ -399,7 +399,7 @@ class FlushingProcess {
private _flushPromise: Promise<void>;
constructor(private readonly doFlush: () => Promise<void>) {
this._flushPromise = this.doFlush();
this._flushPromise = this.doFlush().catch(() => {});
}
waitForCompletion() {
packages/database/prisma/migrations/20240909141925_task_run_attempt_index_on_task_run_id/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- CreateIndex
CREATE INDEX CONCURRENTLY IF NOT EXISTS "TaskRunAttempt_taskRunId_idx" ON "TaskRunAttempt" ("taskRunId");
packages/database/prisma/schema.prisma
+3
View File
@@ -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 {
references/v3-catalog/src/trigger/batch.ts
+45
View File
@@ -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",
};
},
});
references/v3-catalog/src/trigger/crash.ts
+35
View File
@@ -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",
};
},
});
references/v3-catalog/src/trigger/scheduled.ts
+5 -5
View File
@@ -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) => {},
});