Files
Eric Allam 26093896d2 v3: Fixes for using (batch)triggerAndWait with idempotency keys (#1043)
* Fixes various issues with triggerAndWait and batchTriggerAndWait

When using idempotency keys, triggerAndWait and batchTriggerAndWait will still work even if the existing runs have already been completed (or even partially completed, in the case of batchTriggerAndWait)

- TaskRunExecutionResult.id is now the run friendlyId, not the attempt friendlyId
- A single TaskRun can now have many batchItems, in the case of batchTriggerAndWait while using idempotency keys
- A run’s idempotencyKey is now added to the ctx as well as the TaskEvent and displayed in the span view
- When resolving batchTriggerAndWait, the runtimes no longer reject promises, leading to an error in the parent task

* Remove the default queue concurrency limit as we now have env and org concurrency limits

* Use the run friendlyId in the completion result id

* Added some error logging
2024-04-19 10:54:43 +01:00

129 lines
3.6 KiB
TypeScript

import {
TaskRunError,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
TaskRunSuccessfulExecutionResult,
} from "@trigger.dev/core/v3";
import {
BatchTaskRunItemStatus,
TaskRun,
TaskRunAttempt,
TaskRunAttemptStatus,
TaskRunStatus,
} from "@trigger.dev/database";
import { assertNever } from "assert-never";
import { logger } from "~/services/logger.server";
const SUCCESSFUL_STATUSES = [TaskRunStatus.COMPLETED_SUCCESSFULLY];
const FAILURE_STATUSES = [
TaskRunStatus.CANCELED,
TaskRunStatus.INTERRUPTED,
TaskRunStatus.COMPLETED_WITH_ERRORS,
TaskRunStatus.SYSTEM_FAILURE,
TaskRunStatus.CRASHED,
];
export type TaskRunWithAttempts = TaskRun & {
attempts: TaskRunAttempt[];
};
export function executionResultForTaskRun(
taskRun: TaskRunWithAttempts
): TaskRunExecutionResult | undefined {
if (SUCCESSFUL_STATUSES.includes(taskRun.status)) {
// find the last attempt that was successful
const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.COMPLETED);
if (!attempt) {
logger.error("Task run is successful but no successful attempt found", {
taskRunId: taskRun.id,
taskRunStatus: taskRun.status,
taskRunAttempts: taskRun.attempts.map((a) => a.status),
});
return undefined;
}
return {
ok: true,
id: taskRun.friendlyId,
output: attempt.output ?? undefined,
outputType: attempt.outputType,
} satisfies TaskRunSuccessfulExecutionResult;
}
if (FAILURE_STATUSES.includes(taskRun.status)) {
if (taskRun.status === TaskRunStatus.CANCELED) {
return {
ok: false,
id: taskRun.friendlyId,
error: {
type: "INTERNAL_ERROR",
code: "TASK_RUN_CANCELLED",
},
} satisfies TaskRunFailedExecutionResult;
}
const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.FAILED);
if (!attempt) {
logger.error("Task run is failed but no failed attempt found", {
taskRunId: taskRun.id,
taskRunStatus: taskRun.status,
taskRunAttempts: taskRun.attempts.map((a) => a.status),
});
return undefined;
}
const error = TaskRunError.safeParse(attempt.error);
if (!error.success) {
logger.error("Failed to parse error from failed task run attempt", {
taskRunId: taskRun.id,
taskRunStatus: taskRun.status,
taskRunAttempts: taskRun.attempts.map((a) => a.status),
error: attempt.error,
});
return {
ok: false,
id: taskRun.friendlyId,
error: {
type: "INTERNAL_ERROR",
code: "CONFIGURED_INCORRECTLY",
},
} satisfies TaskRunFailedExecutionResult;
}
return {
ok: false,
id: taskRun.friendlyId,
error: error.data,
} satisfies TaskRunFailedExecutionResult;
}
}
export function batchTaskRunItemStatusForRunStatus(status: TaskRunStatus): BatchTaskRunItemStatus {
switch (status) {
case TaskRunStatus.COMPLETED_SUCCESSFULLY:
return BatchTaskRunItemStatus.COMPLETED;
case TaskRunStatus.CANCELED:
case TaskRunStatus.INTERRUPTED:
case TaskRunStatus.COMPLETED_WITH_ERRORS:
case TaskRunStatus.SYSTEM_FAILURE:
case TaskRunStatus.CRASHED:
case TaskRunStatus.COMPLETED_WITH_ERRORS:
return BatchTaskRunItemStatus.FAILED;
case TaskRunStatus.PENDING:
case TaskRunStatus.WAITING_FOR_DEPLOY:
case TaskRunStatus.WAITING_TO_RESUME:
case TaskRunStatus.RETRYING_AFTER_FAILURE:
case TaskRunStatus.EXECUTING:
case TaskRunStatus.PAUSED:
return BatchTaskRunItemStatus.PENDING;
default:
assertNever(status);
}
}