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

Handling failing task runs that cannot create an attempt for whatever reason

Browse Source
This commit is contained in:
Eric Allam
2024-04-30 15:20:09 +01:00
parent 5ed700dad8
commit d1bdd0cf5d
16 changed files with 324 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/v3/authenticatedSocketConnection.server.ts
+8
View File
@@ -72,6 +72,14 @@ export class AuthenticatedSocketConnection {
);
break;
}
case "TASK_RUN_FAILED_TO_RUN": {
await this._consumer.taskRunFailed(
payload.backgroundWorkerId,
payload.data.completion
);
break;
}
case "TASK_HEARTBEAT": {
await this._consumer.taskHeartbeat(payload.backgroundWorkerId, payload.data.id);
break;
apps/webapp/app/v3/eventRepository.server.ts
+2 -1
View File
@@ -147,6 +147,7 @@ export type UpdateEventOptions = {
attributes: TraceAttributes;
endTime?: Date;
immediate?: boolean;
events?: SpanEvents;
};
export class EventRepository {
@@ -217,7 +218,7 @@ export class EventRepository {
isCancelled: false,
status: options?.attributes.isError ? "ERROR" : "OK",
links: event.links ?? [],
events: event.events ?? [],
events: event.events ?? (options?.events as any) ?? [],
duration: calculateDurationFromStart(event.startTime, options?.endTime),
properties: event.properties as Attributes,
metadata: event.metadata as Attributes,
apps/webapp/app/v3/failedTaskRun.server.ts
+109
View File
@@ -0,0 +1,109 @@
import {
ExceptionEventProperties,
TaskRunError,
TaskRunFailedExecutionResult,
} from "@trigger.dev/core/v3";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { marqs } from "~/v3/marqs/index.server";
import { eventRepository } from "./eventRepository.server";
import { BaseService } from "./services/baseService.server";
import { TaskRunStatus } from "@trigger.dev/database";
const FAILABLE_TASK_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "PENDING", "WAITING_FOR_DEPLOY"];
export class FailedTaskRunService extends BaseService {
public async call({
runFriendlyId,
completion,
env,
}: {
runFriendlyId: string;
completion: TaskRunFailedExecutionResult;
env: AuthenticatedEnvironment;
}) {
const taskRun = await this._prisma.taskRun.findUnique({
where: { friendlyId: runFriendlyId },
});
if (!taskRun) {
logger.error("[FailedTaskRunService] Task run not found", {
runFriendlyId,
completion,
});
return;
}
if (!FAILABLE_TASK_RUN_STATUSES.includes(taskRun.status)) {
logger.error("[FailedTaskRunService] Task run is not in a failable state", {
taskRun,
completion,
});
return;
}
// No more retries, we need to fail the task run
logger.debug("[FailedTaskRunService] Failing task run", { taskRun, completion });
await marqs?.acknowledgeMessage(taskRun.id);
// Now we need to "complete" the task run event/span
await eventRepository.completeEvent(taskRun.spanId, {
endTime: new Date(),
attributes: {
isError: true,
},
events: [
{
name: "exception",
time: new Date(),
properties: {
exception: createExceptionPropertiesFromError(completion.error),
},
},
],
});
await this._prisma.taskRun.update({
where: {
id: taskRun.id,
},
data: {
status: "SYSTEM_FAILURE",
},
});
}
}
function createExceptionPropertiesFromError(error: TaskRunError): ExceptionEventProperties {
switch (error.type) {
case "BUILT_IN_ERROR": {
return {
type: error.name,
message: error.message,
stacktrace: error.stackTrace,
};
}
case "CUSTOM_ERROR": {
return {
type: "Error",
message: error.raw,
};
}
case "INTERNAL_ERROR": {
return {
type: "Internal error",
message: [error.code, error.message].filter(Boolean).join(": "),
};
}
case "STRING_ERROR": {
return {
type: "Error",
message: error.raw,
};
}
}
}
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
+18
View File
@@ -4,6 +4,7 @@ import {
TaskRunExecutionLazyAttemptPayload,
TaskRunExecutionPayload,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
serverWebsocketMessages,
} from "@trigger.dev/core/v3";
import { ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
@@ -24,6 +25,7 @@ import {
tracer,
} from "../tracer.server";
import { DevSubscriber, devPubSub } from "./devPubSub.server";
import { FailedTaskRunService } from "../failedTaskRun.server";
const MessageBody = z.discriminatedUnion("type", [
z.object({
@@ -143,6 +145,22 @@ export class DevQueueConsumer {
}
}
public async taskRunFailed(workerId: string, completion: TaskRunFailedExecutionResult) {
this._taskFailures++;
logger.debug("[DevQueueConsumer] taskRunFailed()", { completion });
this._inProgressRuns.delete(completion.id);
const service = new FailedTaskRunService();
await service.call({
runFriendlyId: completion.id,
completion,
env: this.env,
});
}
/**
* @deprecated Use `taskRunHeartbeat` instead
*/
packages/cli-v3/src/apiClient.ts
+52 -60
View File
@@ -15,7 +15,9 @@ import {
GetProjectsResponseBody,
GetProjectResponseBody,
TaskRunExecution,
APIError,
} from "@trigger.dev/core/v3";
import { zodfetch } from "@trigger.dev/core/v3/zodfetch";
export class CliApiClient {
private readonly apiURL: string;
@@ -28,7 +30,7 @@ export class CliApiClient {
}
async createAuthorizationCode() {
return zodfetch(
return wrapZodFetch(
CreateAuthorizationCodeResponseSchema,
`${this.apiURL}/api/v1/authorization-code`,
{
@@ -38,7 +40,7 @@ export class CliApiClient {
}
async getPersonalAccessToken(authorizationCode: string) {
return zodfetch(GetPersonalAccessTokenResponseSchema, `${this.apiURL}/api/v1/token`, {
return wrapZodFetch(GetPersonalAccessTokenResponseSchema, `${this.apiURL}/api/v1/token`, {
method: "POST",
body: JSON.stringify({
authorizationCode,
@@ -51,7 +53,7 @@ export class CliApiClient {
throw new Error("whoAmI: No access token");
}
return zodfetch(WhoAmIResponseSchema, `${this.apiURL}/api/v2/whoami`, {
return wrapZodFetch(WhoAmIResponseSchema, `${this.apiURL}/api/v2/whoami`, {
headers: {
Authorization: `Bearer ${this.accessToken}`,
"Content-Type": "application/json",
@@ -64,7 +66,7 @@ export class CliApiClient {
throw new Error("getProject: No access token");
}
return zodfetch(GetProjectResponseBody, `${this.apiURL}/api/v1/projects/${projectRef}`, {
return wrapZodFetch(GetProjectResponseBody, `${this.apiURL}/api/v1/projects/${projectRef}`, {
headers: {
Authorization: `Bearer ${this.accessToken}`,
"Content-Type": "application/json",
@@ -77,7 +79,7 @@ export class CliApiClient {
throw new Error("getProjects: No access token");
}
return zodfetch(GetProjectsResponseBody, `${this.apiURL}/api/v1/projects`, {
return wrapZodFetch(GetProjectsResponseBody, `${this.apiURL}/api/v1/projects`, {
headers: {
Authorization: `Bearer ${this.accessToken}`,
"Content-Type": "application/json",
@@ -90,7 +92,7 @@ export class CliApiClient {
throw new Error("createBackgroundWorker: No access token");
}
return zodfetch(
return wrapZodFetch(
CreateBackgroundWorkerResponse,
`${this.apiURL}/api/v1/projects/${projectRef}/background-workers`,
{
@@ -109,7 +111,7 @@ export class CliApiClient {
throw new Error("creatTaskRunAttempt: No access token");
}
return zodfetch(TaskRunExecution, `${this.apiURL}/api/v1/runs/${runFriendlyId}/attempts`, {
return wrapZodFetch(TaskRunExecution, `${this.apiURL}/api/v1/runs/${runFriendlyId}/attempts`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.accessToken}`,
@@ -129,12 +131,16 @@ export class CliApiClient {
throw new Error("getProjectDevEnv: No access token");
}
return zodfetch(GetProjectEnvResponse, `${this.apiURL}/api/v1/projects/${projectRef}/${env}`, {
headers: {
Authorization: `Bearer ${this.accessToken}`,
"Content-Type": "application/json",
},
});
return wrapZodFetch(
GetProjectEnvResponse,
`${this.apiURL}/api/v1/projects/${projectRef}/${env}`,
{
headers: {
Authorization: `Bearer ${this.accessToken}`,
"Content-Type": "application/json",
},
}
);
}
async getEnvironmentVariables(projectRef: string) {
@@ -142,7 +148,7 @@ export class CliApiClient {
throw new Error("getEnvironmentVariables: No access token");
}
return zodfetch(
return wrapZodFetch(
GetEnvironmentVariablesResponseBody,
`${this.apiURL}/api/v1/projects/${projectRef}/envvars`,
{
@@ -159,7 +165,7 @@ export class CliApiClient {
throw new Error("initializeDeployment: No access token");
}
return zodfetch(InitializeDeploymentResponseBody, `${this.apiURL}/api/v1/deployments`, {
return wrapZodFetch(InitializeDeploymentResponseBody, `${this.apiURL}/api/v1/deployments`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.accessToken}`,
@@ -174,7 +180,7 @@ export class CliApiClient {
throw new Error("startDeploymentIndexing: No access token");
}
return zodfetch(
return wrapZodFetch(
StartDeploymentIndexingResponseBody,
`${this.apiURL}/api/v1/deployments/${deploymentId}/start-indexing`,
{
@@ -193,7 +199,7 @@ export class CliApiClient {
throw new Error("getDeployment: No access token");
}
return zodfetch(
return wrapZodFetch(
GetDeploymentResponseBody,
`${this.apiURL}/api/v1/deployments/${deploymentId}`,
{
@@ -213,56 +219,42 @@ type ApiResult<TSuccessResult> =
error: string;
};
async function zodfetch<T extends z.ZodTypeAny>(
async function wrapZodFetch<T extends z.ZodTypeAny>(
schema: T,
url: string,
requestInit?: RequestInit
): Promise<ApiResult<z.infer<T>>> {
try {
const response = await fetch(url, requestInit);
const response = await zodfetch(schema, url, requestInit, {
retry: {
minTimeoutInMs: 500,
maxTimeoutInMs: 5000,
maxAttempts: 3,
factor: 2,
randomize: false,
},
});
if ((!requestInit || requestInit.method === "GET") && response.status === 404) {
return {
success: false,
error: `404: ${response.statusText}`,
};
}
if (response.status >= 400 && response.status < 500) {
const body = await response.json();
if (!body.error) {
return { success: false, error: "Something went wrong" };
}
return { success: false, error: body.error };
}
if (response.status !== 200) {
return {
success: false,
error: `Failed to fetch ${url}, got status code ${response.status}`,
};
}
const jsonBody = await response.json();
const parsedResult = schema.safeParse(jsonBody);
if (parsedResult.success) {
return { success: true, data: parsedResult.data };
}
if ("error" in jsonBody) {
return {
success: false,
error: typeof jsonBody.error === "string" ? jsonBody.error : JSON.stringify(jsonBody.error),
};
}
return { success: false, error: parsedResult.error.message };
} catch (error) {
return {
success: false,
error: error instanceof Error ? error.message : JSON.stringify(error),
success: true,
data: response,
};
} catch (error) {
if (error instanceof APIError) {
return {
success: false,
error: error.message,
};
} else if (error instanceof Error) {
return {
success: false,
error: error.message,
};
} else {
return {
success: false,
error: String(error),
};
}
}
}
packages/cli-v3/src/commands/deploy.ts
+3 -1
View File
@@ -187,7 +187,9 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
`Failed to connect to ${authorization.auth?.apiUrl}. Are you sure it's the correct URL?`
);
} else {
throw new Error("You must login first. Use `trigger.dev login` to login.");
throw new Error(
`You must login first. Use the \`login\` CLI command.\n\n${authorization.error}`
);
}
}
packages/cli-v3/src/commands/dev.tsx
+17 -1
View File
@@ -110,7 +110,11 @@ export async function devCommand(dir: string, options: DevCommandOptions) {
)} Connecting to the server failed. Please check your internet connection or contact eric@trigger.dev for help.`
);
} else {
logger.log(`${chalkError("X Error:")} You must login first. Use the \`login\` CLI command.`);
logger.log(
`${chalkError("X Error:")} You must login first. Use the \`login\` CLI command.\n\n${
authorization.error
}`
);
}
process.exitCode = 1;
return;
@@ -317,6 +321,18 @@ function useDev({
}
);
backgroundWorkerCoordinator.onTaskFailedToRun.attach(
async ({ backgroundWorkerId, completion }) => {
await sender.send("BACKGROUND_WORKER_MESSAGE", {
backgroundWorkerId,
data: {
type: "TASK_RUN_FAILED_TO_RUN",
completion,
},
});
}
);
backgroundWorkerCoordinator.onWorkerRegistered.attach(async ({ id, worker, record }) => {
await sender.send("READY_FOR_TASKS", {
backgroundWorkerId: id,
packages/cli-v3/src/commands/whoami.ts
+1 -1
View File
@@ -78,7 +78,7 @@ export async function whoAmI(
options?.profile ?? "default"
}\` to login.`
);
outro("Whoami failed");
outro(`Whoami failed: ${authentication.error}`);
}
}
packages/cli-v3/src/workers/dev/backgroundWorker.ts
+56 -23
View File
@@ -1,4 +1,5 @@
import {
APIError,
BackgroundWorkerProperties,
BackgroundWorkerServerMessages,
CreateBackgroundWorkerResponse,
@@ -12,6 +13,7 @@ import {
TaskRunExecutionLazyAttemptPayload,
TaskRunExecutionPayload,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
childToWorkerMessages,
correctErrorStackTrace,
formatDurationMilliseconds,
@@ -48,6 +50,11 @@ export class BackgroundWorkerCoordinator {
worker: BackgroundWorker;
execution: TaskRunExecution;
}> = new Evt();
public onTaskFailedToRun: Evt<{
backgroundWorkerId: string;
worker: BackgroundWorker;
completion: TaskRunFailedExecutionResult;
}> = new Evt();
public onWorkerRegistered: Evt<{
worker: BackgroundWorker;
id: string;
@@ -73,21 +80,22 @@ export class BackgroundWorkerCoordinator {
private _deprecatedWorkers: Set<string> = new Set();
constructor(private baseURL: string) {
this.onTaskCompleted.attach(async ({ completion, execution }) => {
this.onTaskCompleted.attach(async ({ completion }) => {
if (!completion.ok && typeof completion.retry !== "undefined") {
return;
}
await this.#notifyWorkersOfTaskCompletion(completion, execution);
await this.#notifyWorkersOfTaskCompletion(completion);
});
this.onTaskFailedToRun.attach(async ({ completion }) => {
await this.#notifyWorkersOfTaskCompletion(completion);
});
}
async #notifyWorkersOfTaskCompletion(
completion: TaskRunExecutionResult,
execution: TaskRunExecution
) {
async #notifyWorkersOfTaskCompletion(completion: TaskRunExecutionResult) {
for (const worker of this._backgroundWorkers.values()) {
await worker.taskRunCompletedNotification(completion, execution);
await worker.taskRunCompletedNotification(completion);
}
}
@@ -173,14 +181,43 @@ export class BackgroundWorkerCoordinator {
return;
}
const { completion, execution } = await worker.executeTaskRunLazyAttempt(payload, this.baseURL);
try {
const { completion, execution } = await worker.executeTaskRunLazyAttempt(
payload,
this.baseURL
);
this.onTaskCompleted.post({
completion,
execution,
worker,
backgroundWorkerId: id,
});
this.onTaskCompleted.post({
completion,
execution,
worker,
backgroundWorkerId: id,
});
} catch (error) {
this.onTaskFailedToRun.post({
backgroundWorkerId: id,
worker,
completion: {
ok: false,
id: payload.runId,
retry: undefined,
error:
error instanceof Error
? {
type: "BUILT_IN_ERROR",
name: error.name,
message: error.message,
stackTrace: error.stack ?? "",
}
: {
type: "BUILT_IN_ERROR",
name: "UnknownError",
message: String(error),
stackTrace: "",
},
},
});
}
}
async #executeTaskRun(id: string, payload: TaskRunExecutionPayload) {
@@ -371,12 +408,9 @@ export class BackgroundWorker {
// We need to notify all the task run processes that a task run has completed,
// in case they are waiting for it through triggerAndWait
async taskRunCompletedNotification(
completion: TaskRunExecutionResult,
execution: TaskRunExecution
) {
async taskRunCompletedNotification(completion: TaskRunExecutionResult) {
for (const taskRunProcess of this._taskRunProcesses.values()) {
taskRunProcess.taskRunCompletedNotification(completion, execution);
taskRunProcess.taskRunCompletedNotification(completion);
}
}
@@ -764,24 +798,23 @@ class TaskRunProcess {
return result;
}
taskRunCompletedNotification(completion: TaskRunExecutionResult, execution: TaskRunExecution) {
taskRunCompletedNotification(completion: TaskRunExecutionResult) {
if (!completion.ok && typeof completion.retry !== "undefined") {
return;
}
if (execution.run.id === this.runId) {
if (completion.id === this.runId) {
// We don't need to notify the task run process if it's the same as the one we're running
return;
}
logger.debug(`[${this.runId}] task run completed notification`, {
completion,
execution,
});
this._sender.send("TASK_RUN_COMPLETED_NOTIFICATION", {
version: "v2",
completion,
execution,
});
}
packages/cli-v3/src/workers/dev/worker-facade.ts
+11 -2
View File
@@ -180,8 +180,17 @@ const handler = new ZodMessageHandler({
_isRunning = false;
}
},
TASK_RUN_COMPLETED_NOTIFICATION: async ({ completion, execution }) => {
devRuntimeManager.resumeTask(completion, execution);
TASK_RUN_COMPLETED_NOTIFICATION: async (payload) => {
switch (payload.version) {
case "v1": {
devRuntimeManager.resumeTask(payload.completion, payload.execution.run.id);
break;
}
case "v2": {
devRuntimeManager.resumeTask(payload.completion, payload.completion.id);
break;
}
}
},
CLEANUP: async ({ flush, kill }) => {
if (kill) {
packages/core/package.json
+8
View File
@@ -37,6 +37,14 @@
"require": "./dist/v3/otel/index.js",
"types": "./dist/v3/otel/index.d.ts"
},
"./v3/zodfetch": {
"import": {
"types": "./dist/v3/zodfetch.d.mts",
"default": "./dist/v3/zodfetch.mjs"
},
"require": "./dist/v3/zodfetch.js",
"types": "./dist/v3/zodfetch.d.ts"
},
"./v3/zodMessageHandler": {
"import": {
"types": "./dist/v3/zodMessageHandler.d.mts",
packages/core/src/v3/runtime/devRuntimeManager.ts
+4 -4
View File
@@ -80,18 +80,18 @@ export class DevRuntimeManager implements RuntimeManager {
};
}
resumeTask(completion: TaskRunExecutionResult, execution: TaskRunExecution): void {
const wait = this._taskWaits.get(execution.run.id);
resumeTask(completion: TaskRunExecutionResult, runId: string): void {
const wait = this._taskWaits.get(runId);
if (!wait) {
// We need to store the completion in case the task is awaited later
this._pendingCompletionNotifications.set(execution.run.id, completion);
this._pendingCompletionNotifications.set(runId, completion);
return;
}
wait.resolve(completion);
this._taskWaits.delete(execution.run.id);
this._taskWaits.delete(runId);
}
}
packages/core/src/v3/schemas/messages.ts
+17 -6
View File
@@ -1,5 +1,5 @@
import { z } from "zod";
import { TaskRunExecution, TaskRunExecutionResult } from "./common";
import { TaskRunExecution, TaskRunExecutionResult, TaskRunFailedExecutionResult } from "./common";
import {
EnvironmentType,
@@ -63,6 +63,11 @@ export const BackgroundWorkerClientMessages = z.discriminatedUnion("type", [
completion: TaskRunExecutionResult,
execution: TaskRunExecution,
}),
z.object({
version: z.literal("v1").default("v1"),
type: z.literal("TASK_RUN_FAILED_TO_RUN"),
completion: TaskRunFailedExecutionResult,
}),
z.object({
version: z.literal("v1").default("v1"),
type: z.literal("TASK_HEARTBEAT"),
@@ -109,11 +114,17 @@ export const workerToChildMessages = {
traceContext: z.record(z.unknown()),
metadata: BackgroundWorkerProperties,
}),
TASK_RUN_COMPLETED_NOTIFICATION: z.object({
version: z.literal("v1").default("v1"),
completion: TaskRunExecutionResult,
execution: TaskRunExecution,
}),
TASK_RUN_COMPLETED_NOTIFICATION: z.discriminatedUnion("version", [
z.object({
version: z.literal("v1"),
completion: TaskRunExecutionResult,
execution: TaskRunExecution,
}),
z.object({
version: z.literal("v2"),
completion: TaskRunExecutionResult,
}),
]),
CLEANUP: z.object({
version: z.literal("v1").default("v1"),
flush: z.boolean().default(false),
packages/core/src/v3/zodfetch.ts
+3 -3
View File
@@ -16,12 +16,12 @@ export type ZodFetchOptions = {
retry?: RetryOptions;
};
export async function zodfetch<TResponseBody extends any>(
schema: z.Schema<TResponseBody>,
export async function zodfetch<T extends z.ZodTypeAny>(
schema: T,
url: string,
requestInit?: RequestInit,
options?: ZodFetchOptions
): Promise<TResponseBody> {
): Promise<z.infer<T>> {
return await _doZodFetch(schema, url, requestInit, options);
}
packages/core/tsup.config.ts
+1
View File
@@ -15,5 +15,6 @@ export default defineConfig({
"./src/v3/dev/index.ts",
"./src/v3/prod/index.ts",
"./src/v3/workers/index.ts",
"./src/v3/zodfetch.ts",
],
});
references/v3-catalog/src/trigger/longRunning.ts
+14
View File
@@ -27,3 +27,17 @@ export const longRunningParent = task({
};
},
});
export const longRunningWithDotInName = task({
id: "long.running.with.dot",
run: async (payload: { message: string }) => {
logger.info("Long running payloadd", { payload });
// Wait for 3 minutes
await new Promise((resolve) => setTimeout(resolve, 3 * 60 * 1000));
return {
finished: new Date().toISOString(),
};
},
});